Tassonomia dei failure AI: 12 modi in cui il tuo LLM può rompersi

Tassonomia dei failure AI: 12 modi in cui il tuo LLM può rompersi

Hai messo in produzione il tuo servizio AI. Funziona. Per tre settimane tutto bene. Poi, alle 2:47 di notte, arriva la prima notifica: il modello non risponde. Il tuo on-call apre il dashboard e trova un errore che non ha mai visto prima: non è un 500 generico, non è un problema di rete. È qualcosa di più subdolo: il contesto inviato al modello è troppo lungo, il modello ha semplicemente rifiutato la richiesta senza tornare nemmeno un codice HTTP leggibile.

Il problema dei servizi AI in produzione è che i loro failure mode sono radicalmente diversi da quelli dei servizi tradizionali. Un database va giù: lo vedi subito, hai un errore di connessione pulito, fai failover. Un LLM può rompersi in dodici modi diversi, molti dei quali silenziosi, parziali, o statisticamente distribuiti su una percentuale di richieste - abbastanza da non triggerare un alert, abbastanza da danneggiare l'esperienza utente per ore prima che tu te ne accorga. Questo articolo classifica tutti e 12, con il fingerprint di detection e la mitigation immediata per ciascuno.

Nel mio ml-service (FastAPI + Ollama on-prem) ho misurato la distribuzione reale di questi failure su un arco di tre mesi: il failure mode più frequente non è l'errore API, è il timeout silenzioso - Ollama che supera i 30 secondi il 3-8% delle volte sotto carico. Il secondo è l'output malformato: JSON con virgolette non bilanciate, markdown annidato dentro il JSON, testo prima della struttura attesa. Il terzo è la refusal - il modello che si rifiuta di rispondere per policy interne. Tutti e tre richiedono strategie di fallback diverse.

Sezione 1 - La tassonomia: enum Python come fonte di verità

Il primo passo per gestire i failure è nominarli in modo consistente in tutto il codebase. Se ogni team chiama lo stesso problema in modo diverso - "timeout", "slow response", "model hung" - non puoi aggregare metriche, non puoi costruire alert coerenti, non puoi fare post-mortem significativi. L'enum Python serve esattamente a questo: è il dizionario condiviso dei failure mode. Ogni eccezione, ogni log, ogni metrica Prometheus deve usare questi nomi.

from enum import Enum, auto
from dataclasses import dataclass
from typing import Optional


class LLMFailureMode(Enum):
    """
    Tassonomia dei 12 failure mode LLM in produzione.
    Ogni valore è un caso distinto che richiede mitigation separata.
    """
    # Errori infrastrutturali - il provider non risponde
    HARD_API_ERROR = "hard_api_error"          # HTTP 500, 503: provider down
    RATE_LIMIT = "rate_limit"                  # HTTP 429: quota esaurita temporaneamente
    TIMEOUT = "timeout"                        # Nessuna risposta entro il limite configurato
    REGION_OUTAGE = "region_outage"            # Provider down per area geografica

    # Errori di input - il prompt non è accettabile
    CONTEXT_OVERFLOW = "context_overflow"      # Token count supera la finestra massima
    QUOTA_EXHAUSTED = "quota_exhausted"        # Budget mensile/giornaliero consumato

    # Errori di output - il modello risponde ma male
    MALFORMED_OUTPUT = "malformed_output"      # JSON rotto, markdown inatteso, troncato
    REFUSAL = "refusal"                        # Il modello si rifiuta per policy interna
    HALLUCINATION = "hallucination"            # Output plausibile ma fattualmente errato

    # Errori sistemici - problemi di ciclo di vita del modello
    TOOL_ERROR = "tool_error"                  # Tool call che il modello non esegue correttamente
    MODEL_DEPRECATION = "model_deprecation"    # Il modello richiesto non esiste più
    COST_SPIKE = "cost_spike"                  # Costo per singola richiesta anomalo (token explosion)


@dataclass(frozen=True)
class FailureEvent:
    """Rappresentazione immutabile di un failure catturato."""
    mode: LLMFailureMode
    provider: str           # es. "ollama", "external_api"
    model: str              # es. "llama3.1:8b"
    latency_ms: int         # Latenza al momento del failure (0 se non applicabile)
    raw_error: Optional[str] = None
    token_count: Optional[int] = None
    request_id: Optional[str] = None

