AI API Rate Limiting and Error Handling

AI APIs fail. Networks timeout, providers rate-limit, models return errors, and traffic spikes overwhelm endpoints. Production applications need robust handling for every failure mode. This guide covers battle-tested patterns for rate limiting, retry logic, circuit breakers, and error recovery when using Chinese AI models through TokenEase.

Common AI API Failure Modes

Status	Cause	Retry?
429 Too Many Requests	Rate limit exceeded	Yes, with backoff
500 Internal Server Error	Provider error	Yes, with backoff
502 Bad Gateway	Upstream provider down	Yes, with longer backoff
503 Service Unavailable	Provider overloaded	Yes, with backoff
504 Gateway Timeout	Request took too long	Yes, reduce max_tokens
401 Unauthorized	Invalid API key	No, fix credentials
400 Bad Request	Malformed request	No, fix request

Exponential Backoff with Jitter

The gold standard for retry logic. Wait longer between each retry, plus random jitter to prevent thundering herd:

import random
import time
import requests

def call_with_retry(url, headers, payload, max_retries=3):
    for attempt in range(max_retries + 1):
        try:
            response = requests.post(url, headers=headers, json=payload, timeout=60)
            
            if response.status_code == 200:
                return response.json()
            
            # Don't retry client errors (4xx except 429)
            if 400 <= response.status_code < 500 and response.status_code != 429:
                raise Exception(f"Client error {response.status_code}: {response.text}")
            
            # Server errors and rate limits - retry with backoff
            if attempt < max_retries:
                # Exponential backoff: 1s, 2s, 4s + random jitter
                base_delay = 2 ** attempt
                jitter = random.uniform(0, 1)
                delay = base_delay + jitter
                
                # Respect Retry-After header if present
                retry_after = response.headers.get('Retry-After')
                if retry_after:
                    delay = max(delay, int(retry_after))
                
                print(f"Attempt {attempt + 1} failed ({response.status_code}). Retrying in {delay:.1f}s...")
                time.sleep(delay)
            else:
                raise Exception(f"Max retries exceeded. Last status: {response.status_code}")
                
        except requests.exceptions.Timeout:
            if attempt < max_retries:
                delay = 2 ** attempt + random.uniform(0, 1)
                print(f"Timeout. Retrying in {delay:.1f}s...")
                time.sleep(delay)
            else:
                raise Exception("Request timed out after all retries")
        
        except requests.exceptions.ConnectionError:
            if attempt < max_retries:
                delay = 2 ** attempt + random.uniform(0, 1)
                print(f"Connection error. Retrying in {delay:.1f}s...")
                time.sleep(delay)
            else:
                raise Exception("Connection failed after all retries")
    
    raise Exception("Unexpected end of retry loop")

Circuit Breaker Pattern

When a provider is consistently failing, stop calling it temporarily to prevent cascading failures:

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal operation
    OPEN = "open"          # Failing, reject requests
    HALF_OPEN = "half_open"  # Testing if recovered

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=30):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = None
    
    def call(self, func, *args, **kwargs):
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                print("Circuit breaker: testing recovery...")
            else:
                raise Exception("Circuit breaker is OPEN. Provider temporarily disabled.")
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e
    
    def _on_success(self):
        self.failure_count = 0
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
            print("Circuit breaker: CLOSED (recovered)")
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN
            print(f"Circuit breaker: OPEN ({self.failure_count} failures)")

# Usage
cb = CircuitBreaker(failure_threshold=3, recovery_timeout=60)

