Source: https://aiprimetech.io/zh/blog/rate-limits-and-retries/ 2026 年 23 日，六月 · 7 分钟 · API # 处理 LLM API 的速率限制与重试 A 作者 [AI Prime Tech](https://aiprimetech.io/zh/) · AI 辅助撰写 周二 09:12 时，我们的摘要生成流水线看起来运行正常：96 个工作进程在线，模型延迟中位数约为 2.8 秒，5xx 错误率没有升高。随后，一位客户上传了积压的 38,000 张支持工单。 四分钟内，流水线就“瘫痪”了，但没有任何服务器宕机。我们同时触发了两种不同的限制： - requests_per_minute：API 调用次数过多 - tokens_per_minute：提示词与生成内容的总量过大 简单的重试逻辑让情况变得更糟。每个工作进程都收到了 429，等待一秒后，又一起重试。这形成了一波重试潮，产生了更多 429，继而引发下一波重试。系统无意间针对自身配额发起了一场同步的拒绝服务攻击。 处理 LLM 的速率限制，不只是“加上指数退避”。在生产环境中，你需要将 token、队列、并发、重试、幂等性、备用服务提供商和计费行为作为一个整体系统来考虑。 ## 两种关键限制：RPM 和 TPM 大多数 LLM API 会实施以下限制的某种组合： - RPM：每分钟请求数 - TPM：每分钟 token 数 - RPD / TPD：每天的请求数或 token 数 - 并发请求限制：正在处理的调用数量上限 - 特定模型的限制：较小模型的限额更高，前沿模型的限额更低 关键在于： RPM 和 TPM 触发限制的方式不同. 如果你发送大量很短的提示词，就会先触及 RPM 限制。如果你发送较少但很长的提示词，就会先触及 TPM 限制。 示例： ``` Limit: - 600 requests/minute - 300,000 tokens/minute Workload A: - 1,000 short classification calls/minute - 200 tokens each - 200,000 tokens/minute total Result: - TPM is fine - RPM fails Workload B: - 100 long document analysis calls/minute - 5,000 tokens each - 500,000 tokens/minute total Result: - RPM is fine - TPM fails ``` 一个常见陷阱是：开发者对请求限流，却不对 token 限流。这在提示词较短的预发布环境中行得通。但到了生产环境，用户会粘贴 PDF、聊天记录、日志或法律合同，token 预算便迅速耗尽。 在实践中，我将每次 LLM 调用都视为需要预留两种资源： ``` cost = { requests: 1, tokens: estimated_prompt_tokens + max_output_tokens } ``` 即使模型返回的输出 token 数少于 max_output_tokens，按上限预留也能防止超额分配。 ## 忽视 TPM 实际上会发生什么 假设你有： ``` TPM limit: 120,000 Average request: 4,000 input tokens + 1,000 max output tokens Reserved cost: 5,000 tokens ``` 真正的最大吞吐量并不是“工作进程越多越好”，而是： ``` 120,000 / 5,000 = 24 requests per minute ``` 也就是平均每 2.5 秒一个请求。 如果你运行 100 个工作进程，且服务提供商允许突发流量，那么第一批请求可能都能成功启动。随后，账户就会触及硬性限制。之后的每个请求都会收到 429，重试也随之不断堆积。 解决办法不只是减少工作进程数量。你需要一个能够理解 token 预留机制的调度器。 ## 一个最简的 token 感知限流器 对于单个进程，令牌桶通常就够用了。随着时间推移补充容量，并要求每个请求同时获取请求配额和 token 配额。 下面是一个简化的 Python 示例： ``` import asyncio import time from dataclasses import dataclass @dataclass class Bucket: capacity: float refill_per_second: float tokens: float updated_at: float def refill(self): now = time.monotonic() elapsed = now - self.updated_at self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_per_second) self.updated_at = now async def acquire(self, amount: float): while True: self.refill() if self.tokens >= amount: self.tokens -= amount return missing = amount - self.tokens sleep_for = missing / self.refill_per_second await asyncio.sleep(min(sleep_for, 1.0)) class LLMRateLimiter: def __init__(self, rpm: int, tpm: int): now = time.monotonic() self.request_bucket = Bucket( capacity=rpm, refill_per_second=rpm / 60, tokens=rpm, updated_at=now, ) self.token_bucket = Bucket( capacity=tpm, refill_per_second=tpm / 60, tokens=tpm, updated_at=now, ) self.lock = asyncio.Lock() async def acquire(self, estimated_tokens: int): async with self.lock: await self.request_bucket.acquire(1) await self.token_bucket.acquire(estimated_tokens) ``` 用法： ``` estimated_tokens = prompt_tokens + max_output_tokens await limiter.acquire(estimated_tokens) response = await client.chat.completions.create( model="gpt-5.5", messages=messages, max_tokens=max_output_tokens, ) ``` 这里有意采用了简单设计。在分布式系统中，可以使用 Redis、Postgres 咨询锁，或采用集中式调度的队列。关键原则相同：在调用 API 之前，同时根据请求数量和 token 预算进行限流。 ## 重试：退避是必要条件，但并不充分 重试适用于： - 暂时性的 429 响应 - 网络超时 - 临时性的 500, 502, 503，或 504 错误 - 服务提供商端过载 - 连接重置 以下情况下重试存在风险： - 验证错误 - 上下文长度错误 - 身份验证失败 - 安全策略拒绝 - 必然导致失败的错误提示词 - 非幂等的工具调用 重试策略应首先对错误进行分类。 | 错误类型 | 是否重试？ | 典型操作 | 429 rate_limit_exceeded | 是 | 退避，遵循重试响应头，降低并发 | 500/502/503/504 | 是 | 采用带抖动的退避，限制尝试次数 | 接收响应体前超时 | 视情况而定 | 若支持，则使用幂等键重试 | 超出上下文长度 | 否 | 压缩、分块，或路由到上下文更长的模型 | API 密钥无效 | 否 | 禁用密钥，通知运维人员 | 内容或安全策略拒绝 | 否 | 返回受控的失败结果，或改写请求 | 工具的副作用已执行 | 不重试／视情况而定 | 需要进行幂等性设计 基本的退避形式： ``` import random def retry_delay(attempt: int, base=0.5, cap=30.0): exponential = min(cap, base * (2 ** attempt)) return random.uniform(0, exponential) ``` 其中的 random.uniform(0, exponential) 就是抖动。不要省略它。 不带抖动时： ``` 100 workers fail 100 workers sleep 1s 100 workers retry 100 workers fail 100 workers sleep 2s 100 workers retry ``` 带抖动时： ``` 100 workers fail workers retry across a spread of 0-1s then 0-2s then 0-4s ``` 第二种模式为提供商和你自己的队列留出了恢复的空间。 ### 存在重试响应头时，应遵循其指示 许多 API 会包含这样的提示： ``` Retry-After: 12 ``` 或提供商特有的重置响应头。如果 API 给出了明确的等待时长，就将其作为等待时间的下限： ``` delay = max(header_retry_after_seconds, retry_delay(attempt)) await asyncio.sleep(delay) ``` 在实践中，我还会限制重试的总时长。面向用户的聊天请求不应在用户盯着加载指示器时，悄悄重试两分钟。后台批处理任务则可以等待更长时间。 策略示例： ``` { "interactive_chat": { "max_attempts": 3, "max_total_retry_seconds": 8 }, "batch_summarization": { "max_attempts": 8, "max_total_retry_seconds": 300 } } ``` ## 使用队列优于让工作进程蜂拥而上 如果待处理工作量超过配额，就不应让所有工作进程都直接调用 LLM API。应在前面设置一个队列。 一种实用架构： ``` App/API -> Job queue -> Rate-aware dispatcher -> Provider client -> Result store ``` 调度器负责： - RPM 用量统计 - TPM 用量统计 - 模型选择 - 重试策略 - 提供商回退 - 并发限制 - 队列优先级 这样可以让过载情况变得可见。你看到的不再是成千上万个任务因以下错误而失败： 429，而是队列深度上升、预计等待时间增加。 对于基于 Redis 的系统，我倾向于将“就绪待执行”任务与“延迟重试”任务分开： ``` llm:ready llm:delayed llm:dead_letter ``` 失败的任务不会立即返回到 ready。系统会为它设置一个计划重试时间戳。一个小型调度器会将到期任务移回就绪队列。 ``` { "job_id": "sum_7f13", "model": "claude-sonnet-4.6", "estimated_tokens": 6200, "attempt": 2, "not_before": "2026-02-03T09:14:22Z" } ``` 其中的 not_before 字段可以防止重试风暴。 ## 并发控制与速率限制并不相同 并发限制约束正在执行的工作量。速率限制约束每个时间窗口内的工作量。 两者都需要。 假设每次调用耗时 20 秒，而你的 RPM 为 60。单纯的 RPM 限流器允许每秒发起一个请求。经过 20 秒后，你大约会有 20 个并发调用。如果延迟跃升至 90 秒，相同的 RPM 就会产生 90 个并发调用。这可能超出提供商或本地资源的限制。 使用信号量控制并发： ``` provider_concurrency = asyncio.Semaphore(20) async def call_model(job): async with provider_concurrency: await limiter.acquire(job.estimated_tokens) return await send_request(job) ``` 一个常见的陷阱是：过早获取信号量，会让工作进程在等待速率令牌时，闲置却占用稀缺的并发名额。通常应先获取速率配额，再在网络调用前立即获取并发名额。如果限流器的等待时间可能很长，应避免在等待期间占用工作进程资源。 ## 优雅降级：把高质量能力用在关键之处 遇到速率限制时，除了让请求失败，你还有其他选择。 例如，客服助手可以按以下方式降级： | 压力级别 | 策略 | 对用户的影响 | 正常 | 使用已配置的最强模型 | 最佳质量 | 轻度 TPM 压力 | 降低 max_tokens，精简检索得到的上下文 | 回答略短 | 中度压力 | 将简单任务路由至较小的模型 | 如果分类或路由较容易，影响很小 | 高压力 | 将非紧急任务放入队列 | 后台工作延迟 | 严重压力 | 返回缓存或模板响应 | 质量降低，但仍可用 具体示例： - 将工单标签任务路由至 Haiku 4.5、Qwen 或小型 Llama 变体，而不是使用前沿模型。 - 使用 Sonnet 4.6 进行常规推理，将 Opus 4.8 或 GPT-5.5 留给升级处理路径。 - 只有在提示确实需要相应的上下文窗口时，才将长上下文分析转移至 Fable 5 或 Gemini 3。 - 对于批处理任务，应先暂停低优先级租户的任务，避免影响交互式流量。 代价是复杂性。每条回退路径都需要测试、可观测性以及产品层面的认可。在某些领域，质量较差的回答可能比延迟的回答造成更大的损害。 ## 多密钥与多提供商回退 多密钥和多提供商路由可以提高系统韧性，但也很容易处理不当。 这里有三种不同的模式： | 模式 | 解决的问题 | 风险 | 多个密钥，同一提供商或账户 | 运行隔离 | 如果用于规避限制，可能违反提供商条款 | 多个账户或区域 | 故障隔离 | 增加计费和治理的复杂性 | 多个提供商或模型 | 可用性和成本灵活性 | 输出差异、提示不兼容 不要使用多个密钥偷偷绕过限制。应将其用于合理的隔离：租户、环境、产品或计费边界。 清晰的路由策略可以是这样的： ``` { "task": "support_ticket_summary", "primary": { "provider": "anthropic", "model": "claude-sonnet-4.6" }, "fallbacks": [ { "provider": "openai", "model": "gpt-5.5", "when": ["provider_5xx", "rate_limit_after_retry_budget"] }, { "provider": "google", "model": "gemini-3", "when": ["context_length_needed", "primary_unavailable"] } ], "degrade_to": { "provider": "open_model_cluster", "model": "qwen", "when": ["batch_low_priority"] } } ``` 如果你使用聚合服务或 AI Prime Tech 这类平台来访问多个模型，同样的工程规则依然适用：明确跟踪幂等性、预算、回退条件和最终计费语义。 ## 避免回退时重复计费 回退的难点在于判断第一次请求是否已完成。 考虑以下流程： ``` Client sends request to Provider A Provider A processes it successfully Network connection drops before client receives response Client retries with Provider B Both providers bill Two completions may exist ``` 除非提供商支持幂等键或请求查询，否则你无法仅从客户端彻底解决这个问题。但你可以减少损失。 ### 在支持的情况下使用幂等键 为每个逻辑操作生成一个稳定的键： ``` import hashlib import json def idempotency_key(task_name, tenant_id, input_payload): canonical = json.dumps(input_payload, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(canonical.encode()).hexdigest() return f"{tenant_id}:{task_name}:{digest}" ``` 发送前存储操作记录： ``` { "operation_id": "acme:summarize:9d2a...", "status": "in_flight", "provider": "anthropic", "model": "claude-sonnet-4.6", "created_at": "2026-02-03T09:12:01Z" } ``` 然后在完成后更新记录： ``` { "operation_id": "acme:summarize:9d2a...", "status": "completed", "provider_request_id": "req_abc123", "usage": { "input_tokens": 4180, "output_tokens": 612 } } ``` 在其他地方重试之前，检查操作表。如果另一个工作进程已经完成了该操作，就返回已存储的结果。 ### 区分超时类型 并非所有超时都一样： - 连接超时：请求很可能从未到达提供商 - 收到响应头之前发生读取超时：情况不确定 - 流式接收 token 时超时：提供商几乎肯定已经处理了请求 - 客户端取消：取决于取消操作是否已传递到提供商 对于情况不确定的案例，我倾向于采用保守的重试策略： ``` interactive: - retry connect timeout - do not cross-provider retry after partial stream - show recoverable error if status unknown batch: - retry after reconciliation delay - prefer same provider with idempotency key - fallback only after retry budget expires ``` ### 谨慎处理流式传输 流式传输可以改善用户感知的延迟，但会让重试变得更复杂。一旦向用户发送了部分输出，使用另一个模型重试可能会产生不同的后续内容。 对于流式聊天，我通常将“收到首个 token”视为无法再进行透明重试的分界点。在此之后，失败就会成为用户可见的失败： ``` "The response was interrupted. Retry?" ``` 这不如静默恢复那样优雅，但可以避免将不同模型产生的不兼容输出拼接在一起。 ## 可观测性：真正有用的指标 你需要的不只是成功率。 至少跟踪以下指标： ``` llm_requests_total{provider,model,status} llm_tokens_reserved_total{provider,model} llm_tokens_used_total{provider,model} llm_retries_total{provider,model,error_type} llm_queue_depth{priority,task_type} llm_queue_wait_seconds{priority,task_type} llm_rate_limit_hits_total{provider,model,limit_type} llm_fallbacks_total{from_provider,to_provider,reason} llm_duplicate_suppressed_total{task_type} ``` 在实际事故中，最有用的仪表盘面板往往不是延迟，而是： ``` queue_wait_seconds p50 / p95 rate_limit_hits by provider and model tokens_reserved vs tokens_used fallback rate by reason ``` 如果 tokens_reserved 持续显著高于 tokens_used，说明你的预留量过于保守。如果它太接近实际用量或低于实际用量，说明你正在超额分配，并会意外触及 TPM 限制。 ## 实用的重试封装 下面是一种简洁的模式，仅对符合条件的失败进行重试： ``` import asyncio import random import time RETRYABLE_STATUS = {429, 500, 502, 503, 504} class LLMError(Exception): def __init__(self, status, message="", retry_after=None): self.status = status self.retry_after = retry_after super().__init__(message) def jittered_backoff(attempt, base=0.5, cap=20): return random.uniform(0, min(cap, base * (2 ** attempt))) async def call_with_retries(fn, max_attempts=4, max_elapsed=20): started = time.monotonic() for attempt in range(max_attempts): try: return await fn() except LLMError as error: elapsed = time.monotonic() - started if error.status not in RETRYABLE_STATUS: raise if attempt == max_attempts - 1 or elapsed >= max_elapsed: raise delay = jittered_backoff(attempt) if error.retry_after is not None: delay = max(delay, error.retry_after) remaining = max_elapsed - elapsed await asyncio.sleep(min(delay, remaining)) ``` 这个封装刻意保持简单。复杂逻辑应放在分类、排队、幂等性和路由中，而不是放在一个永远重试的神奇循环里。 ## 实践要点 - 对速率限制建模时，同时考虑 RPM 和 TPM。基于 token 用量的限流可避免大型提示词带来的意外。 - 使用 带随机抖动的指数退避，并在提供商给出重试响应头时遵循它们。 - 设置 队列和调度器 ，置于应用与 LLM 提供商之间，以处理批量或高流量工作负载。 - 控制手段 单独管理并发量 ，将其与速率限制分开；否则，延迟激增可能导致过多请求同时处于处理中。 - 设计 优雅降级 要有意设计：使用更小的模型、更短的上下文、排队执行的后台任务或缓存响应。 - 将 多提供商故障回退 应作为可靠性功能，而不是随意发送重复请求的许可。 - 使用 幂等键和操作记录 可用于减少重复计费，并抑制重复生成。 - 如果流式响应在已经输出 token 后失败，应如实告知用户。 - 围绕以下指标构建仪表盘： 排队等待时间、token 预留量、触及速率限制的次数、重试次数和回退原因. - 保持重试逻辑简单；生产环境中真正需要做的是预算管理、调度，以及判断何时不应重试。 [API](/zh/blog/tag/api/) A [AI Prime Tech](https://aiprimetech.io/zh/) 本博客由 AI Prime Tech 发布。文章借助 AI 辅助起草；在据此采取行动前，请对照官方文档核实模型 ID、价格和限额等特定版本的细节。 以更低价格接入 Claude API 一个 API 密钥即可使用 Claude Opus 5.5、Sonnet 5、Haiku 4.5 和 Fable 5.1，以及 GPT-6 个模型。按量付费，无需订阅。 [获取 API 密钥 →](/register) AI Prime Tech 是独立的第三方 API 网关。Claude™ 和 Anthropic® 是 Anthropic, PBC 的商标。此处不暗示任何关联或背书。