AI Engineering Academy · Les

Fallbackproviders en circuit breakers

Les 3 van 413 stappen

Fallbackproviders en circuit breakers is een gratis AI Engineering Academy-les op CoddyKit. Dit is les 3 van 4. Je kunt de volledige les hieronder gratis lezen en daarna in de browser praktisch oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is. Deze les maakt deel uit van het leertraject AI Engineering Academy. Je voortgang wordt gesynchroniseerd op het web en in de CoddyKit-app. De cursus AI Engineering Academy bevat in totaal 4 lessen.

Risico van één provider

Als je op één LLM-provider vertrouwt, creëer je één storingspunt. OpenAI heeft storingen meegemaakt die minuten tot uren duurden om op te lossen. Als je volledige toepassing ervan afhankelijk is dat GPT-4o beschikbaar is, vertaalt elk incident bij de provider zich onmiddellijk in uitvaltijd voor gebruikers. Een strategie met een terugvalprovider houdt de dienstverlening beschikbaar door verzoeken naar alternatieve providers te leiden wanneer de primaire provider uitvalt.

Een providerketen definiëren

Een providerketen is een geordende lijst van providers en modellen die achtereenvolgens worden geprobeerd. Wanneer de primaire provider uitvalt of een time-out optreedt, probeert het systeem automatisch de volgende provider. Een gebruikelijke keten kan er als volgt uitzien: OpenAI GPT-4o → Anthropic Claude 3.5 Sonnet → een lokaal geïmplementeerd Llama-model. Elk niveau is een terugvalmogelijkheid, waarbij het lokale model als laatste redmiddel dient en niet kan uitvallen.

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),
]

De terugvallus implementeren

Implementeer de terugvallus als een eenvoudige try/except die de keten doorloopt. Vang tijdelijke fouten op (time-outs, 500-fouten, 503-fouten) en ga verder met de volgende provider. Vang geen authenticatiefouten (401) of ongeldige-verzoekfouten (400) op: dit zijn programmeerfouten die onmiddellijk zichtbaar moeten worden in plaats van dat het verzoek naar een andere provider wordt doorgestuurd.

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')

Wat is een stroomonderbreker?

Een stroomonderbreker voorkomt dat een falende service tijdens een storing wordt overspoeld met verzoeken. De naam is ontleend aan elektrische stroomonderbrekers. Het mechanisme heeft drie toestanden: Closed (verzoeken worden normaal doorgelaten), Open (verzoeken worden onmiddellijk afgewezen) en Half-Open (één testverzoek wordt doorgelaten om te controleren of de service is hersteld). Zo bescherm je zowel de achterliggende service als je eigen toepassing tijdens incidenten.

# 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

Een stroomonderbreker implementeren

Hier volgt een minimale implementatie van een stroomonderbreker. Houd het aantal fouten en het tijdstip waarop het circuit is geopend bij. Open het circuit wanneer het aantal fouten de drempel overschrijdt. Laat na een instelbare hersteltime-out één proefverzoek toe. Sluit het circuit als het proefverzoek slaagt. Houd het circuit open en stel de time-out opnieuw in als het mislukt.

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

Stroomonderbrekers met providers integreren

Beheer één stroomonderbreker per provider. Controleer voordat je een provider aanroept of de stroomonderbreker de poging toestaat. Registreer na elke aanroep of deze is geslaagd of mislukt. Wanneer het circuit van een provider opent, slaat de terugvallus deze automatisch over en probeert hij de volgende provider in de keten, zonder bij elke afzonderlijke aanroep op een time-out te wachten.

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')

Trage aanroepen als fouten herkennen

Een provider die na 30 seconden antwoordt, is vanuit het oogpunt van gebruikerservaring bijna net zo slecht als een provider die volledig buiten werking is. Configureer een agressieve time-out per aanroep van een provider en behandel time-outuitzonderingen als fouten in de stroomonderbreker. Met een time-out van 10 seconden wordt de terugval snel genoeg geactiveerd, zodat de gebruiker slechts een korte vertraging ziet en geen vastgelopen scherm.

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')

Gezondheidsdashboard voor providers

Stel een /health/providers-eindpunt beschikbaar dat de huidige toestand van de stroomonderbreker voor elke provider toont, inclusief het aantal fouten, de toestand (closed/open/half-open) en de resterende tijd tot herstel. Zo zie je tijdens een incident in één oogopslag welke providers gezond zijn en kun je bepalen of je handmatig een herstel moet forceren of op automatisch herstel moet wachten.

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()
    }

Uitvoer van providers op elkaar afstemmen

