0Pricing
AI Engineering Academy · Lezione

Provider di fallback e circuit breaker

Costruisca una cascata di provider che esegua automaticamente il failover da OpenAI ad Anthropic e quindi a un modello locale quando il provider primario è lento o non disponibile, usando il pattern circuit breaker.

Provider di fallback e circuit breaker è una lezione AI Engineering Academy gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Engineering Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Engineering Academy include 4 lezioni in totale.

Parti di questa lezione non sono ancora state tradotte e vengono mostrate in inglese.

Single Provider Risk

Relying on a single LLM provider creates a single point of failure. OpenAI has experienced outages that took minutes to hours to resolve. If your entire application depends on GPT-4o being available, any provider incident immediately translates to user-facing downtime. A fallback provider strategy maintains service continuity by routing to alternative providers when the primary fails.

Defining a Provider Cascade

A provider cascade is an ordered list of providers and models tried in sequence. When the primary fails or times out, the system automatically tries the next provider. A typical cascade might be: OpenAI GPT-4o → Anthropic Claude 3.5 Sonnet → a locally deployed Llama model. Each level is a fallback with the local model serving as the last resort that cannot go down.

from dataclasses import dataclass
from typing import Optional

@dataclass
class Provider:
    name: str
    base_url: Optional[str]
    api_key_env: str
    model: str
    priority: int  # lower = higher priority

CASCADE = [
    Provider('openai',    None,                              'OPENAI_API_KEY',    'gpt-4o',              1),
    Provider('anthropic', 'https://api.anthropic.com/v1',   'ANTHROPIC_API_KEY', 'claude-3-5-sonnet',   2),
    Provider('local',     'http://localhost:8000/v1',        'LOCAL_KEY',         'llama-3.1-8b-inst',   3),
]

Implementing the Fallback Loop

Implement the fallback loop as a simple try/except that iterates through the cascade. Catch transient errors (timeouts, 500s, 503s) and move to the next provider. Do not catch authentication errors (401) or invalid request errors (400) — these are programming mistakes that should surface immediately rather than fail over to another provider.

import openai
import os

TRANSIENT_ERRORS = (openai.APITimeoutError, openai.InternalServerError, openai.APIConnectionError)

async def call_with_fallback(messages: list, **kwargs) -> str:
    for provider in CASCADE:
        try:
            client = openai.AsyncOpenAI(
                api_key=os.environ[provider.api_key_env],
                base_url=provider.base_url
            )
            resp = await client.chat.completions.create(
                model=provider.model,
                messages=messages,
                timeout=10.0,
                **kwargs
            )
            return resp.choices[0].message.content
        except TRANSIENT_ERRORS as e:
            print(f'Provider {provider.name} failed: {e}, trying next...')
    raise RuntimeError('All providers failed')

What Is a Circuit Breaker?

A circuit breaker prevents a failing service from being hammered with requests during an outage. Named after electrical circuit breakers, it has three states: Closed (requests pass through normally), Open (requests are immediately rejected), and Half-Open (a single test request is allowed through to check if service has recovered). This protects both the downstream service and your own application during incidents.

# Circuit breaker state machine:
#
# CLOSED --> (failure_count >= threshold) --> OPEN
#    ^                                          |
#    |     (test_request succeeds)              | (timeout expires)
#    +------------ HALF_OPEN <-----------------+
#
# In OPEN state: immediately return fallback/error
# In HALF_OPEN: allow one request through to test recovery
# In CLOSED: normal operation, count failures

Implementing a Circuit Breaker

Here is a minimal circuit breaker implementation. Track the failure count and the time the circuit opened. When failure count exceeds the threshold, open the circuit. After a configurable reset timeout, allow one probe request. If the probe succeeds, close the circuit. If it fails, keep the circuit open and reset the timeout.

import time
from enum import Enum

class State(Enum):
    CLOSED = 'closed'
    OPEN = 'open'
    HALF_OPEN = 'half_open'

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.state = State.CLOSED
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.opened_at = None

    def record_success(self):
        self.failure_count = 0
        self.state = State.CLOSED

    def record_failure(self):
        self.failure_count += 1
        if self.failure_count >= self.failure_threshold:
            self.state = State.OPEN
            self.opened_at = time.time()

    def can_attempt(self) -> bool:
        if self.state == State.CLOSED:
            return True
        if self.state == State.OPEN:
            if time.time() - self.opened_at > self.reset_timeout:
                self.state = State.HALF_OPEN
                return True  # allow one probe
            return False
        return True  # HALF_OPEN: allow probe

Integrating Circuit Breakers with Providers

Maintain one circuit breaker per provider. Before calling a provider, check if its circuit breaker allows the attempt. After each call, record success or failure. When a provider's circuit opens, the fallback loop naturally skips it and tries the next provider in the cascade, without waiting for a timeout on every single call.

breakers = {p.name: CircuitBreaker(failure_threshold=5, reset_timeout=60) for p in CASCADE}

async def call_with_circuit_breaker(messages: list) -> str:
    for provider in CASCADE:
        breaker = breakers[provider.name]
        if not breaker.can_attempt():
            continue  # skip this provider, circuit is open
        try:
            result = await call_provider(provider, messages)
            breaker.record_success()
            return result
        except TRANSIENT_ERRORS as e:
            breaker.record_failure()
            print(f'{provider.name} failed ({breaker.failure_count}/{breaker.failure_threshold})')
    raise RuntimeError('All providers exhausted')

