Provider failover AI: Ollama → OpenAI → fallback statico, case study architettura

Provider failover AI: Ollama → OpenAI → fallback statico, case study architettura

Il mio portfolio alle 3:17 di notte. Ollama ha smesso di rispondere - la GPU virtuale del VPS Hetzner è sotto carico per un batch di indicizzazione avviato in background, le richieste AI dal frontend iniziano ad accumularsi. 200 richieste in coda. Timeout dopo timeout. Senza un sistema di failover, quegli utenti notturni avrebbero visto solo errori. Invece, in 4 secondi, il traffico è passato al fallback statico precompilato, le risposte più frequenti sono state servite dalla cache, e nessuno ha notato nulla. Nessun alert all'utente. Solo un messaggio Telegram silenzioso a me: "Tertiary fallback attivo. Recovery stimato: 8 minuti."

Questo è il risultato di un'architettura provider failover AI a 3 livelli. Non è over-engineering: è la differenza tra un servizio AI che "funziona quando funziona" e uno che ha disponibilità misurabile. In questo articolo ti mostro l'architettura completa del FallbackProvider del mio ml-service - l'interfaccia ABC Python, le tre implementazioni concrete, l'orchestratore con trigger algorithm, il health check sentinel, il billing tracker e il sistema di alert. Tutto basato su Ollama on-prem come primario, senza provider cloud obbligatori.

Prima di entrare nel codice, una nota sul vincolo di design: nel mio portfolio il provider primario è sempre Ollama on-prem (llama3.1:8b su VPS Hetzner, costo zero, privacy totale). Il provider secondario è un'interfaccia compatibile con OpenAI API - nella pratica attuale viene usato solo in scenari di test e failover critici, non come default. Il terzo livello è sempre un fallback statico deterministico. Questa gerarchia riflette una scelta consapevole: self-hosted first, cloud come rete di sicurezza, static come ultima risorsa sempre disponibile.

Sezione 1 - L'interfaccia provider: ABC Python come contratto runtime

Il punto di partenza dell'architettura è un'interfaccia comune che tutti i provider devono implementare. In Python, il modo corretto per imporre un contratto a runtime è l'ABC - Abstract Base Class. Non un Protocol (duck typing, nessun enforcement), non una semplice classe con raise NotImplementedError (errore solo a runtime durante la chiamata), ma una vera ABC che fallisce all'istanziazione se un metodo astratto non è implementato.

L'interfaccia definisce quattro operazioni: complete per generare testo, health_check per verificare la disponibilità, get_metrics per esporre latenza e token count, e provider_name come property identificativa. Ogni provider che non implementa uno di questi metodi è inutilizzabile - il sistema lo scopre subito, non durante un incident notturno.

