免费API限流应对：重试阶梯与熔断设计

重试是一种调度策略，而不是本能反应。免费模型端点在过载时会返回 429。立即重试只会加剧过载。重试阶梯通过指数退避和抖动来分散重试间隔。本教程从零开始构建一个重试阶梯，并加入了熔断器和幂等性保护。每个阶段都附带可运行的验证步骤。完成时你将得到一个可复用的 Python 模块，能够应对速率限制、过载和超时。

为什么会话arangebase.retries.fail

免费端点以可预测的模式失败。速率限制器返回 429。代理返回 503。推理过慢导致超时。不同类型的失败需要不同的响应。

没有退避的 429 会引发雷鸣般的群集效应。十个客户端同时重试，端点持续过载。抖动打破同步。没有退避的 503 持续打击已经故障的服务器。没有验证的超时可能造成重复工作。这个模式在各个提供商间反复出现。免费端点是共享的，一次糟糕的重试会连累其他用户。

决策表就是契约：

第一步：写代码之前先对失败进行分类

将分类逻辑写成一个函数。只有当重试能够起作用时才返回 True。

RETRYABLE = {429, 500, 502, 503, 504}

def should_retry(status):
    return status in RETRYABLE

assert should_retry(429) is True
assert should_retry(503) is True
assert should_retry(400) is False
assert should_retry(422) is False
print("classification ok")

第二步：构建重试阶梯

阶梯有三条规则。退避时间指数增长。抖动防止重试同步化。Retry-After 优先于退避时间。

import random
import time

class RetryLadder:
    def __init__(self, max_attempts=4, base_delay=1.0, max_delay=30.0, jitter=0.3):
        self.max_attempts = max_attempts
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.jitter = jitter

    def delay_for(self, attempt, retry_after=None):
        if retry_after is not None:
            return min(float(retry_after), self.max_delay)
        exp = min(self.base_delay * (2 ** attempt), self.max_delay)
        return exp * (1 + random.uniform(-self.jitter, self.jitter))

    def run(self, call):
        status, payload = None, None
        for attempt in range(self.max_attempts):
            try:
                status, payload = call()
            except TimeoutError:  # Python 3.10+; socket.timeout is TimeoutError
                status, payload = 503, {"error": "timeout"}
            if status not in RETRYABLE:
                return status, payload
            retry_after = None
            if isinstance(payload, dict):
                retry_after = payload.get("retry_after")
            delay = self.delay_for(attempt, retry_after)
            time.sleep(delay)
        return status, payload

数学原理很简单。第 0 次重试等待 base_delay。第 1 次重试等待两倍。第 2 次重试等待四倍。抖动分散了群集效应。Retry-After 是服务端指令，所以不加抖动。

ladder = RetryLadder(base_delay=1.0, jitter=0.0)
assert 1.0 <= ladder.delay_for(0) < 2.0
assert 2.0 <= ladder.delay_for(1) < 4.0
assert 4.0 <= ladder.delay_for(2) < 8.0
print("backoff curve ok")

这个测试固定了曲线。如果有人修改了乘数，断言就会失败。

第三步：加入熔断器

阶梯能处理几次失败，但风暴需要熔断器。当连续五次调用失败时，熔断器打开。新调用快速失败而不触碰端点。经过冷却期后，一个探测请求测试水温。

import time

class CircuitBreaker:
    def __init__(self, failure_threshold=5, cooldown=30.0):
        self.failure_threshold = failure_threshold
        self.cooldown = cooldown
        self.failures = 0
        self.state = "closed"
        self.opened_at = 0.0

    def allow(self):
        if self.state == "open":
            if time.time() - self.opened_at >= self.cooldown:
                self.state = "half_open"
                return True
            return False
        return True

    def record(self, ok):
        if ok:
            self.failures = 0
            self.state = "closed"
        else:
            self.failures += 1
            if self.failures >= self.failure_threshold:
                self.state = "open"
                self.opened_at = time.time()

breaker = CircuitBreaker(failure_threshold=5, cooldown=0.01)
for _ in range(5):
    breaker.record(False)
assert breaker.state == "open"
assert breaker.allow() is False
time.sleep(0.02)
assert breaker.allow() is True
breaker.record(True)
assert breaker.state == "closed"
print("breaker ok")

第四步：让重试具有幂等性

