This langchain fallback chain rate limits tutorial shows how to wire a primary and secondary chat model in LangChain so a 429 from your provider automatically routes to a backup. You’ll build a runnable Python chain that catches rate-limit and timeout exceptions without masking auth or input errors.
Prerequisites
- Python 3.10 or newer
langchain,langchain-openai,langchain-core, andopenaiinstalled- An OpenAI API key for live calls (optional; we’ll include an offline simulation)
- Familiarity with LangChain Expression Language (LCEL) basics
pip install langchain langchain-openai langchain-core openaiIf you run the live examples, export your key:
export OPENAI_API_KEY=sk-your-keyStep 1: Disable internal retries on the models
OpenAI’s Python client retries transient errors by default. In a fallback chain, that delay wastes time before your secondary model kicks in. Set max_retries=0 on each ChatOpenAI instance so the exception surfaces immediately.
from langchain_openai import ChatOpenAI
primary = ChatOpenAI(
model="gpt-4o-mini",
max_retries=0,
request_timeout=10,
)
secondary = ChatOpenAI(
model="gpt-3.5-turbo",
max_retries=0,
request_timeout=10,
)Step 2: Build the fallback chain with LCEL
LangChain’s RunnableWithFallbacks is the correct primitive. It catches specified exceptions from the primary runnable and delegates to the next in the list. Restrict the handled types to RateLimitError and APITimeoutError so a bad API key or malformed prompt still raises.
from langchain_core.runnables import RunnableWithFallbacks
from openai import RateLimitError, APITimeoutError
fallback_chain = primary.with_fallbacks(
[secondary],
exceptions_to_handle=(RateLimitError, APITimeoutError),
)This langchain fallback chain rate limits tutorial deliberately avoids FallbackLLM (the legacy class) because it doesn’t support chat messages cleanly and lacks fine-grained exception filtering.
Step 3: Invoke the chain
A plain invoke works like any other LangChain runnable.
response = fallback_chain.invoke("Explain the circuit breaker pattern in one sentence.")
print(response.content)Expected output when the primary is healthy:
A circuit breaker stops repeated calls to a failing service after a threshold of errors, allowing it to recover.If gpt-4o-mini returns a 429, the same call returns a response from gpt-3.5-turbo without changing your application code.
Step 4: Prove the fallback fires with a simulated 429
You shouldn’t need to burn your rate limit to test. Subclass FakeListChatModel to raise on the first call, then succeed.
from langchain_core.language_models.fake_chat_models import FakeListChatModel
from langchain_core.messages import AIMessage
from openai import RateLimitError
class FlakyModel(FakeListChatModel):
def __init__(self, responses, fail_times=1):
super().__init__(responses=responses)
self.fail_times = fail_times
self.calls = 0
def _generate(self, messages, **kwargs):
self.calls += 1
if self.calls <= self.fail_times:
raise RateLimitError("Rate limit exceeded", response=None, body=None)
return super()._generate(messages, **kwargs)
# Primary fails once, secondary is a stable fake
primary_test = FlakyModel(responses=[AIMessage(content="unused")], fail_times=1)
secondary_test = FakeListChatModel(responses=[AIMessage(content="Recovered via fallback")])
test_chain = primary_test.with_fallbacks(
[secondary_test],
exceptions_to_handle=(RateLimitError,),
)
print(test_chain.invoke("test").content)Expected output:
Recovered via fallbackThe primary raised RateLimitError, the chain caught it, and the secondary produced the answer.
Step 5: Add minimal logging to see which model answered
In production you want to know which model served a request. Wrap each model with a thin RunnableLambda that logs before calling.
from langchain_core.runnables import RunnableLambda
def log_and_call(model, name):
def _run(prompt):
print(f"[model={name}] handling request")
return model.invoke(prompt)
return RunnableLambda(_run)
observable_chain = log_and_call(primary, "gpt-4o-mini").with_fallbacks(
[log_and_call(secondary, "gpt-3.5-turbo")],
exceptions_to_handle=(RateLimitError, APITimeoutError),
)
observable_chain.invoke("Log which model answers")Sample output under degradation:
[model=gpt-4o-mini] handling request
[model=gpt-3.5-turbo] handling requestStep 6: Extend to multiple fallbacks
The with_fallbacks list is ordered. Add a third tier (e.g., a local model or a different provider) the same way.
tertiary = ChatOpenAI(model="gpt-3.5-turbo-0125", max_retries=0)
multi_chain = primary.with_fallbacks(
[secondary, tertiary],
exceptions_to_handle=(RateLimitError, APITimeoutError),
)If both primary and secondary rate-limit, tertiary receives the same input. LangChain does not re-prompt or alter the message list, so keep your prompts model-agnostic.
Step 7: When to use a gateway instead of hand-rolled fallback
If you route through n4n.ai, a single OpenAI-compatible endpoint fronts 240+ models and automatically fails over when a provider is rate-limited or degraded. That removes the need for the chain above when you only care about uptime. The rest of this langchain fallback chain rate limits tutorial is the right call when you must shape prompts per model, enforce cost tiers, or log token usage per provider.
Production caveats
- Never handle
(Exception,)broadly. CatchingAuthenticationErrorwill hide broken keys. - Set
request_timeout. Without it, a hung connection blocks your fallback indefinitely. - Meter tokens per tier. If you bill by usage, wrap
invoketo readresponse.usagefrom each model;ChatOpenAIexposes it on the returnedAIMessage.response_metadata. - Test with the fake models. A CI job running the
FlakyModelpattern catches regressions in your exception filtering. - Consider idempotency. Fallback re-sends the same prompt; ensure your downstream side effects (DB writes, tool calls) are safe under duplicate generation.
The pattern is small but unforgiving: disable retries, narrow the exception filter, and verify with a forced failure before shipping.