7. Multi-Model Structured Output

Overview

In robust production systems, you rarely depend on a single LLM provider. Providers have different prices, strengths, rate limits and availability. On top of that, each one returns data in a slightly different shape.

In this capsule you'll learn: the Adapter pattern for multiple providers, normalizing outputs into a canonical schema, fallback with retry, rate limit handling, and provider selection strategies based on cost and performance.

Why Multi-Model in Production

The Adapter Pattern: A Unified Interface

The Adapter pattern creates a common interface for multiple providers, hiding the API differences behind an abstraction.

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any
import json

@dataclass
class LLMResponse:
    """A normalized response from any provider."""
    content: str
    tokens_input: int
    tokens_output: int
    model: str
    provider: str
    latency_ms: float
    metadata: dict = None
    
    @property
    def estimated_cost_usd(self) -> float:
        """Computes the estimated cost based on the provider and model."""
        prices = {
            "openai": {
                "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
                "gpt-4o": {"input": 0.0025, "output": 0.01}
            },
            "anthropic": {
                "claude-3-5-haiku-20241022": {"input": 0.0008, "output": 0.004},
                "claude-3-5-sonnet-20241022": {"input": 0.003, "output": 0.015}
            }
        }
        
        provider_prices = prices.get(self.provider, {})
        model_prices = provider_prices.get(self.model, {"input": 0.001, "output": 0.002})
        
        cost = (
            (self.tokens_input / 1000) * model_prices["input"] +
            (self.tokens_output / 1000) * model_prices["output"]
        )
        return round(cost, 6)

class LLMAdapter(ABC):
    """The base interface for every LLM adapter."""
    
    @abstractmethod
    def complete(self, system: str, user: str, **kwargs) -> LLMResponse:
        """Generates a completion."""
        pass
    
    @abstractmethod
    def complete_json(self, system: str, user: str, **kwargs) -> dict:
        """Generates a completion in JSON format."""
        pass
    
    @abstractmethod
    def health_check(self) -> bool:
        """Checks that the provider is available."""
        pass
    
    def extract_entities(self, text: str) -> dict:
        """Extracts entities from the text. A generic implementation."""
        system = """
Extract entities from the text. Answer ONLY in JSON with this structure:
{
    "people": ["name1", "name2"],
    "organizations": ["org1", "org2"],
    "locations": ["location1", "location2"],
    "dates": ["date1"],
    "other": ["entity1"]
}
If there are no entities of a type, use an empty list.
"""
        return self.complete_json(system=system, user=text)

The OpenAI adapter

import time
from openai import OpenAI, RateLimitError, APIConnectionError

class OpenAIAdapter(LLMAdapter):
    """An adapter for the OpenAI API."""
    
    def __init__(
        self, 
        model: str = "gpt-4o-mini",
        api_key: str | None = None,
        timeout: int = 30
    ):
        self.client = OpenAI(api_key=api_key) if api_key else OpenAI()
        self.model = model
        self.timeout = timeout
    
    def complete(self, system: str, user: str, **kwargs) -> LLMResponse:
        """
        Generates a completion with OpenAI.
        
        Args:
            system: System prompt
            user: User message
            **kwargs: Additional parameters (temperature, max_tokens, etc.)
        
        Returns:
            A normalized LLMResponse
        """
        start = time.time()
        
        response = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": system},
                {"role": "user", "content": user}
            ],
            **kwargs
        )
        
        latency_ms = (time.time() - start) * 1000
        
        return LLMResponse(
            content=response.choices[0].message.content,
            tokens_input=response.usage.prompt_tokens,
            tokens_output=response.usage.completion_tokens,
            model=self.model,
            provider="openai",
            latency_ms=latency_ms,
            metadata={
                "finish_reason": response.choices[0].finish_reason,
                "response_id": response.id
            }
        )
    
    def complete_json(self, system: str, user: str, **kwargs) -> dict:
        """Generates a completion and parses it as JSON."""
        kwargs["response_format"] = {"type": "json_object"}
        response = self.complete(system=system, user=user, **kwargs)
        return json.loads(response.content)
    
    def complete_with_schema(
        self, 
        system: str, 
        user: str, 
        schema: type,
        **kwargs
    ) -> dict:
        """
        Uses OpenAI's Structured Outputs to guarantee the schema.
        
        Args:
            schema: The Pydantic class for the output
        """
        # OpenAI Structured Outputs with Pydantic
        response = self.client.beta.chat.completions.parse(
            model=self.model,
            messages=[
                {"role": "system", "content": system},
                {"role": "user", "content": user}
            ],
            response_format=schema,
            **kwargs
        )
        return response.choices[0].message.parsed
    
    def health_check(self) -> bool:
        """Checks OpenAI's availability."""
        try:
            self.client.models.retrieve(self.model)
            return True
        except Exception:
            return False

