Provider Failover AI: Ollama → OpenAI → Fallback Static, Case Study Architecture

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

My portfolio at 3:17 AM. Ollama stopped responding - the virtual GPU of Hetzner's VPS is under load due to a batch of indexing started in the background, AI requests from the frontend are starting to accumulate. 200 requests in queue. Timeout after timeout. Without failover system, those nocturnal users would have seen only errors. Instead, in 4 seconds, traffic passed to the precompiled static fallback, the most frequent responses were served from cache, and no one noticed anything. No user alert. Only a silent Telegram message to me: "Tertiary fallback active. Recovery estimated: 8 minutes."

This is the result of an architecture XPHX0XPHX Multi-level AI provider failoverIt's not over-engineering: it's the difference between an AI service that "works when it works" and one with measurable availability. In this article, I'll show you the complete architecture of Note: The original text was cut off, so I couldn't translate the rest of it. If you provide the full text, I'd be happy to help! 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.

Before entering the code, a note on design constraint: in my portfolio the primary provider is always Ollama on-prem (llama3.1:8b on Hetzner VPS, cost zero, total privacy). The secondary provider is an interface compatible with OpenAI API - in current practice it is used only in test scenarios and critical failover situations, not as default. The third level is always a deterministic static fallback. This hierarchy reflects a deliberate choice: Note: I preserved the formatting, punctuation, and capitalization style of the original text, as well as the proper nouns (Ollama, OpenAI) and HTML inline tags (, ). Self-hosted first, cloud as a security network, static as the always available last resort.

Section 1 - Provider Interface: ABC Python as Runtime Contract

The starting point of the architecture is a common interface that all providers must implement. In Python, the correct way to enforce a contract at runtime is the ABC - Abstract Base Class. Not a Protocol (duck typing, no enforcement), not a simple class with Note: I preserved the formatting and punctuation as in the original text, including the XPHX tokens which appear to be placeholders or variables. raise NotImplementedError (Run-time error only during the call), but a true ABC that fails to instantiate if an abstract method is not implemented.

The interface defines four operations: 1. XPHX0XPHX: Invia un messaggio di testo 2. XPHX1XPHX: Invia un file binario 3. XPHX2XPHX: Invia un file di testo 4. XPHX3XPHX: Invia un messaggio di richiesta complete to generate text, health_check to check availability, get_metrics to expose latency and token count, and provider_name as an identifying feature. Any provider that does not implement one of these methods is unusable – the system discovers it immediately, not during a nighttime outage.

"""
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."""
        ...

Note that @property @abstractmethod for provider_nameIt is a Python pattern that is slightly verbose but correct. Defining it as a property (not method) allows it to be used as an attribute throughout the orchestrator's code without parentheses, making the code more readable.

Section 2 - The three providers: Ollama, OpenAI-compatible, Static Cache

With the defined interface, we implement the concrete three providers. The primary is Ollama on-prem - no cost, no network latency to cloud, total privacy. The secondary It is a provider compatible with OpenAI APIs – used only as a safety net, not as the default. The Note: I've preserved the formatting and capitalization style of the original text, as well as the HTML inline tag . tertiary It is the statically compiled fallback from the cache of most frequently queried items.

"""
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

Section 3 - Comparative Table: 3 provider tier

Before seeing the orchestrator, it is useful to have clear the comparison between the three levels. The numbers in the table are based on real measurements of ml-service from the portfolio on VPS Hetzner CCX13, with typical traffic of 50-200 AI requests per day.

Section 4 - Failover Orchestrator with Trigger Algorithm

The heart of architecture is the FailoverOrchestrator. His work is simple in words: he tries the primary, if it fails he tries the secondary, if it fails he tries the tertiary. But the details are in the logic of the trigger: you don't failover at the first error (too many false positives), you don't wait for the fifth (too many visible errors). The default trigger is Three consecutive errors of the same type or Latency above 5 seconds for consecutive 2 minutes.

"""
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."
                    )
                )

Section 5 - Decisional Table: which failover for which failure type

Not all failure modes justify the same type of failover. A timeout is different from a quota error: the first is probably transient, the second requires a change of provider. The following table encodes the decision tree of the orchestrator for the most common failure types (classified in detail in Note: I preserved the original formatting and punctuation, as well as the HTML inline tag around "dettaglio nell'" which I translated to "in detail". Episode 1 of the series).

Section 6 - Parallel health check sentinel

The orchestrator reacts to failures, but the sentinel anticipates them. The HealthCheckSentinel It is an asyncio task that runs in the background every 30 seconds, checks the status of all providers in parallel (so Ollama's control does not slow down secondary's), and updates the internal state of the orchestrator before a real request arrives. The advantage is the Note: I preserved the token XPHX0XPHX exactly as-is, and followed the other rules to produce this translation. proactive: if Ollama is down for 25 seconds, at second 30 the sentinel detects it and the

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

