⬡ API & SDK/2026-05-02Advanced

Building a Cost-Optimized Multi-Provider AI Gateway with Claude API and LiteLLM — Fallback Design, A/B Testing, and Provider Migration Strategy

Learn how to build a production-grade multi-provider AI gateway centered on Claude API using LiteLLM. Covers fallback chain design, A/B testing, cost-based routing, and provider migration strategy with complete code examples.

Claude API¹²⁸LiteLLMmulti-providerAI gatewayfallback⁹ cost optimization¹³A/B testingproduction¹¹¹

✦ Premium Article

If you've run production AI features on a single provider, you've probably hit the wall: a 429 rate_limit_exceeded fires at 3 AM during a batch job and you find out about it from user complaints the next morning. Or a provider incident takes down a feature that's now central to your product's value proposition, and your only option is to wait.

After a few incidents like that on projects I've shipped, I stopped treating multi-provider strategy as a theoretical best practice and started treating it as operational hygiene — the same way you'd treat database replication or retry logic for any network call. This article walks through building an intelligent AI gateway using LiteLLM with Claude API as the primary provider and OpenAI GPT-4o or Google Gemini as fallbacks. I'll cover not just how to set things up, but why each design decision matters and what breaks when you skip it.

What LiteLLM Actually Does (Beyond the Elevator Pitch)

LiteLLM is a Python library and proxy server that unifies 100+ LLM providers behind an OpenAI-compatible interface. The common description — "call any LLM with one interface" — is accurate but undersells what makes it production-useful.

The real value is declarative routing, fallback, and cost tracking across providers. Compare what non-LiteLLM code looks like when you need to support multiple providers:

# Without LiteLLM: provider-specific branches scattered everywhere
import anthropic
import openai
 
def call_ai(provider: str, prompt: str) -> str:
    if provider == "claude":
        client = anthropic.Anthropic()
        message = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            messages=[{"role": "user", "content": prompt}]
        )
        return message.content[0].text
    elif provider == "openai":
        client = openai.OpenAI()
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content
    elif provider == "gemini":
        # ... another SDK, another pattern
        pass
    # Every new provider adds another branch

# With LiteLLM: one function signature, configuration determines behavior
import litellm
 