The Anthropic adapter

import anthropic
import time

class AnthropicAdapter(LLMAdapter):
    """An adapter for the Anthropic API."""
    
    def __init__(
        self,
        model: str = "claude-3-5-haiku-20241022",
        api_key: str | None = None,
        max_tokens: int = 1024
    ):
        self.client = anthropic.Anthropic(api_key=api_key) if api_key else anthropic.Anthropic()
        self.model = model
        self.max_tokens_default = max_tokens
    
    def complete(self, system: str, user: str, **kwargs) -> LLMResponse:
        """
        Generates a completion with Anthropic.
        
        Note: Anthropic requires 'max_tokens' as a parameter.
        """
        start = time.time()
        
        max_tokens = kwargs.pop("max_tokens", self.max_tokens_default)
        
        message = self.client.messages.create(
            model=self.model,
            max_tokens=max_tokens,
            system=system,
            messages=[{"role": "user", "content": user}],
            **kwargs
        )
        
        latency_ms = (time.time() - start) * 1000
        
        return LLMResponse(
            content=message.content[0].text,
            tokens_input=message.usage.input_tokens,
            tokens_output=message.usage.output_tokens,
            model=self.model,
            provider="anthropic",
            latency_ms=latency_ms,
            metadata={
                "stop_reason": message.stop_reason,
                "message_id": message.id
            }
        )
    
    def complete_json(self, system: str, user: str, **kwargs) -> dict:
        """
        Generates a JSON completion with Anthropic.
        
        Anthropic doesn't have a native JSON response_format, 
        but it's very good at following JSON instructions in the system prompt.
        """
        system_json = f"{system}\n\nIMPORTANT: Answer ONLY with valid JSON. No additional text."
        
        response = self.complete(system=system_json, user=user, **kwargs)
        
        # Clean up possible prefixes
        content = response.content.strip()
        
        # Extract the JSON if it's in a code block
        import re
        match = re.search(r"```(?:json)?\s*([\s\S]+?)\s*```", content)
        if match:
            content = match.group(1)
        
        return json.loads(content)
    
    def health_check(self) -> bool:
        """Checks Anthropic's availability."""
        try:
            # Make a minimal call
            self.client.messages.create(
                model=self.model,
                max_tokens=5,
                messages=[{"role": "user", "content": "test"}]
            )
            return True
        except Exception:
            return False

Output Normalization

Different models use different names for the same entities. Normalization guarantees a canonical schema.

from pydantic import BaseModel, field_validator
from typing import Any

class ExtractedEntities(BaseModel):
    """The canonical schema for extracted entities."""
    people: list[str] = []
    organizations: list[str] = []
    locations: list[str] = []
    dates: list[str] = []
    other: list[str] = []
    
    @field_validator("*", mode="before")
    @classmethod
    def ensure_list(cls, v: Any) -> list:
        """Converts non-list values into a list."""
        if v is None:
            return []
        if isinstance(v, str):
            return [v] if v.strip() else []
        if not isinstance(v, list):
            return list(v)
        return [str(item).strip() for item in v if item]

# Field name mapping by provider
FIELD_ALIASES = {
    "people": ["people", "personas", "persons", "nombre", "nombres", "individuals"],
    "organizations": ["organizations", "organizaciones", "orgs", "companies", "empresas"],
    "locations": ["locations", "lugares", "places", "cities", "locaciones", "ubicaciones"],
    "dates": ["dates", "fechas", "timestamps", "times", "datas"],
    "other": ["other", "otros", "others", "misc", "entities", "entidades", "other_entities"]
}

def normalize_entities(data: dict) -> ExtractedEntities:
    """
    Normalizes the output of any provider into the canonical schema.
    
    Args:
        data: A dict with entities in any shape
    
    Returns:
        A normalized ExtractedEntities
    """
    normalized = {}
    
    for canonical_field, aliases in FIELD_ALIASES.items():
        for alias in aliases:
            if alias in data:
                value = data[alias]
                # If it's a string, convert it to a list
                if isinstance(value, str):
                    normalized[canonical_field] = [value] if value.strip() else []
                else:
                    normalized[canonical_field] = value
                break
        
        # If no alias was found, use an empty list
        if canonical_field not in normalized:
            normalized[canonical_field] = []
    
    return ExtractedEntities(**normalized)

