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

在生产环境调用大模型 API 时，最让人头疼的往往不是模型能力，而是稳定性。你一定会遇到这些情况：高峰期请求被 429 Too Many Requests 拒绝、某个推理节点超时返回 502、流式输出中途断开、token 用量突然触顶……如果代码里没有合理的限流与重试机制，一次小波动就可能演变成全线报错。

本文系统整理大模型 API 调用中限流与重试的最佳实践，涵盖限流原理、429 处理、指数退避、抖动、并发控制、熔断降级等关键点，并给出可直接运行的 Python 与 cURL 代码示例。无论你直接调用 OpenAI、Claude，还是通过 EnlyAI 这类聚合平台接入，这些方法都通用。

一、为什么大模型 API 会被限流

限流（Rate Limiting）是 API 提供方保护后端资源的常见手段。大模型 API 的限流通常基于两个维度：

• RPM（Requests Per Minute）：每分钟允许的请求数。例如某账号限制 500 RPM，超过即返回 429。
• TPM（Tokens Per Minute）：每分钟允许处理的 token 总量。即使请求数没超，如果单次请求 prompt 很长，也可能因 TPM 触顶被限流。

除此之外，还有并发数限制（同时进行的请求数）、每日总量配额（daily quota）、以及按模型分别计算的限额。OpenAI、Anthropic、Google 都采用类似的多维限流策略。

关键认知：限流不是错误，而是服务端的“请稍后再试”信号。正确做法是等待并重试，而不是立即报错给用户。

二、读懂 429 响应与 Retry-After 头

当请求被限流时，API 会返回 HTTP 429 Too Many Requests。一个规范的限流响应通常包含以下信息：

HTTP/1.1 429 Too Many Requests
Content-Type: application/json
Retry-After: 12
X-RateLimit-Limit-Requests: 500
X-RateLimit-Remaining-Requests: 0
X-RateLimit-Reset-Requests: 12s

{
  "error": {
    "message": "Rate limit reached for requests",
    "type": "rate_limit_error",
    "code": "rate_limit_exceeded"
  }
}

其中 Retry-After 头告诉你需要等待多少秒才能再次请求，这是最权威的重试时机信号。X-RateLimit-Remaining-Requests 等自定义头则让你能提前感知配额消耗，做预防性降速。

下面是一个用 cURL 触发并观察限流响应的示例：

curl -i https://enlyai.com/v1/chat/completions \
  -H "Authorization: Bearer sk-your-api-key" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "hello"}]
  }'

加上 -i 参数可以打印响应头，方便你确认 Retry-After 和限流相关的 header。生产环境建议把这些 header 记录到日志，用于容量规划。

三、指数退避：重试的核心策略

指数退避（Exponential Backoff）是处理瞬时错误的标准做法：每次重试的等待时间按倍数递增，例如 1s、2s、4s、8s……这样既能给服务端恢复时间，又能避免所有客户端同时重试引发的“惊群效应”。

一个最小可用的指数退避重试实现如下：

import time
import requests

def call_llm_with_retry(payload, max_retries=5):
    url = "https://enlyai.com/v1/chat/completions"
    headers = {
        "Authorization": "Bearer sk-your-api-key",
        "Content-Type": "application/json",
    }

    for attempt in range(max_retries):
        try:
            resp = requests.post(url, headers=headers, json=payload, timeout=60)
            # 429 或 5xx 都应该重试
            if resp.status_code == 429 or resp.status_code >= 500:
                # 优先尊重服务端的 Retry-After
                retry_after = int(resp.headers.get("Retry-After", 0))
                wait = retry_after if retry_after > 0 else 2 ** attempt
                print(f"第 {attempt+1} 次重试，等待 {wait}s")
                time.sleep(wait)
                continue
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as e:
            wait = 2 ** attempt
            print(f"网络异常: {e}，{wait}s 后重试")
            time.sleep(wait)

    raise RuntimeError("重试次数耗尽，请求失败")

# 使用示例
data = call_llm_with_retry({
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "解释什么是指数退避"}]
})
print(data["choices"][0]["message"]["content"])

这段代码做了三件关键的事：优先读取 Retry-After 头、对 429 和 5xx 都重试、网络异常也重试。这是生产环境重试逻辑的最低标准。

四、加入抖动，避免重试风暴

纯指数退避有一个隐患：如果很多请求在同一时刻失败（比如服务端短暂抖动），它们会按相同的退避节奏重试，导致下一波请求再次同时打到服务端，形成周期性尖峰。解决方法是加入随机抖动（Jitter）。

import random
import time

def backoff_with_jitter(attempt, base=1, cap=60):
    # 等待时间 = min(base * 2^attempt, cap) + 随机抖动
    sleep_time = min(base * (2 ** attempt), cap)
    jitter = random.uniform(0, sleep_time / 2)
    return sleep_time + jitter

# 在重试循环里使用
for attempt in range(5):
    # ... 请求逻辑 ...
    wait = backoff_with_jitter(attempt)
    time.sleep(wait)

抖动让每个客户端的重试时间错开，是 AWS、Google 等大型云服务都在推荐的做法。对于大模型 API 这种高并发场景，抖动几乎是必须的。

五、用 tenacity 库简化重试逻辑

手写重试循环容易遗漏边界情况，推荐使用 Python 生态里成熟的 tenacity 库，它用装饰器声明式地描述重试策略，代码更清晰、更易维护。

pip install tenacity

from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
import requests

class RateLimitError(Exception):
    pass