def call_ai(model: str, prompt: str) -> str:
    response = litellm.completion(
        model=model,  # "claude-sonnet-4-6", "gpt-4o", "gemini/gemini-2.5-pro"
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

Claude API works with either the anthropic/ prefix (anthropic/claude-sonnet-4-6) or the model name alone (claude-sonnet-4-6). LiteLLM routes internally to the Anthropic SDK.

Environment Setup and Initial Configuration

Start with installation. Using uv is faster:

# Using uv (recommended — 10x faster than pip)
uv add litellm python-dotenv
 
# Or pip
pip install litellm python-dotenv

Store all API keys in .env — never in source code:

# .env
ANTHROPIC_API_KEY=your_anthropic_api_key
OPENAI_API_KEY=your_openai_api_key
GOOGLE_API_KEY=your_google_api_key
 
# Enable LiteLLM debug logs during development only
# LITELLM_LOG=DEBUG

A basic verification call:

import os
from dotenv import load_dotenv
import litellm
 
load_dotenv()
 
# Basic Claude Sonnet 4.6 call through LiteLLM
response = litellm.completion(
    model="claude-sonnet-4-6",
    messages=[
        {"role": "user", "content": "Explain Python list comprehensions in one sentence."}
    ],
    max_tokens=512,
    temperature=0.3
)
 
print(response.choices[0].message.content)
 
# Always inspect token usage — this is your cost driver
print(f"Input:  {response.usage.prompt_tokens} tokens")
print(f"Output: {response.usage.completion_tokens} tokens")
print(f"Model:  {response.model}")

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Designing a Claude-first fallback chain with per-path retry strategy so a single-provider failure never takes the service down

✦How task-complexity routing combined with semantic caching cut a real monthly API bill by roughly 32%

✦A practical way to handle Anthropic-only features (like Extended Thinking) silently disappearing on the fallback provider

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Fallback Chain Design — The Core of Reliability

The fallback configuration is where multi-provider architecture pays off directly. The design goal: when Claude returns a 429 or 500, route automatically to the next provider without any manual intervention and without the calling code knowing anything changed.

Minimal Fallback

import litellm
 
def generate_with_fallback(prompt: str, max_tokens: int = 1024) -> dict:
    """
    Try Claude Sonnet 4.6 first.
    On failure, try GPT-4o, then Gemini Flash.
    Returns both the response content and which model actually answered.
    """
    response = litellm.completion(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_tokens,
        fallbacks=["gpt-4o", "gemini/gemini-2.5-flash"],
        num_retries=2,      # Retry each model up to 2 times before moving on
        request_timeout=30, # Give each model 30 seconds
    )
    
    return {
        "content": response.choices[0].message.content,
        "model_used": response.model,  # Which model actually responded
        "usage": {
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
        }
    }
 
# Usage
result = generate_with_fallback(
    "What are the tradeoffs between Python asyncio and threading for I/O-bound tasks?"
)
print(f"Answered by: {result['model_used']}")
print(result["content"])

Passing a list to fallbacks is all it takes. LiteLLM manages the exception handling, retry delays, and provider selection internally.

Exception-Type-Aware Fallback Policy

Not all errors call for the same fallback behavior. A rate limit error means "come back later or try a different model"; a 500 means "the provider is having issues." Different situations warrant different responses:

import litellm
 
response = litellm.completion(
    model="claude-sonnet-4-6",
    messages=[{"role": "user", "content": prompt}],
    
    # On rate limit: prefer cheaper models to avoid cost escalation
    fallbacks=[
        {
            "model": "claude-haiku-4-5",
            "condition": "RateLimitError",
        },
        {
            "model": "gpt-4o-mini",
            "condition": "RateLimitError",
        },
    ],
    
    num_retries=3,
    request_timeout=45,
)

The pattern I've settled on after several production deployments: for rate limit fallbacks, use lower-cost models. When the primary is simply overloaded, you often don't need its full capability — you just need a response. Save the premium fallback (same-tier models) for actual availability failures where quality matters.

Async Fallback for High-Throughput Services

In high-concurrency contexts, use the async version:

import asyncio
import litellm
 
async def generate_async(prompt: str) -> str:
    response = await litellm.acompletion(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": prompt}],
        fallbacks=["gpt-4o", "gemini/gemini-2.5-flash"],
        num_retries=2,
        request_timeout=30,
    )
    return response.choices[0].message.content
 
# Process multiple requests concurrently
async def process_batch(prompts: list[str]) -> list[str]:
    tasks = [generate_async(p) for p in prompts]
    return await asyncio.gather(*tasks, return_exceptions=True)

Cost-Based Routing — Cut Your API Bill Without Sacrificing Quality

One of the most impactful changes you can make to a production AI service is routing tasks to appropriately-sized models. Most production workloads mix simple tasks (classification, short QA) with complex ones (long-form generation, multi-step reasoning). Sending everything through the most capable — and most expensive — model is common, but rarely optimal.

Here's a routing layer that classifies task complexity and selects the right model automatically:

from enum import Enum
from dataclasses import dataclass
import litellm
 
class TaskComplexity(Enum):
    SIMPLE = "simple"    # Classification, keyword extraction, short QA
    MEDIUM = "medium"    # Summarization, translation, structured JSON output
    COMPLEX = "complex"  # Long-form generation, code, multi-step reasoning
 
@dataclass
class ModelConfig:
    primary: str
    fallback: str
    max_tokens: int
    estimated_cost_per_1k_input: float  # USD, approximate
 
# Routing table — update as pricing changes
ROUTING_TABLE: dict[TaskComplexity, ModelConfig] = {
    TaskComplexity.SIMPLE: ModelConfig(
        primary="claude-haiku-4-5",
        fallback="gpt-4o-mini",
        max_tokens=512,
        estimated_cost_per_1k_input=0.00025,  # Haiku pricing
    ),
    TaskComplexity.MEDIUM: ModelConfig(
        primary="claude-sonnet-4-6",
        fallback="gpt-4o",
        max_tokens=2048,
        estimated_cost_per_1k_input=0.003,    # Sonnet pricing
    ),
    TaskComplexity.COMPLEX: ModelConfig(
        primary="claude-opus-4-6",
        fallback="gpt-4o",
        max_tokens=8192,
        estimated_cost_per_1k_input=0.015,    # Opus pricing
    ),
}
 
def route_and_complete(
    prompt: str,
    complexity: TaskComplexity,
    system: str | None = None,
) -> dict:
    config = ROUTING_TABLE[complexity]
    
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})
    
    response = litellm.completion(
        model=config.primary,
        messages=messages,
        max_tokens=config.max_tokens,
        fallbacks=[config.fallback],
        num_retries=2,
    )
    
    # Compute actual cost
    from litellm import completion_cost
    actual_cost = completion_cost(completion_response=response)
    
    return {
        "content": response.choices[0].message.content,
        "model_used": response.model,
        "actual_cost_usd": actual_cost,
        "complexity_tier": complexity.value,
    }
 