# Normalization tests
normalization_cases = [
    # Typical OpenAI
    {"personas": ["Juan Perez"], "organizations": ["Google"], "places": ["Mexico City"]},
    # Typical Anthropic  
    {"people": ["Maria Garcia"], "orgs": ["Microsoft"], "locations": ["Monterrey"], "dates": ["2024-01-15"]},
    # An alternative shape
    {"individuals": ["Carlos Lopez"], "companies": ["Meta"], "cities": ["Guadalajara"]},
]

for case in normalization_cases:
    normalized = normalize_entities(case)
    print(f"Input: {list(case.keys())}")
    print(f"Normalized: people={normalized.people}, orgs={normalized.organizations}")
    print()

Fallback with Retry and Backoff

import time
import random
import logging
from typing import Callable, TypeVar

logger = logging.getLogger(__name__)
T = TypeVar("T")

class RetryConfig:
    """The configuration of a retry policy."""
    
    def __init__(
        self,
        max_attempts: int = 3,
        delay_base: float = 1.0,
        delay_max: float = 60.0,
        backoff_factor: float = 2.0,
        jitter: bool = True,
        retryable_exceptions: tuple = (Exception,)
    ):
        self.max_attempts = max_attempts
        self.delay_base = delay_base
        self.delay_max = delay_max
        self.backoff_factor = backoff_factor
        self.jitter = jitter
        self.retryable_exceptions = retryable_exceptions
    
    def compute_delay(self, attempt: int) -> float:
        """Computes the delay with exponential backoff and optional jitter."""
        delay = min(
            self.delay_base * (self.backoff_factor ** (attempt - 1)),
            self.delay_max
        )
        if self.jitter:
            delay *= (0.5 + random.random() * 0.5)  # 50% jitter
        return delay

def with_retry(func: Callable[..., T], config: RetryConfig | None = None, **kwargs) -> T:
    """
    Runs a function with retry and exponential backoff.
    
    Args:
        func: The function to run
        config: The retry configuration (uses the defaults if None)
        **kwargs: The arguments to pass to func
    
    Returns:
        The function's result
    
    Raises:
        The last error if every attempt fails
    """
    config = config or RetryConfig()
    last_error = None
    
    for attempt in range(1, config.max_attempts + 1):
        try:
            return func(**kwargs)
        except config.retryable_exceptions as e:
            last_error = e
            
            if attempt == config.max_attempts:
                logger.error(f"All {config.max_attempts} attempts failed")
                raise
            
            delay = config.compute_delay(attempt)
            logger.warning(
                f"Attempt {attempt}/{config.max_attempts} failed: {e}. "
                f"Retrying in {delay:.2f}s..."
            )
            time.sleep(delay)
    
    raise last_error

def extract_with_fallback(
    text: str,
    adapters: list[LLMAdapter],
    task: str = "entities",
    retry_config: RetryConfig | None = None
) -> dict:
    """
    Extracts information with automatic fallback across providers.
    
    Args:
        text: The text to process
        adapters: A list of adapters in order of preference
        task: The task to run
        retry_config: The retry configuration per provider
    
    Returns:
        The result of the first provider that succeeds
    """
    config = retry_config or RetryConfig(max_attempts=2, delay_base=0.5)
    errors = []
    
    for adapter in adapters:
        provider_name = f"{adapter.provider if hasattr(adapter, 'provider') else type(adapter).__name__}"
        
        try:
            logger.info(f"Trying with {provider_name}...")
            
            def call():
                if task == "entities":
                    return adapter.extract_entities(text)
                else:
                    raise ValueError(f"Unrecognized task: {task}")
            
            result = with_retry(call, config)
            
            logger.info(f"✅ {provider_name} succeeded")
            return result
            
        except Exception as e:
            error_info = {
                "provider": provider_name,
                "error": str(e),
                "type": type(e).__name__
            }
            errors.append(error_info)
            logger.warning(f"❌ {provider_name} failed: {e}")
            continue
    
    raise RuntimeError(
        f"Every provider failed. Errors: {json.dumps(errors, indent=2)}"
    )

A Smart Provider Router

Instead of always using fallback, a smart router picks the best provider for the task.

from enum import Enum

class LLMTask(Enum):
    FAST_CLASSIFICATION = "fast_classification"
    COMPLEX_ANALYSIS = "complex_analysis"
    DATA_EXTRACTION = "data_extraction"
    TEXT_GENERATION = "text_generation"
    CODE_ANALYSIS = "code_analysis"

@dataclass
class ProviderConfig:
    """The configuration of a provider for routing."""
    name: str
    adapter: LLMAdapter
    relative_cost: float  # 1.0 = baseline
    relative_speed: float  # 1.0 = baseline
    optimal_tasks: list[LLMTask]
    available: bool = True

