Claude API Rate Limit 与并发优化完整指南：429 错误处理、重试策略与吞吐量提升

HTTP/1.1 429 Too Many Requests
anthropic-ratelimit-requests-limit: 50
anthropic-ratelimit-requests-remaining: 0
anthropic-ratelimit-requests-reset: 2026-06-30T12:01:00Z
anthropic-ratelimit-tokens-limit: 40000
anthropic-ratelimit-tokens-remaining: 0
anthropic-ratelimit-tokens-reset: 2026-06-30T12:01:00Z
retry-after: 30

import anthropic
import time
import random

client = anthropic.Anthropic(
    api_key="YOUR_API_KEY",
    base_url=" ",  # ClaudeAPI 兼容 Anthropic SDK 格式
)

def call_with_retry(
    messages: list,
    model: str = "claude-sonnet-4-6",
    max_tokens: int = 1024,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
) -> anthropic.types.Message:
    """
    带指数退避重试的 Claude API 调用封装。
    仅对 429（Rate Limit）和 529（Overloaded）进行重试，
    4xx 认证/参数错误不重试。
    """
    for attempt in range(max_retries):
        try:
            response = client.messages.create(
                model=model,
                max_tokens=max_tokens,
                messages=messages,
            )
            return response
        except anthropic.RateLimitError as e:
            if attempt == max_retries - 1:
                raise

            retry_after = getattr(e, "retry_after", None)
            if retry_after:
                wait = float(retry_after)
            else:
                wait = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)

            print(f"[Retry {attempt + 1}/{max_retries}] Rate limited. Waiting {wait:.1f}s...")
            time.sleep(wait)
        except anthropic.APIStatusError as e:
            if e.status_code == 529 and attempt < max_retries - 1:
                wait = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
                print(f"[Retry {attempt + 1}/{max_retries}] Overloaded. Waiting {wait:.1f}s...")
                time.sleep(wait)
            else:
                raise

response = call_with_retry(
    messages=[{"role": "user", "content": "请用 100 字总结大语言模型的发展历史。"}],
    model="claude-sonnet-4-6",
)
print(response.content[0].text)

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

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY,
  baseURL: " ",
});

async function callWithRetry(
  messages: Anthropic.MessageParam[],
  model = "claude-sonnet-4-6",
  maxRetries = 5
): Promise<Anthropic.Message> {
  const baseDelay = 1000;
  const maxDelay = 60000;

  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await client.messages.create({
        model,
        max_tokens: 1024,
        messages,
      });
    } catch (err: unknown) {
      const isRateLimit =
        err instanceof Anthropic.RateLimitError ||
        (err instanceof Anthropic.APIStatusError && err.status === 529);

      if (!isRateLimit || attempt === maxRetries - 1) throw err;

      const jitter = Math.random() * 1000;
      const wait = Math.min(baseDelay * Math.pow(2, attempt) + jitter, maxDelay);

      console.log(`[Retry ${attempt + 1}/${maxRetries}] Waiting ${(wait / 1000).toFixed(1)}s...`);
      await new Promise((resolve) => setTimeout(resolve, wait));
    }
  }

  throw new Error("Max retries exceeded");
}

import asyncio
import anthropic

# 根据你的账户 RPM 配额调整 MAX_CONCURRENCY
MAX_CONCURRENCY = 10

client = anthropic.AsyncAnthropic(
    api_key="YOUR_API_KEY",
    base_url="https://gw.claudeapi.com",
)

async def process_single(sem: asyncio.Semaphore, text: str) -> str:
    async with sem:
        response = await client.messages.create(
            model="claude-haiku-4-5-20251001",  # 高并发场景优先用 Haiku 降低 TPM 消耗
            max_tokens=256,
            messages=[{"role": "user", "content": text}],
        )
        return response.content[0].text

async def batch_process(texts: list[str]) -> list[str]:
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    tasks = [process_single(sem, t) for t in texts]
    return await asyncio.gather(*tasks)

# 运行
results = asyncio.run(batch_process(["摘要任务1", "摘要任务2", "摘要任务3"]))

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "你是一名专业的代码审查助手。以下是项目的全量代码库上下文：\n\n" + large_codebase_context,
            "cache_control": {"type": "ephemeral"},  # 启用 Prompt Caching
        }
    ],
    messages=[{"role": "user", "content": "请审查 src/auth.py 中的安全问题。"}],
)