Detecting Slow Calls as Failures

A provider that responds in 30 seconds is almost as bad as one that is fully down, from a user experience perspective. Configure an aggressive timeout per provider call and treat timeout exceptions as failures in the circuit breaker. A 10-second timeout means the fallback kicks in quickly enough that the user sees only a brief delay, not a hung screen.

async def call_provider(provider: Provider, messages: list) -> str:
    client = openai.AsyncOpenAI(
        api_key=os.environ[provider.api_key_env],
        base_url=provider.base_url
    )
    try:
        resp = await asyncio.wait_for(
            client.chat.completions.create(model=provider.model, messages=messages),
            timeout=10.0  # fail fast, let circuit breaker count it
        )
        return resp.choices[0].message.content
    except asyncio.TimeoutError:
        raise openai.APITimeoutError('Provider timed out')

Provider Health Dashboard

Expose a /health/providers endpoint that shows the current circuit breaker state for each provider, including failure count, state (closed/open/half-open), and time until reset. This makes it easy to see at a glance which providers are healthy during an incident and helps you decide whether to manually force a reset or wait for automatic recovery.

from fastapi import FastAPI

app = FastAPI()

@app.get('/health/providers')
def provider_health():
    return {
        name: {
            'state': cb.state.value,
            'failure_count': cb.failure_count,
            'seconds_until_reset': (
                max(0, cb.reset_timeout - (time.time() - cb.opened_at))
                if cb.state == State.OPEN else None
            )
        }
        for name, cb in breakers.items()
    }

Aligning Provider Outputs

Different providers have different response formats, safety filters, and capabilities. When switching from GPT-4o to Claude, the model might refuse certain requests that GPT-4o would answer. Maintain provider-specific prompt wrappers that adapt your prompts to each provider's conventions. Test each fallback provider independently to ensure it produces acceptable output for your use case.

def adapt_messages_for_provider(provider: Provider, messages: list) -> list:
    if provider.name == 'anthropic':
        # Claude prefers explicit task descriptions
        system = next((m['content'] for m in messages if m['role'] == 'system'), '')
        if 'JSON' not in system:
            messages = [{'role': 'system', 'content': system + ' Respond in JSON.'}] + [
                m for m in messages if m['role'] != 'system'
            ]
    return messages

Testing Fallback Behavior

Write a test that forces the primary provider to fail (by providing a bad API key or a mock that throws errors) and verifies the fallback kicks in and returns a valid response. Also test that the circuit breaker opens correctly after the configured number of failures and that it recovers after the reset timeout. Fallback logic that is never tested is unreliable in a real outage.

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_fallback_on_primary_timeout():
    # Primary provider times out
    with patch('your_module.call_provider', side_effect=[
        openai.APITimeoutError('Timeout'),  # primary fails
        'Claude response'                   # fallback succeeds
    ]):
        result = await call_with_circuit_breaker([{'role': 'user', 'content': 'Hello'}])
    assert result == 'Claude response'

Provider Cascade Cost Considerations

Fallback providers often have different pricing than your primary. Anthropic Claude may cost more or less than OpenAI GPT-4o depending on model tier. Track which provider served each request and compute cost attribution separately. If the fallback is consistently more expensive, investigate whether the primary is under-provisioned and whether upgrading to a higher rate-limit tier would be more cost-effective than frequent fallback usage.

# Approximate costs per 1M tokens (2026):
PROVIDER_COSTS = {
    'openai/gpt-4o':          {'input': 2.50, 'output': 10.00},
    'anthropic/claude-3.5-sonnet': {'input': 3.00, 'output': 15.00},
    'openai/gpt-4o-mini':     {'input': 0.15, 'output': 0.60},
    'local/llama-3.1-8b':     {'input': 0.00, 'output': 0.00},  # infra cost only
}

# If fallback adds $0.50/day and a Tier 2 upgrade costs $100/month:
# Tier 2 pays off if you use fallback > 200 requests/day

Quick Check

Test your understanding of circuit breakers and fallback providers.

Lesson Recap

In this lesson you learned: provider cascades define an ordered fallback sequence from primary to backup LLM providers, circuit breakers prevent hammering a failing provider by quickly short-circuiting after a threshold of failures, and per-provider timeouts ensure slow calls trigger fallback quickly rather than blocking users. Next up we set timeout budgets and implement graceful degradation.

Domande Frequenti

La lezione «Provider di fallback e circuit breaker» è gratuita?

Sì — il testo completo di «Provider di fallback e circuit breaker» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Engineering Academy, passa a CoddyKit PRO. Il corso AI Engineering Academy include 4 lezioni in totale.

Cosa imparerò in «Provider di fallback e circuit breaker»?

Costruisca una cascata di provider che esegua automaticamente il failover da OpenAI ad Anthropic e quindi a un modello locale quando il provider primario è lento o non disponibile, usando il pattern… Eserciti AI Engineering Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Engineering Academy?

Non è richiesta alcuna esperienza precedente. AI Engineering Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.

Quanto tempo richiede la lezione «Provider di fallback e circuit breaker»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Engineering Academy?

Sì. Ogni lezione AI Engineering Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Misurare la latenza degli LLM: TTFT e TPOT
  2. Bilanciamento del carico e strategie multi-key
  3. Provider di fallback e circuit breaker
  4. Budget di timeout e degrado controllato
← Torna a AI Engineering Academy