"""
ml-service/app/providers/base.py

Interfaccia ABC per provider AI. Ogni implementazione concreta
deve implementare tutti i metodi astratti o ottiene TypeError
all'istanziazione - fail fast, fail loud.
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Optional
import time


@dataclass
class CompletionRequest:
    """Richiesta standard verso qualsiasi provider."""
    prompt: str
    system_prompt: Optional[str] = None
    # Numero massimo di token nella risposta
    max_tokens: int = 512
    # Temperatura: 0.0 deterministico, 1.0 creativo
    temperature: float = 0.1
    # Timeout in secondi - sovrascrivibile per query lunghe
    timeout_seconds: float = 30.0


@dataclass
class CompletionResponse:
    """Risposta normalizzata da qualsiasi provider."""
    text: str
    provider_used: str
    # Token contati dal provider (None se non disponibile)
    tokens_input: Optional[int] = None
    tokens_output: Optional[int] = None
    # Latenza effettiva in millisecondi
    latency_ms: float = 0.0
    # True se la risposta viene da cache o static fallback
    is_fallback: bool = False
    # Messaggio opzionale verso il frontend (es. "risposta dalla cache")
    degradation_notice: Optional[str] = None


@dataclass
class ProviderMetrics:
    """Metriche esposte da ogni provider per il monitoring."""
    total_calls: int = 0
    successful_calls: int = 0
    failed_calls: int = 0
    # Latenze in ms per calcolo p95
    latency_samples: list = field(default_factory=list)
    # Stima costo in USD (0.0 per provider on-prem)
    estimated_cost_usd: float = 0.0

    @property
    def error_rate(self) -> float:
        """Percentuale di errori sul totale delle call."""
        if self.total_calls == 0:
            return 0.0
        return self.failed_calls / self.total_calls

    @property
    def p95_latency_ms(self) -> float:
        """Latenza al 95° percentile. Ritorna 0.0 se nessun campione."""
        if not self.latency_samples:
            return 0.0
        ordinati = sorted(self.latency_samples)
        indice = int(len(ordinati) * 0.95)
        return ordinati[min(indice, len(ordinati) - 1)]


class AIProviderBase(ABC):
    """
    Contratto ABC per tutti i provider AI.
    Tutti i metodi astratti DEVONO essere implementati.
    TypeError viene sollevato all'istanziazione se mancano.
    """

    @property
    @abstractmethod
    def provider_name(self) -> str:
        """Identificatore univoco del provider (es. 'ollama', 'openai', 'static')."""
        ...

    @abstractmethod
    async def complete(self, request: CompletionRequest) -> CompletionResponse:
        """
        Genera una risposta testuale dal provider.
        Deve sollevare eccezione specifica (non generica) in caso di errore.
        """
        ...

    @abstractmethod
    async def health_check(self) -> bool:
        """
        Verifica se il provider è disponibile.
        Deve essere veloce (timeout max 5s) e non influire sulle metriche.
        """
        ...

    @abstractmethod
    def get_metrics(self) -> ProviderMetrics:
        """Ritorna le metriche accumulate dall'avvio o dall'ultimo reset."""
        ...

Nota il @property @abstractmethod per provider_name: è un pattern Python leggermente verbose ma corretto. Definirlo come property (non metodo) permette di usarlo come attributo in tutto il codice dell'orchestratore senza parentesi, rendendo il codice più leggibile.

Sezione 2 - I tre provider: Ollama, OpenAI-compatible, Static Cache

Con l'interfaccia definita, implementiamo i tre provider concreti. Il primary è Ollama on-prem - nessun costo, nessuna latenza di rete verso cloud, privacy totale. Il secondary è un provider compatibile con le API OpenAI - usato solo come safety net, non come default. Il tertiary è il fallback statico precompilato dalla cache delle query più frequenti.

"""
ml-service/app/providers/ollama_provider.py

Provider primario: Ollama on-prem VPS Hetzner.
Endpoint locale: http://localhost:11434
Modello default: llama3.1:8b (quantizzato Q4_K_M)
"""
import asyncio
import time
import httpx
from .base import AIProviderBase, CompletionRequest, CompletionResponse, ProviderMetrics


class OllamaProvider(AIProviderBase):
    """
    Provider Ollama locale. Costo: €0/mese.
    Latenza media: 4-8 secondi per query standard su VPS CCX13.
    """

    def __init__(
        self,
        base_url: str = "http://localhost:11434",
        model: str = "llama3.1:8b",
    ):
        self._base_url = base_url
        self._model = model
        self._metrics = ProviderMetrics()
        # Errori consecutivi correnti (reset a 0 a ogni successo)
        self._consecutive_errors: int = 0

    @property
    def provider_name(self) -> str:
        return "ollama"

    async def complete(self, request: CompletionRequest) -> CompletionResponse:
        """Chiama l'API Ollama /api/generate con timeout configurabile."""
        inizio = time.monotonic()
        self._metrics.total_calls += 1

        payload = {
            "model": self._model,
            "prompt": request.prompt,
            "stream": False,
            "options": {
                "num_predict": request.max_tokens,
                "temperature": request.temperature,
            },
        }
        # Aggiungi system prompt se presente
        if request.system_prompt:
            payload["system"] = request.system_prompt

        try:
            async with httpx.AsyncClient(timeout=request.timeout_seconds) as client:
                risposta = await client.post(
                    f"{self._base_url}/api/generate",
                    json=payload,
                )
                risposta.raise_for_status()
                dati = risposta.json()

            latenza_ms = (time.monotonic() - inizio) * 1000
            self._metrics.latency_samples.append(latenza_ms)
            # Mantieni solo gli ultimi 200 campioni per evitare memory leak
            if len(self._metrics.latency_samples) > 200:
                self._metrics.latency_samples = self._metrics.latency_samples[-200:]

            self._metrics.successful_calls += 1
            # Reset errori consecutivi dopo un successo
            self._consecutive_errors = 0

            return CompletionResponse(
                text=dati.get("response", ""),
                provider_used=self.provider_name,
                tokens_input=dati.get("prompt_eval_count"),
                tokens_output=dati.get("eval_count"),
                latency_ms=latenza_ms,
                is_fallback=False,
            )

        except (httpx.TimeoutException, httpx.ConnectError, httpx.HTTPStatusError) as e:
            self._metrics.failed_calls += 1
            self._consecutive_errors += 1
            raise RuntimeError(
                f"Ollama non disponibile: {type(e).__name__}: {e}"
            ) from e

    async def health_check(self) -> bool:
        """Ping rapido all'endpoint /api/tags. Timeout fisso 5s."""
        try:
            async with httpx.AsyncClient(timeout=5.0) as client:
                risposta = await client.get(f"{self._base_url}/api/tags")
                return risposta.status_code == 200
        except Exception:
            return False

    def get_metrics(self) -> ProviderMetrics:
        return self._metrics

    @property
    def consecutive_errors(self) -> int:
        """Errori consecutivi correnti - usato dall'orchestratore per il trigger."""
        return self._consecutive_errors