class LLMRouter:
    """
    A smart router that picks the best provider for the context.
    """
    
    def __init__(self, providers: list[ProviderConfig]):
        self.providers = providers
        self._stats: dict[str, dict] = {p.name: {"successes": 0, "errors": 0} for p in providers}
    
    def select_provider(
        self,
        task: LLMTask,
        optimize_for: str = "balance"  # "cost", "speed", "quality", "balance"
    ) -> ProviderConfig:
        """
        Picks the best available provider for the task.
        
        Args:
            task: The type of task to run
            optimize_for: The optimization criterion
        
        Returns:
            The ProviderConfig of the selected provider
        """
        # Filter the available providers
        available = [p for p in self.providers if p.available]
        
        if not available:
            raise RuntimeError("There are no available providers")
        
        # Prioritize providers for which this is an optimal task
        optimal = [p for p in available if task in p.optimal_tasks]
        candidates = optimal if optimal else available
        
        # Sort by the criterion
        if optimize_for == "cost":
            return min(candidates, key=lambda p: p.relative_cost)
        elif optimize_for == "speed":
            return max(candidates, key=lambda p: p.relative_speed)
        elif optimize_for == "balance":
            # A composite score: low cost + high speed
            def balance_score(p: ProviderConfig) -> float:
                return p.relative_speed / p.relative_cost
            return max(candidates, key=balance_score)
        else:
            return candidates[0]
    
    def run(
        self,
        system: str,
        user: str,
        task: LLMTask = LLMTask.FAST_CLASSIFICATION,
        optimize_for: str = "balance",
        fallback: bool = True
    ) -> LLMResponse:
        """
        Runs an LLM call with routing and automatic fallback.
        """
        primary_provider = self.select_provider(task, optimize_for)
        
        try:
            response = primary_provider.adapter.complete(system=system, user=user)
            self._stats[primary_provider.name]["successes"] += 1
            return response
            
        except Exception as e:
            self._stats[primary_provider.name]["errors"] += 1
            logger.warning(f"The primary provider {primary_provider.name} failed: {e}")
            
            if not fallback:
                raise
            
            # Try the other providers
            other_available = [
                p for p in self.providers 
                if p.name != primary_provider.name and p.available
            ]
            
            for fallback_provider in other_available:
                try:
                    response = fallback_provider.adapter.complete(system=system, user=user)
                    self._stats[fallback_provider.name]["successes"] += 1
                    logger.info(f"Fallback succeeded with {fallback_provider.name}")
                    return response
                except Exception as e2:
                    self._stats[fallback_provider.name]["errors"] += 1
                    continue
            
            raise RuntimeError("Every provider failed")
    
    def stats(self) -> dict:
        """Returns the usage statistics per provider."""
        result = {}
        for name, stats in self._stats.items():
            total = stats["successes"] + stats["errors"]
            result[name] = {
                **stats,
                "success_rate": stats["successes"] / total if total > 0 else 0
            }
        return result

# Router configuration
def create_router() -> LLMRouter:
    openai_adapter = OpenAIAdapter(model="gpt-4o-mini")
    anthropic_adapter = AnthropicAdapter(model="claude-3-5-haiku-20241022")
    
    providers = [
        ProviderConfig(
            name="openai_mini",
            adapter=openai_adapter,
            relative_cost=1.0,
            relative_speed=1.0,
            optimal_tasks=[
                LLMTask.FAST_CLASSIFICATION,
                LLMTask.DATA_EXTRACTION,
                LLMTask.CODE_ANALYSIS
            ]
        ),
        ProviderConfig(
            name="anthropic_haiku",
            adapter=anthropic_adapter,
            relative_cost=1.5,  # Slightly more expensive
            relative_speed=0.9,  # Slightly slower
            optimal_tasks=[
                LLMTask.COMPLEX_ANALYSIS,
                LLMTask.TEXT_GENERATION
            ]
        )
    ]
    
    return LLMRouter(providers)

Structured Extraction with a Canonical Schema

from pydantic import BaseModel, Field
from openai import OpenAI
import anthropic
import json

# The canonical schema for extraction
class ExtractedPerson(BaseModel):
    name: str
    role: str | None = None
    organization: str | None = None

class BusinessNews(BaseModel):
    title: str
    date: str | None = None
    people: list[ExtractedPerson] = Field(default_factory=list)
    organizations: list[str] = Field(default_factory=list)
    amount_usd: float | None = None
    event_type: str | None = None
    summary: str