L'uso di frozen=True nel dataclass non è casuale: un evento di failure è immutabile per definizione. Non vuoi che nessun layer del codice lo modifichi dopo la creazione. Il request_id è fondamentale per correlare il failure con la trace del gateway e il log del modello - senza di esso, il debug in produzione diventa un'ipotesi.

Sezione 2 - I 12 failure mode: frequenza, detection e blast radius

La tabella seguente sintetizza ogni failure mode con la frequenza osservata (range tipico in produzione con provider on-prem e cloud), il segnale di detection primario e il blast radius - ovvero quante richieste vengono impattate quando il failure si manifesta. Un failure con blast radius "tutti" è incomparabilmente più pericoloso di uno con blast radius "singola richiesta", anche se la frequenza è simile.

Una nota sulla hallucination: il 5–15% di frequenza sembra alto, ma considera che "hallucination" in questo contesto include qualsiasi risposta fattualmente non supportata dai documenti di contesto - non solo le invenzioni clamorose. In sistemi RAG con documenti tecnici, la frequenza reale di risposte non groundate si attesta tipicamente tra il 7% e il 12% senza mitigation attiva.

Sezione 3 - Timeout handler: Ollama 30s → cache Redis

Il timeout è il failure mode più insidioso nell'on-prem perché non produce un errore esplicito: produce silenzio. Il processo Python aspetta, il thread è bloccato, l'utente vede un caricamento infinito. La soluzione è asyncio.wait_for con una soglia configurabile, e un fallback su risposta cached identifica con l'hash del prompt normalizzato.

Nel mio ml-service questa strategia gestisce il 3-8% delle call Ollama durante i picchi: la cache Redis con TTL di 2 ore copre la maggior parte delle query frequenti (articoli popolari, query di ricerca ripetute), mentre le query nuove ricevono una risposta di degradation esplicita con un invito a riprovare.

import asyncio
import hashlib
import json
import logging
from typing import Optional, Any
from dataclasses import dataclass

import redis.asyncio as redis

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class TimeoutFallbackConfig:
    """Configurazione immutabile per il timeout handler."""
    timeout_seconds: float = 30.0
    cache_ttl_seconds: int = 7200   # 2 ore
    cache_key_prefix: str = "llm_cache"


class OllamaTimeoutHandler:
    """
    Gestisce timeout di Ollama con fallback su risposta cached.
    Strategia: call Ollama → timeout 30s → cerca cache Redis → risposta statica.
    """

    def __init__(self, redis_client: redis.Redis, config: TimeoutFallbackConfig):
        self._redis = redis_client
        self._config = config

    def _build_cache_key(self, prompt: str, model: str) -> str:
        """Costruisce una chiave cache deterministica da prompt + modello."""
        # Normalizza il prompt: rimuovi spazi multipli e lowercase
        normalized = " ".join(prompt.lower().split())
        content = f"{model}:{normalized}"
        digest = hashlib.sha256(content.encode()).hexdigest()
        return f"{self._config.cache_key_prefix}:{digest}"

    async def call_with_fallback(
        self,
        ollama_coroutine,
        prompt: str,
        model: str,
        static_fallback: str = "Il servizio AI è temporaneamente occupato. Riprova tra qualche minuto."
    ) -> dict[str, Any]:
        """
        Esegue la call Ollama con timeout e fallback a cascata.
        Restituisce sempre una risposta, mai un'eccezione non gestita.
        """
        cache_key = self._build_cache_key(prompt, model)

        try:
            # Tenta la call Ollama con timeout configurato
            result = await asyncio.wait_for(
                ollama_coroutine,
                timeout=self._config.timeout_seconds
            )
            # Salva nella cache per future call identiche
            await self._redis.setex(
                cache_key,
                self._config.cache_ttl_seconds,
                json.dumps(result)
            )
            return {"response": result, "source": "ollama", "cached": False}

        except asyncio.TimeoutError:
            logger.warning(
                "ollama_timeout",
                extra={"model": model, "timeout_s": self._config.timeout_seconds}
            )
            return await self._serve_from_cache_or_static(
                cache_key, static_fallback, reason="timeout"
            )

        except Exception as exc:
            logger.error("ollama_unexpected_error", extra={"error": str(exc)})
            return await self._serve_from_cache_or_static(
                cache_key, static_fallback, reason="error"
            )

    async def _serve_from_cache_or_static(
        self, cache_key: str, static_fallback: str, reason: str
    ) -> dict[str, Any]:
        """Cerca in cache Redis; se assente, restituisce risposta statica."""
        cached = await self._redis.get(cache_key)
        if cached:
            logger.info("serving_cached_response", extra={"reason": reason})
            return {
                "response": json.loads(cached),
                "source": "cache",
                "cached": True,
                "cache_reason": reason,
                "disclaimer": "Risposta dalla cache. Aggiornamento AI disponibile a breve."
            }
        # Nessuna cache disponibile: risposta di degradation
        return {
            "response": static_fallback,
            "source": "static_fallback",
            "cached": False,
            "cache_reason": reason
        }