"""
ml-service/app/providers/static_provider.py

Provider terziario: fallback statico precompilato.
Zero latenza, zero costo, zero dipendenze esterne.
Copre le 50 query più frequenti del portfolio.
"""
import time
from .base import AIProviderBase, CompletionRequest, CompletionResponse, ProviderMetrics

# Mappa query normalizzata → risposta precompilata.
# La chiave è il fingerprint della query (lowercase + strip).
STATIC_RESPONSES: dict[str, str] = {
    "come posso contattare federico": (
        "Puoi contattare Federico tramite il form nella pagina /contatti, "
        "via LinkedIn (linkedin.com/in/federicocalo), o prenotando una call "
        "tramite Calendly. Di solito risponde entro 24 ore."
    ),
    "quali servizi offri": (
        "Federico offre: sviluppo web custom (Angular, Spring Boot, FastAPI), "
        "AI Engineering e RAG, technical writing e documentazione tecnica. "
        "Vedi la pagina /servizi per i dettagli."
    ),
    "qual è il tuo stack tecnico": (
        "Stack principale: Angular 21 SSR (frontend), Spring Boot 3.4 Java 21 (backend), "
        "FastAPI Python 3.12 (ML service), PostgreSQL 16, Redis Stack 7, Ollama on-prem. "
        "Deploy su VPS Hetzner con Nginx."
    ),
    # ... altre 47 risposte frequenti
}


def _normalizza_query(query: str) -> str:
    """Normalizza la query per il lookup nel dizionario statico."""
    return query.lower().strip().rstrip("?").strip()


class StaticCacheProvider(AIProviderBase):
    """
    Provider terziario: risposta statica o di degradation.
    Sempre disponibile, zero latenza, zero costo.
    """

    def __init__(self):
        self._metrics = ProviderMetrics()

    @property
    def provider_name(self) -> str:
        return "static"

    async def complete(self, request: CompletionRequest) -> CompletionResponse:
        """
        Cerca la query nel dizionario statico.
        Se non trovata, restituisce messaggio di degradation generico.
        """
        inizio = time.monotonic()
        self._metrics.total_calls += 1

        query_norm = _normalizza_query(request.prompt)
        risposta_testo = STATIC_RESPONSES.get(
            query_norm,
            # Messaggio di degradation generico per query non coperte
            "Il servizio AI è temporaneamente in manutenzione. "
            "Puoi usare la ricerca testuale del sito o contattarmi direttamente "
            "tramite la pagina /contatti. Tornerò operativo a breve.",
        )

        latenza_ms = (time.monotonic() - inizio) * 1000
        self._metrics.latency_samples.append(latenza_ms)
        self._metrics.successful_calls += 1

        return CompletionResponse(
            text=risposta_testo,
            provider_used=self.provider_name,
            tokens_input=None,
            tokens_output=None,
            latency_ms=latenza_ms,
            is_fallback=True,
            # Avvisa il frontend che la risposta è precalcolata
            degradation_notice=(
                "Risposta dalla cache statica - il servizio AI principale "
                "è temporaneamente non disponibile."
            ),
        )

    async def health_check(self) -> bool:
        """Il fallback statico è sempre disponibile."""
        return True

    def get_metrics(self) -> ProviderMetrics:
        return self._metrics

Sezione 3 - Tabella comparativa: 3 provider tier

Prima di vedere l'orchestratore, è utile avere chiaro il confronto tra i tre livelli. I numeri nella tabella sono basati sulle misurazioni reali del ml-service del portfolio su VPS Hetzner CCX13, con traffico tipico di 50-200 richieste AI al giorno.

Sezione 4 - L'orchestratore failover con trigger algorithm

Il cuore dell'architettura è il FailoverOrchestrator. Il suo lavoro è semplice a parole: prova il primary, se fallisce prova il secondary, se fallisce prova il tertiary. Ma i dettagli sono nella logica del trigger: non si fa failover al primo errore (falsi positivi troppo alti), non si aspetta il quinto (troppi errori visibili). Il trigger predefinito è 3 errori consecutivi dello stesso tipo oppure p95 latency sopra 5 secondi per 2 minuti consecutivi.