def extract_news_openai(text: str) -> BusinessNews:
    """Extracts news data using OpenAI with JSON mode."""
    client = OpenAI()
    
    system = """
Extract structured information from the news story. Answer in JSON with this exact schema:
{
    "title": "string",
    "date": "YYYY-MM-DD or null",
    "people": [{"name": "string", "role": "string or null", "organization": "string or null"}],
    "organizations": ["org1", "org2"],
    "amount_usd": number or null,
    "event_type": "acquisition|funding|partnership|launch|other|null",
    "summary": "2-3 sentences"
}
"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": text}
        ],
        response_format={"type": "json_object"}
    )
    
    data = json.loads(response.choices[0].message.content)
    return BusinessNews(**data)

def extract_news_anthropic(text: str) -> BusinessNews:
    """Extracts news data using Anthropic."""
    client = anthropic.Anthropic()
    
    system = """
Extract structured information from business news stories.
Always answer ONLY with valid JSON, no explanations.
Required schema: title, date (YYYY-MM-DD or null), people (a list of objects with name/role/organization), 
organizations (a list of strings), amount_usd (a number or null), event_type, summary.
"""
    
    message = client.messages.create(
        model="claude-3-5-haiku-20241022",
        max_tokens=1024,
        system=system,
        messages=[{"role": "user", "content": text}]
    )
    
    raw = message.content[0].text.strip()
    
    # Strip the markdown code fence if it's there
    import re
    match = re.search(r"```(?:json)?\s*([\s\S]+?)\s*```", raw)
    if match:
        raw = match.group(1)
    
    data = json.loads(raw)
    return BusinessNews(**data)

def extract_news_multi(text: str) -> tuple[BusinessNews, str]:
    """
    Extracts data with multi-provider fallback.
    
    Returns:
        A tuple (result, provider_used)
    """
    providers = [
        ("openai", extract_news_openai),
        ("anthropic", extract_news_anthropic)
    ]
    
    errors = []
    for name, extractor in providers:
        try:
            result = extractor(text)
            return result, name
        except Exception as e:
            errors.append(f"{name}: {str(e)[:100]}")
            logger.warning(f"The {name} extractor failed: {e}")
    
    raise RuntimeError(f"Every extractor failed: {'; '.join(errors)}")

# Test
news = """
Monterrey, March 15, 2024. The Mexican startup Fintech XYZ announced today 
a $25 million Series A funding round, led by 
SoftBank Latin America Fund. The company's CEO, Carlos Martinez, said
the funds will be used to expand operations into Colombia and Chile.
Acme Capital also participated in the round.
"""

try:
    result, provider = extract_news_multi(news)
    print(f"✅ Extracted with: {provider}")
    print(f"Title: {result.title}")
    print(f"Amount: ${result.amount_usd:,.0f} USD")
    print(f"People: {[p.name for p in result.people]}")
    print(f"Organizations: {result.organizations}")
    print(f"Type: {result.event_type}")
except RuntimeError as e:
    print(f"❌ Error: {e}")

Metrics and Observability

from datetime import datetime
from collections import defaultdict
import statistics

class LLMMetricsCollector:
    """A collector for LLM usage metrics."""
    
    def __init__(self):
        self._calls: list[dict] = []
        self._by_provider: dict[str, list] = defaultdict(list)
    
    def record(self, response: LLMResponse, success: bool = True, error: str | None = None):
        """Records a call to the LLM."""
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "provider": response.provider,
            "model": response.model,
            "tokens_input": response.tokens_input,
            "tokens_output": response.tokens_output,
            "latency_ms": response.latency_ms,
            "cost_usd": response.estimated_cost_usd,
            "success": success,
            "error": error
        }
        
        self._calls.append(entry)
        self._by_provider[response.provider].append(entry)
    
    def summary(self) -> dict:
        """Generates a summary of the metrics."""
        if not self._calls:
            return {"total_calls": 0}
        
        successes = [c for c in self._calls if c["success"]]
        
        return {
            "total_calls": len(self._calls),
            "success_rate": len(successes) / len(self._calls),
            "total_cost_usd": sum(c["cost_usd"] for c in self._calls),
            "total_tokens": sum(c["tokens_input"] + c["tokens_output"] for c in self._calls),
            "average_latency_ms": statistics.mean(c["latency_ms"] for c in self._calls),
            "p95_latency_ms": statistics.quantiles(
                [c["latency_ms"] for c in self._calls], n=20
            )[18] if len(self._calls) >= 2 else 0,
            "by_provider": {
                provider: {
                    "calls": len(calls),
                    "cost_usd": sum(c["cost_usd"] for c in calls),
                    "average_latency_ms": statistics.mean(c["latency_ms"] for c in calls)
                }
                for provider, calls in self._by_provider.items()
            }
        }

# Using the collector
metrics = LLMMetricsCollector()

# Integrate it with the adapter
class MonitoredOpenAIAdapter(OpenAIAdapter):
    def __init__(self, *args, metrics_collector: LLMMetricsCollector | None = None, **kwargs):
        super().__init__(*args, **kwargs)
        self.metrics = metrics_collector
    
    def complete(self, system: str, user: str, **kwargs) -> LLMResponse:
        try:
            response = super().complete(system=system, user=user, **kwargs)
            if self.metrics:
                self.metrics.record(response, success=True)
            return response
        except Exception as e:
            if self.metrics:
                # Create a dummy response for the error metrics
                dummy = LLMResponse("", 0, 0, self.model, "openai", 0)
                self.metrics.record(dummy, success=False, error=str(e))
            raise

Troubleshooting

1. Different schemas across providers

Symptom: OpenAI returns "organizations" but Anthropic returns "organizaciones".

# A robust fix: flexible normalization with several aliases
def normalize_flexible(data: dict, schema_class: type) -> BaseModel:
    """
    Normalizes a dict into a Pydantic schema, tolerating variations.
    It tries several field names before falling back to the default.
    """
    fields = schema_class.model_fields
    normalized = {}
    
    for field, field_info in fields.items():
        # Try the exact field first
        if field in data:
            normalized[field] = data[field]
            continue
        
        # Try the field's aliases
        field_aliases = FIELD_ALIASES.get(field, [field])
        found = False
        
        for alias in field_aliases:
            if alias in data:
                normalized[field] = data[alias]
                found = True
                break
        
        # Use the default if nothing was found
        if not found and field_info.default is not None:
            normalized[field] = field_info.default
    
    return schema_class.model_validate(normalized)

2. Rate limits and 429 errors

from openai import RateLimitError
import time

class RateLimitHandler:
    """Handles rate limits with exponential backoff."""
    
    def __init__(self, max_attempts: int = 5):
        self.max_attempts = max_attempts
    
    def run_with_rate_limit(self, func, *args, **kwargs):
        """Runs a function, handling rate limits automatically."""
        for attempt in range(self.max_attempts):
            try:
                return func(*args, **kwargs)
            except RateLimitError as e:
                if attempt == self.max_attempts - 1:
                    raise
                
                # Pull the wait time from the error if it's available
                retry_after = getattr(e, "retry_after", None)
                if retry_after:
                    wait_time = float(retry_after)
                else:
                    wait_time = (2 ** attempt) + (0.1 * random.random())
                
                logger.warning(f"Rate limit hit. Waiting {wait_time:.1f}s...")
                time.sleep(wait_time)
        
        raise RuntimeError("Rate limit not resolved")

# For Anthropic, the headers include rate limit info
def extract_rate_limit_info(response_headers: dict) -> dict:
    """Extracts rate limit information from the headers."""
    return {
        "requests_limit": response_headers.get("anthropic-ratelimit-requests-limit"),
        "requests_remaining": response_headers.get("anthropic-ratelimit-requests-remaining"),
        "tokens_limit": response_headers.get("anthropic-ratelimit-tokens-limit"),
        "tokens_remaining": response_headers.get("anthropic-ratelimit-tokens-remaining"),
    }

3. Unexpectedly high costs

# Enforce a maximum budget per session or user
class BudgetGuard:
    """Controls the maximum budget for LLM calls."""
    
    def __init__(self, budget_usd: float):
        self.budget_usd = budget_usd
        self._spent = 0.0
    
    def check(self, estimated_cost_usd: float):
        """Checks that there's budget available."""
        if self._spent + estimated_cost_usd > self.budget_usd:
            raise ValueError(
                f"Budget exceeded. Spent: ${self._spent:.4f}, "
                f"Limit: ${self.budget_usd:.4f}"
            )
    
    def record_spend(self, cost_usd: float):
        """Records an expense."""
        self._spent += cost_usd
    
    @property
    def available(self) -> float:
        return max(0, self.budget_usd - self._spent)