# Examples
sentiment = route_and_complete(
    "Classify sentiment: 'Shipping was slow but the product itself is excellent'",
    TaskComplexity.SIMPLE
)
 
summary = route_and_complete(
    "Summarize the key technical points from this 5,000-word article: ...",
    TaskComplexity.MEDIUM
)
 
architecture = route_and_complete(
    "Design a distributed rate-limiting system that handles 100k req/sec...",
    TaskComplexity.COMPLEX
)

Building Cost Visibility Into the Application Layer

Aggregate cost data lets you understand which features are expensive, set per-user budgets, and catch unexpected spikes before they hit your invoice:

import litellm
from litellm import completion_cost
from datetime import datetime
 
async def call_with_cost_tracking(
    user_id: str,
    feature: str,
    model: str,
    messages: list,
    db,
    **kwargs
) -> dict:
    """
    Wrapper that records every API call with cost attribution.
    Use for per-user billing and feature-level cost dashboards.
    """
    start_time = datetime.now()
    
    response = await litellm.acompletion(
        model=model,
        messages=messages,
        **kwargs
    )
    
    latency_ms = (datetime.now() - start_time).total_seconds() * 1000
    cost = completion_cost(completion_response=response)
    
    # Record to database — use this for dashboards and alerts
    await db.execute("""
        INSERT INTO api_usage (
            user_id, feature, model, 
            prompt_tokens, completion_tokens, 
            cost_usd, latency_ms, created_at
        )
        VALUES ($1, $2, $3, $4, $5, $6, $7, NOW())
    """,
        user_id, feature, response.model,
        response.usage.prompt_tokens,
        response.usage.completion_tokens,
        cost,
        latency_ms,
    )
    
    # Alert on expensive calls (adjust threshold per your budget)
    if cost > 0.10:
        logger.warning(
            f"Expensive API call: user={user_id}, feature={feature}, "
            f"cost=${cost:.4f}, model={response.model}"
        )
    
    return {
        "content": response.choices[0].message.content,
        "model": response.model,
        "cost_usd": cost,
        "latency_ms": latency_ms,
    }

A/B Testing — Compare Models on Real Production Traffic

Benchmarks tell you how models perform on standardized test sets. Your production workload will tell you something entirely different. The only reliable way to answer "is Claude Sonnet 4.6 or GPT-4o better for our specific use case?" is to run both on real traffic and measure.

import random
import litellm
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
 
@dataclass
class ABTestResult:
    variant: str
    model: str
    response: str
    latency_ms: float
    cost_usd: float
    timestamp: datetime
 
@dataclass
class ABTestStats:
    """Aggregated statistics for monitoring."""
    total_calls: int = 0
    total_cost_usd: float = 0.0
    avg_latency_ms: float = 0.0
    latencies: list = field(default_factory=list)
 
class ModelABTester:
    def __init__(
        self,
        model_a: str = "claude-sonnet-4-6",
        model_b: str = "gpt-4o",
        traffic_split: float = 0.8,  # 80% to Model A
    ):
        self.model_a = model_a
        self.model_b = model_b
        self.traffic_split = traffic_split
        self._stats: dict[str, ABTestStats] = {
            "A": ABTestStats(),
            "B": ABTestStats(),
        }
    
    def complete(self, messages: list, **kwargs) -> ABTestResult:
        # Consistent assignment via random — for sticky sessions, use user_id hash
        variant = "A" if random.random() < self.traffic_split else "B"
        model = self.model_a if variant == "A" else self.model_b
        
        start = datetime.now()
        try:
            response = litellm.completion(model=model, messages=messages, **kwargs)
            latency_ms = (datetime.now() - start).total_seconds() * 1000
            
            from litellm import completion_cost
            cost = completion_cost(completion_response=response)
            
            # Update in-memory stats
            stats = self._stats[variant]
            stats.total_calls += 1
            stats.total_cost_usd += cost
            stats.latencies.append(latency_ms)
            stats.avg_latency_ms = sum(stats.latencies) / len(stats.latencies)
            
            return ABTestResult(
                variant=variant,
                model=model,
                response=response.choices[0].message.content,
                latency_ms=latency_ms,
                cost_usd=cost,
                timestamp=datetime.now(),
            )
        except Exception as e:
            # Log failure per variant for quality metrics
            raise
    
    def get_comparison(self) -> dict:
        """Return current A/B comparison stats."""
        a, b = self._stats["A"], self._stats["B"]
        return {
            "model_a": {"model": self.model_a, **vars(a)},
            "model_b": {"model": self.model_b, **vars(b)},
        }
 