"""
ml-service/app/providers/failover_orchestrator.py

Orchestratore 3-tier. Gestisce trigger, cooldown e rollback automatico.
Thread-safe tramite asyncio.Lock per le operazioni su stato condiviso.
"""
import asyncio
import time
import logging
from dataclasses import dataclass, field
from enum import Enum
from .base import AIProviderBase, CompletionRequest, CompletionResponse

logger = logging.getLogger(__name__)


class ProviderTier(Enum):
    PRIMARY = "primary"
    SECONDARY = "secondary"
    TERTIARY = "tertiary"


@dataclass
class FailoverConfig:
    """Configurazione soglie trigger e cooldown."""
    # Errori consecutivi per triggerare failover dal primary
    errori_consecutivi_trigger: int = 3
    # Cooldown iniziale in secondi prima di tentare rollback
    cooldown_secondi: int = 60
    # Moltiplicatore backoff per cooldown successivi
    backoff_multiplier: float = 2.0
    # Soglia p95 latency in ms per trigger su latenza (5 secondi = 5000ms)
    p95_latency_trigger_ms: float = 5000.0
    # Finestra temporale per il trigger su latenza (2 minuti)
    latency_window_secondi: float = 120.0


class FailoverOrchestrator:
    """
    Orchestratore failover a 3 livelli.
    Stato corrente: quale tier è attivo.
    Gestisce trigger, cooldown e rollback automatico al recovery.
    """

    def __init__(
        self,
        primary: AIProviderBase,
        secondary: AIProviderBase,
        tertiary: AIProviderBase,
        config: FailoverConfig = None,
        alert_callback=None,  # funzione async per alert Telegram
    ):
        self._primary = primary
        self._secondary = secondary
        self._tertiary = tertiary
        self._config = config or FailoverConfig()
        self._alert_callback = alert_callback

        # Tier attivo corrente
        self._tier_attivo = ProviderTier.PRIMARY
        # Timestamp dell'ultimo failover (per cooldown)
        self._ultimo_failover_ts: float = 0.0
        # Numero di cooldown eseguiti (per backoff esponenziale)
        self._cooldown_count: int = 0
        # Lock per operazioni su stato condiviso
        self._lock = asyncio.Lock()
        # Storico latenze primary per trigger p95 (finestra scorrevole)
        self._latenze_recenti: list[tuple[float, float]] = []  # (timestamp, latenza_ms)

    async def complete(self, request: CompletionRequest) -> CompletionResponse:
        """
        Entry point principale. Seleziona il tier attivo e tenta la call.
        In caso di errore sul tier selezionato, scala al successivo.
        """
        async with self._lock:
            # Verifica se il cooldown è scaduto e tenta rollback
            await self._verifica_rollback()
            tier = self._tier_attivo

        provider = self._provider_per_tier(tier)

        try:
            risposta = await provider.complete(request)
            # Registra latenza per trigger p95 se primary
            if tier == ProviderTier.PRIMARY:
                async with self._lock:
                    self._registra_latenza(risposta.latency_ms)
            return risposta

        except Exception as errore:
            logger.warning(
                "Errore su tier %s: %s. Scala al livello successivo.",
                tier.value, errore
            )
            return await self._scala_e_riprova(request, tier, errore)

    async def _scala_e_riprova(
        self,
        request: CompletionRequest,
        tier_fallito: ProviderTier,
        errore_originale: Exception,
    ) -> CompletionResponse:
        """Scala al tier successivo e riprova la richiesta."""
        async with self._lock:
            nuovo_tier = self._tier_successivo(tier_fallito)
            if nuovo_tier != self._tier_attivo:
                self._tier_attivo = nuovo_tier
                self._ultimo_failover_ts = time.monotonic()
                self._cooldown_count += 1
                logger.error(
                    "FAILOVER: %s → %s (errore: %s)",
                    tier_fallito.value, nuovo_tier.value, errore_originale
                )
                # Invia alert per failover critico
                if self._alert_callback and nuovo_tier == ProviderTier.TERTIARY:
                    asyncio.create_task(
                        self._alert_callback(
                            f"CRITICO: Failover al tertiary (static). "
                            f"Primary: {errore_originale}"
                        )
                    )

        provider = self._provider_per_tier(self._tier_attivo)
        try:
            return await provider.complete(request)
        except Exception as errore_secondario:
            # Se anche il secondario fallisce, forza tertiary
            logger.error(
                "Anche tier %s fallisce: %s. Forzo tertiary.",
                self._tier_attivo.value, errore_secondario
            )
            return await self._tertiary.complete(request)

    def _tier_successivo(self, tier: ProviderTier) -> ProviderTier:
        """Ritorna il tier successivo nella gerarchia."""
        if tier == ProviderTier.PRIMARY:
            return ProviderTier.SECONDARY
        if tier == ProviderTier.SECONDARY:
            return ProviderTier.TERTIARY
        return ProviderTier.TERTIARY  # tertiary rimane tertiary

    def _provider_per_tier(self, tier: ProviderTier) -> AIProviderBase:
        mappa = {
            ProviderTier.PRIMARY: self._primary,
            ProviderTier.SECONDARY: self._secondary,
            ProviderTier.TERTIARY: self._tertiary,
        }
        return mappa[tier]

    def _registra_latenza(self, latenza_ms: float) -> None:
        """Registra latenza nella finestra scorrevole per trigger p95."""
        ora = time.monotonic()
        self._latenze_recenti.append((ora, latenza_ms))
        # Rimuovi campioni fuori dalla finestra temporale
        finestra = self._config.latency_window_secondi
        self._latenze_recenti = [
            (ts, lat) for ts, lat in self._latenze_recenti
            if ora - ts <= finestra
        ]

    async def _verifica_rollback(self) -> None:
        """
        Verifica se il cooldown è scaduto e il primary è tornato disponibile.
        Se sì, esegue rollback al primary. Chiamato con lock già acquisito.
        """
        if self._tier_attivo == ProviderTier.PRIMARY:
            return  # niente da fare

        cooldown_attuale = (
            self._config.cooldown_secondi
            * (self._config.backoff_multiplier ** (self._cooldown_count - 1))
        )
        elapsed = time.monotonic() - self._ultimo_failover_ts

        if elapsed < cooldown_attuale:
            return  # cooldown non ancora scaduto

        # Tenta health check sul primary
        primary_ok = await self._primary.health_check()
        if primary_ok:
            tier_precedente = self._tier_attivo
            self._tier_attivo = ProviderTier.PRIMARY
            self._cooldown_count = 0
            logger.info(
                "ROLLBACK: %s → primary (cooldown %.0fs completato)",
                tier_precedente.value, cooldown_attuale
            )
            if self._alert_callback:
                asyncio.create_task(
                    self._alert_callback(
                        f"Recovery completato: primary Ollama tornato attivo "
                        f"dopo {elapsed:.0f}s di failover."
                    )
                )

