Error handling and fallbacks in LCEL chains

LCEL error handling fallbacks are not optional if you ship LLM features to production. The happy path works in notebooks; the unhappy path determines whether your users see a graceful degradation or a 500 error. This guide walks through the patterns that actually hold up under load, from basic retries to provider-aware fallback chains.

Understanding LCEL’s error surface

LCEL chains are just Runnable objects composed with the | operator. Every component — prompts, models, parsers, retrievers, custom functions — can raise exceptions. The chain itself doesn’t catch them unless you tell it to.

from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI

chain = (
    RunnableLambda(lambda x: {"topic": x["topic"]})
    | ChatOpenAI(model="gpt-4o-mini")
    | RunnableLambda(lambda msg: msg.content.upper())
)

If ChatOpenAI times out, hits a rate limit, or returns malformed JSON, the exception bubbles up uncaught. Your application code sees a raw APIConnectionError, RateLimitError, or OutputParserException. That’s the default behavior — explicit, but brittle.

Basic retry with with_retry

The simplest hardening is retrying transient failures. LCEL provides with_retry on any Runnable:

from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableConfig

llm = ChatOpenAI(model="gpt-4o-mini").with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
    retry_exceptions=(
        ConnectionError,
        TimeoutError,
        # OpenAI SDK specific
        "openai.RateLimitError",
        "openai.APIConnectionError",
        "openai.InternalServerError",
    ),
)

Pitfall: retry_exceptions accepts exception classes or strings. Strings are resolved at runtime, which lets you avoid importing the OpenAI SDK just for error types. But if you mistype the string, retries silently don’t fire. Prefer importing the actual exception classes when possible.

Tradeoff: Retries increase latency and cost. A 3-attempt exponential backoff on a 30-second timeout can hang a request for 90+ seconds. Set stop_after_delay to cap total wall time:

llm = ChatOpenAI(model="gpt-4o-mini", timeout=15).with_retry(
    stop_after_attempt=3,
    stop_after_delay=20,  # hard cap
    wait_exponential_jitter=True,
)

Fallback chains with with_fallbacks

Retries handle transient blips. Fallbacks handle systemic failures — model deprecation, provider outages, quota exhaustion. with_fallbacks takes a list of alternative Runnables:

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

primary = ChatOpenAI(model="gpt-4o")
fallback_1 = ChatOpenAI(model="gpt-4o-mini")
fallback_2 = ChatAnthropic(model="claude-3-5-haiku-20241022")

robust_llm = primary.with_fallbacks([fallback_1, fallback_2])

Each fallback is tried in order only if the previous one raises an exception. Success returns normally. If all fail, the last exception propagates.

Critical detail: Fallbacks must share the same input/output schema. If primary returns a BaseMessage but fallback_1 returns a string (because you wrapped it differently), downstream parsers break. Normalize outputs at the model layer:

from langchain_core.output_parsers import StrOutputParser

def make_llm(model_name: str, **kwargs):
    if "openai" in model_name.lower() or "gpt" in model_name.lower():
        return ChatOpenAI(model=model_name, **kwargs) | StrOutputParser()
    elif "claude" in model_name.lower():
        return ChatAnthropic(model=model_name, **kwargs) | StrOutputParser()
    raise ValueError(f"Unknown model: {model_name}")

primary = make_llm("gpt-4o")
fallback_1 = make_llm("gpt-4o-mini")
fallback_2 = make_llm("claude-3-5-haiku-20241022")

robust_llm = primary.with_fallbacks([fallback_1, fallback_2])

Now every fallback returns str, safe for downstream | JsonOutputParser() or prompt templates.

Per-request fallback control via config

Hardcoding fallbacks in the chain definition works for static policies. Production often needs dynamic routing — e.g., “use cheap model for tier-1 users, premium for tier-2” or “avoid provider X today because of known degradation.”

Pass fallbacks at invoke time through RunnableConfig:

chain = prompt | robust_llm

# At runtime, override fallbacks per request
result = chain.invoke(
    {"topic": "quantum computing"},
    config={"configurable": {"fallbacks": [cheap_llm, premium_llm]}},
)

This requires your robust_llm to be built with a ConfigurableField:

