接入前的准备

429 退避代码本身和 API 来源无关,但稳定的上游接口能让退避策略真正起作用——上游波动越大,退避越频繁,用户感知越差。建议测试时确认:

  • 单 key 跑 10 次连续大输入,记录 RPM 和 ITPM 实际触顶位置
  • 用 curl -i 看一次正常响应,确认 anthropic-ratelimit-* 头格式
  • 如果是国内调 Anthropic 官方,先备一条 独立开发者用得起的 Claude 4.7 / GPT-5.5 中转,减少网络层超时与限流叠加

Anthropic Rate Limits 官方文档(访问于 2026-05-19)明确说明 429 会带 retry-after 头和 anthropic-ratelimit-tokens-reset 时间戳,这是 2026 年 1 月以后所有 SDK 都会读的字段。

Python 完整模板

下面是生产级 Python 退避,基于官方 SDK,200 行内可独立运行:

import random
import time
from typing import Callable, TypeVar
import anthropic
from anthropic import APIStatusError, RateLimitError

T = TypeVar("T")

def with_backoff(
    fn: Callable[[], T],
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: float = 0.2,
) -> T:
    """指数退避调用器。返回值或抛 RateLimitError。"""
    for attempt in range(max_retries):
        try:
            return fn()
        except RateLimitError as e:
            # 优先用 retry-after 头
            retry_after = e.response.headers.get("retry-after")
            if retry_after:
                delay = float(retry_after)
            else:
                delay = min(base_delay * (2**attempt), max_delay)
            # 加 ±jitter 抖动
            delay *= 1 + random.uniform(-jitter, jitter)
            if attempt == max_retries - 1:
                raise
            time.sleep(delay)
        except APIStatusError as e:
            # 529 过载:用纯指数退避,不消耗 429 配额
            if e.status_code == 529:
                delay = min(base_delay * (2**attempt), max_delay)
                delay *= 1 + random.uniform(-jitter, jitter)
                time.sleep(delay)
                continue
            raise
    raise RuntimeError("Unreachable")

client = anthropic.Anthropic()

def call_claude(prompt: str) -> str:
    def _go():
        msg = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )
        return msg.content[0].text

    return with_backoff(_go, max_retries=5)

if __name__ == "__main__":
    print(call_claude("一句话讲清楚指数退避"))

要点:

  1. RateLimitError 单独处理,优先用 retry-after 头精确等
  2. APIStatusError.status_code == 529 是过载,不算账号配额,纯指数退避
  3. 最后一次失败直接抛,让上层决定换 key 还是返回错误

TypeScript 完整模板

import Anthropic from '@anthropic-ai/sdk';

const client = new Anthropic();

async function withBackoff<T>(
  fn: () => Promise<T>,
  maxRetries = 5,
  baseDelay = 1000,
  maxDelay = 60_000,
  jitter = 0.2,
): Promise<T> {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await fn();
    } catch (err: any) {
      const status = err?.status ?? err?.response?.status;

      // 429: 优先读 retry-after
      if (status === 429) {
        const retryAfter = err?.headers?.['retry-after'];
        let delay = retryAfter
          ? parseFloat(retryAfter) * 1000
          : Math.min(baseDelay * 2 ** attempt, maxDelay);
        delay *= 1 + (Math.random() * 2 - 1) * jitter;
        if (attempt === maxRetries - 1) throw err;
        await new Promise((r) => setTimeout(r, delay));
        continue;
      }

      // 529: 过载,纯退避
      if (status === 529) {
        let delay = Math.min(baseDelay * 2 ** attempt, maxDelay);
        delay *= 1 + (Math.random() * 2 - 1) * jitter;
        await new Promise((r) => setTimeout(r, delay));
        continue;
      }

      throw err;
    }
  }
  throw new Error('Unreachable');
}

export async function callClaude(prompt: string) {
  return withBackoff(async () => {
    const msg = await client.messages.create({
      model: 'claude-sonnet-4-6',
      max_tokens: 1024,
      messages: [{ role: 'user', content: prompt }],
    });
    return (msg.content[0] as any).text;
  });
}

TS 版要注意 SDK 在不同版本里 error shape 不一样:err.status 是 5.x 起的官方位置,旧版本要看 err.response.status,两个都兜底最稳。

Python vs TS 的常见差异

维度Python SDKTypeScript SDK
错误类RateLimitError 单独类APIError 子类,看 status
retry-after 位置e.response.headerserr.headers
默认 max_retries22
异步默认否(用 AsyncAnthropic)是(Promise)
流式退避中断后从头重试同左,SSE 不支持续传
Jitter 内置否否
推荐外置退避是是

中转方多 key fallback 钩子

如果走 独立开发者可用的 Claude / OpenAI API 中转(访问于 2026-05-19)这类多 key 中转,可在每次 429 后切 key:

KEYS = ["sk-relay-1", "sk-relay-2", "sk-relay-3"]
BASE_URL = "https://your-relay.example.com/v1"

def call_with_key_rotation(prompt: str) -> str:
    last_err = None
    for key in KEYS:
        client = anthropic.Anthropic(api_key=key, base_url=BASE_URL)
        try:
            return with_backoff(
                lambda: client.messages.create(
                    model="claude-sonnet-4-6",
                    max_tokens=1024,
                    messages=[{"role": "user", "content": prompt}],
                ).content[0].text,
                max_retries=2,
            )
        except (RateLimitError, APIStatusError) as e:
            last_err = e
            continue
    raise last_err or RuntimeError("All keys exhausted")

中转方层做主路退避,你的应用层做 key 池切换——两段不重叠,既不会拉长延迟,也不会丢失重试。

数据:退避前后的 429 抛错率

AI Free API 2026 年 1 月统计(访问于 2026-05-19)显示,纯重试(无退避)在 Tier 1 账号上跑批量任务时,429 抛错率达到 18-22%;接入 retry-after + 指数退避后降到 3-5%,再加 jitter 降到 1% 以下。Anthropic 2026 年 4 月把 Tier 1-4 的 token 限额上调 10 倍后,这个数字进一步降到 0.4%。

上线后的监控

  • 把退避代码写成共享 lib,所有 LLM 调用统一走它
  • 监控 P95 重试次数,> 2 次说明配额不够该升 Tier 或加中转 fallback
  • 日志记录 retry_after_received 值,看 Anthropic 给的退避建议时长分布
  • max_retries 别超过 5,5 次后 ≈ 31 秒,用户感知极差,直接报错让 UI 引导用户

相关阅读