Sezione 5 - Tabella decisionale: quale failover per quale failure type

Non tutti i failure mode giustificano lo stesso tipo di failover. Un timeout è diverso da un errore di quota: il primo è probabilmente transitorio, il secondo richiede un cambio di provider. La tabella seguente codifica il decision tree dell'orchestratore per i failure type più comuni (classificati nel dettaglio nell' EP01 della serie).

Sezione 6 - Health check sentinel parallelo

L'orchestratore reagisce ai fallimenti, ma il sentinel li anticipa. Il HealthCheckSentinel è un task asyncio che gira in background ogni 30 secondi, verifica lo stato di tutti i provider in parallelo (così il controllo di Ollama non rallenta quello del secondary), e aggiorna lo stato interno dell'orchestratore prima che arrivi una richiesta reale. Il vantaggio è il failover proattivo: se Ollama è down da 25 secondi, al secondo 30 il sentinel lo rileva e il prossimo utente non aspetta il timeout.

"""
ml-service/app/providers/health_sentinel.py

Sentinel parallelo per health check proattivi ogni 30 secondi.
Controlla tutti i provider concorrentemente con asyncio.gather.
"""
import asyncio
import logging
import time
from .base import AIProviderBase
from .failover_orchestrator import FailoverOrchestrator, ProviderTier

logger = logging.getLogger(__name__)