@retry(
    stop=stop_after_attempt(6),
    wait=wait_exponential_jitter(initial=1, max=60),
    retry=retry_if_exception_type((RateLimitError, requests.RequestException)),
    reraise=True
)
def call_llm(messages, model="gpt-4o-mini"):
    resp = requests.post(
        "https://enlyai.com/v1/chat/completions",
        headers={"Authorization": "Bearer sk-your-api-key"},
        json={"model": model, "messages": messages},
        timeout=60,
    )
    if resp.status_code == 429:
        raise RateLimitError("被限流")
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

print(call_llm([{"role": "user", "content": "用一句话介绍量子计算"}]))

wait_exponential_jitter 内置了指数退避加抖动，retry_if_exception_type 精确控制哪些异常才重试（避免对 400 这类参数错误做无意义重试）。这是生产代码里最推荐的重试写法。

六、客户端主动限流：令牌桶算法

重试是被动应对，更主动的做法是客户端自己控制请求速率，从源头避免触发 429。令牌桶（Token Bucket）是最常用的限流算法：以固定速率往桶里放令牌，每个请求消耗一个令牌，桶空了就等待。

import time
import threading

class TokenBucket:
    def __init__(self, rate, capacity):
        # rate: 每秒生成的令牌数；capacity: 桶容量
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self):
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last
            self.last = now
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            if self.tokens < 1:
                need = 1 - self.tokens
                time.sleep(need / self.rate)
                self.tokens = 0
            else:
                self.tokens -= 1

# 假设限额 300 RPM，即 5 请求/秒，桶容量设为 10 允许小幅突发
bucket = TokenBucket(rate=5, capacity=10)

def safe_call(payload):
    bucket.acquire()  # 先拿令牌，再发请求
    # ... 调用 LLM API ...
    pass

令牌桶的好处是允许一定程度的突发流量（桶里有积累的令牌时可以瞬间消耗），同时保证长期平均速率不超限。对于批量处理、爬虫式调用场景非常合适。

七、并发控制与超时设置

除了速率，并发数也是触发限流的常见原因。如果你的应用用线程池或异步并发调用 API，务必限制最大并发量，并给每个请求设置合理的超时。

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

def process_one(item):
    try:
        resp = requests.post(
            "https://enlyai.com/v1/chat/completions",
            headers={"Authorization": "Bearer sk-your-api-key"},
            json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": item}]},
            timeout=(10, 120),  # 连接超时 10s，读取超时 120s
        )
        resp.raise_for_status()
        return resp.json()["choices"][0]["message"]["content"]
    except Exception as e:
        return f"失败: {e}"

tasks = ["总结这段文字", "翻译这句话", "生成标题"] * 20

# 限制并发为 5，避免瞬间打满 RPM
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(process_one, t) for t in tasks]
    for f in as_completed(futures):
        print(f.result())

关于超时，推荐使用二元超时：timeout=(connect_timeout, read_timeout)。连接超时设短一些（5-10s），读取超时要给足（60-120s），因为大模型生成长文本本身就慢。

八、熔断与降级：失败时优雅退化

当某个模型或通道持续失败时，继续重试只会雪上加霜。熔断器（Circuit Breaker）模式会在失败率达到阈值后“断开”一段时间，直接走降级逻辑，避免拖垮整个系统。

一个简单的降级策略是：主模型失败时切换到更便宜、更稳定的备用模型。例如 GPT-4o 失败时降级到 GPT-4o-mini，或者通过 EnlyAI 切换到 Claude、Gemini 等其他通道。

def call_with_fallback(messages):
    # 主模型 -> 备用模型1 -> 备用模型2
    models = ["gpt-4o", "claude-3-5-sonnet", "gemini-1.5-pro"]
    for model in models:
        try:
            return call_llm(messages, model=model)
        except Exception as e:
            print(f"{model} 失败: {e}，尝试下一个")
    raise RuntimeError("所有模型均不可用")

通过 EnlyAI 的统一接口，切换模型只需改一个字符串，无需为每个模型维护不同的 SDK 和鉴权逻辑，这让多模型降级变得极其简单。

九、限流重试最佳实践清单

把前面的内容浓缩成一份可执行的检查清单：

维度	建议
重试触发	仅对 429、5xx、网络超时重试；4xx 参数错误不重试
重试间隔	指数退避 + 随机抖动，优先尊重 Retry-After 头
重试次数	3-6 次为宜，过多会放大延迟、浪费配额
客户端限流	用令牌桶主动控速，留 10%-20% 安全余量
并发控制	限制最大并发数，避免瞬间打满 RPM
超时设置	二元超时，连接短、读取长
熔断降级	失败率超阈值时切换备用模型或返回缓存
可观测性	记录 Retry-After、剩余配额、重试次数到日志
幂等性	重试时注意请求幂等，避免重复扣费或重复生成

十、总结

大模型 API 的稳定性是一个系统工程，限流与重试是其中最基础也最重要的一环。核心原则可以概括为：被动重试用指数退避加抖动，主动限流用令牌桶控速，失败时用熔断降级保住用户体验。

实际落地时，建议直接使用 tenacity 这类成熟库处理重试，把精力放在业务逻辑上。同时，选择一个稳定的 API 入口同样关键——EnlyAI 提供 OpenAI 兼容的统一接口，内置多模型路由与故障转移能力，当某个上游通道出现限流或抖动时，可以快速切换到其他模型，大幅降低你自己在客户端实现复杂容错逻辑的成本。