Claude API を Python で本番運用する — レート制限・リトライ・タイムアウトの設計

import random
import time
from anthropic import Anthropic, APIError, APIStatusError, APIConnectionError, APITimeoutError
 
client = Anthropic(timeout=60.0, max_retries=0)
 
RETRYABLE_STATUS = {429, 503, 504, 529}
MAX_ATTEMPTS = 5
BASE_DELAY = 1.0
MAX_DELAY = 30.0
 
def call_with_retry(payload: dict) -> dict:
    attempt = 0
    while True:
        attempt += 1
        try:
            return client.messages.create(**payload)
 
        except APIStatusError as e:
            if e.status_code not in RETRYABLE_STATUS or attempt >= MAX_ATTEMPTS:
                raise
            sleep_for = _retry_delay(e, attempt)
 
        except (APIConnectionError, APITimeoutError) as e:
            if attempt >= MAX_ATTEMPTS:
                raise
            sleep_for = _backoff(attempt)
 
        time.sleep(sleep_for)
 
def _backoff(attempt: int) -> float:
    # full jitter
    cap = min(BASE_DELAY * 2 ** (attempt - 1), MAX_DELAY)
    return random.uniform(0, cap)
 
def _retry_delay(err: APIStatusError, attempt: int) -> float:
    # Retry-After があれば最優先
    retry_after = getattr(err.response, "headers", {}).get("retry-after")
    if retry_after:
        try:
            return min(float(retry_after), MAX_DELAY)
        except ValueError:
            pass
    return _backoff(attempt)

except APIStatusError as e:
    log.warning({
        "event": "anthropic_status_error",
        "status": e.status_code,
        "message": getattr(e, "message", None),
        "request_id": getattr(e, "request_id", None),
        "retry_after": getattr(e.response, "headers", {}).get("retry-after"),
        "attempt": attempt,
    })

import httpx
from anthropic import Anthropic
 
http_client = httpx.Client(
    timeout=httpx.Timeout(
        connect=5.0,
        read=120.0,
        write=10.0,
        pool=5.0,
    ),
    limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
)
 
client = Anthropic(http_client=http_client, max_retries=0)

import time
 
def call_with_budget(payload: dict, total_budget_s: float) -> dict:
    start = time.monotonic()
    attempt = 0
    while True:
        attempt += 1
        remaining = total_budget_s - (time.monotonic() - start)
        if remaining <= 0:
            raise TimeoutError("budget exhausted")
 
        per_attempt_timeout = min(remaining, 30.0)
        try:
            return client.with_options(timeout=per_attempt_timeout).messages.create(**payload)
        except (APIConnectionError, APITimeoutError) as e:
            if attempt >= MAX_ATTEMPTS or (time.monotonic() - start) >= total_budget_s:
                raise
            time.sleep(_backoff(attempt))

import asyncio
from anthropic import AsyncAnthropic
 
client = AsyncAnthropic()
sem = asyncio.Semaphore(10)  # 同時並列上限
 
async def call_async(payload: dict):
    async with sem:
        return await client.messages.create(**payload)

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last = time.monotonic()
        self.lock = asyncio.Lock()
 
    async def acquire(self):
        async with self.lock:
            now = time.monotonic()
            elapsed = now - self.last
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last = now
            if self.tokens < 1:
                wait = (1 - self.tokens) / self.rate
                await asyncio.sleep(wait)
                self.tokens = 0
            else:
                self.tokens -= 1

def call_safely(payload: dict):
    request_id = None
    try:
        res = client.messages.create(**payload)
        return res
    except APIStatusError as e:
        request_id = getattr(e, "request_id", None)
        log.error({
            "event": "claude_api_status_error",
            "status": e.status_code,
            "request_id": request_id,
            "payload_model": payload.get("model"),
            "payload_max_tokens": payload.get("max_tokens"),
            "input_tokens_estimate": _estimate_tokens(payload),
        })
        raise
    except (APIConnectionError, APITimeoutError) as e:
        log.error({
            "event": "claude_api_network_error",
            "error_type": type(e).__name__,
        })
        raise

async def stream_with_idle_guard(payload, idle_limit_s=20.0):
    last = time.monotonic()
    async with client.messages.stream(**payload) as stream:
        async for event in stream:
            now = time.monotonic()
            if now - last > idle_limit_s:
                raise TimeoutError("stream idle exceeded")
            last = now
            yield event