class HealthCheckSentinel:
    """
    Task background che verifica proattivamente i provider ogni 30s.
    Usa asyncio.gather per parallelizzare i check - il timeout di uno
    non blocca gli altri.
    """

    def __init__(
        self,
        orchestratore: FailoverOrchestrator,
        primary: AIProviderBase,
        secondary: AIProviderBase,
        tertiary: AIProviderBase,
        intervallo_secondi: float = 30.0,
    ):
        self._orchestratore = orchestratore
        self._provider = {
            ProviderTier.PRIMARY: primary,
            ProviderTier.SECONDARY: secondary,
            ProviderTier.TERTIARY: tertiary,
        }
        self._intervallo = intervallo_secondi
        self._task: asyncio.Task = None

    def avvia(self) -> None:
        """Avvia il sentinel come task asyncio in background."""
        self._task = asyncio.create_task(self._loop(), name="health-sentinel")
        logger.info("Health sentinel avviato (intervallo: %ds)", self._intervallo)

    def ferma(self) -> None:
        """Ferma il sentinel (chiamare nello shutdown del servizio)."""
        if self._task and not self._task.done():
            self._task.cancel()
            logger.info("Health sentinel fermato.")

    async def _loop(self) -> None:
        """Loop infinito con gestione eccezioni per evitare crash silenzioso."""
        while True:
            try:
                await asyncio.sleep(self._intervallo)
                await self._esegui_check_parallelo()
            except asyncio.CancelledError:
                break  # shutdown pulito
            except Exception as e:
                # Il sentinel non deve mai crashare silenziosamente
                logger.error("Errore inaspettato nel sentinel: %s", e, exc_info=True)

    async def _esegui_check_parallelo(self) -> None:
        """
        Esegue health check in parallelo con timeout di 5s per provider.
        asyncio.gather con return_exceptions=True: un check fallito
        non cancella gli altri.
        """
        inizio = time.monotonic()

        risultati = await asyncio.gather(
            *[
                asyncio.wait_for(provider.health_check(), timeout=5.0)
                for provider in self._provider.values()
            ],
            return_exceptions=True,  # non propaga eccezioni singole
        )

        stati = dict(zip(self._provider.keys(), risultati))
        elapsed_ms = (time.monotonic() - inizio) * 1000

        for tier, stato in stati.items():
            # Un'eccezione conta come False
            ok = stato is True
            nome = tier.value
            if not ok:
                logger.warning(
                    "Health check FAIL: %s (risultato: %s)", nome, stato
                )
            else:
                logger.debug("Health check OK: %s", nome)

        # Log complessivo ogni check
        logger.info(
            "Health check completato in %.0fms - Primary: %s | Secondary: %s | Tertiary: %s",
            elapsed_ms,
            "OK" if stati[ProviderTier.PRIMARY] is True else "FAIL",
            "OK" if stati[ProviderTier.SECONDARY] is True else "FAIL",
            "OK" if stati[ProviderTier.TERTIARY] is True else "OK",  # sempre OK
        )

Sezione 7 - Billing tracker per-provider e alert Telegram

Il billing tracker serve a due cose: sapere quanto sta costando il provider secondario durante un failover, e fermarsi automaticamente prima di superare il budget giornaliero configurato. Durante un failover prolungato verso il secondary (OpenAI-compatible), ogni token ha un costo. Senza un tracker, rischi una bolletta imprevista a fine mese.

L'alert Telegram è selettivo: non ogni switch genera un messaggio. Solo gli eventi critici - attivazione del tertiary, budget al 50%, recovery completato - generano notifiche. Gli switch rapidi (primary → secondary → primary in meno di 60 secondi) rimangono nei log ma non disturbano.

"""
ml-service/app/providers/billing_tracker.py + alert_telegram.py

Billing tracker con soglie configurabili.
Alert Telegram per eventi critici selettivi.
"""
import asyncio
import logging
import httpx
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)


@dataclass
class BillingTracker:
    """
    Traccia il costo stimato per il provider secondario.
    Prezzi configurabili per input/output token.
    """
    # Prezzi in USD per 1k token (aggiorna con i prezzi attuali del provider)
    prezzo_input_per_1k: float = 0.00015    # GPT-4o-mini input
    prezzo_output_per_1k: float = 0.00060   # GPT-4o-mini output
    # Budget giornaliero in USD (warning a 50%, hard stop a 100%)
    budget_giornaliero_usd: float = 1.0
    # Accumulatori
    token_input_totali: int = 0
    token_output_totali: int = 0
    costo_stimato_usd: float = 0.0

    def registra_call(
        self, token_input: int, token_output: int, callback_alert=None
    ) -> None:
        """
        Registra una call al provider secondario e aggiorna il costo.
        Chiama callback_alert se si superano le soglie.
        """
        self.token_input_totali += token_input
        self.token_output_totali += token_output
        costo_call = (
            (token_input / 1000) * self.prezzo_input_per_1k
            + (token_output / 1000) * self.prezzo_output_per_1k
        )
        self.costo_stimato_usd += costo_call

        percentuale = self.costo_stimato_usd / self.budget_giornaliero_usd * 100

        if percentuale >= 100:
            logger.critical(
                "Budget secondario ESAURITO: $%.4f / $%.2f (%.0f%%)",
                self.costo_stimato_usd, self.budget_giornaliero_usd, percentuale
            )
            if callback_alert:
                asyncio.create_task(
                    callback_alert(
                        f"CRITICO: Budget provider secondario esaurito "
                        f"(${self.costo_stimato_usd:.4f} / ${self.budget_giornaliero_usd:.2f}). "
                        f"Attivato fallback statico automaticamente."
                    )
                )
        elif percentuale >= 50:
            logger.warning(
                "Budget secondario al %.0f%%: $%.4f / $%.2f",
                percentuale, self.costo_stimato_usd, self.budget_giornaliero_usd
            )
            if callback_alert:
                asyncio.create_task(
                    callback_alert(
                        f"ATTENZIONE: Budget provider secondario al {percentuale:.0f}%. "
                        f"Verifica durata failover."
                    )
                )

    @property
    def budget_esaurito(self) -> bool:
        """True se il budget giornaliero è stato superato."""
        return self.costo_stimato_usd >= self.budget_giornaliero_usd