from langchain_core.runnables import ConfigurableField

primary = make_llm("gpt-4o")
fallback_1 = make_llm("gpt-4o-mini")

robust_llm = primary.configurable_fields(
    fallbacks=ConfigurableField(
        id="llm_fallbacks",
        name="LLM Fallbacks",
        description="Alternative models to try on failure",
    )
).with_fallbacks([fallback_1])  # default fallback

Now callers can inject entirely different fallback lists without redeploying chain code.

Handling streaming errors

Streaming introduces a failure mode that with_fallbacks doesn’t cover: the stream starts successfully, then dies mid-way. The caller has already received partial tokens. You can’t “fallback” a half-streamed response transparently.

Two strategies:

1. Buffer and validate before yielding — collect the full stream, validate, then yield. Defeats the purpose of streaming for latency-sensitive UIs.

2. Checkpoint and resume — persist partial output, then on failure, invoke a fallback with a prompt that continues from the checkpoint.

from langchain_core.runnables import RunnableGenerator
from langchain_core.messages import AIMessageChunk

async def stream_with_checkpoint(chain, input, config):
    buffer = []
    async for chunk in chain.astream(input, config):
        buffer.append(chunk)
        yield chunk
    
    # If we get here, stream completed successfully
    # Persist buffer for potential resume
    await save_checkpoint(config["configurable"]["session_id"], buffer)

# Usage: wrap your chain
streaming_chain = RunnableGenerator(stream_with_checkpoint).bind(chain=robust_chain)

Tradeoff: Checkpointing adds storage and complexity. For most chat applications, accepting partial failure (showing “response interrupted, retrying…”) is simpler than building resume logic.

Circuit breakers for provider outages

Retries and fallbacks assume the provider might recover. Circuit breakers assume it won’t — at least for a while. Stop hammering a down provider and fail fast to the next fallback.

LangChain doesn’t ship a circuit breaker, but pybreaker integrates cleanly:

import pybreaker
from langchain_openai import ChatOpenAI

breaker = pybreaker.CircuitBreaker(
    fail_max=5,           # open after 5 failures
    reset_timeout=60,     # try again after 60s
    exclude=[ValueError], # don't count validation errors
)

class CircuitBreakerLLM(ChatOpenAI):
    def invoke(self, input, config=None, **kwargs):
        return breaker.call(super().invoke, input, config, **kwargs)
    
    async def ainvoke(self, input, config=None, **kwargs):
        return breaker.call(super().ainvoke, input, config, **kwargs)

primary = CircuitBreakerLLM(model="gpt-4o")

Now the first 5 failures trip the breaker. Subsequent calls fail immediately with pybreaker.CircuitBreakerError, which with_fallbacks catches and routes to the next model.

Pitfall: Circuit breakers are process-local. In a multi-worker deployment (Gunicorn, Kubernetes), each worker has its own breaker state. For coordinated breaking, use a shared backend (Redis) with pybreaker’s StateStorage interface.

Structured error handling for observability

Raw exceptions lose context. Wrap failures in a typed error that captures what failed, which fallback was attempted, and why:

from dataclasses import dataclass
from typing import Optional
from langchain_core.runnables import Runnable, RunnableConfig

@dataclass
class LLMError(Exception):
    provider: str
    model: str
    attempt: int
    original_error: Exception
    fallback_tried: Optional[str] = None

def with_observability(runnable: Runnable, provider: str, model: str) -> Runnable:
    async def _ainvoke(input, config: RunnableConfig):
        attempt = config.get("configurable", {}).get("_attempt", 0)
        try:
            return await runnable.ainvoke(input, config)
        except Exception as e:
            raise LLMError(
                provider=provider,
                model=model,
                attempt=attempt,
                original_error=e,
            ) from e
    return RunnableLambda(_ainvoke).with_config(runnable.config_specs)

Apply to each fallback:

primary = with_observability(make_llm("gpt-4o"), "openai", "gpt-4o")
fallback_1 = with_observability(make_llm("gpt-4o-mini"), "openai", "gpt-4o-mini")
fallback_2 = with_observability(make_llm("claude-3-5-haiku"), "anthropic", "claude-3-5-haiku")

robust_llm = primary.with_fallbacks([fallback_1, fallback_2])