# Usage
tester = ModelABTester(
    model_a="claude-sonnet-4-6",
    model_b="gpt-4o",
    traffic_split=0.9,  # 90% Claude while validating GPT-4o
)
 
result = tester.complete(
    messages=[{"role": "user", "content": "Write a detailed code review comment for: ..."}]
)
print(f"Variant {result.variant} ({result.model}): {result.latency_ms:.0f}ms, ${result.cost_usd:.5f}")
 
# Periodically log the comparison
print(tester.get_comparison())

For statistically meaningful results, aim for at least 1,000 calls per variant before drawing conclusions on latency and cost. Quality evaluation usually requires human annotation or an LLM-as-judge setup on a sample.

LiteLLM Proxy Server — A Shared Gateway for Teams

The library approach works well for single services. Once multiple teams or microservices need AI, running LiteLLM as a centralized proxy becomes much more maintainable. API key management, routing policy, caching, and spend tracking all live in one place.

# litellm_config.yaml
model_list:
  - model_name: "claude-primary"
    litellm_params:
      model: "claude-sonnet-4-6"
      api_key: os.environ/ANTHROPIC_API_KEY
      
  - model_name: "claude-haiku"
    litellm_params:
      model: "claude-haiku-4-5"
      api_key: os.environ/ANTHROPIC_API_KEY
      
  - model_name: "gpt-4o-fallback"
    litellm_params:
      model: "gpt-4o"
      api_key: os.environ/OPENAI_API_KEY
 
router_settings:
  routing_strategy: "least-busy"  # Route to least-loaded model instance
  num_retries: 3
  fallbacks:
    - {"claude-primary": ["claude-haiku", "gpt-4o-fallback"]}
  
  # Budget guardrails per model
  model_group_alias:
    "production-primary": ["claude-primary"]
    "production-backup": ["gpt-4o-fallback"]
 
litellm_settings:
  # Enable Redis caching for prompt cache compatibility
  cache: true
  cache_params:
    type: "redis"
    host: os.environ/REDIS_HOST
    port: 6379
    ttl: 3600  # 1 hour cache TTL
 
general_settings:
  master_key: os.environ/LITELLM_MASTER_KEY
  database_url: os.environ/DATABASE_URL
  
  # Per-key budget limits (optional — for per-team cost allocation)
  max_budget: 500.0  # $500 total budget
  budget_duration: "30d"

# Start the proxy
litellm --config litellm_config.yaml --port 8000
 
# Or with Docker
docker run -p 8000:8000 \
  -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY \
  -e OPENAI_API_KEY=$OPENAI_API_KEY \
  -v $(pwd)/litellm_config.yaml:/app/config.yaml \
  ghcr.io/berriai/litellm:main \
  --config /app/config.yaml

All services use the proxy via the standard OpenAI SDK — no LiteLLM dependency in application code:

import openai
 
# Every service points to the same proxy
client = openai.OpenAI(
    api_key="team-key-from-config",
    base_url="http://your-litellm-proxy:8000"
)
 
response = client.chat.completions.create(
    model="claude-primary",  # Name from litellm_config.yaml
    messages=[{"role": "user", "content": "Your request here"}]
)

The key architectural benefit: swap the backend model by editing litellm_config.yaml, with zero changes to any application code. This is especially useful for phased rollouts and emergencies.

Three Production Pitfalls and How to Avoid Them

Pitfall 1: Streaming + Fallback Requires Explicit Handling

Fallback behavior differs between stream=False and stream=True. When a fallback fires mid-stream, the client connection breaks without a clean error. Handle this at the caller level:

from typing import Generator
 
def stream_with_fallback(prompt: str) -> Generator[str, None, None]:
    """
    Try to stream from Claude. On error, fall back to a non-streaming response.
    This degrades gracefully — the user sees the full response slightly delayed
    rather than a broken stream.
    """
    try:
        for chunk in litellm.completion(
            model="claude-sonnet-4-6",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
            request_timeout=30,
        ):
            delta = chunk.choices[0].delta.content
            if delta:
                yield delta
    except Exception as e:
        # Log the exception with context
        logger.warning(f"Streaming fallback triggered: {type(e).__name__}")
        
        # Non-streaming fallback — yields the full response as one chunk
        response = litellm.completion(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
        )
        yield response.choices[0].message.content