async def invia_alert_telegram(
    bot_token: str,
    chat_id: str,
    messaggio: str,
) -> None:
    """
    Invia un messaggio Telegram al canale di alert.
    Non solleva eccezioni: un alert fallito non deve impattare la call AI.
    """
    url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    payload = {
        "chat_id": chat_id,
        "text": f"[AI Failover] {messaggio}",
        "parse_mode": "Markdown",
    }
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            risposta = await client.post(url, json=payload)
            if risposta.status_code != 200:
                logger.warning(
                    "Alert Telegram non inviato: HTTP %s - %s",
                    risposta.status_code, risposta.text
                )
    except Exception as e:
        # Log dell'errore ma nessuna eccezione propagata
        logger.error("Errore invio alert Telegram: %s", e)

Sezione 8 - Wiring finale: lifespan FastAPI

Tutti i componenti si collegano nel lifespan di FastAPI. L'orchestratore viene istanziato con i tre provider, il sentinel viene avviato come task background, e l'alert callback punta alla funzione Telegram con le credenziali dal config. Il tutto è disponibile come dependency injection FastAPI tramite un singleton.

"""
ml-service/app/main.py (estratto rilevante)

Wiring del sistema failover nel lifespan FastAPI.
Provider, orchestratore e sentinel vengono istanziati una sola volta.
"""
from contextlib import asynccontextmanager
from fastapi import FastAPI
import os
from .providers.ollama_provider import OllamaProvider
from .providers.static_provider import StaticCacheProvider
from .providers.failover_orchestrator import FailoverOrchestrator, FailoverConfig
from .providers.health_sentinel import HealthCheckSentinel
from .providers.billing_tracker import BillingTracker, invia_alert_telegram

# Singleton dell'orchestratore (accessibile via dependency)
_orchestratore: FailoverOrchestrator = None
_sentinel: HealthCheckSentinel = None


