n4nAI

Multi-provider LLM fallback: a Python code pattern

A hands-on Python tutorial for building a multi provider llm fallback python pattern with OpenAI and Anthropic SDKs, plus graceful degradation.

n4n Team2 min read501 words

Audio narration

Coming soon — every post will get a voice note here.

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
  • openai and anthropic packages installed (pip install openai anthropic)
  • API keys for OpenAI and Anthropic exported as OPENAI_API_KEY and ANTHROPIC_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. Your is_retryable check 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.

Tagsfallbackpythonmulti-providercode-pattern

Written by

n4n Team

The team building n4n — a single OpenAI-compatible API in front of 240+ models, with automatic fallback, load balancing and pay-per-token metering.

More from n4n Team →

All multi-provider fallback code patterns posts →