免费 LLM API 稳定性：429 应对策略

免费 LLM 端点迟早会返回 429 Too Many Requests。问题不在于是否会发生——而在于你的客户端能否存活。大多数 Agent 代码把模型 API 当作可靠依赖：一次调用，一次响应，没有意外。在免费层级上，这个假设会在三种可预测的方式上失效，每一种都需要不同的应对措施。

这篇文章是给在免费模型层级上构建应用的人的准备了一份可靠性手册。最终产物是一个小小的 Python 客户端，结合了指数退避、熔断器和配额感知降级——这三种模式把一个不可靠的免费端点变成可以接受的端点。

免费层级的三种失败模式

免费端点的失败方式与付费端点不同。付费 API 很少失败且持续时间短。免费层则持续、可预测地失败。你需要区分它们，因为修复方式不同。

速率限制（429）——端点正常，但你超出了每分钟或每秒的配额。修复方式是退避和重试。

配额耗尽（429 但响应体不同，或者 403）——每月的 token 额度已用完。修复方式是降级，而不是重试；重试一个已耗尽的配额是浪费资源。

基础设施抖动（5xx、超时）——免费服务器过载或正在重启。修复方式是熔断器，这样你就不会持续锤击一个奄奄一息的服务。

单一的 retry 循环对以上三种都处理不好。重试速率限制是可行的。重试已耗尽的配额会让情况更糟。重试 5xx 风暴会放大导致风暴的负载。

最重要的模式是熔断器。它把一个持续失败的依赖从重复异常的来源转换为一个触发降级的快速失败路径。

实现：一个弹性客户端

以下是三种模式的最小化实现。它封装了任何 OpenAI 兼容的端点，这样可以保持 provider 可替换。

# resilient_client.py — backoff + circuit breaker + fallback for LLM endpoints
import random
import threading
import time
from datetime import datetime, timedelta
from openai import OpenAI

class CircuitBreaker:
    """Opens after N consecutive failures, then allows a probe after cooldown."""
    def __init__(self, failure_threshold=5, cooldown_seconds=60):
        self.failure_threshold = failure_threshold
        self.cooldown_seconds = cooldown_seconds
        self.failures = 0
        self.open_until = None
        self.lock = threading.Lock()

    def allow(self):
        with self.lock:
            if self.open_until is None:
                return True
            if datetime.now() >= self.open_until:
                self.open_until = None
                self.failures = 0
                return True
            return False

    def record_failure(self):
        with self.lock:
            self.failures += 1
            if self.failures >= self.failure_threshold:
                self.open_until = datetime.now() + timedelta(seconds=self.cooldown_seconds)

    def record_success(self):
        with self.lock:
            self.failures = 0
            self.open_until = None

class QuotaBudget:
    """Tracks token usage and flips to degraded mode before exhaustion."""
    def __init__(self, limit, warn_at=0.8):
        self.limit = limit
        self.warn_at = warn_at
        self.used = 0
        self.lock = threading.Lock()

    def record(self, prompt_tokens, completion_tokens):
        with self.lock:
            self.used += prompt_tokens + completion_tokens

    @property
    def degraded(self):
        with self.lock:
            return self.used >= self.limit * self.warn_at

class ResilientLLMClient:
    def __init__(self, base_url, api_key, model,
                 quota_limit=10_000_000,
                 max_retries=5, base_delay=1.0, max_delay=30.0,
                 fallback=None):
        self.client = OpenAI(base_url=base_url, api_key=api_key)
        self.model = model
        self.breaker = CircuitBreaker()
        self.quota = QuotaBudget(quota_limit)
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
        self.fallback = fallback  # (client, model) tuple or None

    def complete(self, messages, tools=None, temperature=0):
        # Quota degradation: stop calling the free tier before it is spent.
        if self.quota.degraded:
            return self._degrade(messages, tools, temperature)

        # Circuit breaker: fast-fail when the endpoint is unhealthy.
        if not self.breaker.allow():
            return self._degrade(messages, tools, temperature)

        for attempt in range(self.max_retries):
            try:
                resp = self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    tools=tools,
                    temperature=temperature,
                )
                self.breaker.record_success()
                self.quota.record(
                    resp.usage.prompt_tokens,
                    resp.usage.completion_tokens,
                )
                return resp
            except Exception as e:
                self.breaker.record_failure()
                status = getattr(e, "status_code", None)
                # Quota exhaustion is not retryable.
                if status in (403,) or (status == 429 and "quota" in str(e).lower()):
                    return self._degrade(messages, tools, temperature)
                if attempt == self.max_retries - 1:
                    return self._degrade(messages, tools, temperature)
                delay = min(self.base_delay * (2 ** attempt) + random.uniform(0, 1), self.max_delay)
                time.sleep(delay)

    def _degrade(self, messages, tools, temperature):
        if self.fallback is not None:
            fallback_client, fallback_model = self.fallback
            return fallback_clie