# Estimating the cost before calling
def estimate_call_cost(
    system: str,
    user: str,
    model: str = "gpt-4o-mini",
    max_output_tokens: int = 500
) -> float:
    """Estimates the cost of a call before running it."""
    try:
        import tiktoken
        enc = tiktoken.encoding_for_model(model)
        tokens_input = len(enc.encode(system + user))
    except Exception:
        tokens_input = len((system + user).split()) * 1.3
    
    prices_per_1k = {
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "claude-3-5-haiku-20241022": {"input": 0.0008, "output": 0.004},
    }
    
    prices = prices_per_1k.get(model, {"input": 0.001, "output": 0.002})
    
    cost = (
        (tokens_input / 1000) * prices["input"] +
        (max_output_tokens / 1000) * prices["output"]
    )
    
    return round(cost, 6)

Exercises

Exercise 1: Implement an Adapter for Google Gemini

Create a GeminiAdapter that implements the same LLMAdapter interface using the Google Generative AI API.

# pip install google-generativeai

class GeminiAdapter(LLMAdapter):
    """An adapter for the Google Gemini API."""
    
    def __init__(self, model: str = "gemini-1.5-flash", api_key: str | None = None):
        import google.generativeai as genai
        
        if api_key:
            genai.configure(api_key=api_key)
        # If there's no api_key, it uses GOOGLE_API_KEY from the environment
        
        self.genai = genai
        self.model_name = model
        self.model = genai.GenerativeModel(model)
    
    def complete(self, system: str, user: str, **kwargs) -> LLMResponse:
        import time
        start = time.time()
        
        # Gemini combines system and user into one message
        full_prompt = f"{system}\n\n{user}"
        
        response = self.model.generate_content(
            full_prompt,
            generation_config=self.genai.types.GenerationConfig(
                max_output_tokens=kwargs.get("max_tokens", 1024),
                temperature=kwargs.get("temperature", 0.7)
            )
        )
        
        latency_ms = (time.time() - start) * 1000
        
        return LLMResponse(
            content=response.text,
            tokens_input=response.usage_metadata.prompt_token_count,
            tokens_output=response.usage_metadata.candidates_token_count,
            model=self.model_name,
            provider="google",
            latency_ms=latency_ms
        )
    
    def complete_json(self, system: str, user: str, **kwargs) -> dict:
        system_json = f"{system}\n\nAnswer ONLY with valid JSON, no additional text."
        response = self.complete(system=system_json, user=user, **kwargs)
        
        content = response.content.strip()
        import re
        match = re.search(r"```(?:json)?\s*([\s\S]+?)\s*```", content)
        if match:
            content = match.group(1)
        
        return json.loads(content)
    
    def health_check(self) -> bool:
        try:
            models = [m for m in self.genai.list_models()]
            return any(self.model_name in m.name for m in models)
        except Exception:
            return False