Pitfall 2: Anthropic-Specific Parameters Need `extra_body`

Claude's Extended Thinking, betas, and other Anthropic-specific fields don't map to standard OpenAI parameters. Pass them via extra_body:

# Extended Thinking via LiteLLM
response = litellm.completion(
    model="claude-opus-4-6",
    messages=[{"role": "user", "content": "Solve this multi-step optimization problem..."}],
    max_tokens=16000,
    extra_body={
        "thinking": {
            "type": "enabled",
            "budget_tokens": 8000
        }
    }
)
 
# Access thinking content if needed
# Note: LiteLLM normalizes the response, so thinking blocks may need extra handling

Pitfall 3: Tool Call Response Shapes Differ Between Providers

LiteLLM normalizes tool call responses to a common format, but edge cases appear with complex schemas, nested structures, or when Claude's Extended Thinking is active alongside tool use. Build defensive parsing:

import json
from typing import Any
 
def safe_parse_tool_call(response) -> dict[str, Any]:
    """
    Parse tool call results defensively across Claude and GPT-4o.
    LiteLLM normalizes most cases, but validate regardless.
    """
    message = response.choices[0].message
    
    if not message.tool_calls:
        return {"error": "no tool call in response", "model": response.model}
    
    tool_call = message.tool_calls[0]
    
    try:
        args = json.loads(tool_call.function.arguments)
        
        # Type validation — add your schema checks here
        if not isinstance(args, dict):
            return {
                "error": f"Expected dict, got {type(args).__name__}",
                "raw": tool_call.function.arguments
            }
        
        return {
            "name": tool_call.function.name,
            "args": args,
            "model": response.model,
        }
    except json.JSONDecodeError as e:
        # Can occur with Claude when thinking is enabled alongside tool use
        logger.error(f"Tool call JSON parse error from {response.model}: {e}")
        return {
            "error": f"JSON parse error: {e}",
            "raw": tool_call.function.arguments,
            "model": response.model,
        }

Phased Provider Migration — Moving Between Providers Without Risk

If you ever need to migrate from Claude to another provider — or onboard Claude into an existing stack that uses OpenAI — LiteLLM's traffic splitting makes this safe:

import os
import random
import litellm
 
# Migration phases — advance when quality metrics are satisfactory
MIGRATION_PHASES = {
    "phase_0": {"claude": 1.00, "gpt4o": 0.00},  # Baseline
    "phase_1": {"claude": 0.90, "gpt4o": 0.10},  # 10% to new provider
    "phase_2": {"claude": 0.70, "gpt4o": 0.30},  # Validate at 30%
    "phase_3": {"claude": 0.50, "gpt4o": 0.50},  # Parity validation
    "phase_4": {"claude": 0.00, "gpt4o": 1.00},  # Migration complete
}
 
def get_model_for_request() -> str:
    """Select model based on current migration phase."""
    phase = os.getenv("MIGRATION_PHASE", "phase_0")
    weights = MIGRATION_PHASES.get(phase, MIGRATION_PHASES["phase_0"])
    
    return "claude-sonnet-4-6" if random.random() < weights["claude"] else "gpt-4o"
 
# Advance phases by setting the env var — no deployment required
# MIGRATION_PHASE=phase_1 → Start directing 10% of traffic to GPT-4o
# MIGRATION_PHASE=phase_2 → Increase to 30% once phase_1 metrics look good

Monitor these metrics per phase before advancing:

Error rate — are failures increasing in the new provider?
Latency p50/p99 — is response time acceptable?
Quality score — use LLM-as-judge or human annotation on a sample

Hard cutover migrations carry significant risk. This phased approach adds a few weeks to the timeline but lets you catch problems at 10% traffic rather than at 100%.

Observability: Knowing What's Happening in Production

A gateway without observability is a black box. Wire LiteLLM into your existing monitoring:

import litellm
from litellm.integrations.custom_logger import CustomLogger
 