这段代码中有三个细节是刻意为之的：

配额检查在 API 调用之前进行，而不是之后。 等到你收到配额耗尽的响应时，你已经浪费了这次请求。在 80% 使用量时降级，为正在途中的请求留出了缓冲空间。

熔断器在 5 次连续失败后打开。 这样可以防止 retry 循环放大 5xx 风暴。一旦打开，它会在 60 秒内拒绝所有流量，然后放行一个探测请求。

配额耗尽是通过检查错误响应体来检测的，而不是仅看状态码。 一些 provider 对速率限制和配额耗尽都返回 429；前者可以重试，后者重试是浪费时间。

测试：注入失败，而不是指望它们

验证弹性设计的唯一方法是强制触发失败。一个简单的故障注入包装器可以模拟每种模式，而无需触碰真实端点。

# fault_injector.py — wrap any OpenAI client and inject failures
import random
from openai import OpenAI

class FaultInjector:
    def __init__(self, client, fail_rate=0.3, quota_exhausted=False):
        self.client = client
        self.fail_rate = fail_rate
        self.quota_exhausted = quota_exhausted
        self.calls = 0

    def chat(self):
        return _FaultyCompletions(self)

class _FaultyCompletions:
    def __init__(self, injector):
        self.injector = injector
        self.model = "faulty"

    def create(self, **kwargs):
        self.injector.calls += 1
        if self.injector.quota_exhausted:
            raise Exception("429 quota exceeded")
        if random.random() < self.injector.fail_rate:
            raise Exception("500 Internal Server Error")
        return _FakeResponse()

class _FakeResponse:
    class Usage:
        prompt_tokens = 100
        completion_tokens = 50
    usage = Usage()

    class Choice:
        class Message:
            content = "ok"
            tool_calls = None
        message = Message()
    choices = [Choice()]

然后对每种失败模式下客户端的行为做出断言：

# test_resilience.py
from resilient_client import ResilientLLMClient
from fault_injector import FaultInjector

injector = FaultInjector(client=None, fail_rate=0.5)
client = ResilientLLMClient(
    base_url="http://localhost:9999", api_key="test", model="test",
    max_retries=3, base_delay=0.1, max_delay=0.5,
)
client.client = injector  # swap in the faulty client

# Expect: retries absorb the 50% failure rate, no exception raised.
resp = client.complete([{"role": "user", "content": "hi"}])
assert resp.choices[0].message.content == "ok"

# Expect: quota exhaustion degrades immediately, no retry loop.
injector.quota_exhausted = True
client.quota.used = client.quota.limit  # force degraded state
resp = client.complete([{"role": "user", "content": "hi"}])
# Falls through to _degrade; with no fallback configured, raises RuntimeError.

这个测试套件才是真正的交付物。它把决策表编码成可执行的断言，这样将来对重试策略的改动不会悄无声息地破坏降级路径。

免费层级的定位

为了针对真实的免费端点运行这个手册，我使用了 MonkeyCode 的免费层级，在撰写本文时它提供 10M token 额度和一个免费服务器选项。上述弹性客户端适用于任何 OpenAI 兼容的 base URL，所以 provider 的选择只是一个环境变量，而不是架构层面的承诺。

披露：本文是 MonkeyCode 产品推广的一部分。

精确的配额、速率限制和模型可用性会随时间变化——在设计中依赖它们之前，请查阅官方文档核实当前条款。本文的代码不依赖任何特定 provider 行为，仅依赖 OpenAI 兼容接口。

这不是基准测试。故障注入测试验证的是模拟失败下的行为；它们不说明任何特定免费层级的真实延迟或吞吐量。

熔断器不是修复方案。它是一种遏制策略。如果免费端点长期不健康，正确的做法是更换 provider，而不是调整熔断器参数。

配额跟踪是近似值。QuotaBudget 类统计的是 API 响应中报告的 token 数，可能与 provider 的内部计账不匹配。把 80% 视为安全边际，而不是精确阈值。

不要将这个模式用于实时用户面向的功能、有合同 SLA 的工作负载或高并发生产 Agent。免费层级适用于批处理任务、个人自动化和评估循环。

免费 LLM 访问是一个带有失败配置文件的预算约束。把它当作约束来处理，它就变成了一个有用的工程输入：它迫使你去设计降级路径、配额感知和快速失败行为——而这些你的付费 API 代码可能缺乏。构建一次弹性层，免费层级就不再是风险，而只是同一个客户端背后的另一个端点。

如果你想针对真实的免费端点尝试这个模式，MonkeyCode 的免费层级是一个合理的选择——客户端代码只需要一个 base URL 和一个 API key。但这个手册本身是 provider 无关的：下次任何依赖开始返回 429 时，你就已经有了正确的应对方案。