# Usage
# gemini_adapter = GeminiAdapter(model="gemini-1.5-flash")
# result = gemini_adapter.extract_entities("Google announced...")

Exercise 2: Fallback with preference and metrics logging

Implement extract_with_fallback_v2 that records which provider was used, the tokens consumed and the cost.

from dataclasses import dataclass

@dataclass
class ResultWithMeta:
    data: dict
    provider_used: str
    total_tokens: int
    cost_usd: float
    attempts: int
    previous_errors: list[str]

def extract_with_fallback_v2(
    text: str,
    adapters: list[LLMAdapter],
    metrics: LLMMetricsCollector | None = None
) -> ResultWithMeta:
    """
    Extracts data with fallback and metrics tracking.
    """
    previous_errors = []
    
    for i, adapter in enumerate(adapters):
        provider = type(adapter).__name__.replace("Adapter", "").lower()
        
        try:
            # The actual call to the adapter
            system = """Extract entities. Answer in JSON: 
{"people": [], "organizations": [], "locations": [], "dates": []}"""
            
            start = time.time()
            data = adapter.complete_json(system=system, user=text)
            latency = (time.time() - start) * 1000
            
            # Create an LLMResponse for the metrics
            response_mock = LLMResponse(
                content=json.dumps(data),
                tokens_input=len(text.split()) * 2,  # Estimated
                tokens_output=len(json.dumps(data).split()),
                model=adapter.model if hasattr(adapter, 'model') else "unknown",
                provider=provider,
                latency_ms=latency
            )
            
            if metrics:
                metrics.record(response_mock, success=True)
            
            return ResultWithMeta(
                data=data,
                provider_used=provider,
                total_tokens=response_mock.tokens_input + response_mock.tokens_output,
                cost_usd=response_mock.estimated_cost_usd,
                attempts=i + 1,
                previous_errors=previous_errors
            )
            
        except Exception as e:
            error_msg = f"{provider}: {str(e)[:100]}"
            previous_errors.append(error_msg)
            logger.warning(f"The {provider} adapter failed: {e}")
    
    raise RuntimeError(f"Every adapter failed: {previous_errors}")

# Test
metrics_collector = LLMMetricsCollector()

openai_adapter = OpenAIAdapter()
anthropic_adapter = AnthropicAdapter()

adapters = [openai_adapter, anthropic_adapter]

text = "Tim Cook, Apple's CEO, unveiled the new iPhone in Cupertino on September 15."
result = extract_with_fallback_v2(text, adapters, metrics_collector)