class ProductionLogger(CustomLogger):
    """Send LiteLLM events to your monitoring infrastructure."""
    
    def log_success_event(self, kwargs, response_obj, start_time, end_time):
        latency_ms = (end_time - start_time).total_seconds() * 1000
        
        # Send to your metrics system (Datadog, Prometheus, etc.)
        metrics.histogram("litellm.latency_ms", latency_ms, tags={
            "model": kwargs.get("model"),
            "fallback_triggered": str(kwargs.get("_is_fallback", False)),
        })
        
        # Track cost per model for budget dashboards
        from litellm import completion_cost
        cost = completion_cost(completion_response=response_obj)
        metrics.increment("litellm.cost_usd", cost, tags={"model": response_obj.model})
    
    def log_failure_event(self, kwargs, response_obj, start_time, end_time):
        logger.error(
            "LiteLLM request failed",
            extra={
                "model": kwargs.get("model"),
                "exception": str(kwargs.get("exception")),
                "fallback_triggered": kwargs.get("_is_fallback", False),
            }
        )
        metrics.increment("litellm.errors", tags={
            "model": kwargs.get("model"),
            "error_type": type(kwargs.get("exception")).__name__,
        })
 
# Register the logger globally
litellm.callbacks = [ProductionLogger()]

Putting It Together: A Production-Ready Gateway Module

Here's a minimal but complete gateway module that combines everything covered in this article:

# ai_gateway.py
import os
import random
import litellm
from litellm import completion_cost
from enum import Enum
from dataclasses import dataclass
from datetime import datetime
from typing import Any
 
class TaskTier(Enum):
    FAST = "fast"       # <100ms target, low cost
    STANDARD = "standard"  # Standard quality
    DEEP = "deep"          # Best quality, complex reasoning
 
TIER_CONFIG = {
    TaskTier.FAST: ("claude-haiku-4-5", "gpt-4o-mini", 512),
    TaskTier.STANDARD: ("claude-sonnet-4-6", "gpt-4o", 4096),
    TaskTier.DEEP: ("claude-opus-4-6", "gpt-4o", 16384),
}
 
class AIGateway:
    def __init__(self, db=None):
        self.db = db
    
    def complete(
        self,
        prompt: str,
        tier: TaskTier = TaskTier.STANDARD,
        system: str | None = None,
        user_id: str | None = None,
    ) -> dict[str, Any]:
        primary, fallback, max_tokens = TIER_CONFIG[tier]
        
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})
        
        start = datetime.now()
        response = litellm.completion(
            model=primary,
            messages=messages,
            max_tokens=max_tokens,
            fallbacks=[fallback],
            num_retries=2,
            request_timeout=30,
        )
        latency_ms = (datetime.now() - start).total_seconds() * 1000
        cost = completion_cost(completion_response=response)
        
        result = {
            "content": response.choices[0].message.content,
            "model": response.model,
            "tier": tier.value,
            "cost_usd": cost,
            "latency_ms": latency_ms,
            "fallback_used": response.model != primary,
        }
        
        if self.db and user_id:
            self._log_async(user_id, result, response.usage)
        
        return result
    
    def _log_async(self, user_id, result, usage):
        # Implement async DB logging here
        pass
 
# Usage
gateway = AIGateway()
result = gateway.complete(
    "Analyze the architectural tradeoffs of event-sourcing vs CQRS for this system...",
    tier=TaskTier.DEEP,
)
print(f"Cost: ${result['cost_usd']:.5f} | Latency: {result['latency_ms']:.0f}ms | Fallback: {result['fallback_used']}")

Caching Strategy: Reducing Costs with Prompt Caching

Claude API supports prompt caching for large system prompts — when a system prompt exceeds ~1,024 tokens, marking it as cacheable can reduce input costs by up to 90% on repeated calls. LiteLLM passes this through cleanly:

import litellm
 
# Large system prompt that's constant across requests
SYSTEM_PROMPT = """
You are an expert code reviewer with deep knowledge of Python, TypeScript, and Rust.
Your reviews focus on: correctness, security, performance, and maintainability.
 
[... assume this is 2000+ tokens of detailed instructions ...]
"""
 
def review_code(code: str, language: str) -> str:
    response = litellm.completion(
        model="claude-sonnet-4-6",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": SYSTEM_PROMPT,
                        "cache_control": {"type": "ephemeral"}  # Mark for caching
                    },
                    {
                        "type": "text",
                        "text": f"Review this {language} code:\n\n```{language}\n{code}\n```"
                    }
                ]
            }
        ],
        max_tokens=2048,
    )
    
    # cache_read_input_tokens tells you how much was served from cache
    usage = response.usage
    if hasattr(usage, 'cache_read_input_tokens') and usage.cache_read_input_tokens > 0:
        savings_pct = (usage.cache_read_input_tokens / usage.prompt_tokens) * 100
        print(f"Cache hit: {savings_pct:.0f}% of input tokens served from cache")
    
    return response.choices[0].message.content

