Automatic fallback LLM outages are the difference between a degraded experience and a hard outage. When your primary model provider hits rate limits, latency spikes, or goes dark entirely, you need traffic to shift without human intervention. This guide walks through building a fallback system that actually works in production — from health checks and routing logic to the subtle failures that catch teams off guard.
Start with a routing abstraction
Hardcoding model names across your codebase makes fallback impossible. Every call site should reference a logical task — “summarization”, “code-generation”, “chat” — not a specific model identifier. The routing layer resolves tasks to concrete models at request time.
# routing.py
from dataclasses import dataclass
from enum import Enum
from typing import Optional
class Task(str, Enum):
SUMMARIZATION = "summarization"
CODE_GEN = "code-generation"
CHAT = "chat"
@dataclass
class ModelCandidate:
provider: str
model_id: str
priority: int # lower = preferred
max_latency_ms: int
cost_per_1k_tokens: float
ROUTE_TABLE: dict[Task, list[ModelCandidate]] = {
Task.SUMMARIZATION: [
ModelCandidate("openai", "gpt-4o-mini", 1, 3000, 0.15),
ModelCandidate("anthropic", "claude-3-haiku", 2, 4000, 0.25),
ModelCandidate("together", "meta-llama-3.1-8b", 3, 5000, 0.08),
],
Task.CODE_GEN: [
ModelCandidate("openai", "gpt-4o", 1, 5000, 5.00),
ModelCandidate("anthropic", "claude-3.5-sonnet", 2, 6000, 3.00),
ModelCandidate("deepseek", "deepseek-coder", 3, 8000, 0.14),
],
}The route table lives in config, not code. When you need to add a new provider or reorder priorities, you deploy config — not a full service rollout.
Health checks that reflect reality
A model is “healthy” only if it meets your SLAs right now. Provider status pages lag reality by minutes. Build your own probes that exercise the actual code path: authenticate, send a minimal request, validate the response shape, measure latency.
# health.py
import asyncio
import time
from dataclasses import dataclass
from typing import Protocol
class LLMClient(Protocol):
async def complete(self, prompt: str, model: str) -> str: ...
@dataclass
class HealthResult:
model: str
healthy: bool
latency_ms: int
error: str | None = None
async def probe_model(client: LLMClient, model: str, timeout_ms: int) -> HealthResult:
start = time.perf_counter()
try:
# Minimal valid request for the provider
await asyncio.wait_for(
client.complete("ping", model=model),
timeout=timeout_ms / 1000
)
latency = int((time.perf_counter() - start) * 1000)
return HealthResult(model, True, latency)
except asyncio.TimeoutError:
return HealthResult(model, False, timeout_ms, "timeout")
except Exception as e:
latency = int((time.perf_counter() - start) * 1000)
return HealthResult(model, False, latency, str(e))Run probes on a background loop — every 10-30 seconds for critical paths, every 2-5 minutes for background workloads. Store results in a shared cache (Redis, in-memory with TTL) so request handlers don’t block on health checks.
Pitfall: Probing too aggressively gets you rate-limited by the provider, creating a self-inflicted outage. Respect provider limits; use exponential backoff on probe failures.
The fallback decision logic
When a request arrives, the router selects the highest-priority healthy candidate. If the primary fails mid-request, retry on the next candidate — but only for idempotent operations.
# router.py
import random
from dataclasses import dataclass
@dataclass
class RouteDecision:
provider: str
model: str
attempt: int = 0
class Router:
def __init__(self, route_table: dict, health_cache: dict):
self.route_table = route_table
self.health = health_cache # model_id -> HealthResult
def select(self, task: Task) -> RouteDecision | None:
candidates = self.route_table.get(task, [])
for c in candidates:
key = f"{c.provider}:{c.model_id}"
hr = self.health.get(key)
if hr and hr.healthy and hr.latency_ms <= c.max_latency_ms:
return RouteDecision(c.provider, c.model_id)
return None
def next_fallback(self, task: Task, failed: RouteDecision) -> RouteDecision | None:
candidates = self.route_table.get(task, [])
start_idx = next(
(i for i, c in enumerate(candidates)
if c.provider == failed.provider and c.model_id == failed.model),
-1
)
for c in candidates[start_idx + 1:]:
key = f"{c.provider}:{c.model_id}"
hr = self.health.get(key)
if hr and hr.healthy:
return RouteDecision(c.provider, c.model_id, attempt=failed.attempt + 1)
return NoneTradeoff: Strict priority ordering is simple but brittle. A model that’s technically “healthy” but running at 95th-percentile latency will degrade your p99. Consider adding a latency budget check — if the primary’s p99 exceeds your threshold, treat it as unhealthy for new requests.
Retry with idempotency keys
Not every failure warrants fallback. Network blips, 5xx errors, and timeouts on idempotent calls can retry on the same model. Reserve cross-provider fallback for: 429 rate limits, 503 unavailable, validated model degradation, or explicit routing directives from the client.
# client.py
from dataclasses import dataclass
from typing import Optional
@dataclass
class CompletionRequest:
task: Task
prompt: str
idempotency_key: Optional[str] = None
max_fallbacks: int = 2
routing_hint: Optional[str] = None # e.g., "prefer:anthropic"
async def complete_with_fallback(
router: Router,
clients: dict[str, LLMClient],
req: CompletionRequest
) -> str:
decision = router.select(req.task)
if not decision:
raise RuntimeError(f"No healthy models for {req.task}")
last_error = None
for attempt in range(req.max_fallbacks + 1):
client = clients[decision.provider]
try:
return await client.complete(req.prompt, model=decision.model)
except RateLimitError:
last_error = "rate_limited"
except ProviderUnavailableError:
last_error = "unavailable"
except ValidationError as e:
# Model returned garbage — don't retry same model
last_error = f"validation_failed: {e}"
except Exception as e:
# Unknown error — retry same model once if idempotent
if attempt == 0 and req.idempotency_key:
continue
last_error = f"error: {e}"
# Exhausted retries on this model, try fallback
decision = router.next_fallback(req.task, decision)
if not decision:
break
raise RuntimeError(f"All fallbacks exhausted for {req.task}: {last_error}")Pitfall: Falling back on validation errors (malformed JSON, truncated output) without changing the prompt often repeats the failure. If you fallback, consider simplifying the prompt or reducing max_tokens for the secondary model.
Preserve context across fallbacks
When you switch providers mid-conversation, the new model needs the same context. If you’re using a gateway that normalizes message formats, this is automatic. If you’re calling providers directly, you must translate.
# context.py
from typing import TypedDict
class Message(TypedDict):
role: str
content: str
def translate_messages(messages: list[Message], target: str) -> list[dict]:
"""Convert normalized messages to provider-specific format."""
if target.startswith("anthropic:"):
# Anthropic expects system prompt separate, no "system" role in messages
system = next((m["content"] for m in messages if m["role"] == "system"), "")
user_msgs = [m for m in messages if m["role"] != "system"]
return {"system": system, "messages": user_msgs}
elif target.startswith("openai:") or target.startswith("together:"):
return {"messages": messages}
elif target.startswith("google:"):
# Gemini uses "user"/"model" roles
converted = []
for m in messages:
role = "user" if m["role"] == "user" else "model"
converted.append({"role": role, "parts": [m["content"]]})
return {"contents": converted}
raise ValueError(f"Unknown provider: {target}")Tradeoff: Full context preservation increases token spend on fallback calls. For long conversations, consider summarizing history before fallback — but test that the summary doesn’t lose critical details.
Meter usage per model, not per request
If you bill customers or enforce quotas, you need per-model token counts. Provider response formats differ; normalize them at the gateway layer.
# metering.py
from dataclasses import dataclass
@dataclass
class Usage:
prompt_tokens: int
completion_tokens: int
total_tokens: int
model: str
provider: str
def extract_usage(response: dict, provider: str, model: str) -> Usage:
if provider == "openai":
u = response["usage"]
return Usage(u["prompt_tokens"], u["completion_tokens"], u["total_tokens"], model, provider)
elif provider == "anthropic":
u = response["usage"]
return Usage(u["input_tokens"], u["output_tokens"], u["input_tokens"] + u["output_tokens"], model, provider)
elif provider == "together":
u = response["usage"]
return Usage(u["prompt_tokens"], u["completion_tokens"], u["total_tokens"], model, provider)
# Fallback: estimate from character count
text = response.get("text") or response.get("content") or ""
est = len(text) // 4
return Usage(est, est, est * 2, model, provider)Log every completion with its Usage record. This lets you answer: “How much did the fallback to Claude cost us during the OpenAI outage last Tuesday?”
Test fallback paths before you need them
Chaos engineering for LLM routing means deliberately degrading providers in staging.
# chaos/test_fallback.sh
#!/bin/bash
set -euo pipefail
# Simulate OpenAI latency spike
tc qdisc add dev eth0 root netem delay 8000ms 1000ms distribution normal
# Run integration test suite
pytest tests/integration/test_fallback.py -v
# Clean up
tc qdisc del dev eth0 root# tests/integration/test_fallback.py
import pytest
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
async def test_fallback_on_rate_limit(router, clients):
# Primary returns 429
clients["openai"].complete = AsyncMock(side_effect=RateLimitError("429"))
# Secondary succeeds
clients["anthropic"].complete = AsyncMock(return_value="fallback response")
result = await complete_with_fallback(router, clients, CompletionRequest(
task=Task.CHAT,
prompt="Hello",
idempotency_key="test-123"
))
assert result == "fallback response"
clients["anthropic"].complete.assert_called_once()Run these in CI on every deploy. If your fallback path hasn’t been exercised in 30 days, it’s broken.
Common pitfalls that cause cascading failures
1. Thundering herd on recovery. When the primary recovers, all traffic snaps back instantly, overwhelming it. Add a cooldown period — keep traffic on the fallback for 2-5 minutes after the primary reports healthy.
# Add to HealthResult
@dataclass
class HealthResult:
...
recovered_at: float | None = None # timestamp when health flipped to True
def is_eligible(hr: HealthResult, cooldown_sec: int = 120) -> bool:
if not hr.healthy:
return False
if hr.recovered_at and (time.time() - hr.recovered_at) < cooldown_sec:
return False
return True2. Silent data corruption. A model returns valid JSON but wrong schema. Your validator passes, downstream code crashes. Version your response schemas and validate strictly on every hop.
3. Cache poisoning. If you cache responses by prompt hash, a fallback model’s different output format pollutes the cache for the primary. Include model/provider in the cache key.
4. Ignoring client routing directives. Some callers need specific models for compliance, latency, or quality. Honor routing_hint headers — but validate them against your route table so clients can’t request models you don’t support.
Observability you’ll actually use
Dashboards that show “requests per model” are noise. Build views that answer operational questions:
- Fallback rate by task:
sum(rate(fallback_total[5m])) by (task) / sum(rate(requests_total[5m])) by (task) - Latency penalty:
histogram_quantile(0.99, rate(fallback_latency_bucket[5m])) - histogram_quantile(0.99, rate(primary_latency_bucket[5m])) - Cost impact:
sum(increase(fallback_cost_usd[1h])) by (provider) - Health flip-flop count:
increase(model_health_changes_total[1h])— high values mean your thresholds are too sensitive
Alert on fallback rate > 5% for any task, or latency penalty > 2x baseline. Page on fallback rate > 25% — that’s a major incident.
When to use a gateway vs. build your own
If you’re routing across 3+ providers, handling 10k+ requests/day, or need per-customer routing policies, the gateway approach pays off. n4n.ai handles the health checks, fallback logic, usage metering, and cache-control forwarding across 240+ models behind one OpenAI-compatible endpoint — so your application code stays clean.
For simpler cases (2 providers, low volume, no per-tenant policies), the router pattern above is ~200 lines of Python you own and understand. The key insight: fallback is a routing concern, not a model concern. Keep it out of your business logic.