AI API 限流重试策略实战：生产环境优雅处理 429 错误完全指南

在真实的生产环境里，429 错误从来不是偶发异常，而是常态。调用 Claude Opus 4.8 跑批量文档分析、用 GPT-5.4 处理高并发对话，或是深夜跑 DeepSeek V3 的推理任务——只要量上来，AI API 限流就是绕不过去的坎。处理得好，系统稳如磐石；处理得差，一个瞬间流量峰值就能让整条链路雪崩。本文聚焦三个核心机制：指数退避（Exponential Backoff）、令牌桶（Token Bucket）、请求队列（Request Queue），把它们在 AI API 场景下的具体实现讲清楚。

429 错误的本质：搞清楚你在对抗什么

大多数 AI API 的限流维度不止一个，常见的有三层：

维度	典型限制	说明
RPM（Requests Per Minute）	60 ~ 5000	每分钟请求次数
TPM（Tokens Per Minute）	10K ~ 2M	每分钟处理 token 量
TPD（Tokens Per Day）	1M ~ 无限	每日总 token 配额

OpenAI 和 Anthropic 都会在 429 响应头里返回关键信息：x-ratelimit-limit-requests、x-ratelimit-remaining-tokens、retry-after。先读响应头，再决定重试间隔，这是最基础的操作，但大量线上代码直接忽略了这个字段。

另一个容易踩的坑：TPM 触发的 429 和 RPM 触发的 429 行为不同。RPM 限制通常在 60 秒窗口重置，退避 5~10 秒往往够用；TPM 限制取决于排队中的 token 消耗速度，盲目快速重试反而会加剧拥塞。

指数退避：最基础的重试策略

指数退避的核心思路是：每次重试的等待时间按 base * 2^attempt + jitter 计算，避免所有客户端在同一时刻重试（即"惊群效应"）。

import time
import random
import httpx

def call_with_backoff(
    request_fn,
    max_retries: int = 6,
    base_delay: float = 1.0,
    max_delay: float = 64.0
):
    for attempt in range(max_retries):
        try:
            return request_fn()
        except httpx.HTTPStatusError as e:
            if e.response.status_code != 429:
                raise  # 非限流错误直接抛出

            # 优先读 retry-after 响应头
            retry_after = e.response.headers.get("retry-after")
            if retry_after:
                wait = float(retry_after)
            else:
                # 指数退避 + 随机抖动（jitter）
                wait = min(base_delay * (2 ** attempt), max_delay)
                wait += random.uniform(0, wait * 0.2)

            if attempt == max_retries - 1:
                raise  # 超过最大重试次数

            time.sleep(wait)

关键参数：base_delay 建议从 1 秒起步，max_delay 设 64 秒足够覆盖绝大多数场景，jitter 幅度控制在 10%~20%。Anthropic 官方文档推荐的 jitter 范围正是 [0, base_delay]，完全随机化，避免周期性重试风暴。

不要直接 time.sleep 在异步代码里。用 asyncio.sleep 替换，否则会阻塞整个事件循环：

import asyncio

async def call_with_backoff_async(request_fn, max_retries=6, ...):
    ...
    await asyncio.sleep(wait)

令牌桶算法：在客户端主动控速

指数退避是"撞了再退"，令牌桶是"提前不撞"。两者不是替代关系，而是配合使用：令牌桶做预防，退避做兜底。

令牌桶的工作原理：桶里按固定速率补充令牌（比如每秒补 50 个），每次请求消耗一定数量的令牌。桶空了就等，直到令牌补够再发请求。

import asyncio
import time

class TokenBucket:
    def __init__(self, rate: float, capacity: float):
        """
        rate: 每秒补充的令牌数（对应 RPM/60 或 TPM/60）
        capacity: 桶的最大容量（允许的突发量）
        """
        self.rate = rate
        self.capacity = capacity
        self._tokens = capacity
        self._last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self, tokens: float = 1.0):
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self._last_refill
            self._tokens = min(
                self.capacity,
                self._tokens + elapsed * self.rate
            )
            self._last_refill = now

            if self._tokens < tokens:
                wait = (tokens - self._tokens) / self.rate
                await asyncio.sleep(wait)
                self._tokens = 0
            else:
                self._tokens -= tokens

# 使用示例：按 TPM 限速
# 假设 TPM 上限 100K，换算为每秒约 1667 token
bucket = TokenBucket(rate=1667, capacity=5000)

async def call_api(prompt: str):
    estimated_tokens = len(prompt) / 3.5  # 粗估 token 数
    await bucket.acquire(estimated_tokens)
    # ... 发起实际请求

实践中，capacity（突发容量）通常设为 rate * 3，允许 3 秒的流量突发，平衡响应速度和限流安全。对 token 数量的预估可以用 tiktoken（OpenAI 兼容模型）或 anthropic.count_tokens() 做精确计算，替代上面的粗估公式。

请求队列：管理并发与优先级

当系统有多个业务线同时调用同一个 API key，单纯的令牌桶不够用——还需要一个中心化的请求队列来协调并发。

核心设计要点：

并发上限控制。用 asyncio.Semaphore 限制同时在途的请求数。对大多数免费/标准层 API，并发数控制在 5~10 是安全范围；企业层可以适当放开到 50~100。

semaphore = asyncio.Semaphore(10)  # 最大 10 并发

async def controlled_request(fn):
    async with semaphore:
        return await fn()

优先级队列。交互式请求（用户实时对话）优先于批处理任务。用 asyncio.PriorityQueue 实现，优先级数字越小越先处理：

from asyncio import PriorityQueue

queue: PriorityQueue = PriorityQueue()

# 实时请求优先级 0，批处理优先级 10
await queue.put((0, realtime_request))
await queue.put((10, batch_request))

熔断器（Circuit Breaker）集成。当连续失败次数超过阈值（比如 5 次），暂停所有请求 30 秒，避免在 API 服务降级时持续消耗重试预算。这是生产级系统必备的稳定性兜底，tenacity 库提供了现成的 stop_after_attempt + wait_exponential 组合，配合自定义 retry_if_exception 可以快速落地。

把三层策略组合起来

完整的生产方案是三层叠加：

请求发起
  → 令牌桶限速（预防触发限流）
  → 优先级队列 + 并发 Semaphore（管理并发）
  → 实际 API 调用
      → 成功：返回结果
      → 429：指数退避重试（最多 6 次）
      → 连续失败：熔断器触发，暂停 30s

几个值得注意的细节：日志里记录每次重试的 attempt、wait_time、remaining_tokens（从响应头读），这是后续调优限速参数的数据来源。另外，不同模型的限流策略差异较大——GPT-5.4-mini 的 TPM 上限通常比 GPT-5.5 高出 3~5 倍，批处理任务优先选轻量模型不只是成本考量，也能获得更宽松的速率限制。

---

如果你的项目同时接多个模型（比如主力用 Claude Opus 4.8，降级用 Sonnet 4.6 或 DeepSeek V3），管理多套 API key 和限流配置会很繁琐。我日常用的是 XycAi 词元平台，一个 OpenAI 兼容接口接入 200+ 模型，限流策略只需维护一份，Claude Code、Codex、Gemini CLI 都能直接接入。全球 CN2 节点延迟低至 5ms，价格比官方低不少，企业合规和发票也有覆盖，省去了不少运维麻烦。