In practice, a code review service that runs the same 2,000-token system prompt thousands of times per day will see the cache hit rate stabilize above 95% within the first hour of operation, reducing effective input costs dramatically. Pair this with LiteLLM's Redis caching layer (configured in litellm_config.yaml) for semantic caching — where similar (but not identical) prompts also return cached results.

Load Testing Your Gateway Before It Goes Live

Before a new gateway configuration reaches production, load test it against realistic traffic patterns. The most common surprises are fallback cascades (where Claude hits rate limits and GPT-4o also gets overwhelmed) and latency spikes under concurrent load.

import asyncio
import time
import litellm
from statistics import mean, quantiles
 
async def single_request(prompt: str, model: str) -> dict:
    start = time.perf_counter()
    try:
        response = await litellm.acompletion(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=256,
            request_timeout=15,
        )
        latency = (time.perf_counter() - start) * 1000
        return {"success": True, "latency_ms": latency, "model": response.model}
    except Exception as e:
        latency = (time.perf_counter() - start) * 1000
        return {"success": False, "latency_ms": latency, "error": str(e)}
 
async def load_test(
    concurrency: int = 20,
    total_requests: int = 200,
    model: str = "claude-sonnet-4-6",
) -> dict:
    """
    Run a load test and report latency distribution and error rate.
    Start with concurrency=5 and increase gradually.
    """
    test_prompt = "Summarize the tradeoffs between REST and GraphQL in 2 sentences."
    
    semaphore = asyncio.Semaphore(concurrency)
    
    async def bounded_request():
        async with semaphore:
            return await single_request(test_prompt, model)
    
    tasks = [bounded_request() for _ in range(total_requests)]
    results = await asyncio.gather(*tasks)
    
    successes = [r for r in results if r["success"]]
    failures = [r for r in results if not r["success"]]
    latencies = [r["latency_ms"] for r in successes]
    
    if not latencies:
        return {"error": "All requests failed"}
    
    sorted_latencies = sorted(latencies)
    p50, p95, p99 = quantiles(sorted_latencies, n=100)[49], quantiles(sorted_latencies, n=100)[94], quantiles(sorted_latencies, n=100)[98]
    
    # Count fallback usage
    primary_count = sum(1 for r in successes if model in r.get("model", ""))
    fallback_count = len(successes) - primary_count
    
    return {
        "total": total_requests,
        "success_rate": f"{len(successes)/total_requests*100:.1f}%",
        "error_rate": f"{len(failures)/total_requests*100:.1f}%",
        "latency_p50_ms": f"{p50:.0f}",
        "latency_p95_ms": f"{p95:.0f}",
        "latency_p99_ms": f"{p99:.0f}",
        "fallback_rate": f"{fallback_count/len(successes)*100:.1f}%" if successes else "N/A",
    }
 
# Run the load test
if __name__ == "__main__":
    results = asyncio.run(load_test(concurrency=10, total_requests=100))
    for k, v in results.items():
        print(f"{k}: {v}")

Run this against your staging environment before every major configuration change. The numbers to watch are p99 latency (should stay under your SLA) and fallback rate (anything above 5% warrants investigation into why the primary model is failing).

Security Considerations for a Shared Gateway

When multiple teams or services share a single LiteLLM proxy, access control becomes important. LiteLLM supports per-key rate limits and budget caps that prevent one team from starving others:

# Add to litellm_config.yaml
general_settings:
  master_key: os.environ/LITELLM_MASTER_KEY
 
# Issue team-specific keys with individual budgets
# via the LiteLLM admin API: POST /key/generate
# {
#   "key_alias": "team-data-engineering",
#   "max_budget": 50.0,       # $50/month limit
#   "budget_duration": "30d",
#   "max_parallel_requests": 10,
#   "models": ["claude-primary", "claude-haiku"],  # Restrict access
#   "tpm_limit": 100000,  # Tokens per minute
#   "rpm_limit": 100,     # Requests per minute
# }

From an API key security standpoint: the LiteLLM proxy means your application services never hold Anthropic API keys directly. They hold LiteLLM virtual keys, which you can revoke instantly without rotating the underlying provider keys. This separation is particularly valuable when onboarding external contractors or running in multi-tenant environments.

One additional consideration specific to Claude API: system prompts often contain sensitive business logic (pricing rules, internal knowledge, behavioral constraints). When logging requests through LiteLLM, make sure your log sanitization strips or truncates system prompt content before it reaches log aggregation tools.