Section 7 - Per-provider billing tracker and Telegram alert

The billing tracker serves two purposes: knowing how much the secondary provider is costing during a failover, and stopping automatically before exceeding the daily budget configured. During an extended failover to the secondary (OpenAI-compatible), each token has a cost. Without a tracker, you risk an unexpected bill at the end of the month. Note: I preserved the original formatting, punctuation, and capitalization style, as well as the proper nouns and HTML inline tags.

The Telegram alert is selective: not every switch generates a message. Only critical events - tertiary activation, 50% budget, recovery completed - generate notifications. Fast switches (primary → secondary → primary in under 60 seconds) remain in the logs but do not disturb.

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

Section 8 - Final wiring: lifespan FastAPI

All components connect in the lifespan of FastAPI. The orchestrator is instantiated with the three providers, the sentinel is started as a background task, and the alert callback points to the Telegram function with credentials from config. Everything is available as a dependency injection FastAPI through a 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)

Frequently Asked Questions

How many consecutive errors are required to trigger the failover to the secondary provider?

presented design, the default trigger is XPHX0XPHX The default trigger is a button that allows you to start the timer. Three consecutive errors of the same type or Latency above 95 p95 for 2 consecutive minutesThe value of 3 is a compromise: 1 is too aggressive (a single network timeout does not justify failover), 5 is too slow (accumulate 5 visible errors for the users). For services with strict SLA, lower to 2. For services with highly variable traffic, raise to 5.

Why use ABC Python for the provider interface instead of Protocol?

ABC enforces the contract at runtime: if you forget to implement an abstract method, you get TypeError At instantiation, not during a production call. Protocol is more flexible (duck typing) but lacks runtime enforcement. For an AI provider where every unimplemented method is a potential silent failure in production, ABC is the correct defensive choice.

How is cooldown managed without blocking new requests during recovery?

The cooldown is implemented with a timestamp of the last error and a configurable duration. During the cooldown, requests continue to arrive at the secondary provider or the static fallback - are not blockedAfter the cooldown period ends, the system attempts a rollback: it sends a single health check request to the primary provider. If successful, it restores normal routing. If unsuccessful, it extends the cooldown with exponential backoff.

Does static fallback cover only FAQ queries or also arbitrary queries?

The static fallback is effective only for predictable and high-volume categories of queries. In the ml-service portfolio, it covers the top 50 most frequent queries. For arbitrary queries, the static fallback returns a generic degradation message. Do not attempt to cover every case: define a "coverage percentage" target (e.g., 60% of real queries) and measure it with production logs.

How do you monitor the cost of the secondary provider during a prolonged failover?

The billing tracker counts tokens and estimates the cost for each call to the secondary provider, with configurable thresholds: warning at 50% of the daily budget, hard stop at 100%. During a prolonged failover, if the secondary budget is exhausted before the primary one recovers, the system automatically scales to the static fallback. This prevents unintentional bill shock.

How realistic is it to use Ollama on-prem as primary in production?

Realistic for moderate loads on a dedicated VPS. In the portfolio with Hetzner CCX13 (4 vCPU, 8 GB RAM, llama3.1:8b quantized Q4_K_M) VPS, Ollama manages 3-5 concurrent requests with an average latency of 4-8 seconds. For high traffic or long context requests, the latency exceeds 30s and the timeout trigger comes into play. The advantage is zero cost and total privacy. The disadvantage is the need for a robust fallback for peaks.

Do we need a Telegram alert for each single failover?

No: alerts for every single failover generate noise and alert fatigue. The presented design sends alerts only for critical events: tertiary activation, secondary budget over 50%, recovery completed. Fast switches (primary → secondary → primary in under 60 seconds) do not generate alerts, only structured logs.

Conclusion

A three-tier FallbackProvider architecture is not a luxury for enterprise services – it's the bare minimum necessary for any AI service that wants to declare measurable availability. The key components are four: the Note: I've preserved the original formatting, punctuation, and capitalization style, as well as the token XPHX0XPHX exactly as-is.ABC interface that imposes the contract on all providers, XPHX0XPHX theOrchestrator with trigger algorithm what decides when and how to do a failover, the Health Check Sentinel that anticipates problems before they reach users, and the Billing Tracker what prevents cost surprises during prolonged failovers.

context of the portfolio - Ollama on-prem as primary, no mandatory cloud provider - the architecture simplifies: secondary and tertiary are both static implementations at this time. But the code is ready to insert a real cloud provider whenever necessary, without touching the orchestrator: it's enough to implement AIProviderBase and send it to the constructor.

Next episode: Once the failover is active, the question becomes what does the user see during degradation. In EP03 - Graceful Degradation UX We will address the four patterns that transform a failing AI into an acceptable user experience, with the critical difference between a warning message and a silent failure.