La chiave del design è che il metodo call_with_fallback non lancia mai eccezioni verso il caller: gestisce internamente tutti i casi e restituisce sempre un dizionario con il campo source che indica la provenienza della risposta. Il chiamante può quindi decidere autonomamente se mostrare o meno il disclaimer all'utente.

Sezione 4 - Hallucination detector: tre segnali combinati

La hallucination è l'unico failure mode che non produce un errore tecnico: il sistema funziona perfettamente, restituisce una risposta ben formata, e quella risposta è sbagliata. Non c'è un'eccezione da catturare. Devi costruire un layer di detection esplicito basato su segnali semantici.

import re
from dataclasses import dataclass
from typing import Optional


# Pattern linguistici che indicano alta confidenza del modello
# (spesso associati a hallucination: il modello mente con sicurezza)
OVERCONFIDENCE_PATTERNS = [
    r"\b(certamente|assolutamente|senza dubbio|è un fatto che|come tutti sanno)\b",
    r"\b(il \d{1,4} [a-z]+ \d{4})\b",   # date specifiche inventate
    r"\b(fonte:|secondo [A-Z][a-z]+ [A-Z][a-z]+)\b",  # citazioni di persone/fonti
]


@dataclass(frozen=True)
class HallucinationSignal:
    """Segnali di detection hallucination con score aggregato."""
    confidence_score_low: bool       # Segnale 1: confidence < 0.6 (se disponibile)
    not_grounded_in_sources: bool    # Segnale 2: nessun documento RAG supporta la risposta
    overconfidence_detected: bool    # Segnale 3: pattern linguistici di falsa certezza
    composite_risk: str              # "low" | "medium" | "high"


def detect_hallucination_risk(
    response_text: str,
    rag_sources: list[str],
    confidence_score: Optional[float] = None,
    grounding_threshold: float = 0.3
) -> HallucinationSignal:
    """
    Valuta il rischio di hallucination su tre segnali combinati.
    Non è infallibile: riduce i falsi negativi combinando più euristiche.
    """
    # Segnale 1: confidence score bassa (solo se il modello lo espone)
    confidence_low = (
        confidence_score is not None and confidence_score < 0.6
    )

    # Segnale 2: verifica grounding - almeno il 30% delle frasi deve avere
    # un termine chiave presente in almeno un documento RAG
    sentences = [s.strip() for s in response_text.split(".") if len(s.strip()) > 20]
    grounded_count = 0
    for sentence in sentences:
        words = set(sentence.lower().split())
        for source in rag_sources:
            source_words = set(source.lower().split())
            # Grounded se più del 20% delle parole della frase sono nel documento
            overlap = len(words & source_words) / max(len(words), 1)
            if overlap > 0.2:
                grounded_count += 1
                break

    grounding_ratio = grounded_count / max(len(sentences), 1)
    not_grounded = grounding_ratio < grounding_threshold

    # Segnale 3: pattern linguistici di overconfidence
    overconfident = any(
        re.search(pattern, response_text, re.IGNORECASE)
        for pattern in OVERCONFIDENCE_PATTERNS
    )

    # Score composito: 2+ segnali attivi = alto rischio
    active_signals = sum([confidence_low, not_grounded, overconfident])
    if active_signals >= 2:
        risk = "high"
    elif active_signals == 1:
        risk = "medium"
    else:
        risk = "low"

    return HallucinationSignal(
        confidence_score_low=confidence_low,
        not_grounded_in_sources=not_grounded,
        overconfidence_detected=overconfident,
        composite_risk=risk
    )

Sezione 5 - JSON malformato: parser con fallback a tre livelli