Verschillende providers hebben verschillende antwoordindelingen, veiligheidsfilters en mogelijkheden. Wanneer je van GPT-4o naar Claude overschakelt, kan het model bepaalde verzoeken weigeren die GPT-4o wel zou beantwoorden. Beheer providerspecifieke promptomhulsels die je prompts aanpassen aan de conventies van elke provider. Test elke terugvalprovider afzonderlijk om te controleren of deze voor jouw gebruikssituatie aanvaardbare uitvoer produceert.

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

Ter обvalgedrag testen

Schrijf een test die de primaire provider expres laat mislukken (door een ongeldige API-sleutel op te geven of een mock te gebruiken die fouten genereert) en controleert of de terugval wordt geactiveerd en een geldig antwoord retourneert. Test ook of de stroomonderbreker correct opent na het geconfigureerde aantal fouten en of deze na de hersteltime-out weer herstelt. Terugvallogica die nooit wordt getest, is bij een echte storing onbetrouwbaar.

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'

Kostenoverwegingen bij een providercascade

Fallbackproviders hanteren vaak andere prijzen dan uw primaire provider. Anthropic Claude kan, afhankelijk van de modelklasse, meer of minder kosten dan OpenAI GPT-4o. Houd bij welke provider elk verzoek heeft afgehandeld en bereken de kosten afzonderlijk per provider. Als de fallback consequent duurder is, onderzoek dan of de primaire provider onvoldoende capaciteit heeft en of een upgrade naar een hogere limietklasse kosteneffectiever zou zijn dan veelvuldig gebruik van de fallback.

# 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

Korte controle

Toets uw begrip van circuitbreakers en fallbackproviders.

Beantwoord dit in de interactieve les

Samenvatting van de les

In deze les hebt u geleerd dat providercascades een geordende fallbackvolgorde definiëren van primaire naar back-up-LLM-providers, dat circuitbreakers voorkomen dat een falende provider voortdurend wordt aangeroepen door na een drempelwaarde aan fouten snel kort te sluiten, en dat time-outs per provider ervoor zorgen dat trage aanroepen snel een fallback activeren in plaats van gebruikers te blokkeren. Hierna stellen we time-outbudgetten in en implementeren we gecontroleerde kwaliteitsvermindering.

Gratis beginnen

Leer Python met een AI-tutor — gratis

Schrijf echte code en voer die uit in je browser, krijg direct hulp van een AI-tutor die 24/7 beschikbaar is en ga verder waar je gebleven bent op het web of in de app.

Cursussen: 30
Lessen: 120

Gratis beginnen met leren Bekijk de volledige cursus

Veelgestelde vragen

Is de les “Fallbackproviders en circuit breakers” gratis?

Ja — de volledige tekst van “Fallbackproviders en circuit breakers” kun je hier gratis op het web lezen. Als je interactief wilt oefenen met een ingebouwde code-editor en een AI-begeleider die 24/7 beschikbaar is, en de rest van de cursus AI Engineering Academy wilt ontgrendelen, kun je upgraden naar CoddyKit PRO. De cursus AI Engineering Academy bevat in totaal 4 lessen.

Wat leer ik in “Fallbackproviders en circuit breakers”?

Bouw een providercascade die automatisch overschakelt van OpenAI naar Anthropic en vervolgens naar een lokaal model wanneer de primaire provider traag of niet beschikbaar is, met gebruik van het circ… Je oefent met AI Engineering Academy door code rechtstreeks in de browser uit te voeren. Een AI-begeleider die 24/7 beschikbaar is beantwoordt je vragen terwijl je de les doorwerkt.

Heb ik ervaring nodig om met AI Engineering Academy te beginnen?

Ervaring vooraf is niet nodig. AI Engineering Academy op CoddyKit is opgebouwd voor beginners tot gevorderden, zodat je hier of bij het begin kunt starten en in je eigen tempo kunt leren. Dit is les 3 van 4.

Hoe lang duurt de les “Fallbackproviders en circuit breakers”?

De meeste lessen van CoddyKit duren ongeveer 5–10 minuten. Elke les is kort en interactief, zodat je gestaag vooruitgaat en op het web en in de app precies verdergaat waar je was gebleven.

Kan ik code schrijven en uitvoeren in deze les over AI Engineering Academy?

Ja. Elke les over AI Engineering Academy bevat een ingebouwde code-editor, zodat je rechtstreeks in je browser echte code kunt schrijven en uitvoeren en direct feedback van AI krijgt — lokale installatie is niet nodig.

Alle lessen in deze cursus

LLM-latentie meten: TTFT en TPOT
Load balancing en strategieën met meerdere sleutels
Fallbackproviders en circuit breakers
Time-outbudgetten en geleidelijke degradatie

← Terug naar AI Engineering Academy