Fallback strategy

Your current client has a single point of failure: if OpenAI goes down, your app goes down. In real production, that's unacceptable. The solution is automatic fallback: when a provider fails, it tries the next one automatically.

But "try the next one" isn't uniform — it depends on the type of error. If it was a rate limit, it's worth waiting and retrying. If it was an invalid API key, no — that one is never going to work. If it was a timeout, jump to the next provider quickly. This capsule teaches you to do it right.

By the end you'll be able to:

• Implement automatic fallback with the list of providers defined in config.fallback
• Distinguish transient errors (retry helps) from permanent ones (retry doesn't help)
• Apply a simple circuit breaker to avoid hammering downed providers
• Report clear errors when all providers fail

Mental model: when to retry, when to go next, when to abort

Three types of errors with three different responses:

The difference matters: retrying on AuthError burns requests uselessly. Retrying on RateLimitError with the same force doesn't work because the rate is still active.

Circuit breaker in one sentence

If a provider fails N times in a row, mark that provider as "open" for a while (e.g., 60s). During that time, you skip the provider without trying it — because you know it's down. After the time, you give it another chance.

Benefit: when OpenAI has an outage, you don't pay timeout × each request × each user. You jump straight to the fallback.

Implementation: extending UnifiedClient

Add this to unified_ai_client/client.py:

# unified_ai_client/client.py
import time
import logging
from dataclasses import dataclass, field
from pathlib import Path
import yaml

from .models import Message, ChatResponse, ClientConfig
from .factory import ProviderFactory
from .adapters.base import BaseAdapter
from .exceptions import (
    ConfigError,
    AuthError,
    RateLimitError,
    TimeoutError,
    ProviderError,
    AllProvidersFailedError,
)

logger = logging.getLogger(__name__)


@dataclass
class CircuitState:
    """Circuit breaker state for a provider."""
    consecutive_failures: int = 0
    open_until: float = 0  # epoch seconds


class UnifiedClient:
    """Unified client with automatic fallback and circuit breaker."""

    # How many failures before opening the circuit
    CIRCUIT_THRESHOLD = 3
    # For how long (seconds) to ignore a provider after reaching the threshold
    CIRCUIT_DURATION_S = 60

    def __init__(self, config: ClientConfig):
        self.config = config
        if config.primary not in config.providers:
            raise ConfigError(f"Primary '{config.primary}' does not exist in providers")
        for fb in config.fallback:
            if fb not in config.providers:
                raise ConfigError(f"Fallback '{fb}' does not exist in providers")

        # Ordered list: primary first, then fallbacks
        self.adapters: list[BaseAdapter] = [
            ProviderFactory.create(config.providers[config.primary])
        ] + [
            ProviderFactory.create(config.providers[name]) for name in config.fallback
        ]

        # Circuit breaker state per provider name
        self.circuit: dict[str, CircuitState] = {
            a.name: CircuitState() for a in self.adapters
        }

    @classmethod
    def from_yaml(cls, path: str | Path) -> "UnifiedClient":
        with open(path) as f:
            return cls(ClientConfig(**yaml.safe_load(f)))

    @classmethod
    def from_dict(cls, data: dict) -> "UnifiedClient":
        return cls(ClientConfig(**data))

    # ===========================================
    # Public API
    # ===========================================
    def chat(
        self,
        prompt: str,
        *,
        system: str | None = None,
        max_tokens: int = 256,
        temperature: float = 0.7,
    ) -> ChatResponse:
        messages: list[Message] = []
        if system:
            messages.append(Message(role="system", content=system))
        messages.append(Message(role="user", content=prompt))
        return self.chat_with_messages(messages, max_tokens=max_tokens, temperature=temperature)

    def chat_with_messages(
        self,
        messages: list[Message],
        max_tokens: int = 256,
        temperature: float = 0.7,
    ) -> ChatResponse:
        errors: dict[str, Exception] = {}

        for adapter in self.adapters:
            if self._circuit_open(adapter.name):
                logger.info(f"Skipping '{adapter.name}': circuit breaker open")
                errors[adapter.name] = ProviderError(
                    adapter.name, "Circuit breaker open (provider considered down)"
                )
                continue

            try:
                return self._try_with_retry(
                    adapter, messages, max_tokens, temperature
                )
            except AuthError as e:
                # Permanent: do NOT try others if auth is globally wrong
                # But it may be that only this provider has a problem; let's continue.
                logger.error(f"Auth error on '{adapter.name}': {e}")
                errors[adapter.name] = e
                self._record_failure(adapter.name)
                continue
            except (RateLimitError, TimeoutError, ProviderError) as e:
                logger.warning(f"'{adapter.name}' failed: {e}. Trying next.")
                errors[adapter.name] = e
                self._record_failure(adapter.name)
                continue

        # If we get here, no adapter worked
        raise AllProvidersFailedError(errors)

    # ===========================================
    # Internal: retry and circuit breaker
    # ===========================================
    def _try_with_retry(
        self,
        adapter: BaseAdapter,
        messages: list[Message],
        max_tokens: int,
        temperature: float,
        max_retries: int = 2,
    ) -> ChatResponse:
        """Tries an adapter with retry for transient errors."""
        for attempt in range(max_retries + 1):
            try:
                response = adapter.chat(messages, max_tokens, temperature)
                self._record_success(adapter.name)
                return response
            except RateLimitError:
                if attempt < max_retries:
                    wait = 2 ** attempt  # 1s, 2s, 4s
                    logger.info(f"Rate limit on '{adapter.name}', waiting {wait}s")
                    time.sleep(wait)
                    continue
                raise
            except AuthError:
                # Permanent: no retry
                raise
            except (TimeoutError, ProviderError):
                # No retry on the same provider; jump to the next
                raise

        raise RuntimeError("unreachable")  # type narrow

    def _circuit_open(self, provider_name: str) -> bool:
        state = self.circuit[provider_name]
        return state.open_until > time.time()

    def _record_failure(self, provider_name: str) -> None:
        state = self.circuit[provider_name]
        state.consecutive_failures += 1
        if state.consecutive_failures >= self.CIRCUIT_THRESHOLD:
            state.open_until = time.time() + self.CIRCUIT_DURATION_S
            logger.warning(
                f"Circuit open for '{provider_name}' for {self.CIRCUIT_DURATION_S}s "
                f"({state.consecutive_failures} consecutive failures)"
            )

    def _record_success(self, provider_name: str) -> None:
        # Success resets the counter and closes the circuit
        self.circuit[provider_name] = CircuitState()

How it works in practice

Case 1 — Primary works normally:

chat("...") → OpenAI ✓ → return ChatResponse

Case 2 — Primary has a transient rate limit:

chat("...") → OpenAI 429 → wait 1s → retry OpenAI ✓ → return ChatResponse

Case 3 — Primary goes down, fallback works:

chat("...") → OpenAI timeout → retry doesn't apply (timeout)
            → next: OpenRouter ✓ → return ChatResponse

Case 4 — Primary goes down repeatedly (circuit breaker):

1st request: OpenAI timeout → OpenRouter ✓
2nd request: OpenAI timeout → OpenRouter ✓
3rd request: OpenAI timeout (circuit OPENS for OpenAI) → OpenRouter ✓
4th request (next 60s): skip OpenAI → OpenRouter ✓ (without touching OpenAI)
65s later: OpenAI is tried again, if OK it closes the circuit

Case 5 — All providers fail:

chat("...") → OpenAI fail → OpenRouter fail → Ollama fail
            → AllProvidersFailedError(errors={...})

Verification with a realistic example

Config with 3 providers in a chain:

# examples/clients_fallback.yaml
primary: openai-mini
fallback:
  - openrouter-mistral
  - ollama-mistral

providers:
  openai-mini:
    name: openai-mini
    type: openai
    model: gpt-4o-mini
    api_key_env: OPENAI_API_KEY
    price_input_per_1m: 0.15
    price_output_per_1m: 0.60

  openrouter-mistral:
    name: openrouter-mistral
    type: openai_compatible
    model: mistralai/mistral-7b-instruct
    api_key_env: OPENROUTER_API_KEY
    base_url: https://openrouter.ai/api/v1
    price_input_per_1m: 0.07
    price_output_per_1m: 0.07

  ollama-mistral:
    name: ollama-mistral
    type: openai_compatible
    model: mistral
    base_url: http://localhost:11434/v1
    api_key_env: OLLAMA_DUMMY_KEY

Test script:

# examples/test_fallback.py
import logging
from unified_ai_client import UnifiedClient, AllProvidersFailedError

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")

client = UnifiedClient.from_yaml("examples/clients_fallback.yaml")

# Normal case
try:
    r = client.chat("Say hi in 5 words", max_tokens=20)
    print(f"\n→ OK from {r.provider}: {r.text}")
except AllProvidersFailedError as e:
    print(f"\n✗ All failed: {e}")

Force the primary to fail to verify fallback:

Change OPENAI_API_KEY to an invalid value and run again:

OPENAI_API_KEY=invalid python examples/test_fallback.py

You should see:

ERROR ... : Auth error on 'openai-mini': ...
WARNING ... : 'openai-mini' failed: ...
→ OK from openrouter-mistral: Hi from the other shore.

Configurable policy: opt-in fallback per request

Sometimes you want to disable fallback for a specific request (debugging, A/B testing a provider). Add a parameter:

def chat(
    self,
    prompt: str,
    *,
    system: str | None = None,
    max_tokens: int = 256,
    temperature: float = 0.7,
    use_fallback: bool = True,
) -> ChatResponse:
    ...
    return self.chat_with_messages(
        messages,
        max_tokens=max_tokens,
        temperature=temperature,
        use_fallback=use_fallback,
    )

def chat_with_messages(
    self,
    messages: list[Message],
    max_tokens: int = 256,
    temperature: float = 0.7,
    use_fallback: bool = True,
) -> ChatResponse:
    adapters = self.adapters if use_fallback else [self.adapters[0]]
    # ... rest the same, iterating over `adapters` instead of `self.adapters`

Usage:

# Default: uses fallback
r = client.chat("Hi")

# No fallback: only tries primary
r = client.chat("Hi", use_fallback=False)

Common traps

Trap 1 — "Retry forever on rate limit." Without a cap, you get stuck. My code uses max_retries=2 with exponential backoff. For your real case, consider something more sophisticated (jitter, max total time).

Trap 2 — "A circuit breaker that never closes." Without _record_success resetting the counter, the circuit stays open forever after 3 failures. Verify that your success logic does close it.

Trap 3 — "Fallback from OpenAI → OpenAI." If your fallback is the same provider (e.g., a different model), an OpenAI outage takes down both. The fallback must be from a different provider to be genuinely useful.

Trap 4 — "My log spams during a long outage." 3 providers × 100 requests × a 1-hour outage = 30,000 log lines. With a circuit breaker it drops drastically, but verify that the "circuit open" log is INFO or DEBUG, not WARNING per request, only when it opens/closes.

Trap 5 — "I tested fallback only in a demo, not in tests." Demos test the happy path. You need unit tests that mock primary failures and verify that fallback is invoked (capsule 07).

Exercise

Modify the code to:

1. Make CIRCUIT_THRESHOLD and CIRCUIT_DURATION_S configurable per instance (not class constants)
2. Add a client.reset_circuit(provider_name) method to force-reset a circuit (useful for tests and debugging)
3. Add a client.circuit_status() method that returns a dict with the state of each circuit

class UnifiedClient:
    def __init__(
        self,
        config: ClientConfig,
        circuit_threshold: int = 3,
        circuit_duration_s: int = 60,
    ):
        # ... rest the same
        self.circuit_threshold = circuit_threshold
        self.circuit_duration_s = circuit_duration_s
        # ...

    def _record_failure(self, provider_name: str) -> None:
        state = self.circuit[provider_name]
        state.consecutive_failures += 1
        if state.consecutive_failures >= self.circuit_threshold:
            state.open_until = time.time() + self.circuit_duration_s
            # ...

    def reset_circuit(self, provider_name: str) -> None:
        if provider_name not in self.circuit:
            raise ValueError(f"Unknown provider: {provider_name}")
        self.circuit[provider_name] = CircuitState()

    def circuit_status(self) -> dict[str, dict]:
        now = time.time()
        return {
            name: {
                "consecutive_failures": state.consecutive_failures,
                "open": state.open_until > now,
                "seconds_until_close": max(0, state.open_until - now),
            }
            for name, state in self.circuit.items()
        }


# Usage
client = UnifiedClient.from_yaml("clients.yaml", circuit_threshold=5, circuit_duration_s=120)
print(client.circuit_status())
client.reset_circuit("openai-mini")

Summary

You learned:

• ✅ Iterate over primary + fallbacks with error handling differentiated by type
• ✅ Retry with backoff for transient errors (rate limit)
• ✅ Skip a provider for permanent errors (auth, config)
• ✅ A simple circuit breaker that avoids hammering downed providers
• ✅ AllProvidersFailedError with detail of what failed on each one
• ✅ Opt-in fallback per request (for debugging and A/B testing)

Checkpoint: if you forced the primary to fail and your client automatically fell back to the fallback without your intervention, you're ready.

Next capsule

05 — Cost optimization. So far fallback corrects errors. We're going to use the same structure for priority routing: given a profile ("cost-first" / "quality-first" / "balanced"), choose the optimal provider automatically without the application code having to know which one.

Resources

1. Martin Fowler — Circuit Breaker pattern — the pattern explained.
2. Tenacity — Python library for serious retry/backoff.
3. pybreaker — production-ready circuit breaker in Python.
4. Exponential Backoff and Jitter (AWS) — why jitter matters for distributed retry.