Your error tracker (Sentry, Datadog, custom) now gets structured fields for alerting and dashboards.

Provider-aware fallback ordering

Not all fallbacks are equal. Order them by:

1. Same provider, smaller model — fastest failover, same auth, same latency profile
2. Different provider, comparable model — survives provider-wide outage
3. Different provider, smaller/cheaper model — cost-controlled degradation
4. Local/offline model — last resort, runs in your VPC

fallbacks = [
    make_llm("gpt-4o-mini"),           # same provider, cheaper
    make_llm("claude-3-5-haiku"),      # different provider, fast
    make_llm("llama-3.1-8b-instant"),  # different provider (Groq), very fast
    local_llm,                          # self-hosted, no external deps
]

If you run an inference gateway that abstracts provider routing (like n4n.ai), the fallback list collapses to a single endpoint with automatic provider failover — your chain only needs one with_fallbacks to a local model for total isolation.

Testing failure paths

Don’t ship without simulating failures. Use RunnableLambda to inject faults:

import random
from langchain_core.runnables import RunnableLambda

def flaky_llm(inner, failure_rate=0.3, error_type=ConnectionError):
    async def _ainvoke(input, config):
        if random.random() < failure_rate:
            raise error_type("Simulated failure")
        return await inner.ainvoke(input, config)
    return RunnableLambda(_ainvoke)

# In tests
test_chain = prompt | flaky_llm(robust_llm, failure_rate=0.5)
# Verify fallback_1 gets called, output is valid

Test these scenarios:

• Primary succeeds (baseline)
• Primary fails once, fallback_1 succeeds
• Primary fails, fallback_1 fails, fallback_2 succeeds
• All fail — verify error type and observability fields
• Stream interrupted mid-way — verify partial handling

Common pitfalls summary

Putting it together: production-ready chain

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import (
    RunnableLambda, ConfigurableField, RunnableConfig
)
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
import pybreaker

# --- Models with circuit breakers ---
def make_circuit_breaker(model_name: str, **kwargs):
    breaker = pybreaker.CircuitBreaker(fail_max=5, reset_timeout=60)
    cls = ChatOpenAI if "gpt" in model_name.lower() else ChatAnthropic
    base = cls(model=model_name, **kwargs)
    
    class BreakerLLM(base.__class__):
        def invoke(self, *a, **kw):
            return breaker.call(super().invoke, *a, **kw)
        async def ainvoke(self, *a, **kw):
            return breaker.call(super().ainvoke, *a, **kw)
    
    return BreakerLLM(model=model_name, **kwargs)

# --- Normalized fallbacks ---
def llm_with_parser(model_name: str):
    llm = make_circuit_breaker(model_name, timeout=15)
    return llm.with_retry(
        stop_after_attempt=2,
        stop_after_delay=20,
        wait_exponential_jitter=True,
    ) | StrOutputParser()

primary = llm_with_parser("gpt-4o").configurable_fields(
    fallbacks=ConfigurableField(id="llm_fallbacks")
)
fallback_1 = llm_with_parser("gpt-4o-mini")
fallback_2 = llm_with_parser("claude-3-5-haiku-20241022")
fallback_3 = llm_with_parser("llama-3.1-8b-instant")  # via Groq

robust_llm = primary.with_fallbacks([fallback_1, fallback_2, fallback_3])

# --- Prompt + parser ---
prompt = ChatPromptTemplate.from_messages([
    ("system", "Extract key entities as JSON: {schema}"),
    ("user", "{text}"),
])
parser = JsonOutputParser(pydantic_object=EntitySchema)

# --- Final chain ---
chain = prompt | robust_llm | parser

# --- Invoke with dynamic fallbacks ---
result = chain.invoke(
    {"text": "...", "schema": EntitySchema.model_json_schema()},
    config={
        "configurable": {
            "llm_fallbacks": [fallback_1, fallback_2],  # override for this request
            "session_id": "user-123-conv-456",
        }
    },
)

This chain: retries transient errors, circuit-breaks persistent ones, falls back across three providers, normalizes output for the parser, accepts per-request fallback overrides, and carries a session ID for checkpointing. That’s the baseline for any LLM feature you charge money for.