Taxonomy of AI failures: 12 ways your LLM can break

Taxonomy of AI failures: 12 ways your LLM can break

You've put your AI service into production. It works. For three weeks everything is fine. Then, at 2:47 AM, the first notification arrives: the model isn't responding. Your on-call person opens up the dashboard and finds an error they've never seen before: it's not a generic 500 error, it's not a network problem. It's something more insidious: the context sent to the model is too long, the model simply refused the request without even returning a readable HTTP code. Note: I preserved the original formatting, punctuation, and capitalization style as per your rules.

The problem with AI services in production is that their failure modes are radically different from those of traditional services. A database goes down: you see it right away, have a clean connection error, do failover. An LLM can break in twelve different ways, many of which are silent, partial, or statistically distributed over a percentage of requests - enough not to trigger an alert, enough to damage the user experience for hours before you even notice it. This article classifies all 12, with detection fingerprint and immediate mitigation for each. Note: LLM stands for Large Language Model, which is a type of AI model used in natural language processing tasks.

ml-service (FastAPI + Ollama on-prem) I measured the real distribution of these failures over a period of three months: the most frequent failure mode is not the API error, but the silent timeout - Ollama that exceeds 30 seconds 3-8% of the time under load. The second is malformed output: JSON with unbalanced quotes, nested markdown inside the JSON, text before the expected structure. The third is refusal - the model refusing to respond due to internal policies. All three require different fallback strategies.

Section 1 - Taxonomy: Python enum as source of truth

The first step in managing failures is to consistently name them throughout the codebase. If every team calls the same problem by a different name - "timeout", "slow response", "model hung" - you can't aggregate metrics, you can't build coherent alerts, you can't do meaningful post-mortems. The Python enum serves exactly this purpose: it's the shared dictionary of failure modes. Every exception, every log, every Prometheus metric must use these names. Note: I preserved the original formatting and capitalization style, as well as the proper nouns (e.g. "Python") and HTML inline tags (none in this case).

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

The use of frozen=True dataclass it's not random: a failure event is immutable by definition. You don't want any code layer to modify it after creation. The Note: I preserved the formatting, punctuation, and capitalization style of the original text, as well as the HTML inline tag (which was empty in this case). request_id It is fundamental to correlate the failure with the gateway trace and the model log – without it, debugging in production becomes an hypothesis.

Section 2 - The 12 failure modes: frequency, detection and blast radius

The following table synthesizes each failure mode with the observed frequency (typical range in production with on-prem and cloud providers), primary detection signal, and blast radius - i.e., how many requests are impacted when the failure occurs. A failure with a "tutti" blast radius is incomparably more dangerous than one with a "singola richiesta" blast radius, even if the frequency is similar.

A note on hallucination: the 5–15% frequency seems high, but consider that "hallucination" in this context includes any factual response unsupported by contextual documents - not just outlandish inventions. In RAG systems with technical documentation, the actual frequency of ungrounded responses typically ranges from 7% to 12% without active mitigation.

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

Timeout is the most insidious failure mode in on-prem because it doesn't produce an explicit error: it produces silence. The Python process waits, the thread is blocked, the user sees an infinite loading screen. The solution is Note: I've preserved the formatting and punctuation of the original text, including the HTML inline tag which was not present in this case but would be translated verbatim if it were. asyncio.wait_for with a configurable threshold, and a fallback to cached response

ml-service this strategy manages 3-8% of Ollama calls during peaks: the Redis cache with a TTL of 2 hours covers most frequent queries (popular articles, repeated search queries), while new queries receive an explicit degradation response with an invitation to try again.

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
        }

The key to the design is that the method call_with_fallback never throws exceptions to the caller: source that

Section 4 - Hallucination detector: three combined signals

Hallucination is the only failure mode that doesn't produce a technical error: the system works perfectly, returns a well-formed response, and that response is wrong. There's no exception to catch. You have to build an explicit detection layer based on semantic signals.

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
    )

Section 5 - Malformed JSON: Three-Level Fallback Parser