def make_api_call():
    response = requests.post(API_URL, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

try:
    result = cb.call(make_api_call)
except Exception as e:
    # Fallback: use different model or cached response
    result = fallback_to_cached_or_alternative_model()

Model Fallback Strategy

When one model provider fails, automatically switch to another. This is where TokenEase's unified API shines:

MODEL_PRIORITY = ["deepseek", "qwen", "zhipu", "kimi", "doubao"]

def call_with_fallback(messages, preferred_model="deepseek"):
    # Try preferred model first
    models_to_try = [preferred_model] + [m for m in MODEL_PRIORITY if m != preferred_model]
    
    last_error = None
    for model in models_to_try:
        try:
            response = requests.post(
                f"{BASE_URL}/chat/completions",
                headers={"Authorization": f"Bearer {API_KEY}"},
                json={
                    "model": model,
                    "messages": messages,
                    "max_tokens": 500
                },
                timeout=30
            )
            
            if response.status_code == 200:
                print(f"Success with model: {model}")
                return response.json()
            
            # If rate limited or server error, try next model
            if response.status_code in [429, 500, 502, 503, 504]:
                print(f"{model} returned {response.status_code}, trying fallback...")
                continue
            else:
                # Client error - don't retry with other models
                response.raise_for_status()
                
        except requests.exceptions.Timeout:
            print(f"{model} timed out, trying fallback...")
            continue
        except Exception as e:
            last_error = e
            print(f"{model} failed: {e}")
            continue
    
    raise Exception(f"All models failed. Last error: {last_error}")

# Usage - seamless fallback across 5 providers
result = call_with_fallback(messages, preferred_model="deepseek")

Token Bucket Rate Limiter (Client-Side)

Prevent yourself from hitting provider rate limits by throttling your own requests:

import time
import threading

class TokenBucket:
    def __init__(self, rate_per_minute=60):
        self.rate = rate_per_minute / 60.0  # tokens per second
        self.tokens = rate_per_minute
        self.last_update = time.time()
        self.lock = threading.Lock()
    
    def acquire(self, tokens=1):
        with self.lock:
            now = time.time()
            elapsed = now - self.last_update
            self.tokens = min(self.rate * 60, self.tokens + elapsed * self.rate)
            self.last_update = now
            
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            else:
                wait_time = (tokens - self.tokens) / self.rate
                return wait_time
    
    def wait_and_acquire(self, tokens=1):
        result = self.acquire(tokens)
        if result is not True:
            time.sleep(result)
            self.acquire(tokens)

# Usage: limit to 30 requests per minute
bucket = TokenBucket(rate_per_minute=30)

for query in batch_queries:
    bucket.wait_and_acquire(1)
    response = call_api(query)

Handling Streaming Errors

Streaming connections have unique failure modes:

def safe_stream(model, messages, on_token, on_error):
    try:
        response = requests.post(
            f"{BASE_URL}/chat/completions",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={"model": model, "messages": messages, "stream": True},
            stream=True,
            timeout=(5, 60)  # (connect_timeout, read_timeout)
        )
        
        if response.status_code != 200:
            on_error(f"HTTP {response.status_code}: {response.text}")
            return
        
        for line in response.iter_lines():
            if line:
                try:
                    data = line.decode('utf-8')[6:]  # Remove "data: "
                    if data == '[DONE]':
                        break
                    chunk = json.loads(data)
                    if chunk["choices"][0]["delta"].get("content"):
                        on_token(chunk["choices"][0]["delta"]["content"])
                except Exception as e:
                    # Skip malformed chunks, continue stream
                    continue
                    
    except requests.exceptions.ChunkedEncodingError:
        on_error("Stream interrupted. Connection lost.")
    except requests.exceptions.ReadTimeout:
        on_error("Stream timed out. Model generation took too long.")
    except Exception as e:
        on_error(f"Stream error: {str(e)}")

Monitoring and Alerting

Track API health to detect issues before users complain:

import datetime

class APIMonitor:
    def __init__(self):
        self.metrics = {
            "total_requests": 0,
            "successful": 0,
            "failed": 0,
            "rate_limited": 0,
            "avg_latency": 0,
            "errors_by_model": {}
        }
    
    def record(self, model, status_code, latency, success):
        self.metrics["total_requests"] += 1
        
        if success:
            self.metrics["successful"] += 1
        else:
            self.metrics["failed"] += 1
            if status_code == 429:
                self.metrics["rate_limited"] += 1
        
        if model not in self.metrics["errors_by_model"]:
            self.metrics["errors_by_model"][model] = {"success": 0, "fail": 0}
        
        key = "success" if success else "fail"
        self.metrics["errors_by_model"][model][key] += 1
        
        # Update rolling average latency
        n = self.metrics["total_requests"]
        self.metrics["avg_latency"] = (self.metrics["avg_latency"] * (n-1) + latency) / n
    
    def health_check(self):
        failure_rate = self.metrics["failed"] / max(self.metrics["total_requests"], 1)
        
        if failure_rate > 0.1:  # >10% failure rate
            print(f"ALERT: High failure rate {failure_rate:.1%}")
            return "unhealthy"
        elif failure_rate > 0.05:
            return "degraded"
        return "healthy"

# Usage
monitor = APIMonitor()

start = time.time()
try:
    result = call_api(model="deepseek", messages=messages)
    monitor.record("deepseek", 200, time.time() - start, True)
except Exception as e:
    monitor.record("deepseek", 500, time.time() - start, False)

Production Checklist

• Set connection timeouts (5s) and read timeouts (30-60s) appropriate for your use case
• Implement exponential backoff with jitter for all retryable errors
• Use circuit breakers to prevent cascading failures during provider outages
• Implement model fallback across multiple providers for high availability
• Add client-side rate limiting to stay within provider quotas
• Log all errors with context (model, request size, timestamp) for debugging
• Monitor error rates per model and alert when failure rate exceeds 5%
• Cache successful responses for identical or similar queries
• Handle streaming errors gracefully — never crash mid-stream

Conclusion

Reliable AI integrations require more than just calling an API. They need retry logic, circuit breakers, rate limiting, and graceful degradation. The patterns in this guide have been proven in production systems handling millions of requests per day.

Start with exponential backoff for retries, add circuit breakers for resilience, and implement model fallback for high availability. Monitor everything, and your AI integration will be as reliable as any other production service.