API Error Handling for LLM Applications: A Practical Guide

Why LLM error handling is different

Traditional API error handling follows predictable patterns: 200 means success, 4xx means client error, 5xx means server error. LLM APIs add complexity:

• Streaming errors mid-response — The connection drops after partial content
• Token limit errors — Your prompt + completion exceeds the model's context window
• Content filter triggers — The model refuses to respond based on safety filters
• Rate limits at multiple levels — Per-minute, per-day, per-model
• Silent model fallbacks — You request one model, the gateway returns a different one
• Variable latency — Responses can take 0.5s or 30s depending on load

This guide covers practical strategies for each scenario.

The error hierarchy

Client Side          → Network timeout, DNS failure, connection reset
Gateway Side         → Rate limit, auth failure, model not found, billing issue
Provider Side        → 502/503, model overload, content filter, context overflow
Application Side     → Empty response, malformed JSON, unexpected model

Retry strategies

Basic retry with exponential backoff

import time
import requests

def chat_with_retry(model, messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.post(
                "https://api.rivenai.io/v1/chat/completions",
                headers={"Authorization": "Bearer rvn_..."},
                json={"model": model, "messages": messages, "max_tokens": 100},
                timeout=30
            )
            
            if response.status_code == 200:
                return response.json()
            
            # Don't retry on client errors (except 429)
            if response.status_code in (400, 401, 403, 404):
                raise Exception(f"Client error {response.status_code}: {response.text}")
            
            # Retry on 429 (rate limit) and 5xx (server error)
            if response.status_code == 429:
                wait = min(2 ** attempt, 60)  # Cap at 60s
                print(f"Rate limited, waiting {wait}s...")
                time.sleep(wait)
                continue
                
            if response.status_code >= 500:
                wait = 2 ** attempt
                print(f"Server error {response.status_code}, retrying in {wait}s...")
                time.sleep(wait)
                continue
                
        except requests.exceptions.Timeout:
            if attempt < max_retries - 1:
                print(f"Timeout, retrying...")
                time.sleep(2 ** attempt)
            else:
                raise
    
    raise Exception(f"Failed after {max_retries} retries")

Circuit breaker pattern

For high-traffic applications, use a circuit breaker to stop retrying when a provider is consistently failing:

from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failures = 0
        self.failure_threshold = failure_threshold
        self.last_failure = None
        self.recovery_timeout = recovery_timeout
        self.state = "closed"  # closed, open, half-open
    
    def record_failure(self):
        self.failures += 1
        self.last_failure = datetime.now()
        if self.failures >= self.failure_threshold:
            self.state = "open"
    
    def record_success(self):
        self.failures = 0
        self.state = "closed"
    
    def can_proceed(self):
        if self.state == "closed":
            return True
        if self.state == "open":
            if datetime.now() - self.last_failure > timedelta(seconds=self.recovery_timeout):
                self.state = "half-open"
                return True
            return False
        return True  # half-open: allow one attempt

Handling streaming errors

Streaming responses (SSE) can fail mid-stream. Handle this gracefully:

import json
import requests

def stream_chat_with_recovery(model, messages, on_chunk, on_error):
    """Stream chat with automatic recovery for mid-stream failures."""
    try:
        response = requests.post(
            "https://api.rivenai.io/v1/chat/completions",
            headers={
                "Authorization": "Bearer rvn_...",
                "Content-Type": "application/json"
            },
            json={
                "model": model,
                "messages": messages,
                "stream": True,
                "max_tokens": 1000
            },
            stream=True,
            timeout=60
        )
        
        if response.status_code != 200:
            on_error(f"HTTP {response.status_code}: {response.text}")
            return
        
        buffer = ""
        for line in response.iter_lines():
            if line:
                line_str = line.decode("utf-8")
                if line_str.startswith("data: "):
                    data = line_str[6:]
                    if data == "[DONE]":
                        break
                    try:
                        chunk = json.loads(data)
                        delta = chunk["choices"][0].get("delta", {})
                        content = delta.get("content", "")
                        if content:
                            on_chunk(content)
                    except json.JSONDecodeError:
                        pass  # Skip malformed chunks
                        
    except requests.exceptions.ConnectionError:
        on_error("Connection lost during streaming")
    except requests.exceptions.Timeout:
        on_error("Stream timed out")
    except Exception as e:
        on_error(f"Unexpected error: {str(e)}")

Detecting silent fallbacks

When using a gateway like Riven, the response model might differ from the requested model:

response = requests.post(
    "https://api.rivenai.io/v1/chat/completions",
    headers={"Authorization": "Bearer rvn_..."},
    json={"model": "gpt-5.6", "messages": [...]}
)

data = response.json()
returned_model = data.get("model", "")
requested_model = "gpt-5.6"

if returned_model != requested_model:
    # Silent fallback detected — gateway routed to a different provider
    print(f"Warning: requested {requested_model}, got {returned_model}")
    # Decide: accept the fallback or retry with explicit provider

Context window management

Prevent context overflow errors by tracking token counts:

import tiktoken

def count_tokens(text, model="gpt-5.6"):
    """Estimate token count for a text string."""
    try:
        encoding = tiktoken.encoding_for_model(model)
    except KeyError:
        encoding = tiktoken.get_encoding("cl100k_base")
    return len(encoding.encode(text))

def truncate_messages(messages, max_tokens, model="gpt-5.6"):
    """Truncate conversation to fit within context window."""
    model_context_limits = {
        "gpt-5.6": 128000,
        "claude-opus-5": 200000,
        "glm-5.2": 128000,
        "deepseek-v3": 64000,
    }
    
    context_limit = model_context_limits.get(model, 32000)
    available = context_limit - max_tokens  # Reserve space for completion
    
    total = 0
    truncated = []
    
    # Keep system message, then work backwards through conversation
    if messages and messages[0]["role"] == "system":
        system_tokens = count_tokens(messages[0]["content"], model)
        total += system_tokens
        truncated.append(messages[0])
        messages = messages[1:]
    
    for msg in reversed(messages):
        msg_tokens = count_tokens(msg["content"], model)
        if total + msg_tokens > available:
            break
        truncated.insert(1 if truncated and truncated[0]["role"] == "system" else 0, msg)
        total += msg_tokens
    
    return truncated

Fallback model chains

Configure automatic fallback to cheaper or more reliable models:

FALLBACK_CHAINS = {
    "gpt-5.6": ["claude-opus-5", "glm-5.2", "deepseek-v3"],
    "claude-opus-5": ["gpt-5.6", "glm-5.2"],
}

def chat_with_fallback(model, messages, max_tokens=100):
    """Try the primary model, fall back to alternatives on failure."""
    chain = [model] + FALLBACK_CHAINS.get(model, [])
    
    for m in chain:
        try:
            response = requests.post(
                "https://api.rivenai.io/v1/chat/completions",
                headers={"Authorization": "Bearer rvn_..."},
                json={
                    "model": m,
                    "messages": messages,
                    "max_tokens": max_tokens
                },
                timeout=30
            )
            
            if response.status_code == 200:
                return response.json()
            
            print(f"Model {m} failed: {response.status_code}")
            continue
            
        except requests.exceptions.Timeout:
            print(f"Model {m} timed out")
            continue
    
    raise Exception("All models in fallback chain failed")

Monitoring and alerting

Track these metrics for your LLM integration:

| Metric | Alert threshold | |---|---| | Error rate (non-200) | > 5% over 5 min | | P99 latency | > 30s | | Rate limit hits (429) | > 10/min | | Silent fallback rate | > 1% | | Empty response rate | > 0.5% | | Circuit breaker trips | > 0 |

Best practices summary

1. Always set timeouts — LLM calls can hang indefinitely
2. Use streaming for long responses — Prevents timeout on long completions
3. Implement exponential backoff — Never retry immediately
4. Check the response model — Detect silent fallbacks
5. Track token usage — Prevent context overflow before it happens
6. Have fallback models — Don't depend on a single provider
7. Log everything — You need full request/response for debugging
8. Set per-key budgets — Prevent runaway costs
9. Use a gateway — Riven handles failover, rate limiting, and billing

Getting started

Riven's gateway handles most of these concerns for you — automatic failover, rate limiting, budget caps, and usage tracking are built in. Sign up free and start building.