Building a resilient LLM integration means planning for provider outages and rate limits. The multi provider llm fallback python pattern lets you chain calls across vendors so a single 429 doesn’t take down your feature. This tutorial walks through a concrete implementation using the OpenAI and Anthropic SDKs, with runnable code at each step.
Prerequisites
- Python 3.10 or newer
openaiandanthropicpackages installed (pip install openai anthropic)- API keys for OpenAI and Anthropic exported as
OPENAI_API_KEYandANTHROPIC_API_KEY - Comfort with Python exceptions and standard library
typing
If you only have one provider key, the simulated failure section still works.
Define a uniform result
Don’t pass raw strings through your fallback layer. Capture provenance and token counts so you can meter spend per vendor.
from dataclasses import dataclass
@dataclass
class CompletionResult:
text: str
provider: str
prompt_tokens: int
completion_tokens: int
Provider interface
Use a Protocol to keep adapters swappable. The multi provider llm fallback python logic only depends on this contract.
from typing import Protocol, runtime_checkable
@runtime_checkable
class LLMProvider(Protocol):
def complete(self, prompt: str) -> CompletionResult:
...
OpenAI adapter
The OpenAI SDK exposes chat.completions.create. Set an explicit timeout so a hung connection doesn’t block the chain.
import os
from openai import OpenAI, RateLimitError, APITimeoutError
class OpenAIProvider:
def __init__(self, model: str = "gpt-4o-mini"):
self.client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
self.model = model
def complete(self, prompt: str) -> CompletionResult:
resp = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
timeout=10,
)
msg = resp.choices[0].message.content
return CompletionResult(
text=msg,
provider="openai",
prompt_tokens=resp.usage.prompt_tokens,
completion_tokens=resp.usage.completion_tokens,
)
Checkpoint: run it alone.
p = OpenAIProvider()
r = p.complete("Say hello in one word.")
print(r.text, r.provider, r.prompt_tokens)
Expected output resembles: Hello openai 12
Anthropic adapter
Anthropic’s messages API differs in shape but maps cleanly to the same result.
from anthropic import Anthropic, RateLimitError as AnthropicRateLimitError
class AnthropicProvider:
def __init__(self, model: str = "claude-3-5-sonnet-20241022"):
self.client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
self.model = model
def complete(self, prompt: str) -> CompletionResult:
resp = self.client.messages.create(
model=self.model,
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
timeout=10,
)
return CompletionResult(
text=resp.content[0].text,
provider="anthropic",
prompt_tokens=resp.usage.input_tokens,
completion_tokens=resp.usage.output_tokens,
)
The fallback chain
Iterate providers, catch only retryable errors, and return the first success. Non-retryable errors (bad request, auth) should bubble immediately.
from openai import RateLimitError, APITimeoutError
from anthropic import RateLimitError as AnthropicRateLimitError
def is_retryable(e: Exception) -> bool:
return isinstance(e, (RateLimitError, APITimeoutError, AnthropicRateLimitError))
class FallbackChain:
def __init__(self, providers: list[LLMProvider]):
self.providers = providers
def complete(self, prompt: str) -> CompletionResult:
last_err = None
for p in self.providers:
try:
return p.complete(prompt)
except Exception as e:
if not is_retryable(e):
raise
last_err = e
print(f"[warn] {type(p).__name__} degraded: {e}")
raise RuntimeError("All providers failed") from last_err
Checkpoint with both providers live:
chain = FallbackChain([OpenAIProvider(), AnthropicProvider()])
result = chain.complete("Explain fallback in one sentence.")
print(result.provider, "->", result.text)
You’ll see output from whichever provider answered first. No warnings printed.
Simulate degradation
Force the first provider to fail to prove the chain skips.
class FailingProvider:
def complete(self, prompt: str) -> CompletionResult:
raise RateLimitError(
message="Simulated 429",
response=None,
body=None,
)
chain = FallbackChain([FailingProvider(), AnthropicProvider()])
r = chain.complete("Say hi.")
print(r.provider, r.text)
Expected console:
[warn] FailingProvider degraded: Simulated 429
anthropic -> Hi.
That’s the multi provider llm fallback python pattern working.
Order providers by latency and cost
Provider order is a business decision. Put the cheapest or fastest first; let the chain absorb its quota limits.
# Cheapest first, premium backup
chain = FallbackChain([
OpenAIProvider(model="gpt-4o-mini"),
AnthropicProvider(model="claude-3-5-sonnet-20241022"),
])
Async for concurrent serving
In a web server you shouldn’t block the event loop. Use AsyncOpenAI and AsyncAnthropic.
import asyncio
from openai import AsyncOpenAI
class AsyncOpenAIProvider:
def __init__(self, model="gpt-4o-mini"):
self.client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
self.model = model
async def complete(self, prompt: str) -> CompletionResult:
resp = await self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
timeout=10,
)
return CompletionResult(
text=resp.choices[0].message.content,
provider="openai-async",
prompt_tokens=resp.usage.prompt_tokens,
completion_tokens=resp.usage.completion_tokens,
)
class AsyncFallbackChain:
def __init__(self, providers: list):
self.providers = providers
async def complete(self, prompt: str) -> CompletionResult:
last_err = None
for p in self.providers:
try:
return await p.complete(prompt)
except Exception as e:
if not is_retryable(e):
raise
last_err = e
raise RuntimeError("All providers failed") from last_err
Run it:
async def main():
chain = AsyncFallbackChain([AsyncOpenAIProvider()])
r = await chain.complete("Ping")
print(r.text)
asyncio.run(main())
Track usage for cost allocation
Because CompletionResult carries token counts, you can emit metrics per provider without extra calls.
def log_usage(r: CompletionResult):
print(f"provider={r.provider} in={r.prompt_tokens} out={r.completion_tokens}")
log_usage(chain.complete("Status?"))
If you run this in production, ship those numbers to your metrics backend tagged by provider.
When a gateway is simpler
Hand-rolling fallback is reasonable when you need fine-grained control or run on-prem models. If you’d rather not maintain the error taxonomy, n4n.ai exposes one OpenAI-compatible endpoint that addresses 240+ models and applies automatic fallback when a provider is rate-limited or degraded, collapsing the multi provider llm fallback python pattern into a single client call.
from openai import OpenAI
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key=os.environ["N4N_KEY"])
resp = client.chat.completions.create(
model="auto",
messages=[{"role": "user", "content": "Hello"}],
)
print(resp.choices[0].message.content)
Tests
A minimal pytest confirms the chain skips a failing node.
def test_fallback_skips_failure():
chain = FallbackChain([FailingProvider(), OpenAIProvider()])
r = chain.complete("test")
assert r.provider == "openai"
Run with pytest -q. The test hits the network; mark it accordingly in CI.
Edge cases you must handle
- Partial responses: streaming APIs may emit tokens then error. Decide whether to discard or return partials.
- Schema validation: if you request JSON, a backup provider may return malformed JSON. Validate after the chain.
- Timeouts differ: set per-provider timeouts based on observed p95 latency, not a guess.
- Auth errors: never retry
401. Youris_retryablecheck already excludes them.
The multi provider llm fallback python approach is not a substitute for capacity planning, but it converts a hard dependency into a soft one. Ship the chain, meter the usage, and sleep better.