When a single LLM provider throws a 429 or silently degrades, your app goes down. Implementing openai python sdk multi-model fallback lets you chain candidate models behind one OpenAI-compatible client so a request survives provider outages without rewriting your call sites.
Prerequisites
- Python 3.10 or newer
- The
openaiPython package (v1.x) - API key and base URL for an OpenAI-compatible endpoint (OpenAI, a self-hosted vLLM server, or a gateway)
- At least two model identifiers you can call (e.g.,
gpt-4o-mini,gpt-3.5-turbo)
If you point the SDK at a gateway such as n4n.ai, you get automatic fallback across 240+ models when a provider is rate-limited. The code below instead gives you explicit, auditable control over the fallback order.
Install and configure
pip install openai>=1.0.0Set credentials via environment variables to avoid hardcoding:
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ["LLM_API_KEY"],
base_url=os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1"),
)The base_url is the only thing that changes if you target a non-OpenAI endpoint. The openai python sdk multi-model fallback pattern works identically because the request shape is the same.
Define the model chain
Pick models in order of preference. Usually a strong model first, then cheaper/faster ones:
MODEL_CHAIN = [
"gpt-4o-mini",
"gpt-3.5-turbo",
"mistralai/mistral-7b-instruct", # example cross-provider name on some gateways
]Keep the list short. Three candidates is enough for most workloads; more increases tail latency.
Classify errors that justify fallback
Not every exception should trigger the next model. A 400 from malformed input will fail on all models. Fallback only on transient or capacity errors:
from openai import (
APIConnectionError,
RateLimitError,
APIStatusError,
)
def is_fallbackable(exc: Exception) -> bool:
if isinstance(exc, RateLimitError):
return True
if isinstance(exc, APIConnectionError):
return True
if isinstance(exc, APIStatusError):
# 500, 502, 503, 504, and sometimes 404 if model missing at provider
return exc.status_code in (500, 502, 503, 504, 404)
return FalseImplement the router
A small wrapper loops over the chain, catches fallbackable errors, and returns the first success:
from dataclasses import dataclass
from openai.types.chat import ChatCompletion
@dataclass
class FallbackResult:
content: str
model_used: str
attempts: int
def chat_with_fallback(
client: OpenAI,
model_chain: list[str],
messages: list[dict],
max_retries_per_model: int = 1,
) -> FallbackResult:
last_exc: Exception | None = None
for attempt, model in enumerate(model_chain, start=1):
for retry in range(max_retries_per_model):
try:
resp: ChatCompletion = client.chat.completions.create(
model=model,
messages=messages,
timeout=20.0, # per-request timeout
)
return FallbackResult(
content=resp.choices[0].message.content or "",
model_used=model,
attempts=attempt,
)
except Exception as exc:
last_exc = exc
if not is_fallbackable(exc):
raise # bad request, auth error: don't burn other models
# optional: sleep before retry on same model
if retry < max_retries_per_model - 1:
import time
time.sleep(0.5 * (retry + 1))
raise RuntimeError(f"All models failed, last error: {last_exc}")The timeout parameter is critical. Without it a hung TCP connection blocks the entire chain.
Run the example
if __name__ == "__main__":
messages = [{"role": "user", "content": "Summarize: fallback chains keep LLM apps online."}]
try:
result = chat_with_fallback(client, MODEL_CHAIN, messages)
print(f"Model used: {result.model_used}")
print(f"Attempt index: {result.attempts}")
print(f"Reply: {result.content}")
except RuntimeError as e:
print(f"Fatal: {e}")Expected output
If the first model is healthy:
Model used: gpt-4o-mini
Attempt index: 1
Reply: Fallback chains improve reliability by routing around failed model providers...If gpt-4o-mini returns 429 and gpt-3.5-turbo succeeds:
Model used: gpt-3.5-turbo
Attempt index: 2
Reply: They help maintain uptime by trying alternative models when one is unavailable.If all fail, you get the RuntimeError with the last exception.
Streaming changes the calculus
The above uses non-streaming calls. With stream=True, tokens may arrive before an error occurs mid-stream. You cannot un-send those tokens to the end user. For streaming fallback, buffer the entire response in memory and only flush after the stream completes, or accept that a failed stream aborts the request. For most production services, run fallback on non-streaming and stream only after a model is locked in.
Production hardening
- Per-model timeouts: smaller models can have tighter timeouts.
- Backoff with jitter: replace the fixed
sleepwithrandom.uniform(0, 2**retry). - Usage metering: capture
resp.usageand logprompt_tokens/completion_tokensper model. Costs differ across the chain. - Cache-control headers: if your endpoint honors
cache-controlhints, forward them viaextra_headers={"cache-control": "max-age=300"}on the create call. Some gateways pass these to the provider. - Observability: emit a metric tagged with
model_usedso you can see fallback frequency.
When to use a gateway instead
Hand-rolled fallback is transparent and debuggable, but it adds latency and code to maintain. An OpenAI-compatible inference gateway can apply fallback automatically at the edge, return per-token usage, and respect routing directives you send in headers. If you already depend on such a gateway, you may skip the loop and just call one model alias that the gateway resolves. The openai python sdk multi-model fallback logic still applies when you need fine-grained control over which model answers.
Final notes
Test the chain with fault injection: temporarily set the first model to a nonexistent ID and confirm the second serves. The pattern scales to any number of OpenAI-compatible endpoints by swapping base_url per client if you need cross-provider fallback beyond what a single gateway offers.
Keep the chain short, fail fast on non-transient errors, and log which model actually answered. That’s the difference between a resilient integration and a slow one that hides outages.