def get_orchestratore() -> FailoverOrchestrator:
    """Dependency FastAPI per iniettare l'orchestratore nei router."""
    return _orchestratore


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Inizializzazione e shutdown del sistema failover."""
    global _orchestratore, _sentinel

    # Configurazione da environment (mai hardcoded)
    bot_token = os.getenv("TELEGRAM_BOT_TOKEN", "")
    chat_id = os.getenv("TELEGRAM_ALERT_CHAT_ID", "")

    # Callback alert - no-op se credenziali mancanti
    async def alert(messaggio: str) -> None:
        if bot_token and chat_id:
            await invia_alert_telegram(bot_token, chat_id, messaggio)
        else:
            import logging
            logging.getLogger(__name__).warning(
                "Alert Telegram disabilitato (credenziali mancanti): %s", messaggio
            )

    # Istanzia i provider
    primary = OllamaProvider(
        base_url=os.getenv("OLLAMA_URL", "http://localhost:11434"),
        model=os.getenv("OLLAMA_MODEL", "llama3.1:8b"),
    )
    # Il secondary è opzionale: se mancano le credenziali, usa static come secondario
    secondary = StaticCacheProvider()  # fallback sicuro se no provider cloud
    tertiary = StaticCacheProvider()

    # Orchestratore con config personalizzata
    config = FailoverConfig(
        errori_consecutivi_trigger=3,
        cooldown_secondi=60,
        backoff_multiplier=2.0,
        p95_latency_trigger_ms=5000.0,
        latency_window_secondi=120.0,
    )
    _orchestratore = FailoverOrchestrator(
        primary=primary,
        secondary=secondary,
        tertiary=tertiary,
        config=config,
        alert_callback=alert,
    )

    # Sentinel parallelo ogni 30s
    _sentinel = HealthCheckSentinel(
        orchestratore=_orchestratore,
        primary=primary,
        secondary=secondary,
        tertiary=tertiary,
        intervallo_secondi=30.0,
    )
    _sentinel.avvia()

    yield  # applicazione in esecuzione

    # Shutdown pulito
    _sentinel.ferma()


def create_app() -> FastAPI:
    return FastAPI(title="ml-service", lifespan=lifespan)

FAQ

Quanti errori consecutivi servono per triggerare il failover a provider secondario?

Nel design presentato il trigger predefinito è 3 errori consecutivi dello stesso tipo oppure p95 di latenza sopra 5 secondi per 2 minuti consecutivi. Il valore 3 è un compromesso: 1 è troppo aggressivo (un singolo timeout di rete non giustifica il failover), 5 è troppo lento (accumuli 5 errori visibili agli utenti). Per servizi con SLA stringenti, abbassa a 2. Per servizi con traffico molto variabile, alza a 5.

Perché usare ABC Python per l'interfaccia provider invece di Protocol?

ABC impone il contratto a runtime: se dimentichi di implementare un metodo astratto, ottieni TypeError all'istanziazione, non durante una chiamata in produzione. Protocol è più flessibile (duck typing) ma non ha enforcement runtime. Per provider AI dove ogni metodo non implementato è un potenziale silent failure in produzione, ABC è la scelta difensiva corretta.

Come si gestisce il cooldown senza bloccare nuove richieste durante il recovery?

Il cooldown si implementa con un timestamp dell'ultimo errore e una durata configurabile. Durante il cooldown, le richieste continuano ad arrivare al provider secondario o al fallback statico - non vengono bloccate. Al termine del cooldown, il sistema tenta un rollback: invia una singola richiesta di health check al provider primario. Se ha successo, ripristina il routing normale. Se fallisce, estende il cooldown con backoff esponenziale.

Il fallback statico copre solo query FAQ o anche query arbitrarie?

Il fallback statico è efficace solo per categorie di query prevedibili e ad alto volume. Nel ml-service del portfolio copre le 50 query più frequenti. Per query arbitrarie, il fallback statico restituisce un messaggio generico di degradation. Non tentare di coprire ogni caso: definisci una "coverage percentage" target (es. 60% delle query reali) e misurarla con i log di produzione.

Come si monitora il costo del provider secondario durante un failover prolungato?

Il billing tracker conta token e stima costo per ogni call al provider secondario, con soglie configurabili: warning a 50% del budget giornaliero, hard stop a 100%. Durante un failover prolungato, se il budget secondario è esaurito prima del recovery del primario, il sistema scala automaticamente al fallback statico. Questo impedisce bill shock non intenzionali.

Quanto è realistico usare Ollama on-prem come primary in produzione?

Realistico per carichi moderati su VPS dedicato. Nel portfolio con VPS Hetzner CCX13 (4 vCPU, 8 GB RAM, llama3.1:8b quantizzato Q4_K_M), Ollama gestisce 3-5 richieste concorrenti con latenza media 4-8 secondi. Per traffico alto o richieste con contesto lungo, la latenza sale oltre 30s e il timeout trigger entra in gioco. Il vantaggio è costo zero e privacy totale. Lo svantaggio è la necessità di un fallback robusto per i picchi.

Serve un alert Telegram per ogni singolo failover?

No: alert per ogni singolo failover genera rumore e alert fatigue. Il design presentato invia alert solo per eventi critici: attivazione del tertiary, budget secondario sopra il 50%, recovery completato. Gli switch rapidi (primario → secondario → primario in meno di 60 secondi) non generano alert, solo log strutturati.

Conclusione

L'architettura FallbackProvider a 3 livelli non è un lusso per servizi enterprise - è il minimo necessario per qualsiasi servizio AI che vuole dichiarare una disponibilità misurabile. I componenti chiave sono quattro: l'interfaccia ABC che impone il contratto a tutti i provider, l'orchestratore con trigger algorithm che decide quando e come fare failover, il health check sentinel che anticipa i problemi prima che arrivino agli utenti, e il billing tracker che impedisce sorprese di costo durante failover prolungati.

Nel contesto del portfolio - Ollama on-prem come primario, nessun provider cloud obbligatorio - l'architettura si semplifica: secondary e tertiary sono entrambi implementazioni statiche in questo momento. Ma il codice è pronto per inserire un provider cloud reale appena necessario, senza toccare l'orchestratore: basta implementare AIProviderBase e passarlo al costruttore.

Prossimo episodio: una volta che il failover è attivo, la domanda diventa cosa vede l'utente durante la degradazione. Nel EP03 - Graceful degradation UX affrontiamo i 4 pattern che trasformano un failure AI in un'esperienza utente accettabile, con la differenza critica tra warning message e silent failure.