LLM 调用天然不具备幂等性。同一个 prompt 可能产生不同的补全。重复的代码生成调用可能返回不同的补丁。按请求哈希存储第一个成功响应，在重复重试时返回它。

import hashlib
import json

class ResponseCache:
    def __init__(self):
        self.store = {}

    def key(self, request):
        raw = json.dumps(request, sort_keys=True).encode()
        return hashlib.sha256(raw).hexdigest()

    def get(self, request):
        return self.store.get(self.key(request))

    def put(self, request, response):
        self.store[self.key(request)] = response

cache = ResponseCache()
req = {"prompt": "summarize this log", "max_tokens": 50}
cache.put(req, {"text": "first completion"})
assert cache.get(req) == {"text": "first completion"}
print("idempotency ok")

第五步：用 Mock 端点验证

不要一开始就对真实端点测试重试。先构建一个返回预定失败序列的 Mock。这让测试变得确定性。预定序列就是真相来源。它从测试中移除了随机性。每次运行都走同一条路径。

import json
from http.server import BaseHTTPRequestHandler, HTTPServer

SCRIPT = [
    (429, {"error": "rate limited", "retry_after": 0.1}),
    (429, {"error": "rate limited", "retry_after": 0.1}),
    (503, {"error": "overloaded"}),
    (200, {"text": "ok"}),
]

class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        code, payload = SCRIPT.pop(0) if SCRIPT else (200, {"text": "ok"})
        body = json.dumps(payload).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(body)

HTTPServer(("127.0.0.1", 8765), Handler).serve_forever()

在一个终端运行 Mock，在另一个终端运行阶梯。

import json
import urllib.error
import urllib.request

def call_model():
    req = urllib.request.Request(
        "http://127.0.0.1:8765/v1/complete",
        data=json.dumps({"prompt": "hello"}).encode(),
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(req, timeout=5) as resp:
            return resp.status, json.loads(resp.read())
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read())
    except TimeoutError:
        raise

ladder = RetryLadder(max_attempts=4, base_delay=0.2, max_delay=2.0)
status, payload = ladder.run(call_model)
print(status, payload)

预期输出：200 {'text': 'ok'}。阶梯挺过了两次 429 和一次 503。它等待、退避，最终落在成功上。在 call_model 内部加一条 print 来计数重试次数。模式应该是：失败、失败、失败、成功。

第六步：在真实免费端点旁运行

Mock 证明了逻辑。真实端点证明了假设。让同一个 call_model 指向免费模型端点。把阶梯放在一台小服务器上，这样笔记本休眠时它也能运行。MonkeyCode 的免费服务器就适合这里：它把阶梯放在模型端点旁边，免费模型访问让你有一个真实的端点来保护。披露：本文是 MonkeyCode 产品推广的一部分。

记录三个数字。每请求尝试次数。每次尝试的延迟。最终状态。将它们与 Mock 运行结果对比。准备好惊讶。真实端点与 Mock 有差异。Retry-After 值可能缺失。超时可能先触发。调整阶梯，然后重新运行 Mock 确认修改。如果真实端点返回 200 但 payload 损坏，阶梯不是正确的工具。这是验证问题，不是传输问题。

最终流程是什么样的

完整流程是四层。分类决定重试是否有帮助。阶梯分散重试间隔。熔断器阻止风暴。缓存去重结果。每一层都可以独立测试。它们一起把不稳定的端点变成可预测的。代价是延迟。收益是在故障窗口内存活而不需要人工干预。

重试不会创造配额。如果 429 意味着"额度已用完"，退避只是延迟了失败。配额要单独规划。熔断器给每个请求增加了延迟。只有当许多请求共享一个端点时它才值得。一个用户跑一个脚本可能永远不需要它。LLM 重试不具备幂等性。缓存减少重复但不能保证补全结果相同。在信任重试输出之前先验证它。超时是猜测。先测量 p95 延迟。在 30 秒端点上设 10 秒超时每次都会失败。本教程中的数字是起点。你的端点有自己的限制。先测量再调优。

谁不应该用这个

需要一秒内返回响应的实时 UI。每次调用都修改外部状态的系统。有付费 SLA 的团队应该使用提供商原生的重试 SDK。如果你不能容忍延迟响应，重试阶梯是错误的一层。

轮到你了：修改 SCRIPT 列表，观察阶梯如何适应。然后将它指向真实端点并绘制尝试次数图。429 风暴是一位老师。阶梯只是让课程可重复。