Claude API の 429 エラーを 6 サイト同時運用で観測して書き直した本番リトライの実装

import time
import random
import hashlib
import json
import anthropic
from anthropic import RateLimitError, APIStatusError
 
client = anthropic.Anthropic()
 
def idempotency_key(model: str, messages: list, max_tokens: int) -> str:
    """Same request -> same key. Dedup is the caller's job."""
    payload = json.dumps({"m": model, "ms": messages, "t": max_tokens}, sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()[:24]
 
def call_with_retry(model, messages, max_tokens=1024, max_retries=5):
    """Jittered exponential backoff with Retry-After honoring."""
    base = 1.0  # base seconds
    key = idempotency_key(model, messages, max_tokens)
 
    for attempt in range(max_retries):
        try:
            resp = client.messages.create(
                model=model,
                max_tokens=max_tokens,
                messages=messages,
                extra_headers={"X-Idempotency-Key": key},
            )
            return resp
        except RateLimitError as e:
            retry_after = e.response.headers.get("Retry-After") if e.response else None
            if retry_after:
                wait = float(retry_after)
            else:
                wait = base * (2 ** attempt)
 
            # 1. positive jitter (avoids window-edge collisions)
            wait = wait * (1.2 + random.random() * 0.3)
            # 2. cap (prevents runaway 60s+ sleeps)
            wait = min(wait, 60.0)
 
            if attempt == max_retries - 1:
                raise
            time.sleep(wait)
        except APIStatusError as e:
            # 529 overloaded -- same treatment as 429
            if e.status_code != 529:
                raise
            time.sleep(base * (2 ** attempt) * (1.2 + random.random() * 0.3))
 
    raise RuntimeError("retry exhausted")

import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
async function sleep(ms: number) {
  return new Promise((r) => setTimeout(r, ms));
}
 
export async function callWithRetry(
  model: string,
  messages: Anthropic.MessageParam[],
  maxTokens = 1024,
  maxRetries = 5
) {
  const base = 1000; // ms
 
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      return await client.messages.create({
        model,
        max_tokens: maxTokens,
        messages,
      });
    } catch (err) {
      const status = (err as any)?.status;
      const retryAfter = (err as any)?.headers?.["retry-after"];
 
      if (status !== 429 && status !== 529) throw err;
      if (attempt === maxRetries - 1) throw err;
 
      let waitMs = retryAfter
        ? parseFloat(retryAfter) * 1000
        : base * 2 ** attempt;
      // positive jitter + cap
      waitMs = Math.min(waitMs * (1.2 + Math.random() * 0.3), 60_000);
 
      await sleep(waitMs);
    }
  }
  throw new Error("retry exhausted");
}

import time
import threading
 
class TokenBucket:
    """Self-imposed rate limit. Survives bursts; smooths long-running batches."""
    def __init__(self, capacity: int, refill_per_sec: float):
        self.capacity = capacity
        self.tokens = float(capacity)
        self.refill = refill_per_sec
        self.last = time.monotonic()
        self.lock = threading.Lock()
 
    def take(self, cost: int = 1, timeout: float = 30.0) -> bool:
        deadline = time.monotonic() + timeout
        while True:
            with self.lock:
                now = time.monotonic()
                delta = now - self.last
                self.tokens = min(self.capacity, self.tokens + delta * self.refill)
                self.last = now
                if self.tokens >= cost:
                    self.tokens -= cost
                    return True
                short = cost - self.tokens
                wait = short / self.refill
            if time.monotonic() + wait > deadline:
                return False
            time.sleep(min(wait, 1.0))
 
    def slow_down(self, factor: float = 0.5):
        """Called when RateLimit-Remaining-Tokens drops below threshold."""
        with self.lock:
            self.refill *= factor
 
# usage
tpm_bucket = TokenBucket(capacity=90_000, refill_per_sec=90_000 / 60)
 
def call(model, messages, max_tokens=1024):
    estimated = sum(len(m["content"]) for m in messages) // 3 + max_tokens
    if not tpm_bucket.take(estimated, timeout=60):
        raise RuntimeError("local bucket starved -- backpressure upstream")
    resp = call_with_retry(model, messages, max_tokens)
    remaining = int(resp.response.headers.get("RateLimit-Remaining-Tokens", 0))
    if remaining < 10_000:
        tpm_bucket.slow_down(0.6)
    return resp

import anthropic
 
client = anthropic.Anthropic()
 
# 何度も使い回す大きな共通コンテキスト（数千〜数万トークン）
SHARED_CONTEXT = """（製品ドキュメントやスタイルガイドなど、毎回同じ前置き）"""
 
def ask_with_cache(question: str) -> anthropic.types.Message:
    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=[
            {"type": "text", "text": "あなたは丁寧なアシスタントです。"},
            {
                "type": "text",
                "text": SHARED_CONTEXT,
                "cache_control": {"type": "ephemeral"},  # ここをキャッシュ
            },
        ],
        messages=[{"role": "user", "content": question}],
    )
    u = resp.usage
    # cache_read_input_tokens が伸びていれば TPM 負荷が減っている
    print(u.input_tokens, u.cache_read_input_tokens, u.cache_creation_input_tokens)
    return resp