LangChain RunnableWithFallbacks explained

This langchain runnablewithfallbacks tutorial shows how to make an LLM chain survive provider errors without hand-rolled try/except blocks. We’ll compose a primary model and one or more fallbacks using RunnableWithFallbacks, run it against a real prompt, and inspect what happens when the first model fails. The pattern is the backbone of any production system that can’t afford a hard 500 from a single vendor.

Prerequisites

• Python 3.10 or newer
• langchain-core and langchain-openai (both v0.2+)
• A valid OPENAI_API_KEY in your environment
• Basic familiarity with ChatPromptTemplate and runnables

pip install langchain-core langchain-openai python-dotenv

# .env
OPENAI_API_KEY=sk-...

If you want to test cross-provider fallback, set a second variable for an alternate base URL. For this walkthrough we’ll use a second OpenAI model as the fallback to keep the code minimal.

The naive chain

Start with a plain prompt → model → parser sequence. This is what most LangChain apps ship before they hit their first rate limit.

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a terse senior engineer. Answer in one sentence."),
    ("user", "{question}")
])

primary_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

chain = prompt | primary_model | parser

print(chain.invoke({"question": "What is a mutex?"}))

Expected output (truncated):

A mutex is a synchronization primitive that prevents concurrent access to a shared resource by multiple threads.

Forcing a failure

To see fallback behavior without waiting for an outage, point the primary at a model name that doesn’t exist. OpenAI returns a 404 and LangChain raises openai.NotFoundError.

broken_model = ChatOpenAI(model="gpt-99-nonexistent", temperature=0)
broken_chain = prompt | broken_model | parser

try:
    broken_chain.invoke({"question": "What is a mutex?"})
except Exception as e:
    print(type(e).__name__, str(e)[:80])

Expected output:

NotFoundError b'{"error":{"message":"The model `gpt-99-nonexistent` does not exist..."}'

Wrapping with RunnableWithFallbacks

RunnableWithFallbacks takes a runnable (the primary) and a list of fallbacks. It invokes the primary; if that raises, it walks the fallback list in order. Each fallback must accept the same input schema as the primary.

from langchain_core.runnables import RunnableWithFallbacks

fallback_model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
fallback_chain = prompt | fallback_model | parser

resilient_chain = RunnableWithFallbacks(
    runnable=broken_chain,
    fallbacks=[fallback_chain],
)

print(resilient_chain.invoke({"question": "What is a mutex?"}))

Expected output:

A mutex (mutual exclusion) is a lock that ensures only one thread can access a resource at a time.

The broken primary threw, LangChain caught it, and the gpt-3.5-turbo fallback produced a response. No try/except in your business logic.

Inspecting which path was taken

In production you want telemetry on fallback events. Wrap the fallback in a RunnableLambda that logs, or use callbacks. Minimal logging approach:

import logging
logging.basicConfig(level=logging.INFO)

def log_fallback(inputs):
    logging.info("Primary failed, using fallback")
    return fallback_chain.invoke(inputs)

fallback_with_log = RunnableLambda(log_fallback)

resilient_chain = RunnableWithFallbacks(
    runnable=broken_chain,
    fallbacks=[fallback_with_log],
)

resilient_chain.invoke({"question": "Explain RAID 0"})

Console:

INFO:root:Primary failed, using fallback

Multiple fallbacks in order

You can stack more than one. The first that succeeds wins.

fallback_model_b = ChatOpenAI(model="gpt-3.5-turbo-0125", temperature=0)
chain_b = prompt | fallback_model_b | parser

resilient_chain = RunnableWithFallbacks(
    runnable=broken_chain,
    fallbacks=[fallback_chain, chain_b],
)

If gpt-3.5-turbo is also degraded, it moves to chain_b. The fallbacks list is evaluated lazily—only invoked if needed.

Streaming with fallbacks

RunnableWithFallbacks supports streaming, but the fallback only triggers if the primary raises before producing the first chunk. If the primary dies mid-stream, the exception propagates. Design primary models to fail fast on auth/rate-limit checks.

for chunk in resilient_chain.stream({"question": "What is a socket?"}):
    print(chunk, end="")

If the primary is broken, you’ll see the fallback’s stream start after a brief delay.

Batch and async

The same object works with batch and ainvoke. For batch, each input is independently fallen back—one bad input won’t kill the whole batch.

results = resilient_chain.batch([
    {"question": "What is a deadlock?"},
    {"question": "What is a semaphore?"},
])
print(results)

Beyond app-level fallback

If you’d rather not maintain model lists in code, an OpenAI-compatible gateway such as n4n.ai fronts 240+ models and applies automatic fallback when a provider is rate-limited or degraded, while honoring client routing directives. You can point a single ChatOpenAI(base_url="https://api.n4n.ai/v1") at it and skip RunnableWithFallbacks for simple redundancy. The LangChain pattern still wins when fallback logic depends on app state—e.g., “if the user is enterprise, use model X, else model Y.”

Caveats engineers miss

Input compatibility. Fallbacks must accept identical input. If your primary prompt includes a user_id and the fallback chain expects only question, you’ll get a validation error that looks like a model failure.

Stateful chains. If the primary mutated external state before raising (e.g., wrote to a vector store), the fallback will re-run from the start. Wrap side effects in idempotent runnables.

Latency. Fallback adds tail latency. A 30s primary timeout before fallback means the user waits 30s + fallback time. Set request_timeout on the model client:

primary_model = ChatOpenAI(model="gpt-4o-mini", request_timeout=10)

Cost. Fallbacks consume tokens on the secondary model even if the primary would have succeeded on retry. For transient 429s, consider a retry policy (RunnableRetry) before falling back to a weaker model.

Full runnable script

import logging
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableWithFallbacks, RunnableLambda
from langchain_openai import ChatOpenAI

logging.basicConfig(level=logging.INFO)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a terse senior engineer. Answer in one sentence."),
    ("user", "{question}")
])
parser = StrOutputParser()

primary = prompt | ChatOpenAI(model="gpt-99-nonexistent") | parser
fb1 = prompt | ChatOpenAI(model="gpt-3.5-turbo") | parser

def log_and_run(inputs):
    logging.info("fallback path")
    return fb1.invoke(inputs)

chain = RunnableWithFallbacks(
    runnable=primary,
    fallbacks=[RunnableLambda(log_and_run)],
)

print(chain.invoke({"question": "What is a lambda?"}))

Run it. You’ll get a log line and a coherent answer from the fallback. That’s the entire mechanism—no custom retry loops, no scattered exception handling. For a langchain runnablewithfallbacks tutorial, this is the smallest useful surface; from here, add batching, async, and routing as your SLA demands.