大模型 API 限流与重试最佳实践

在生产环境中调用大模型 API，几乎每个开发者都会遇到请求被拒、返回 429 Too Many Requests 的情况。限流是 API 提供方保护服务稳定性的重要手段，而如何在客户端优雅地处理限流并保证任务最终完成，就成了工程落地的关键一环。本文将梳理 API 限流的常见机制，并给出重试策略的最佳实践，帮助你在开发中少踩坑。

## 理解 API 限流的两种常见模式

不同模型厂商的限流策略存在差异，但通常可以归为两类：

1. **速率限制（Rate Limit）** 以每秒请求数（RPS）、每分钟请求数（RPM）或每分钟令牌数（TPM）为单位。比如当前 API Key 限制为 60 RPM，意味着每分钟最多可发送 60 个请求。超出的请求会立即返回 429。

2. **并发限制（Concurrency Limit）** 对同时处于 in-flight 状态的请求数量进行限制。比较典型的是 OpenAI 的某些模型，允许的最大并发数可能从 5 到数千不等。一旦超出并发上限，新请求会排队或被直接拒绝，同样返回 429。

很多 API 会同时应用这两种限制，比如每分钟最多 500 个请求，同时最大并发 20。

## 响应头是你的重要线索

当收到 429 错误时，不要盲目重试。合规的 API 通常会在响应头中携带必要的限流信息，例如：

- `Retry-After`：建议等待多少秒后再重试（可以是秒数或 HTTP-date） - `X-RateLimit-Limit`：当前窗口的总配额 - `X-RateLimit-Remaining`：当前窗口剩余配额 - `X-RateLimit-Reset`：下一个窗口重置的时间戳

根据这些信息动态调整重试间隔，比写死一个固定等待时间要有效得多。例如，OpenAI 的 API 在返回 429 时，通常会附上 `Retry-After` 秒数；而一些中转站或第三方网关则可能使用 `X-RateLimit-*` 系列头。

## 最佳实践：指数退避 + 抖动 + 响应头解析

我们要实现的并不是单纯“失败就重试”，而是一个智能的重试控制器。基本结构如下：

- 发生 429 时，优先使用 `Retry-After` 头给出的秒数作为等待时间。 - 如果未提供 `Retry-After`，则采用指数退避（exponential backoff）算法，以初始延迟为基数，逐次翻倍。 - 在退避基础上加入随机抖动（jitter），避免多个客户端同时重试导致请求洪峰。 - 遇到 5xx 网络错误时，也可采用类似退避策略，但要设置上限（如最多重试 3 次）。 - 所有重试逻辑放在一个可复用的包装层中，避免业务代码重复。

下面是一个 Python 示例，使用 `httpx` 和 `tenacity` 库实现上述策略：

```python import random import httpx from tenacity import ( retry, retry_if_result, wait_exponential_jitter, stop_after_attempt, after_log ) import logging

logger = logging.getLogger(__name__)

def is_rate_limited(response): return response.status_code == 429

@retry( retry=retry_if_result(is_rate_limited), wait=wait_exponential_jitter(initial=1, max=60, jitter=3), stop=stop_after_attempt(5), after=after_log(logger, logging.WARNING) ) async def call_api(client, messages): resp = await client.post( "https://api.openai.com/v1/chat/completions", json={ "model": "gpt-3.5-turbo", "messages": messages } )

# 如果服务端给出了 Retry-After，优先使用 if resp.status_code == 429 and "Retry-After" in resp.headers: retry_after = int(resp.headers["Retry-After"]) raise RetryAfterException(retry_after) # 自定义异常，供上层处理

resp.raise_for_status() return resp.json() ```

`wait_exponential_jitter` 会在 1~1.5 秒、2~3 秒……范围内随机波动，避免固定间隔。若希望完全利用响应头，可以自定义 wait 函数，从异常中提取 `Retry-After` 值。对于高频调用，还可以在客户端侧实现令牌桶或滑动窗口，提前预判限制，减轻服务端压力。

## 避免重试循环的常见陷阱

- 永远不要无限重试。必须设置最大重试次数或总耗时上限，防止因配额耗尽而卡死。 - 对于非幂等的请求（如某些文件上传接口），需谨慎重试，或者通过检查响应状态实现“至少一次”语义。 - 如果使用多个 API Key 横向扩展，可在重试策略中增加 Key 轮转逻辑：当某个 Key 达到限流阈值时，切换至备用 Key，并记录其可用窗口。

## 选择更灵活的 API 接入方式

自己处理限流逻辑虽然可行，但在实际迭代中往往会遇到多模型切换、多 Key 管理、并发控制等复杂问题。如果你不希望在这些底层细节上耗费太多精力，可以考虑使用专业的 API 中转服务，它们通常已经内置了智能限流、自动重试和负载均衡。

比如 [TokenPocket API 中转站](https://tokenpocket.site) 就提供了 DeepSeek、Qwen、Claude、Gemini 等一系列主流模型的统一接口，按量计费、无需预先购买高额套餐，注册即送免费额度，非常适合开发测试和中小规模应用。你只需关心 prompt 和业务逻辑，剩下的限流与重试都会在网关层自动处理。无论是个人开发者还是小而美的团队，都可以借此大幅降低接入成本，把时间花在真正的产品创新上。