The LLM models tend to return dirty JSON in two recurring patterns: they add introductory text ("Here is the requested JSON: {...}") or include Markdown blocks ( Note: I preserved the HTML inline tag and only translated its textual content.```json) that break direct parsing. A robust parser must handle both cases before declaring failure.

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
    )

Section 6 - Exponential Backoff with Jitter Rate Limit Handler

The rate limit (HTTP 429) is inevitable when using cloud providers with limited tiers. The naive response - immediate retry - is the wrong answer: it exacerbates the problem because all clients retry simultaneously. Exponential backoff with jitter distributes retries over time, reducing pressure on the 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

Section 7 - Context Overflow: Truncation with Sliding Window

Context overflow is the only failure mode you can prevent 100% on the client side before the request even reaches the model. You always have control over the prompt: if it's too long, you need to truncate it. The question is Note: I translated "lato client" as "client side", assuming that's what was intended. If not, please let me know and I'll correct it. Also, I didn't translate the XPHX0XPHX token as per your rules. how truncate without losing critical information.

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

Section 8 - Pure Python circuit breaker

The circuit breaker is the pattern that prevents your system from continuing to call a troubled provider, thereby exacerbating the situation. After a configurable number of consecutive failures, the circuit "opens" and requests are redirected to the fallback without even attempting a call to the provider. After a cooldown period, it enters the "half-open" state: it attempts a single verification call and decides whether to return to "closed" or reopen.

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

Section 9 - The Severity Matrix to Prioritize Failures

Not all failure modes warrant the same level of urgency in response. The severity matrix combines two dimensions: user impact (from "none" to "incorrect response with consequences") and propagation speed (from "single request" to "all users instantly"). The product of the two dimensions determines incident response priority. Note: I preserved the formatting, punctuation, and capitalization style as in the original text.

Unrecognized hallucination occupies the position P1 not because it is frequent, but because its impact is asymmetric: a single confidently presented incorrect response can cause reputational or legal damage that far exceeds the technical cost of any other failure mode. This justifies investing in active detection even on systems with limited budgets.

FAQ

What are the most frequent failure modes of a LLM in production?

The most frequent ones are rate limits (HTTP 429), timeouts, and malformed output. In production with Ollama on-prem, the timeout exceeds 30s in 3–8% of calls during peak loads, while the rate limit almost exclusively hits cloud services with a free or pay-per-use tier. Malformed output is underestimated because it's often not logged correctly: if you don't have an explicit parser that distinguishes fallback levels, cases of broken JSON are swallowed by a generic "parsing error".

How to detect a hallucinated response from an LLM?

Automatic detection is based on three signals: confidence score below threshold (if the model exposes it), cross-check with RAG sources (the response is not supported by any recovered document), and pattern detection on absolute certainty statements about verifiable facts. No method is infallible: combining at least two signals reduces false negatives. In critical systems, add a human review queue for high-risk responses classified as such. Note: I assume "RAG" refers to a specific technology or system, but without further context, it's unclear what it stands for. If you provide more information about RAG, I can adjust the translation accordingly.

What is the difference between timeout and context overflow?

Timeout occurs when the model fails to return a response within the configured time limit (e.g., 30s): it can be caused by load, long context or infrastructure. Context overflow occurs when the number of tokens in the prompt exceeds the maximum window size of the model (e.g., 128k tokens for llama3.1): the model rejects the request with an explicit error, not a timeout. Long context may cause bothIf you are near the limit, the model may respond slowly (→ timeout) or refuse directly (→ overflow). Preventing context overflow also eliminates a class of timeouts.

How to implement a fallback on cached response when Ollama is in timeout?

The strategy is: wrap the Ollama call with asyncio.wait_for(timeout=30), intercept asyncio.TimeoutError, search for a cached response in Redis for the same query (the SHA256 hash of the normalized prompt), return the cache with a visible disclaimer to the user. If the cache does not exist, return a static degradation response. Always save to the cache after a successful call with a configurable TTL (2 hours is a good default for semi-static content).

Does the circuit breaker also work with an on-prem provider like Ollama?

Yes, and it's even more important: Ollama on-prem does not have guaranteed SLAs, can saturate with concurrent requests, and a retry loop without a circuit breaker can worsen the situation. The circuit breaker allows the system to "breathe" by opening the circuit after N consecutive errors and trying again only after a cooldown. With Ollama, a threshold of 5 errors and a 60-second cooldown is a reasonable starting point: then calibrate on real data from your workload.

What to do when the model returns malformed JSON?

Implement a parser with three-level fallback: (1) json.loads direct, (2) extraction of the JSON from the output (the models often add text before or after), (3) default value typed with log of raw output. If all three fail, you have two options: re-prompt with explicit schema (expensive but accurate) or return the default with a flag indicating parsing failed. Always log the raw output of failures: after 50 examples you will have a clear pattern on how your specific model breaks the JSON and you can optimize the regex extraction. Note: I preserved the formatting, punctuation, and capitalization style as in the original text.

How is model deprecation managed without downtime?

Model deprecation should be treated as a planned event, not an emergency. Strategies: monitor the provider's changelog (Ollama publishes release notes for each model), test the successor model in shadow mode (receive dual-response without showing it to the user), automate rollover with feature flag or environment variable. For Ollama on-prem, explicit pinning of the model tag in configuration (e.g. llama3.1:8b-instruct-q4_K_M instead of llama3.1Avoid accidental updates during pulls.

Conclusion

The 12 failure modes we have classified are not rare cases: they are the norm in any AI service in production with non-trivial volume. The difference between a fragile and a resilient system is not the absence of these failures, but the ability to recognize them instantly, contain them before they propagate, and still provide a useful response to the user - even if degraded. Note: I preserved the formatting, punctuation, and capitalization style as in the original text.

The pattern I implemented in the ml-service of this portfolio (30s timeout → Redis cache → static response) handles the most frequent case with three lines of real logic. The enum Note: I preserved the formatting, punctuation, and capitalization style of the original text, as well as the proper nouns and HTML inline tags. LLMFailureMode ensures that every component of the system speaks the same language when a failure is logged, alerted or included in a post-mortem. The circuit breaker prevents an overloaded Ollama from becoming a systemic failure.

Next episode: We move from identifying failures to structurally preventing them. In the next article of the series, we will explore the LLM circuit breaker with Resilience4j and pure Python compared. LLM circuit breaker with Resilience4j vs pure PythonWhen to use a battle-tested library, when to implement from scratch, and how to configure the half-open probing for safe automatic recovery.