I modelli LLM tendono a restituire JSON sporco in due pattern ricorrenti: aggiungono testo introduttivo ("Ecco il JSON richiesto: {…}") o includono blocchi markdown (```json) che rompono il parsing diretto. Un parser robusto deve gestire entrambi i casi prima di dichiarare fallimento.

import json
import re
import logging
from typing import Any, Optional


logger = logging.getLogger(__name__)

# Pattern per estrarre JSON da output con testo circostante
JSON_EXTRACTION_PATTERNS = [
    r"```json\s*([\s\S]*?)\s*```",  # Blocco markdown ```json ... ```
    r"```\s*([\s\S]*?)\s*```",      # Blocco markdown generico
    r"(\{[\s\S]*\})",               # JSON object standalone
    r"(\[[\s\S]*\])",               # JSON array standalone
]


def parse_llm_json(
    raw_output: str,
    expected_schema: Optional[dict] = None,
    default_value: Any = None
) -> tuple[Any, bool]:
    """
    Parser JSON con fallback a tre livelli per output LLM malformati.
    Restituisce (valore_parsato, successo: bool).

    Livello 1: json.loads diretto
    Livello 2: estrazione regex del JSON dall'output
    Livello 3: default_value con log dell'errore
    """
    raw_stripped = raw_output.strip()

    # Livello 1: parsing diretto
    try:
        result = json.loads(raw_stripped)
        return result, True
    except json.JSONDecodeError:
        logger.debug("json_parse_level1_failed", extra={"preview": raw_stripped[:100]})

    # Livello 2: estrazione regex
    for pattern in JSON_EXTRACTION_PATTERNS:
        match = re.search(pattern, raw_stripped, re.DOTALL)
        if match:
            candidate = match.group(1).strip()
            try:
                result = json.loads(candidate)
                logger.info("json_parse_level2_success", extra={"pattern": pattern})
                return result, True
            except json.JSONDecodeError:
                continue  # Prova il pattern successivo

    # Livello 3: fallback al valore di default
    logger.error(
        "json_parse_all_levels_failed",
        extra={
            "raw_preview": raw_stripped[:200],
            "raw_length": len(raw_stripped)
        }
    )
    return default_value, False


def parse_llm_json_with_schema_retry(
    raw_output: str,
    schema_description: str,
    llm_caller_fn,
    max_retries: int = 1
) -> tuple[Any, bool]:
    """
    Se il parsing fallisce, ri-chiede al modello un output conforme allo schema.
    Usare solo quando il costo del re-prompt è accettabile.
    """
    result, success = parse_llm_json(raw_output)
    if success:
        return result, True

    if max_retries <= 0:
        return None, False

    # Re-prompt con schema esplicito
    repair_prompt = (
        f"Il tuo output precedente non era JSON valido. "
        f"Rispondi SOLO con JSON valido conforme a questo schema:\n{schema_description}\n"
        f"Output precedente (da correggere):\n{raw_output[:500]}"
    )
    repaired_output = llm_caller_fn(repair_prompt)
    return parse_llm_json_with_schema_retry(
        repaired_output, schema_description, llm_caller_fn, max_retries - 1
    )

Sezione 6 - Rate limit handler con backoff esponenziale e jitter

Il rate limit (HTTP 429) è inevitabile quando usi provider cloud con tier limitati. La risposta naive - fare retry immediato - è la risposta sbagliata: aggrava il problema perché tutti i client riprovano simultaneamente. Il backoff esponenziale con jitter distribuisce i retry nel tempo, riducendo la pressione sul provider.

import asyncio
import random
import logging
from dataclasses import dataclass
from typing import TypeVar, Callable, Awaitable

logger = logging.getLogger(__name__)

T = TypeVar("T")


@dataclass(frozen=True)
class BackoffConfig:
    """Configurazione immutabile per exponential backoff."""
    max_retries: int = 5
    base_delay_seconds: float = 1.0
    max_delay_seconds: float = 60.0
    jitter_factor: float = 0.25   # ±25% di variazione casuale


def compute_backoff_delay(attempt: int, config: BackoffConfig) -> float:
    """
    Calcola il delay con exponential backoff e jitter.
    Formula: min(base * 2^attempt, max) * (1 ± jitter)
    """
    exponential = config.base_delay_seconds * (2 ** attempt)
    capped = min(exponential, config.max_delay_seconds)
    jitter = capped * config.jitter_factor * (2 * random.random() - 1)
    return max(0.1, capped + jitter)  # Mai sotto 100ms


async def with_rate_limit_retry(
    fn: Callable[[], Awaitable[T]],
    config: BackoffConfig,
    rate_limit_status: int = 429
) -> T:
    """
    Esegue fn con retry automatico in caso di rate limit.
    Usa il header Retry-After se presente, altrimenti backoff esponenziale.
    """
    last_exception = None

    for attempt in range(config.max_retries):
        try:
            return await fn()

        except Exception as exc:
            last_exception = exc
            # Determina se è un rate limit (basato su messaggio o codice HTTP)
            is_rate_limit = "429" in str(exc) or "rate limit" in str(exc).lower()

            if not is_rate_limit or attempt == config.max_retries - 1:
                raise  # Non è un rate limit o abbiamo esaurito i retry

            delay = compute_backoff_delay(attempt, config)
            logger.warning(
                "rate_limit_backoff",
                extra={
                    "attempt": attempt + 1,
                    "max_retries": config.max_retries,
                    "delay_seconds": round(delay, 2)
                }
            )
            await asyncio.sleep(delay)

    raise last_exception

Sezione 7 - Context overflow: truncation con sliding window

Il context overflow è l'unico failure mode che puoi prevenire al 100% lato client, prima che la richiesta arrivi al modello. Hai sempre il controllo del prompt: se è troppo lungo, devi truncarlo. La domanda è come truncare senza perdere le informazioni critiche.

from dataclasses import dataclass
from typing import Optional
import re


@dataclass(frozen=True)
class ContextWindowConfig:
    """Configurazione limiti contesto per modello."""
    max_tokens: int               # Finestra massima del modello
    safety_margin: float = 0.85   # Usa solo l'85%: lascia spazio per la risposta
    chars_per_token: float = 3.5  # Stima euristica: 1 token ≈ 3.5 caratteri IT


def estimate_token_count(text: str, chars_per_token: float = 3.5) -> int:
    """Stima approssimativa dei token senza tokenizer esterno."""
    return int(len(text) / chars_per_token)


def truncate_context_sliding_window(
    system_prompt: str,
    conversation_history: list[dict],
    user_query: str,
    config: ContextWindowConfig
) -> tuple[list[dict], bool]:
    """
    Trunca la history di conversazione con sliding window.
    Strategia: mantieni sempre system_prompt e user_query intatti;
    rimuovi i messaggi più vecchi fino a rientrare nel limite.

    Restituisce (history_truncata, was_truncated: bool).
    """
    max_chars = int(config.max_tokens * config.chars_per_token * config.safety_margin)
    fixed_chars = len(system_prompt) + len(user_query) + 100  # Buffer sicurezza

    available_chars = max_chars - fixed_chars
    if available_chars <= 0:
        # System prompt + query già occupano tutto: history vuota
        return [], True

    # Sliding window: parti dall'ultimo messaggio e aggiungi all'indietro
    truncated_history = []
    chars_used = 0
    was_truncated = False

    for message in reversed(conversation_history):
        msg_chars = len(message.get("content", ""))
        if chars_used + msg_chars > available_chars:
            was_truncated = True
            break  # I messaggi più vecchi vengono scartati
        truncated_history.insert(0, message)
        chars_used += msg_chars

    return truncated_history, was_truncated

Sezione 8 - Circuit breaker in Python puro

Il circuit breaker è il pattern che impedisce al tuo sistema di continuare a chiamare un provider in difficoltà, aggravando la situazione. Dopo un numero configurabile di fallimenti consecutivi, il circuito si "apre" e le richieste vengono reindirizzate al fallback senza nemmeno tentare la call al provider. Dopo un cooldown, entra in stato "half-open": tenta una singola call di verifica e decide se tornare in "closed" o riaprire.

import time
import asyncio
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Awaitable, TypeVar

logger = logging.getLogger(__name__)

T = TypeVar("T")


class CircuitState(Enum):
    CLOSED = "closed"       # Normale: le call passano
    OPEN = "open"           # Failure: le call vengono bloccate
    HALF_OPEN = "half_open" # Verifica: una singola call di prova


@dataclass
class CircuitBreakerState:
    """Stato mutabile del circuit breaker (intentionally mutable per tracking)."""
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    last_failure_time: float = 0.0
    success_count_half_open: int = 0


@dataclass(frozen=True)
class CircuitBreakerConfig:
    """Configurazione immutabile del circuit breaker."""
    failure_threshold: int = 5      # Apre dopo N fallimenti consecutivi
    cooldown_seconds: float = 60.0  # Attende prima di passare a HALF_OPEN
    half_open_success: int = 2      # Successi in HALF_OPEN per tornare a CLOSED


class LLMCircuitBreaker:
    """
    Circuit breaker per call LLM. Previene cascade failure su provider instabile.
    """

    def __init__(self, name: str, config: CircuitBreakerConfig):
        self._name = name
        self._config = config
        self._state = CircuitBreakerState()

    @property
    def is_open(self) -> bool:
        return self._state.state == CircuitState.OPEN

    def _check_and_transition(self) -> CircuitState:
        """Valuta se transitare da OPEN a HALF_OPEN dopo il cooldown."""
        if self._state.state == CircuitState.OPEN:
            elapsed = time.monotonic() - self._state.last_failure_time
            if elapsed >= self._config.cooldown_seconds:
                self._state.state = CircuitState.HALF_OPEN
                self._state.success_count_half_open = 0
                logger.info("circuit_half_open", extra={"name": self._name})
        return self._state.state

    def _on_success(self) -> None:
        if self._state.state == CircuitState.HALF_OPEN:
            self._state.success_count_half_open += 1
            if self._state.success_count_half_open >= self._config.half_open_success:
                self._state.state = CircuitState.CLOSED
                self._state.failure_count = 0
                logger.info("circuit_closed", extra={"name": self._name})
        elif self._state.state == CircuitState.CLOSED:
            self._state.failure_count = 0  # Reset su ogni successo

    def _on_failure(self) -> None:
        self._state.failure_count += 1
        self._state.last_failure_time = time.monotonic()
        if self._state.state == CircuitState.HALF_OPEN:
            # Un fallimento in HALF_OPEN ri-apre il circuito
            self._state.state = CircuitState.OPEN
            logger.warning("circuit_reopened", extra={"name": self._name})
        elif self._state.failure_count >= self._config.failure_threshold:
            self._state.state = CircuitState.OPEN
            logger.error(
                "circuit_opened",
                extra={"name": self._name, "failures": self._state.failure_count}
            )

    async def call(
        self,
        fn: Callable[[], Awaitable[T]],
        fallback: Callable[[], Awaitable[T]]
    ) -> T:
        """
        Esegue fn se il circuito è chiuso/half-open, altrimenti chiama fallback.
        """
        current_state = self._check_and_transition()

        if current_state == CircuitState.OPEN:
            logger.debug("circuit_open_fallback", extra={"name": self._name})
            return await fallback()

        try:
            result = await fn()
            self._on_success()
            return result
        except Exception as exc:
            self._on_failure()
            logger.warning(
                "circuit_breaker_failure",
                extra={"name": self._name, "error": str(exc)}
            )
            return await fallback()

Sezione 9 - La severity matrix per prioritizzare i failure

Non tutti i failure mode meritano la stessa urgenza di risposta. La severity matrix combina due dimensioni: l'impatto sull'utente (da "nessuno" a "risposta errata con conseguenze") e la velocità di propagazione (da "singola richiesta" a "tutti gli utenti istantaneamente"). Il prodotto delle due dimensioni determina la priorità di incident response.

La hallucination non rilevata occupa la posizione P1 non perché sia frequente, ma perché il suo impatto è asimmetrico: una sola risposta errata presentata con sicurezza può causare danni reputazionali o legali che superano di gran lunga il costo tecnico di qualsiasi altro failure mode. Questo giustifica l'investimento in detection attiva anche su sistemi con budget limitato.

FAQ

Quali sono i failure mode più frequenti di un LLM in produzione?

I più frequenti sono rate limit (HTTP 429), timeout e malformed output. In produzione con Ollama on-prem, il timeout supera il 30s nel 3–8% delle call durante picchi di carico, mentre il rate limit colpisce quasi esclusivamente servizi cloud con tier free o pay-per-use. Il malformed output è sottostimato perché spesso non viene loggato correttamente: se non hai un parser esplicito che distingue i livelli di fallback, i casi di JSON rotto vengono inghiottiti da un generico "errore di parsing".

Come si detecta una risposta hallucinated da un LLM?

La detection automatica si basa su tre segnali: confidence score sotto soglia (se il modello lo espone), cross-check con RAG sources (la risposta non è supportata da nessun documento recuperato), e pattern detection su affermazioni di certezza assoluta su fatti verificabili. Nessun metodo è infallibile: combinare almeno due segnali riduce i falsi negativi. In sistemi critici, aggiungi una human review queue per le risposte classificate ad alto rischio.

Qual è la differenza tra timeout e context overflow?

Il timeout avviene quando il modello non restituisce risposta entro il limite configurato (es. 30s): può essere causato da carico, contesto lungo o infrastruttura. Il context overflow avviene quando il numero di token del prompt supera la finestra massima del modello (es. 128k token per llama3.1): il modello rifiuta la richiesta con errore esplicito, non con un timeout. Il contesto lungo può causare entrambi: se sei vicino al limite, il modello potrebbe rispondere lentamente (→ timeout) oppure rifiutare direttamente (→ overflow). La prevenzione del context overflow elimina anche una classe di timeout.

Come implementare un fallback su risposta cached quando Ollama è in timeout?

La strategia è: wrappa la call Ollama con asyncio.wait_for(timeout=30), intercetta asyncio.TimeoutError, cerca in Redis una risposta cached per la stessa query (hash SHA256 del prompt normalizzato), restituisci la cache con un disclaimer visibile all'utente. Se la cache non esiste, restituisci una risposta statica di degradation. Salva sempre nella cache dopo una call riuscita con TTL configurabile (2 ore è un buon default per contenuti semi-statici).

Il circuit breaker serve anche con un provider on-prem come Ollama?

Sì, anzi è ancora più importante: Ollama on-prem non ha SLA garantiti, può saturarsi con richieste concorrenti, e un loop di retry senza circuit breaker può aggravare la situazione. Il circuit breaker permette al sistema di "respirare" aprendo il circuito dopo N errori consecutivi e riprovando solo dopo un cooldown. Con Ollama, un threshold di 5 errori e un cooldown di 60 secondi è un punto di partenza ragionevole: calibra poi su dati reali del tuo carico.

Cosa fare quando il modello restituisce JSON malformato?

Implementa un parser con fallback a tre livelli: (1) json.loads diretto, (2) estrazione regex del JSON dall'output (i modelli spesso aggiungono testo prima o dopo), (3) valore di default tipizzato con log del raw output. Se tutti e tre falliscono, hai due opzioni: re-prompt con schema esplicito (costoso ma accurato) oppure restituire il default con un flag che indica parsing fallito. Logga sempre il raw output dei fallimenti: dopo 50 esempi avrai un pattern chiaro su come il tuo modello specifico rompe il JSON e potrai ottimizzare l'estrazione regex.

Come si gestisce la model deprecation senza downtime?

La model deprecation va trattata come un evento pianificato, non un'emergenza. Strategie: monitora i changelog del provider (Ollama pubblica release notes per ogni modello), testa il modello successore in shadow mode (ricevi dual-response senza mostrarla all'utente), automatizza il rollover con feature flag o variabile d'ambiente. Per Ollama on-prem, il pinning esplicito del tag modello in configurazione (es. llama3.1:8b-instruct-q4_K_M invece di llama3.1) evita aggiornamenti accidentali durante i pull.

Conclusione

I 12 failure mode che abbiamo classificato non sono casi rari: sono la norma in qualsiasi servizio AI in produzione con volume non banale. La differenza tra un sistema fragile e uno resiliente non è l'assenza di questi failure, ma la capacità di riconoscerli istantaneamente, contenerli prima che si propaghino e servire comunque una risposta utile all'utente - anche se degradata.

Il pattern che ho implementato nel ml-service di questo portfolio (timeout 30s → cache Redis → risposta statica) gestisce il caso più frequente con tre righe di logica reale. L'enum LLMFailureMode garantisce che ogni componente del sistema parli la stessa lingua quando un failure viene loggato, alertato o incluso in un post-mortem. Il circuit breaker impedisce che un Ollama momentaneamente saturo diventi un failure sistemico.

Prossimo episodio: passiamo dall'identificazione dei failure alla loro prevenzione strutturale. Nel prossimo articolo della serie esploreremo il circuit breaker LLM con Resilience4j e Python puro a confronto: quando usare una libreria battle-tested, quando implementare da zero, e come configurare il half-open probing per un recovery automatico sicuro.