用 Python 实现 Smart AI Proxy 轮换大规模下的健康评分

大多数代理轮换器回答的是错的问题。它们回答的是"轮到谁了？"，而真正决定你的流水线能否跑完的问题是"这个请求该走哪条路由？"轮询给每条路由相同份额的流量，只有在所有路由一样好时这才成立。代理池从来没有这么齐整：一条路由很快，直到被封；另一条很慢，却从不失败；第三条成片地失败，然后又恢复。

一旦接受这一点，轮换就不再是调度问题，而是反馈问题。轮换器观察每个请求的结果，对每条路由的表现维持一个滚动估计，并让这些估计引导下一次选择。这就是健康评分的全部想法：代理池告诉你它正在做什么，而路由策略在听。

本文用 Python 构建这样一个轮换器。指数加权估计跟踪成功率与延迟，熔断器负责那些平均值反应太慢的骤然故障，基准测试则把结果与朴素轮询作对比。其中一条路由是 Crawlbase Smart AI Proxy，它值得被放进来，正是因为它把问题的一整层收走了：你不再为一个个住宅 IP 打分，而是为一条自己处理轮换与反爬破解的路由打分。

为什么轮换需要健康评分

代理池并不均质，也不会静止。各条路由在延迟、可靠性以及某个目标站点对它们的态度上都不同，而这三样都可能在任务运行途中变化。把它们当作可互换的，就等于让策略对唯一要紧的那些条件视而不见。

一个考虑健康的轮换器需要三种信号，而它们并不是同一种信号的不同灵敏度：

1. 存活。 此刻哪些路由是成功的，流量又该从哪些路由撤走。
2. 延迟。 在能用的路由里优先选快的。一个花了九秒才成功的请求，照样花掉了你九秒。
3. 失败的稳定性。 反复失败应当把一条路由移出正常流量，并让它通过一次受控的探测回来，而不是立刻放回池子。

轮询这三样一个都不提供。不管路由是健康、迟缓还是已死，它都均匀地摊派请求。这恰恰使它成为合适的对照组：把两种策略跑在同一份负载上，差值就是反馈的价值。

系统的形状

三份职责，刻意彼此隔开。一条 路由 执行一次抓取，并报告一个规范化的结果：成功与否、HTTP 状态、耗时。一个 健康模型 把这串结果变成分数。一条 策略 读取分数并挑出下一条路由。因为它们是分开的，你可以在不碰传输层的情况下替换策略，或者在完全相同的路由上比较两种策略。

Smart AI Proxy 之所以在这张图里占有一席之地，是因为它吸收了一整层。它对外只给一个接入点，在其背后处理住宅 IP 轮换与反爬破解，于是你的应用评的是一条路由，而不是一群地址。该由哪条获取路径承接流量，仍然由你的策略决定。

下面每段代码都摘自配套仓库 ScraperHub/smart-ai-proxy-rotation-in-python-health-scoring-at-scale，其中可运行的实现放在 final/ ，分阶段的检查点放在 steps/.

运行环境

Python 3.11 或更新版本，以及一个用于代理路由的 Crawlbase 账号。这里用 Normal token 就够了；只有当目标需要渲染才能产出内容时才涉及 JavaScript token，而那是与路由无关的另一个问题。两者都可以在 控制台设置中找到。

git clone https://github.com/ScraperHub/smart-ai-proxy-rotation-in-python-health-scoring-at-scale.git
cd smart-ai-proxy-rotation-in-python-health-scoring-at-scale/final

python -m venv .venv && source .venv/bin/activate  # Windows: .venv\Scripts\activate
pip install -r requirements.txt

cp .env.example .env  # then set CRAWLBASE_TOKEN

token 属于运行环境而不属于源码，这既是寻常的卫生习惯，也划出了下一节所依赖的那条配置边界。

把配置当作契约

