LangChain fallback chains for provider outages

When your production LLM calls start failing because OpenAI hits capacity or Anthropic degrades, you need a langchain fallback provider outage tutorial that shows how to keep requests flowing without manual intervention. LangChain’s RunnableWithFallbacks gives you a composable way to chain models across providers so a single outage doesn’t cascade into user-facing errors. This guide walks through building a production-ready fallback strategy, from basic chaining to routing-aware configurations that respect cost and latency constraints.

Start with a minimal fallback chain

The simplest pattern wraps a primary model with one or more alternatives. LangChain treats fallbacks as first-class runnables, so you can compose them like any other component.

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_core.runnables import RunnableWithFallbacks

primary = ChatOpenAI(model="gpt-4o", temperature=0)
fallback = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0)

chain = primary.with_fallbacks([fallback])

This chain tries gpt-4o first. On any exception — timeout, rate limit, 5xx, validation error — it automatically retries the request with Claude. The fallback receives the exact same input, so prompts and message history transfer cleanly.

Pitfall: The default behavior retries on any exception. A malformed prompt that raises a validation error on the primary will also fail on the fallback, wasting latency budget. Handle input validation upstream or catch specific exception types (see the error-classification section below).

Add multiple fallbacks with ordering

Real systems need more than two models. Order fallbacks by your actual priority: latency, cost, capability, or provider diversity.

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_mistralai import ChatMistralAI

models = [
    ChatOpenAI(model="gpt-4o", temperature=0),
    ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0),
    ChatGoogleGenerativeAI(model="gemini-1.5-pro", temperature=0),
    ChatMistralAI(model="mistral-large-latest", temperature=0),
]

chain = models[0].with_fallbacks(models[1:])

Each fallback attempt incurs full latency. If your p99 budget is 8 seconds and each model takes 3 seconds, three fallbacks blow past your SLO. Cap the chain length or add per-attempt timeouts.

Configure per-attempt timeouts and retries

LangChain’s RunnableWithFallbacks doesn’t enforce timeouts itself — it relies on the underlying client libraries. Set timeouts at the model level so each attempt fails fast.

from langchain_openai import ChatOpenAI
import httpx

http_client = httpx.Client(timeout=httpx.Timeout(connect=5.0, read=15.0))

primary = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    http_client=http_client,
    max_retries=0,  # disable client-side retries; let fallback chain handle it
)

Disable the provider SDK’s built-in retries (max_retries=0). Otherwise a single provider’s retry loop consumes your entire latency budget before the fallback chain ever activates. The fallback chain should be the only retry mechanism.

Tradeoff: Aggressive timeouts increase fallback frequency, which raises cost if fallbacks use more expensive models. Tune timeouts to your p50 latency, not p99 — you want to fall back on genuine degradation, not tail latency.

Classify errors to avoid wasted fallbacks

Not every error warrants a fallback. A 400 from a bad schema should fail fast. A 429 or 503 should trigger the chain. Subclass RunnableWithFallbacks to filter exceptions.

from langchain_core.runnables import RunnableWithFallbacks
from langchain_core.runnables.config import RunnableConfig
from typing import Any, List, Type
from httpx import HTTPStatusError

class SelectiveFallbacks(RunnableWithFallbacks):
    retryable_statuses = {429, 500, 502, 503, 504}
    
    def _is_retryable(self, error: BaseException) -> bool:
        if isinstance(error, HTTPStatusError):
            return error.response.status_code in self.retryable_statuses
        if isinstance(error, TimeoutError):
            return True
        # Network errors, connection resets
        if isinstance(error, (ConnectionError, OSError)):
            return True
        return False
    
    def invoke(self, input: Any, config: RunnableConfig | None = None, **kwargs: Any) -> Any:
        last_error = None
        for i, runnable in enumerate([self.runnable] + self.fallbacks):
            try:
                return runnable.invoke(input, config, **kwargs)
            except Exception as e:
                last_error = e
                if i == len(self.fallbacks) or not self._is_retryable(e):
                    break
        raise last_error

Use this wrapper instead of .with_fallbacks() when you need fine-grained control. The same pattern works for ainvoke, stream, and astream — override all four for full coverage.

Preserve streaming across fallbacks

Streaming is where fallback chains get tricky. If the primary model streams 80% of a response then errors, you can’t seamlessly continue from the fallback — the conversation context has diverged. Two practical approaches:

Option 1: Disable streaming for fallback chains. Buffer the full response. Simpler, higher latency for the first token.

chain = primary.with_fallbacks([fallback])
# Use .invoke() not .stream()
response = chain.invoke(messages)

Option 2: Stream from primary, fall back to non-streaming. If the primary fails mid-stream, discard partial output and retry the full request on the fallback.

async def stream_with_fallback(chain, input, config):
    try:
        async for chunk in chain.runnable.astream(input, config):
            yield chunk
    except Exception as e:
        if not _is_retryable(e):
            raise
        # Fallback: non-streaming retry
        for fb in chain.fallbacks:
            try:
                result = await fb.ainvoke(input, config)
                yield result  # single chunk
                return
            except Exception:
                continue
        raise e

Pitfall: Partial streamed output may have already reached the user (via WebSocket or SSE). You can’t “unsend” tokens. Design your UX to handle this — either buffer until complete, or accept that fallback responses may appear as a new message.

Route by model capability, not just availability

Different models excel at different tasks. A fallback chain that blindly routes coding tasks to a weak model produces garbage. Tag models with capabilities and select the appropriate fallback.

from dataclasses import dataclass
from typing import Literal

@dataclass
class ModelSpec:
    runnable: Any
    capabilities: set[str]  # e.g., {"coding", "reasoning", "vision", "long_context"}
    cost_per_1k_tokens: float
    avg_latency_ms: int

MODELS = [
    ModelSpec(
        runnable=ChatOpenAI(model="gpt-4o"),
        capabilities={"coding", "reasoning", "vision", "long_context"},
        cost_per_1k_tokens=0.03,
        avg_latency_ms=1200,
    ),
    ModelSpec(
        runnable=ChatAnthropic(model="claude-3-5-sonnet-20241022"),
        capabilities={"coding", "reasoning", "long_context"},
        cost_per_1k_tokens=0.015,
        avg_latency_ms=1500,
    ),
    ModelSpec(
        runnable=ChatGoogleGenerativeAI(model="gemini-1.5-pro"),
        capabilities={"reasoning", "long_context", "vision"},
        cost_per_1k_tokens=0.007,
        avg_latency_ms=2000,
    ),
]

def select_fallbacks(required_caps: set[str], exclude: set[str] = None) -> list[ModelSpec]:
    exclude = exclude or set()
    return [
        m for m in MODELS
        if required_caps.issubset(m.capabilities) and m.runnable not in exclude
    ]

# Usage: coding task needs coding + reasoning
fallbacks = select_fallbacks({"coding", "reasoning"})
chain = fallbacks[0].runnable.with_fallbacks([f.runnable for f in fallbacks[1:]])

This prevents routing a function-calling workload to a model that doesn’t support tools. Extend the spec with max_context_tokens to avoid sending 100k-token contexts to 8k-window models.

Add observability: log every fallback event

You can’t improve what you don’t measure. Wrap the chain to emit structured logs on each fallback attempt.

import structlog
from langchain_core.runnables import RunnableLambda

logger = structlog.get_logger()

def with_fallback_logging(chain: RunnableWithFallbacks):
    @RunnableLambda
    async def logged_ainvoke(input, config):
        for i, runnable in enumerate([chain.runnable] + chain.fallbacks):
            model_name = getattr(runnable, "model_name", str(runnable))
            try:
                result = await runnable.ainvoke(input, config)
                if i > 0:
                    logger.warning(
                        "fallback_succeeded",
                        attempt=i,
                        model=model_name,
                        primary_model=getattr(chain.runnable, "model_name", "unknown"),
                    )
                return result
            except Exception as e:
                logger.warning(
                    "fallback_attempt_failed",
                    attempt=i,
                    model=model_name,
                    error_type=type(e).__name__,
                    error=str(e)[:200],
                )
                if i == len(chain.fallbacks):
                    logger.error("all_fallbacks_exhausted", error=str(e))
                    raise
    return logged_ainvoke

Correlate these logs with your metrics dashboard. Alert on fallback rate spikes — they signal upstream degradation before users complain.

Handle provider-specific quirks

Each provider returns different error shapes. Anthropic uses overloaded_error, OpenAI uses rate_limit_error, Google uses ResourceExhausted. Normalize them before your fallback logic sees them.

from langchain_core.runnables import RunnableLambda

