LLM Rate Limiting and Retry Patterns for Production with Anthropic SDK

import anthropic
 
# The Anthropic Python SDK auto-retries:
# - 429 (rate limit) — with exponential backoff
# - 529 (API overloaded)
# - Network timeouts
# Default: max_retries=2
 
client = anthropic.Anthropic(max_retries=5)    # Increase default retries

import anthropic
import time
import random
import logging
from anthropic import RateLimitError, APIStatusError
 
logger = logging.getLogger(__name__)
 
 
def create_with_retry(
    client: anthropic.Anthropic,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    **kwargs
) -> anthropic.types.Message:
    """Call messages.create with full retry logic."""
 
    last_error = None
    for attempt in range(max_retries):
        try:
            return client.messages.create(**kwargs)
 
        except RateLimitError as e:
            last_error = e
            # Check retry-after header if present
            retry_after = float(e.response.headers.get("retry-after", base_delay))
            delay = min(retry_after * (2 ** attempt) + random.uniform(0, 1), max_delay)
 
            logger.warning(f"Rate limited (attempt {attempt+1}/{max_retries}). Waiting {delay:.1f}s")
            time.sleep(delay)
 
        except APIStatusError as e:
            if e.status_code == 529:    # Overloaded
                delay = min(base_delay * (2 ** attempt) + random.uniform(0, 2), max_delay)
                logger.warning(f"API overloaded (attempt {attempt+1}/{max_retries}). Waiting {delay:.1f}s")
                time.sleep(delay)
                last_error = e
            else:
                raise    # Don't retry other errors
 
        except Exception as e:
            raise    # Don't retry unexpected errors
 
    raise last_error
 
 
# Usage
client = anthropic.Anthropic()
response = create_with_retry(
    client,
    model="claude-sonnet-5",
    max_tokens=1000,
    messages=[{"role": "user", "content": "Hello"}]
)

import time
import threading
 
 
class TokenBucketRateLimiter:
    """Token bucket for rate limiting API calls."""
 
    def __init__(self, calls_per_minute: int = 50, tokens_per_minute: int = 100_000):
        self.calls_per_minute = calls_per_minute
        self.tokens_per_minute = tokens_per_minute
 
        self.calls_bucket = calls_per_minute
        self.token_bucket = tokens_per_minute
 
        self.last_refill = time.time()
        self.lock = threading.Lock()
 
    def _refill(self):
        now = time.time()
        elapsed = now - self.last_refill
        if elapsed >= 60:
            self.calls_bucket = self.calls_per_minute
            self.token_bucket = self.tokens_per_minute
            self.last_refill = now
 
    def acquire(self, estimated_tokens: int = 1000, timeout: float = 120) -> bool:
        deadline = time.time() + timeout
        while time.time() < deadline:
            with self.lock:
                self._refill()
                if self.calls_bucket >= 1 and self.token_bucket >= estimated_tokens:
                    self.calls_bucket -= 1
                    self.token_bucket -= estimated_tokens
                    return True
            time.sleep(0.5)
        raise TimeoutError(f"Could not acquire rate limit token within {timeout}s")
 
 
# Usage
limiter = TokenBucketRateLimiter(calls_per_minute=40, tokens_per_minute=80_000)
 
def rate_limited_create(prompt: str, estimated_tokens: int = 1000) -> str:
    limiter.acquire(estimated_tokens=estimated_tokens)
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=min(estimated_tokens, 4000),
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text

import queue
import threading
from dataclasses import dataclass, field
from typing import Callable
 
 
@dataclass
class LLMRequest:
    prompt: str
    callback: Callable
    priority: int = 5    # 1=highest, 10=lowest
    estimated_tokens: int = 1000
 
    def __lt__(self, other):
        return self.priority < other.priority
 
 
class RateLimitedQueue:
    """Process LLM requests through a rate-limited queue."""
 
    def __init__(self, calls_per_minute: int = 40):
        self.queue = queue.PriorityQueue()
        self.calls_per_minute = calls_per_minute
        self.interval = 60 / calls_per_minute    # Seconds between calls
        self.client = anthropic.Anthropic()
        self._start_worker()
 
    def _start_worker(self):
        thread = threading.Thread(target=self._process_queue, daemon=True)
        thread.start()
 
    def _process_queue(self):
        while True:
            priority, request = self.queue.get()
            try:
                response = create_with_retry(
                    self.client,
                    model="claude-haiku-4-5",
                    max_tokens=request.estimated_tokens,
                    messages=[{"role": "user", "content": request.prompt}]
                )
                request.callback(response.content[0].text, None)
            except Exception as e:
                request.callback(None, e)
            finally:
                time.sleep(self.interval)    # Enforce rate limit
 
    def submit(self, request: LLMRequest):
        self.queue.put((request.priority, request))
 
 
# Usage
queue_processor = RateLimitedQueue(calls_per_minute=40)
 
results = {}
 
def handle_result(text, error, request_id):
    if error:
        results[request_id] = f"Error: {error}"
    else:
        results[request_id] = text
 
# Submit 100 requests — they process at rate limit pace
for i in range(100):
    req = LLMRequest(
        prompt=f"Analyze log entry {i}",
        callback=lambda t, e, rid=i: handle_result(t, e, rid),
        priority=5,
        estimated_tokens=500
    )
    queue_processor.submit(req)

import time
 
 
class RateLimitMonitor:
    def __init__(self):
        self.requests_last_minute = []
        self.rate_limit_hits = 0
        self.total_requests = 0
 
    def record_request(self, tokens_used: int):
        now = time.time()
        self.requests_last_minute = [t for t in self.requests_last_minute if now - t < 60]
        self.requests_last_minute.append(now)
        self.total_requests += 1
 
    def record_rate_limit_hit(self):
        self.rate_limit_hits += 1
 
    def get_metrics(self) -> dict:
        return {
            "requests_per_minute": len(self.requests_last_minute),
            "total_requests": self.total_requests,
            "rate_limit_hit_rate": self.rate_limit_hits / max(self.total_requests, 1),
            "headroom_pct": max(0, (50 - len(self.requests_last_minute)) / 50 * 100)
        }
 
 
monitor = RateLimitMonitor()