---
title: "代理群体的出口策略：集中限流、缓存与去重"
canonical: "https://siyuanailab.com/curation/translations/fb46ee1055b5a5920d5108ea67973474-2e78d57acb97c967c791e5196d85d3199db825c6381ccae5e776c81539f28163/"
dateModified: "2026-10-26"
datePublished: "2026-10-26"
---

[返回日报条目](https://siyuanailab.com/curation/#entry/fb46ee1055b5a5920d5108ea67973474)

 

资料节译

 

# 代理群体的出口策略：集中限流、缓存与去重

 

[阅读原文](https://dev.to/dima_kruglov_608741d9a416/your-agent-is-a-guest-egress-policy-for-agent-fleets-2baa)

 

现有资料无法确认原文完整，本页只翻译已取得的材料。

 

现有原件无法确认图片是否齐全。

 

我审阅的大多数智能体代码，对沙箱*内部*的行为很谨慎，对离开沙箱时的行为却很随意。然后它遇到一个网站：凡是 User-Agent 中带有 `python-httpx` 的请求都返回 403，它便重试四十次，耗掉整个下午。

 

网络对智能体越来越不宽容。出版商直接屏蔽爬虫，CDN 会根据 TLS 和请求头进行指纹识别，越来越多的网站开始采取“要么付费使用 API，要么走开”的做法。这不是一时的风气——而是常态。如果你的智能体集群要从你不拥有的域名获取任何内容，出站流量就必须作为一个子系统来设计，而不是在工具里随手塞一个 `client.get()` 调用。

 

下面是从代理层看到的情况：哪些会出故障、该记录什么，以及该在智能体前面部署什么。

 

有两种智能体工作负载，在笔记本上看起来都没问题，但到了集群规模都会出故障：

 

-

 

## **广度优先研究。**数千个智能体实例分别针对公共来源追查略有不同的查询。OpenAI 的生物数据工作是这种模式中广为人知的例子；规模化的难点在于去重，而不是计算。

 

**深度优先工具调用。**一个智能体围绕一项任务循环执行，每一步调用十个外部 API。规模化的难点在于，一个行为不当的循环就可能产生比整个测试套件还多的流量。

 

这两种情况下的故障都一样：每个智能体都有自己的 HTTP 客户端和重试策略，却没有共享的“我们已经向这个主机问过这个问题”的概念。每个智能体各自的速率限制器并不是真正的速率限制器，而是一个倍增器。

 

把出站流量集中到一个地方——边车、代理进程，或供智能体调用的单一服务——这样速率限制器、`robots.txt` 缓存和去重索引就能共享。用于智能体集群的沙箱运行环境（近期出现的 OpenShell 就是其中之一）通常会提供一个网络边界，你可以借此实现这一点。用好这个边界；不要让智能体直接连出网络。

 

拒绝请求的状态不全是 403；不是 403 的那些情况更棘手：

 

| 信号 | 含义 | 正确应对 |
| --- | --- | --- |
| `403` | 策略或指纹识别导致的拒绝 | 停止。不要重试。升级交由人工处理。 |
| `429` +`Retry-After` | 明确的背压信号 | 遵从请求头指示，然后恢复请求 |
| `200` 并带有挑战页面 | 软封锁，状态码逻辑无法发现 | 检测到后按失败处理 |
| 连接重置 / TLS 错误 | 指纹识别，或确实不稳定 | 加入随机抖动重试两次，然后停止 |

 

软封锁最容易让人栽跟头。你的提取器返回零行，智能体得出“没有结果”的结论，然后对一个空页面写出信心十足的摘要。检查每个主机的解析产出量——如果某个主机过去每页会返回 40 个链接，现在却在状态码为 200 时返回 0 个，那就是封锁，而不是内容变了。

 

通过轮换 IP 或伪装浏览器 User-Agent 来绕过封锁，是一场你必输的军备竞赛，也会带来合同问题：你是在规避一种表达运营方意愿的技术措施。工程上的应对办法是购买 API、取得数据许可，或者弃用该来源。

 

Python 3.10+，`httpx`。这个实现刻意保持精简：一个类、一个共享缓存，遇到未知主机时默认拒绝。

 

```
import time
import urllib.robotparser
from collections import defaultdict
from urllib.parse import urlparse
import httpx
class EgressGate:
    """Shared politeness + robots.txt gate. One instance per fleet, not per agent."""
    def __init__(self, user_agent: str, min_interval_s: float = 1.0):
        self.ua = user_agent
        self.min_interval = min_interval_s
        self._last_hit: dict[str, float] = defaultdict(float)
        self._robots: dict[str, tuple[float, urllib.robotparser.RobotFileParser]] = {}
        self._client = httpx.Client(
            headers={"User-Agent": user_agent},   # identify yourself, always
            timeout=httpx.Timeout(10.0, connect=5.0),
            follow_redirects=True,
        )
    def _robots_for(self, url: str) -> urllib.robotparser.RobotFileParser:
        host = urlparse(url).netloc
        cached = self._robots.get(host)
        # Cache for an hour. Re-fetching robots.txt per request is itself abuse.
        if cached and time.monotonic() - cached[0] < 3600:
            return cached[1]
        rp = urllib.robotparser.RobotFileParser()
        rp.set_url(f"https://{host}/robots.txt")
        try:
            rp.read()
        except Exception:
            rp.disallow_all = True  # fail closed: unknown host, stay out
        self._robots[host] = (time.monotonic(), rp)
        return rp
    def fetch(self, url: str) -> httpx.Response | None:
        host = urlparse(url).netloc
        if not self._robots_for(url).can_fetch(self.ua, url):
            return None  # policy skip, not an error - log it separately
        wait = self.min_interval - (time.monotonic() - self._last_hit[host])
        if wait > 0:
            time.sleep(wait)
        self._last_hit[host] = time.monotonic()
        for attempt in range(3):
            resp = self._client.get(url)
            if resp.status_code == 429:
                # Retry-After is sometimes an HTTP date; parse defensively in prod.
                delay = float(resp.headers.get("Retry-After", 2 ** attempt))
                time.sleep(min(delay, 60.0))
                continue
            break
        if resp.status_code == 403:
            raise PermissionError(f"{host} refused agent traffic")
        return resp
```

 

有两点值得指出。`min_interval` 是每个主机的*全局*间隔下限，因此两个智能体同时请求同一域名时会排队等待。触发 403 时抛出异常是刻意为之：403 应该向上传递并终止子任务，而不是被三层调用栈下面的 `try/except: pass` 吞掉。

 

如果你要让一千个智能体针对固定语料库运行，最省钱的优化就是不要重复请求。对 `(method, url, normalized_query)` 元组进行哈希，在共享存储中通过原子 `SETNX` 抢占任务；未抢到的请求等待抢到者的结果，而不是重新推导。如果没有这一步，故障场面会很难看：2,000 个智能体各自发现相同的 40 个页面，接着 2,000 个礼貌的速率限制器全都以每秒一次请求的速度猛敲同一个主机。

 

按 URL 和内容哈希为抓取到的载荷建立索引，并根据来源内容的易变程度设置 TTL。不要存储你没有权利保留的正文——你的网关应能按主机强制执行这项策略，而不是把它留给提取时自行判断。

 

每次抓取都记录：主机、状态码、字节数、延迟、策略决策（`allowed` / `robots_skip` / `blocked`）、去重键，以及发起请求的智能体 ID。然后监控两个比率：每个主机的策略跳过率，以及每个主机的解析产出量。两者同时异常就是软封锁的典型信号。这两项都不会显示在状态码仪表板上。

 

保留智能体 ID。有人问你的集群为什么访问了他们的网站 8,000 次时，“哪个智能体、执行什么操作、属于哪项任务”是唯一能结束这场对话的答案。

 

-

 

## **在请求路径中获取 `robots.txt`。**你把一次请求变成了两次，还给每次请求增加了对某个主机的依赖，而该主机可能会限制请求速率。

 

## **每个智能体进程各自配置速率限制器。**十个智能体，十个限制器，达到预期速率的十倍。

 

## **重试 403。**它不会变成 200，而且你现在把自己是个麻烦的信号加倍放大了。

 

## **将 `200` 视为成功。**挑战页面、同意提示页和 JS 外壳都会返回 200，但没有内容。

 

## **忽略 `Retry-After`，并使用固定的 1s 退避。**固定退避会让你的重试同步成惊群；应加入随机抖动。

 

## **把轮换出站 IP 当作解决办法。**它能管用一周，却会把速率限制问题变成法律问题。

 

**所有请求都使用同一个 User-Agent 字符串。**你将无法解释自己的流量，也无法申请加入允许名单。

 

- 所有智能体的 HTTP 流量都经过同一个出站组件；智能体不能直接访问网络。
 
- `robots.txt` 每个主机只获取一次并缓存，且

 

AI 翻译；具体表述请与原文对照。