def normalize_errors(runnable):
    @RunnableLambda
    async def wrapper(input, config):
        try:
            return await runnable.ainvoke(input, config)
        except Exception as e:
            # Map provider-specific exceptions to a common hierarchy
            raise _map_provider_error(e) from e
    return wrapper

def _map_provider_error(e: Exception) -> Exception:
    error_msg = str(e).lower()
    if "rate limit" in error_msg or "429" in error_msg:
        return RateLimitError(str(e))
    if "overloaded" in error_msg or "503" in error_msg:
        return ProviderOverloadedError(str(e))
    if "timeout" in error_msg:
        return TimeoutError(str(e))
    return e

class RateLimitError(Exception): pass
class ProviderOverloadedError(Exception): pass

Wrap each model before adding to the chain: normalize_errors(ChatOpenAI(...)). Your selective fallback logic then catches RateLimitError | ProviderOverloadedError | TimeoutError cleanly.

Test fallback behavior under load

Unit tests with mocked exceptions verify logic. Integration tests with real providers verify latency and cost. Use a test harness that simulates degraded providers.

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_fallback_on_rate_limit():
    primary = AsyncMock()
    primary.ainvoke.side_effect = RateLimitError("429")
    
    fallback = AsyncMock()
    fallback.ainvoke.return_value = "fallback response"
    
    chain = primary.with_fallbacks([fallback])
    result = await chain.ainvoke("test")
    
    assert result == "fallback response"
    assert primary.ainvoke.call_count == 1
    assert fallback.ainvoke.call_count == 1

@pytest.mark.asyncio
async def test_no_fallback_on_validation_error():
    primary = AsyncMock()
    primary.ainvoke.side_effect = ValueError("bad schema")
    
    fallback = AsyncMock()
    
    chain = SelectiveFallbacks(primary, [fallback])
    
    with pytest.raises(ValueError):
        await chain.ainvoke("test")
    
    assert fallback.ainvoke.call_count == 0

Run chaos tests in staging: inject latency, return 503s, truncate responses. Verify your fallback chain activates within your SLO and doesn’t create thundering herds on the fallback provider.

Consider a routing layer for complex topologies

When you have dozens of models across providers, hardcoded chains become unmaintainable. A routing layer separates model selection from execution logic. n4n.ai provides this as a managed service — one OpenAI-compatible endpoint that addresses 240+ models with automatic fallback when a provider is rate-limited or degraded, per-token usage metering, and client routing directives — but you can build a simpler version yourself.

class ModelRouter:
    def __init__(self, model_specs: list[ModelSpec]):
        self.specs = {s.runnable: s for s in model_specs}
    
    def route(self, required_caps: set[str], strategy: Literal["cheapest", "fastest", "best"] = "best"):
        candidates = [s for s in self.specs.values() if required_caps.issubset(s.capabilities)]
        if not candidates:
            raise ValueError(f"No models support capabilities: {required_caps}")
        
        if strategy == "cheapest":
            candidates.sort(key=lambda s: s.cost_per_1k_tokens)
        elif strategy == "fastest":
            candidates.sort(key=lambda s: s.avg_latency_ms)
        else:  # best: balance cost, latency, capability
            candidates.sort(key=lambda s: (s.cost_per_1k_tokens * 0.6 + s.avg_latency_ms / 1000 * 0.4))
        
        primary = candidates[0]
        fallbacks = candidates[1:4]  # cap at 3 fallbacks
        return primary.runnable.with_fallbacks([f.runnable for f in fallbacks])

The router encapsulates policy. Your application code just calls router.route({"coding"}).ainvoke(messages). Change routing strategy globally without touching call sites.

Common pitfalls summary

Production checklist

Before deploying a fallback chain to production:

1. Disable client-side retries on every model (max_retries=0)
2. Set aggressive timeouts (connect 5s, read 15-20s) via custom httpx.Client
3. Classify errors — only fallback on 429, 5xx, timeouts, network errors
4. Cap fallback depth at 3 attempts maximum
5. Tag models with capabilities and filter fallbacks per request
6. Log every fallback event with model, attempt number, error type
7. Alert on fallback rate > 5% of traffic (adjust for your baseline)
8. Load test with injected failures to verify SLO compliance
9. Document the routing policy so on-call engineers know which model serves which traffic

Fallback chains aren’t a silver bullet — they add complexity, latency variance, and cost unpredictability. But for any LLM application with real availability requirements, they’re the difference between a brief degradation and a full outage. Build them deliberately, instrument them thoroughly, and test them like your uptime depends on it — because it does.