配置是系统的第一份运行期契约。当必需的依赖缺失时，轮换器应当拒绝启动，而不是带着一条不可用的路由起来，等到负载跑到一半才发现。

def _required(name: str) -> str:
    value = os.environ.get(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value

@dataclass(frozen=True)
class Config:
    crawlbase_token: str = field(default_factory=lambda: _required("CRAWLBASE_TOKEN"))
    smart_proxy_host: str = field(default_factory=lambda: os.environ.get("SMART_PROXY_HOST", "smartproxy.crawlbase.com"))
    smart_proxy_port: int = field(default_factory=lambda: int(os.environ.get("SMART_PROXY_PORT", "8012")))
    success_decay: float = field(default_factory=lambda: float(os.environ.get("SUCCESS_DECAY", "0.3")))
    latency_decay: float = field(default_factory=lambda: float(os.environ.get("LATENCY_DECAY", "0.3")))
    breaker_threshold: int = field(default_factory=lambda: int(os.environ.get("BREAKER_THRESHOLD", "3")))
    breaker_cooldown_s: float = field(default_factory=lambda: float(os.environ.get("BREAKER_COOLDOWN_S", "15")))

注意这条分界说明了什么。 CRAWLBASE_TOKEN 是凭据，且为必需。其余全是策略：成功估计反应多快、延迟估计反应多快、多少次连续失败会打开熔断器、被打开的路由要在外面待多久。把 dataclass 冻结意味着路由路径读到的是一份固定配置，而不是在请求时去翻环境变量。

值得被超越的基线

这里所谓路由，就是任何能取一个 URL 并报告发生了什么的东西。实现里给了两条：直连，以及 Smart AI Proxy。

class CrawlbaseRoute(Route):
    name = "crawlbase-smart-proxy"

    def __init__(self, config: Config) -> None:
        proxy_url = (
            f"http://{config.crawlbase_token}:@"
            f"{config.smart_proxy_host}:{config.smart_proxy_port}"
        )
        self._client = httpx.Client(proxy=proxy_url, verify=False, timeout=config.request_timeout_s, follow_redirects=True)

token 作为代理用户名传入，密码留空，Smart AI Proxy 就是这样做认证的。 verify=False 不是偷懒，值得理解而不是照抄：代理会终结 TLS 以便加上自己的请求头，因此你的客户端看到的是 Crawlbase 的证书而不是目标站点的，严格校验会把它拒掉。这是该接入点被记录在案的行为，而不是对配置错误的绕行。

DirectRoute 是同一套接口，只是不走代理。面对未设防的目标它往往更快，而一旦出现反爬控制它会最先劣化，这恰好让它成为健康模型要消费的那种失败信号的有用来源。

基线策略刻意忽略上述每一个信号：

class NaiveRotator:
    def __init__(self, routes: list[Route]) -> None:
        self._routes = routes
        self._cycle = itertools.cycle(routes)

    def fetch(self, url: str) -> tuple[str, FetchResult]:
        route = next(self._cycle)
        return route.name, route.fetch(url)

这就是策略的全部。它是确定性的，也完全公平，而公平正是问题所在：在各条路由不再一样好之后，这份公平依然照旧。

给健康打分：两个平均值加一个熔断器

指数加权移动平均适合这里，因为它给近期观测更大权重，并逐渐忘掉旧的。十分钟前失败过的路由不该被永久惩罚；三十秒前开始失败的路由则应当迅速掉下去。衰减系数就是这两者之间的旋钮。

def observe(self, ok: bool, latency_s: float) -> None:
    self.samples += 1
    outcome = 1.0 if ok else 0.0
    self.success_ewma = self.success_decay * outcome + (1 - self.success_decay) * self.success_ewma
    if ok:
        self.latency_ewma_s = self.latency_decay * latency_s + (1 - self.latency_decay) * self.latency_ewma_s
        self.consecutive_failures = 0
        if self.breaker_state is BreakerState.HALF_OPEN:
            self.breaker_state = BreakerState.CLOSED
    else:
        self.consecutive_failures += 1
        if self.consecutive_failures >= self.breaker_threshold:
            self.breaker_state = BreakerState.OPEN
            self.opened_at = time.monotonic()

其中有两个刻意的选择。延迟只在成功时更新，因为一个失败请求耗费的时间描述的是这次失败，而不是这条路由正常工作时的响应速度；把它们算进去，会让一条失败得很快的路由看上去很快。而稳定性交给熔断器而不是平均值，因为连续失败与缓慢漂移的均值是不同性质的证据，值得更果断的回应。

随后这两个估计塌缩成一个数：

def value(self) -> float:
    if self.breaker_state is BreakerState.OPEN:
        return 0.0
    latency_term = 1.0 / (1.0 + self.latency_ewma_s)
    return self.success_ewma * latency_term

这里的设计决定是相乘而非相加，它编码了一个合取关系：一条路由必须既可靠又快，才能拿到高分。把两项相加，一条大部分时候都失败的快路由，仍能靠延迟那一半攒出一个像样的分数。把它们相乘，接近零的成功率就会把整个值拽到接近零，无论失败得多快。熔断器一旦打开，值被短路成恰好 0.0，这让摘除路由成为显式机制，而不是某种涌现结果。

回路，一次一个请求

每个请求都走同样的四步：找出可用路由、按健康值挑一条、执行、把结果送回去。

def fetch(self, url: str) -> tuple[str, FetchResult]:
    candidates = self._eligible() or self._scored
    chosen = self._select(candidates)
    result = chosen.route.fetch(url)
    chosen.health.observe(result.ok, result.latency_s)
    return chosen.route.name, result

选择会把每条可用路由的健康值提升到一个指数 gamma再作为权重。指数为零时选择是均匀的；随着它变大，流量向评分最高的路由集中。把它保持得温和些。过大的指数会造出一个死死咬住最先看起来不错的那条路由的轮换器，之后它再也没有办法发现被降权的路由已经恢复，因为它根本不再往那边发请求。

第一行里的兜底比看上去更重要：当熔断器把所有路由都打开时， _eligible() 为空，轮换器会回退到完整的评分集合，而不是抛出异常。一个整体都不健康的池子，仍然应该去试那个最不糟的选项。

基准测试真正说明了什么

python src/main.py benchmark 20

Policy comparison:
policy              requests    success rate    mean latency
round-robin               20          100.0%          1.589s
health-weighted           20          100.0%          0.289s

这段要仔细读，因为最抢眼的那个数字反而最没意思。面对像 example.com 这样未设防的目标，两条路由都会成功，于是成功率完全相同，全部差异都落在延迟上。轮询继续把一半流量发往较慢的那条路由，因为公平就是这个意思。按健康加权的策略注意到了，于是停手。

绝对延迟值会随运行和目标而变，那并不是本文的论点。论点关于行为：一种策略会对证据作出反应，另一种则做不到。把同一个基准测试指向一个有防护的目标，同样的机制就会改从成功率上体现出来：直连路由的成功 EWMA 下滑，连续失败触发熔断器，流量随之转移，而轮询仍在往一条正在拒绝它们的路由上喂请求。

推向生产环境

基准测试是单进程且同步的，因为那样控制回路才读得清楚。当情况不再如此，有四件事会变。

健康状态必须共享。 放在内存里，每个 worker 都持有自己对池子的私见，于是一个进程可能认定某条路由已坏，另一个却照用不误，熔断器在任何地方都不会一致地打开。把成功 EWMA、延迟 EWMA、熔断器状态和失败计数挪进类似 Redis 的地方，按路由用一致的键组织，才能让这个信号成为共同认知，而不是各进程自己的传说。

更新必须是并发安全的。 真实负载会并行地观察结果，而 observe 里的每个字段都是读改写。没有同步或原子操作，两次同时发生的失败可能读到同一个计数器，又写回同一个增量，于是阈值三悄悄变成了阈值五。

衰减系数需要针对你的流量调。 0.3 是一个站得住脚的起点，不是答案。更高的值追着近期观测跑、反应更快，代价是对一次短暂抖动反应过度。更低的值更稳，但更晚才注意到真正的变化。你更愿意犯哪种错，取决于你的目标有多嘈杂。

健康未必是你唯一的目标。 各条路由在成本上的差别不亚于质量，最快的那条未必就是你想让它承接每个请求的那条。给这个标量加上一项成本，策略就能在可靠性、延迟与花费之间权衡，而不是只优化一个维度，然后被账单吓一跳。

有一条边界落在模型之外：只把它指向你有权访问的目标，并尊重它们的条款、频率预期与各种限制。基准测试之所以用 example.com ，正是因为它是一个不提出上述任何要求的受控目标。

关键要点

当选择由观察到的行为驱动，而不是由它在列表中的位置驱动，轮换器就开始有用了。三个机制在做事：对成功的 EWMA 反映近期可靠性，对延迟的 EWMA 反映响应速度，熔断器负责平均值会抹平的骤然故障。把前两者相乘，使一条路由在两个维度上都必须诚实；熔断器则接管"渐进是错误速度"的那种情形。

轮询仍留在画面里，作为让改进可被度量的对照组。而 Smart AI Proxy 是以一条路由的身份、而非一份责任的身份嵌进来的：它吸收了 IP 轮换与反爬破解，好让它之上的策略始终只是一个路由决定。

选择、执行、观察、更新、再选择。这篇文章里其余的一切，都只是每一步做得有多好的细节。

常见问题（FAQ）

既然 Smart AI Proxy 已经在轮换，我为什么还要自己给路由打分？

因为两者工作在不同层面。Smart AI Proxy 在它自己的池内轮换 IP，所以你的应用不需要为一个个地址建模。你的轮换器评的是 路由：直连对比走代理、一个区域接入点对比另一个、便宜的路径对比昂贵的路径。这个决定只有你的应用能做，因为只有它知道这些流量是干什么用的。本文的模型刻意做得通用，好套用到你实际运营的任意一组路由上。

衰减系数该用多少？

两个都从 0.3 开始。每次新观测为更新后的估计贡献 30%，先前的估计承担其余 70%。当目标变化很快、策略需要更早察觉时调高它；当测量噪声大、你不希望一次慢响应就把流量挪走时调低它。

什么时候熔断器比单靠成功 EWMA 更管用？

当故障是突然到来的时候。EWMA 天生就是个平滑器，所以一条彻底死掉的路由仍需要好几次观测才会掉到足够低、开始产生影响，而这几次观测每一次都是被浪费掉的请求。熔断器则以连续失败为依据，立刻把该路由撤下，冷却之后再给一次受控的半开探测，而不是干等平均值慢慢爬回来。两者互补：平均值应对劣化，熔断器应对崩塌。

这个做法需要 JavaScript token 吗？

不需要。这里用到的 Smart AI Proxy 路由，Normal token 就够了。JavaScript token 是给那些必须先渲染才会有内容可返回的目标准备的，那是关于目标的问题，不是关于路由的问题。两种情况下轮换逻辑完全一样。

同一个模型能路由两个以上的选项吗？

可以，而且那样更有用。健康模型和选择策略都没有假定只有两条路由：两者都在一个列表上工作。两条路由只是让基准测试更容易读懂。加上区域接入点或第二家供应商，无非是往路由列表里追加，而加权选择会在现有的所有项之间分配流量。

NZ

Neil Zamora

高级架构师 · Crawlbase

Crawlbase 高级架构师，专注大规模爬取背后的系统：代理轮换、反机器人韧性，以及隐藏这些复杂性的 API。