LangChain fallback chains model routing lets you keep a pipeline alive when a model endpoint throws rate limits, 5xx errors, or hangs. By pointing ChatOpenAI at an OpenAI-compatible gateway and attaching fallbacks via with_fallbacks, you get graceful degradation without rewriting prompt logic. This guide builds a runnable chain that tries a primary model, then falls back to two alternatives, and shows how to verify which model actually served the request.
Step 1: Install dependencies
Use langchain-openai and langchain-core (LangChain 0.2+). The fallback behavior relies on LCEL runnables, not the legacy LLMChain class. Do not mix the old FallbackChain from langchain.chains with the new RunnableWithFallbacks pattern; they serialize differently and the old one is deprecated.
pip install langchain-openai langchain-core python-dotenvLoad your API key from environment. The gateway expects a bearer token; treat it like an OpenAI key.
import os
from dotenv import load_dotenv
load_dotenv()
API_KEY = os.environ["N4N_API_KEY"] # or OPENAI_API_KEY if you mirror that var
GATEWAY_BASE = "https://api.n4n.ai/v1" # single OpenAI-compatible endpointStep 2: Configure primary and fallback models
Define three ChatOpenAI instances against the same base URL but with different model strings. When you point LangChain at n4n.ai’s OpenAI-compatible endpoint, one base URL covers 240+ models and the gateway already does provider-level automatic fallback on degradation. We still add application-level fallbacks because we want control over which models stand in for which tasks, and because we may want to fall back to a cheaper model before an expensive one.
from langchain_openai import ChatOpenAI
primary = ChatOpenAI(
model="openai/gpt-4o-mini",
api_key=API_KEY,
base_url=GATEWAY_BASE,
temperature=0,
timeout=10,
max_retries=1,
)
secondary = ChatOpenAI(
model="anthropic/claude-3-5-sonnet",
api_key=API_KEY,
base_url=GATEWAY_BASE,
temperature=0,
timeout=10,
max_retries=1,
)
tertiary = ChatOpenAI(
model="mistral/mistral-large",
api_key=API_KEY,
base_url=GATEWAY_BASE,
temperature=0,
timeout=10,
max_retries=1,
)Model names follow the gateway’s provider/model convention. Swap them for any of the 240+ available; the code does not change. Setting timeout and max_retries low is important—otherwise a hung primary blocks the fallback for minutes.
Step 3: Build the fallback chain
LangChain fallback chains model routing is implemented with with_fallbacks, which returns a RunnableWithFallbacks. The first runnable is the primary; the list passed to with_fallbacks is tried in order on exception. Only exceptions trigger fallback, not low-quality output.
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
prompt = ChatPromptTemplate.from_messages([
("system", "Answer concisely."),
("user", "{question}")
])
# Attach fallbacks at the model layer
model_with_fallback = primary.with_fallbacks([secondary, tertiary])
chain = prompt | model_with_fallback | StrOutputParser()You can also wrap the whole chain: (prompt | primary).with_fallbacks([prompt | secondary, prompt | tertiary]). Wrapping the model is cleaner when the prompt is static. If you need async, the same runnables expose ainvoke:
async def run_chain(question: str) -> str:
return await chain.ainvoke({"question": question})Step 4: Pass routing directives and cache hints
The gateway honors client routing directives and forwards provider cache-control hints. If you want to bias the primary call toward a specific upstream or enable Anthropic prompt caching, pass extra_body or model_kwargs. This does not affect LangChain’s fallback order; it only influences the gateway’s model selection and upstream headers.
primary_routed = ChatOpenAI(
model="anthropic/claude-3-5-sonnet",
api_key=API_KEY,
base_url=GATEWAY_BASE,
temperature=0,
extra_body={
"routing": {"prefer": "anthropic"},
"cache_control": {"type": "ephemeral"}
}
)n4n.ai forwards provider cache-control hints, so the cache_control block reaches Anthropic unchanged. If the request falls back to a provider that ignores it, the gateway strips or ignores it safely. You can also pass routing directives per-call via the config object if your runnable reads RunnableConfig:
from langchain_core.runnables import RunnableConfig
config = RunnableConfig(
metadata={"routing": {"prefer": "openai"}}
)
chain.invoke({"question": "Hi"}, config=config)Step 5: Invoke and handle failures
Call the chain inside a try/except to catch the case where all models fail (e.g., invalid key). On success, inspect response_metadata to see which model answered. Wrapping the model only gives you the parsed string; to get metadata, call the model-with-fallback directly before the parser.
from langchain_core.runnables import RunnableWithFallbacks
question = "What is the difference between TCP and UDP in one sentence?"
try:
response = (prompt | model_with_fallback).invoke({"question": question})
answer = StrOutputParser().invoke(response)
print("ANSWER:", answer)
except Exception as e:
print("All fallbacks exhausted:", repr(e))For streaming, use stream on the runnable; fallback still works because the stream is opened only after the primary commits to a response.
for chunk in (prompt | model_with_fallback).stream({"question": question}):
print(chunk.content, end="", flush=True)Step 6: Verify success and metering
Success means you got a non-empty string and response.response_metadata.model matches one of your configured models. Because the gateway provides per-token usage metering, you can assert on token counts:
usage = response.response_metadata.get("usage", {})
assert usage.get("total_tokens", 0) > 0, "No tokens metered"
print(f"Metered {usage['total_tokens']} tokens on {response.response_metadata['model']}")Run the script with the primary model intentionally broken (e.g., wrong model string) to confirm the fallback engages:
primary = ChatOpenAI(model="openai/does-not-exist", api_key=API_KEY, base_url=GATEWAY_BASE)
# re-build chain...You should see the answer arrive from anthropic/claude-3-5-sonnet or mistral/mistral-large and the usage block reflect that model. A minimal pytest check:
def test_fallback_runs():
resp = (prompt | primary.with_fallbacks([secondary, tertiary])).invoke(
{"question": "ping"}
)
assert resp.content
assert "model" in resp.response_metadataTroubleshooting
- 401 Unauthorized: Key not in env, or using wrong var name. The gateway expects the key in
Authorization: Bearer. - 404 on model: Model string must use
provider/model. List available models from the gateway’s/v1/modelsendpoint. - Fallback not triggering:
with_fallbacksonly catches exceptions, not empty responses or content-filter blocks that return 200. If you need to fallback on junk output, wrap the parser with a custom validator that raises. - Latency stacking: Each fallback adds full round-trip time. Set
timeouton eachChatOpenAIinstance to fail fast. - Wrong model in metadata: Some providers return a normalized name. Compare with
inrather than exact equality. - Cache hint ignored: Ensure
extra_bodyis supported by yourlangchain-openaiversion (0.1.15+).
primary = ChatOpenAI(
model="openai/gpt-4o-mini",
api_key=API_KEY,
base_url=GATEWAY_BASE,
timeout=10,
max_retries=1
)Scaling the pattern
Once this works, extend it: put the fallback runnable behind a RunnableParallel to call two models concurrently and pick the first valid response, or compose it with a RouterChain that selects the primary based on input complexity. The LangChain fallback chains model routing approach stays the same—you are just changing what sits upstream of with_fallbacks. Because the gateway already handles provider degradation and per-token metering, your application code focuses purely on model preference and order.
You now have a single LCEL chain that implements LangChain fallback chains model routing across heterogeneous providers through one OpenAI-compatible base URL. The application controls fallback priority; the gateway handles provider degradation and token metering underneath. Swap the prompt for a more complex sequence without touching the fallback wiring.