import litellm
 
def sanitize_for_logging(kwargs: dict) -> dict:
    """
    Remove sensitive content before logging request details.
    Call this before sending to your log aggregator.
    """
    safe = dict(kwargs)
    messages = safe.get("messages", [])
    
    sanitized_messages = []
    for msg in messages:
        if msg.get("role") == "system":
            # Keep structure but redact content
            sanitized_messages.append({
                "role": "system",
                "content": f"[REDACTED — {len(msg.get('content', ''))} chars]"
            })
        else:
            sanitized_messages.append(msg)
    
    safe["messages"] = sanitized_messages
    return safe

Field Notes the Docs Won't Tell You

After running a multi-provider setup in production for about half a year, here are the things that only became clear from operating it — not from reading the documentation.

Fallback Makes Requests Slower, Not Faster — Design Around That

The most commonly overlooked detail is the latency spike at the moment a fallback fires.

In a batch job for a project I run as an indie developer, Claude Sonnet 4.6 responds in around 1,400ms under normal conditions. But once a 429 hits, two retries (num_retries=2) elapse, and the request falls back to GPT-4o, the same call took roughly 4,800ms — the retry backoff stacks up.

So fallback is insurance, not acceleration. On latency-sensitive paths like interactive chat, cap num_retries at 1 and shorten the timeout to about 15 seconds so you hand off to the next provider quickly. On batch paths, letting it persist with num_retries=3 actually lowered my total cost. Splitting the retry strategy per path is the realistic answer.

# Split retry strategy by request path
INTERACTIVE = dict(num_retries=1, request_timeout=15)   # prioritize perceived speed
BATCH       = dict(num_retries=3, request_timeout=60)   # prioritize completion
 
def complete_for_path(path: str, prompt: str):
    policy = INTERACTIVE if path == "interactive" else BATCH
    return litellm.completion(
        model="claude-sonnet-4-6",
        messages=[{"role": "user", "content": prompt}],
        fallbacks=["claude-haiku-4-5", "gpt-4o"],
        **policy,
    )

The "30% Cost Reduction" Is Measured, Not Marketing

I wrote "30% reduction" earlier — that figure is from my own workload. On a system where classification and short-form QA made up roughly 60% of all requests, routing those to Haiku brought the monthly API bill down by about 32%.

What amplifies this is pairing routing with a semantic cache design. Adding a "don't call at all" layer before the "send to the cheaper model" layer widens the savings further. LiteLLM's cache: true is exact-match only, so when you want to collapse semantically similar requests, a dedicated cache layer is worth the extra moving part.

Anthropic-Only Features Vanish on the Fallback Provider

Bake this into your design from the start. Any path built around Extended Thinking or Claude's Tool Use loses that capability the instant it falls back to GPT-4o.

On paths that rely on combining Extended Thinking with Tool Use, either make the fallback provider perform equivalent reasoning, or deliberately decide "this path never falls back outside Anthropic." I restrict reasoning-critical paths to an Anthropic-internal chain of claude-opus-4-6 → claude-sonnet-4-6 → claude-haiku-4-5 to keep quality from swinging.

Add It While You're Still Only "Considering" Migration

Introducing LiteLLM in the middle of a large migration like moving from OpenAI is painful. Routing through litellm.completion() while a single provider still suffices turns a future migration into "change a number in a config file." I once deferred this abstraction and paid for it with duplicated work during a migration, so I strongly recommend putting it in early.

Start Small, Expand Based on Evidence

LiteLLM with Claude API delivers three measurable improvements in production: resilience against single-provider failures, cost visibility that makes budgeting accurate instead of estimated, and model selection based on data rather than assumptions.

The lowest-friction starting point: replace one anthropic.Anthropic().messages.create() call with litellm.completion() in a non-critical service. Add a single fallback model. Log the cost output. Most teams that do this expand it to their entire stack within a month because the operational benefits — specifically, the combination of reduced on-call incidents and clearer cost attribution — are immediately visible.

Provider lock-in in AI infrastructure is a solvable problem. The infrastructure described here adds a single library dependency and a few hundred lines of configuration, and in return gives you the flexibility to respond to pricing changes, quality improvements, or availability issues without being hostage to any single provider's roadmap.

Thank You for Reading

Claude Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.

✦Copy-paste ready implementation code
✦New advanced guides published daily
✦$5/mo or $10 for lifetime access

View Membership →