处理 LLM API 的速率限制与重试

两种关键限制：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 错误
• 服务提供商端过载
• 连接重置

以下情况下重试存在风险：

• 验证错误
• 上下文长度错误
• 身份验证失败
• 安全策略拒绝
• 必然导致失败的错误提示词
• 非幂等的工具调用

重试策略应首先对错误进行分类。

基本的退避形式：

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)

一个常见的陷阱是：过早获取信号量，会让工作进程在等待速率令牌时，闲置却占用稀缺的并发名额。通常应先获取速率配额，再在网络调用前立即获取并发名额。如果限流器的等待时间可能很长，应避免在等待期间占用工作进程资源。

优雅降级：把高质量能力用在关键之处

遇到速率限制时，除了让请求失败，你还有其他选择。

例如，客服助手可以按以下方式降级：

具体示例：

• 将工单标签任务路由至 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 预留量、触及速率限制的次数、重试次数和回退原因.
• 保持重试逻辑简单；生产环境中真正需要做的是预算管理、调度，以及判断何时不应重试。