print(f"Provider: {result.provider_used}")
print(f"Attempts: {result.attempts}")
print(f"Cost: ${result.cost_usd:.6f}")
print(f"Data: {result.data}")

print("\nGlobal metrics:")
print(json.dumps(metrics_collector.summary(), indent=2))

Exercise 3: A router that adapts to rate limits

Implement a router that, when it detects a rate limit on a provider, marks it temporarily unavailable for N seconds.

import time
from openai import RateLimitError as OpenAIRateLimitError

class AdaptiveRouter:
    """A router that learns from rate limits and avoids them temporarily."""
    
    def __init__(self, providers: list[ProviderConfig]):
        self.providers = {p.name: p for p in providers}
        self._cooldown_until: dict[str, float] = {}
        self._default_cooldown_seconds = 60
    
    def _is_available(self, name: str) -> bool:
        """Checks whether the provider is out of cooldown."""
        cooldown_until = self._cooldown_until.get(name, 0)
        return time.time() > cooldown_until
    
    def _apply_cooldown(self, name: str, seconds: int | None = None):
        """Puts the provider in cooldown."""
        seconds = seconds or self._default_cooldown_seconds
        self._cooldown_until[name] = time.time() + seconds
        logger.warning(f"Provider {name} in cooldown for {seconds}s")
    
    def run(self, system: str, user: str) -> LLMResponse:
        """Runs with adaptive routing."""
        available = [
            p for name, p in self.providers.items()
            if self._is_available(name) and p.available
        ]
        
        if not available:
            raise RuntimeError("There are no providers available outside of cooldown")
        
        for provider in available:
            try:
                return provider.adapter.complete(system=system, user=user)
            except OpenAIRateLimitError as e:
                # Pull retry-after if it's available
                retry_after = getattr(e, 'retry_after', 60)
                self._apply_cooldown(provider.name, retry_after)
            except Exception as e:
                logger.error(f"{provider.name} error: {e}")
                # A short cooldown for non-rate-limit errors
                self._apply_cooldown(provider.name, 10)
        
        raise RuntimeError("Every provider failed or is in cooldown")
    
    def status(self) -> dict:
        """The current availability status."""
        now = time.time()
        return {
            name: {
                "available": now > self._cooldown_until.get(name, 0),
                "cooldown_remaining_s": max(0, self._cooldown_until.get(name, 0) - now)
            }
            for name in self.providers
        }

Exercise 4: Compare two providers' outputs with an LLM judge

Implement a function that runs the same query with two providers and uses a third LLM to evaluate which answer is better.

from openai import OpenAI

client = OpenAI()

def evaluate_with_llm(
    query: str,
    answer_a: str,
    answer_b: str,
    criterion: str = "accuracy and clarity"
) -> dict:
    """
    Uses an LLM as a judge to evaluate two answers.
    
    Returns:
        A dict with the winner ("A", "B", "TIE") and a rationale
    """
    judge_system = """
You are an expert, impartial evaluator. 
Your task is to compare two answers and determine which one is better.
Be objective and base your evaluation on the criteria specified.
Answer ONLY in JSON: {"winner": "A|B|TIE", "rationale": "string", "score_a": 0-10, "score_b": 0-10}
"""
    
    judge_user = f"""
Query: {query}

Evaluation criterion: {criterion}

Answer A:
{answer_a}

Answer B:
{answer_b}

Objectively evaluate which answer is better according to the given criterion.
"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": judge_system},
            {"role": "user", "content": judge_user}
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    
    return json.loads(response.choices[0].message.content)

# A real comparison
query = "What are the best practices for handling errors in FastAPI?"
system = "You are an expert in FastAPI and Python. Answer concisely and practically."

openai_adapter = OpenAIAdapter()
anthropic_adapter = AnthropicAdapter()

resp_openai = openai_adapter.complete(system=system, user=query)
resp_anthropic = anthropic_adapter.complete(system=system, user=query)

evaluation = evaluate_with_llm(
    query=query,
    answer_a=resp_openai.content,
    answer_b=resp_anthropic.content,
    criterion="technical accuracy, code examples, and clarity"
)

print(f"Winner: {evaluation['winner']}")
print(f"OpenAI score: {evaluation['score_a']}/10")
print(f"Anthropic score: {evaluation['score_b']}/10")
print(f"Rationale: {evaluation['rationale']}")

Summary

Further resources

1. OpenAI API Reference
2. Anthropic API Reference
3. Google Generative AI Python SDK
4. Design Patterns: Adapter
5. LiteLLM - A unified proxy for multiple providers
6. OpenAI Structured Outputs