Debug trace analysis: root-cause in 5 minuti, trace similarity, error pattern detection

Debug trace analysis: root-cause in 5 minuti, trace similarity, error pattern detection

Sono le 14:30 di un mercoledi qualunque, un utente premium del portfolio apre un ticket: "il tool /internal/ai/recommend mi ha consigliato un articolo che non esiste, titolo inventato di sana pianta". Apro Langfuse, filtro per user_id, trovo la trace incriminata: 200 span totali, 50 tool call, 12 retrieval annidate, 8 step di agent reasoning. La trace è un albero che riempie 4 schermate verticali. Dove sta la root cause? Nel retrieval che ha portato il documento sbagliato? Nel rerank che ha promosso noise? Nel generator che ha hallucinato a partire da context valido? In una orchestrazione tipica LLM la diagnosi manuale di una singola trace può richiedere 2-3 ore per un engineer esperto, e moltiplicato per 30-40 ticket settimana è un full time job.

Eppure con il tooling giusto la stessa diagnosi scende a 5-8 minuti. La differenza non sta nell'esperienza dell'engineer, ma nell'avere strumenti automatizzati che (1) confrontano trace fallite con trace successo per evidenziare il delta, (2) clusterizzano errori simili per riconoscere pattern ricorrenti, (3) calcolano similarity fra trace per trovare casi già diagnosticati, (4) applicano decision tree strutturati per inferire root cause, (5) rilevano span anomali statisticamente, (6) permettono replay deterministico locale per testare fix prima del deploy. Questo articolo costruisce tutti e sette questi tool da zero in Python, con esempi diretti dal mio ml-service portfolio in produzione su VPS Hetzner con Ollama llama3.1:8b.

Caso reale che racconto in dettaglio: ad Aprile 2026 ho dovuto debuggare proprio questo bug del retrieve+rerank, downvote rate al 7% sul tool recommend. La pipeline di debug applicata, partendo dalla raccolta delle 50 trace flagged fino al fix in produzione, ha richiesto 4 ore totali di cui solo 30 minuti per il primo trace diff manuale. Il resto è stato tooling automatico che oggi gira ogni notte come job batch e identifica pattern di errore prima che diventino reclami utente. Il risparmio di MTTR è la vera metrica: da 2-3 ore a 5-8 minuti per incident debug.

Sezione 1 - Trace diff tool: comparare 2 trace LLM in Python

Il primo strumento è anche il più semplice concettualmente: dato due JSON di trace, una fallita e una successo "simile", evidenzia automaticamente cosa differisce. Il problema reale è che le trace LLM contengono tonnellate di campi sempre diversi (timestamp, span_id, trace_id, request_id, latency esatta in ms, costi al centesimo) che generano centinaia di diff irrilevanti. La libreria DeepDiff consente di escludere path specifici e normalizzare valori prima del confronto, restituendo solo il delta semantico.

Lo script seguente prende come input due trace serializzate da Langfuse (export JSON), normalizza i campi numerici in bucket discreti, esclude il rumore di basso livello, e produce un report leggibile con il delta semantico evidenziato. L'output ha tipicamente 5-15 differenze rilevanti invece delle 200-400 grezze.

# trace_diff.py - confronta due trace LLM ed estrae il delta semantico
# Uso: python trace_diff.py trace_failed.json trace_success.json

import json
import math
from typing import Any
from deepdiff import DeepDiff

# Path da escludere perché sempre diversi e irrilevanti per root-cause
EXCLUDE_PATHS = {
    "root['trace_id']",
    "root['request_id']",
    "root['user_id']",
    # Pattern wildcard per span annidati
}

EXCLUDE_REGEX = [
    r"root\['spans'\]\[\d+\]\['span_id'\]",
    r"root\['spans'\]\[\d+\]\['parent_id'\]",
    r"root\['spans'\]\[\d+\]\['start_time'\]",
    r"root\['spans'\]\[\d+\]\['end_time'\]",
]


def normalize_trace(trace: dict) -> dict:
    """Normalizza valori numerici in bucket discreti per ridurre rumore.

    Esempio: latency 123ms diventa bucket "100-200ms",
    cost 0.0234 EUR diventa bucket 0.02 EUR.
    Così due trace 'praticamente uguali' producono diff vuoto.
    """
    normalized = json.loads(json.dumps(trace))  # deep copy

    for span in normalized.get("spans", []):
        # Bucket latency in fasce 100ms
        if "latency_ms" in span:
            span["latency_ms"] = int(span["latency_ms"] // 100) * 100
        # Arrotonda costo al decimo di centesimo
        if "cost_cents" in span:
            span["cost_cents"] = round(span["cost_cents"], 1)
        # Bucket token in fasce 50
        if "tokens_in" in span:
            span["tokens_in"] = int(span["tokens_in"] // 50) * 50
        if "tokens_out" in span:
            span["tokens_out"] = int(span["tokens_out"] // 50) * 50

    return normalized


def compute_semantic_diff(trace_failed: dict, trace_success: dict) -> dict:
    """Calcola diff semantico fra due trace ignorando il rumore."""
    norm_failed = normalize_trace(trace_failed)
    norm_success = normalize_trace(trace_success)

    diff = DeepDiff(
        norm_success,  # baseline = trace successo
        norm_failed,   # variante = trace fallita
        exclude_paths=EXCLUDE_PATHS,
        exclude_regex_paths=EXCLUDE_REGEX,
        ignore_order=False,  # ordine span è significativo
        report_repetition=True,
    )

    return diff.to_dict()


def render_human_readable(diff: dict) -> str:
    """Genera report markdown-friendly dei differenziali."""
    lines = ["# Trace Diff Report\n"]

    if "values_changed" in diff:
        lines.append("## Valori modificati\n")
        for path, change in diff["values_changed"].items():
            lines.append(f"- `{path}`: {change['old_value']} -> {change['new_value']}")

    if "iterable_item_added" in diff:
        lines.append("\n## Span aggiunti nella trace fallita\n")
        for path, item in diff["iterable_item_added"].items():
            op = item.get("operation_name", "?")
            lines.append(f"- `{path}`: operation={op}")

    if "iterable_item_removed" in diff:
        lines.append("\n## Span presenti solo nella trace successo\n")
        for path, item in diff["iterable_item_removed"].items():
            op = item.get("operation_name", "?")
            lines.append(f"- `{path}`: operation={op}")

    return "\n".join(lines)


if __name__ == "__main__":
    import sys
    with open(sys.argv[1]) as f:
        trace_failed = json.load(f)
    with open(sys.argv[2]) as f:
        trace_success = json.load(f)

    diff = compute_semantic_diff(trace_failed, trace_success)
    print(render_human_readable(diff))

L'output tipico su una trace di 200 span è ridotto a 8-12 differenze evidenti come "model nel span generate cambiato da llama3.1:8b a llama3.1:8b-q4", "rerank_score nel span retrieve sceso da 0.92 a 0.41", "operation_name 'fallback_response' presente solo nella fallita". Questo già restringe enormemente l'investigazione manuale. Per il caso reale del bug retrieve+rerank di Aprile, il diff ha mostrato immediatamente che le trace fallite avevano retrieval_cosine_score bucket "0.4-0.5" mentre le successo erano "0.7-0.8", segnale netto che il problema era upstream nella retrieval, non nella generation come sembrava dal sintomo (titolo articolo inventato).

Una raffinatezza che applico in produzione è la scelta intelligente della trace successo "baseline" da confrontare. Non basta prendere a caso una trace andata a buon fine: la baseline deve essere semanticamente vicina alla fallita (stessa categoria di query, stesso modello, stesso periodo temporale per evitare drift di prompt template). Il workflow che funziona nel ml-service è il seguente: dalla trace fallita estraggo l'embedding della query utente, faccio similarity search nelle trace etichettate come "success" delle ultime 48 ore, prendo le top-3 più simili, e per ciascuna eseguo il diff. Se 3 diff su 3 evidenziano la stessa differenza chiave, ho identificato con alta confidenza il vero delta semantico. Se invece i 3 diff divergono, significa che la fallita ha caratteristiche peculiari che richiedono debug individuale più approfondito.

Un altro pattern utile è il diff aggregato N-vs-N: invece di confrontare 1 fallita vs 1 successo, prendo 50 fallite del cluster e 50 successo simili, calcolo il diff su ogni coppia, e poi conto la frequenza con cui ogni campo path appare nei diff. I path che compaiono in >80% dei diff sono i veri "marker semantici" del cluster di errore. I path che compaiono solo in <20% dei diff sono rumore residuo da escludere in iterazioni successive del filtro EXCLUDE_PATHS. Questa procedura iterativa di tuning del filtro porta nel giro di 2-3 settimane a una configurazione molto stabile che produce diff puliti e immediatamente leggibili.

Sezione 2 - Error pattern detection con clustering scikit-learn

Quando hai 1000+ trace fallite a settimana, comparare a una a una contro casi successo non scala. Servirebbe raggruppare automaticamente le fallite in error categories, ognuna con runbook dedicato e priorità di fix. Questo è un classico problema di clustering non supervisionato. La domanda è che algoritmo usare: KMeans richiede di specificare K a priori (numero categorie), ma in ambito LLM le categorie evolvono ogni settimana. DBSCAN invece scopre i cluster automaticamente e identifica gli outlier come noise (etichetta -1), che corrispondono ai veri "casi unici" da debuggare singolarmente.

Lo script seguente legge un dataset CSV con feature estratte dalle trace fallite, costruisce il vector di feature, applica DBSCAN con metric cosine, e produce per ogni cluster una "scheda" con le statistiche aggregate (modello dominante, latency media, error type modale, esempi di trace_id). Ogni cluster diventa una error category nella dashboard.

# error_pattern_detector.py - clusterizza trace fallite in error categories
# Uso: python error_pattern_detector.py failed_traces.csv

import pandas as pd
import numpy as np
from sklearn.cluster import DBSCAN
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from collections import Counter


def build_feature_matrix(df: pd.DataFrame) -> np.ndarray:
    """Costruisce feature matrix combinando categoriche + numeriche normalizzate.

    Feature usate:
    - operation_name (one-hot): retrieve, rerank, generate, tool_call, ...
    - model (one-hot): llama3.1:8b, llava:7b, nomic-embed, ...
    - latency_ms (log + scaler)
    - tokens_total (log + scaler)
    - cost_cents (log + scaler)
    - retrieval_top_k (raw + scaler)
    - error_type (one-hot): timeout, json_decode, hallucination, ...
    """
    # Log-transform su distribuzioni log-normali
    df["log_latency"] = np.log1p(df["latency_ms"])
    df["log_tokens"] = np.log1p(df["tokens_total"])
    df["log_cost"] = np.log1p(df["cost_cents"])

    preprocessor = ColumnTransformer(
        transformers=[
            ("cat", OneHotEncoder(handle_unknown="ignore"),
             ["operation_name", "model", "error_type"]),
            ("num", StandardScaler(),
             ["log_latency", "log_tokens", "log_cost", "retrieval_top_k"]),
        ]
    )

    return preprocessor.fit_transform(df).toarray()


def cluster_traces(features: np.ndarray, eps: float = 0.3,
                   min_samples: int = 5) -> np.ndarray:
    """Applica DBSCAN con metric cosine per cluster trace simili.

    eps=0.3 funziona bene su feature normalizzate per RAG-style trace.
    min_samples=5 evita micro-cluster spuri.
    Outlier (noise) = label -1 = casi unici da debug manuale.
    """
    clusterer = DBSCAN(eps=eps, min_samples=min_samples, metric="cosine", n_jobs=-1)
    return clusterer.fit_predict(features)


def summarize_clusters(df: pd.DataFrame, labels: np.ndarray) -> dict:
    """Per ogni cluster, calcola statistiche aggregate per error category."""
    df["cluster"] = labels
    summary = {}

    for cluster_id in sorted(set(labels)):
        if cluster_id == -1:
            label_name = "noise (casi unici)"
        else:
            label_name = f"cluster_{cluster_id}"

        subset = df[df["cluster"] == cluster_id]

        summary[label_name] = {
            "size": len(subset),
            "model_dominant": Counter(subset["model"]).most_common(1)[0][0],
            "operation_dominant": Counter(subset["operation_name"]).most_common(1)[0][0],
            "error_type_dominant": Counter(subset["error_type"]).most_common(1)[0][0],
            "latency_p50_ms": int(subset["latency_ms"].median()),
            "cost_mean_cents": round(subset["cost_cents"].mean(), 2),
            "sample_trace_ids": subset["trace_id"].head(3).tolist(),
        }

    return summary


if __name__ == "__main__":
    import sys
    df = pd.read_csv(sys.argv[1])
    features = build_feature_matrix(df)
    labels = cluster_traces(features)
    summary = summarize_clusters(df, labels)

    for cluster_name, stats in summary.items():
        print(f"\n=== {cluster_name} ({stats['size']} trace) ===")
        for k, v in stats.items():
            if k != "size":
                print(f"  {k}: {v}")

Su un dataset reale di 8000 trace fallite della scorsa settimana del ml-service, l'output produce tipicamente 12-15 cluster significativi più un cluster noise di 200-300 outlier. Esempi di cluster reali emersi: cluster_0 (timeout su Ollama generate, 1200 trace, latency_p50 30s), cluster_3 (rerank score basso ma generation procede, 800 trace, il bug del recommend), cluster_7 (JSONDecodeError su tool output strutturato, 400 trace, problema parsing). Ogni cluster diventa un "incidente aggregato" con runbook dedicato, evitando di debuggare 1200 trace individualmente quando hanno tutte la stessa root cause.

Il tuning di eps e min_samples richiede iterazione: ho cominciato con eps=0.5 e min_samples=10 ottenendo 4 macro-cluster troppo eterogenei (un singolo cluster conteneva sia timeout che hallucination), poi sceso a eps=0.3 e min_samples=5 trovando granularità giusta. Per validare la qualità del clustering uso il silhouette score di scikit-learn: se >0.4 il clustering è solido, se <0.2 i cluster sono sovrapposti e devo rivedere le feature. Un'altra metrica utile è la stabilità temporale: lancio il clustering su settimana corrente e settimana precedente, confronto la distribuzione delle dimensioni dei cluster, se cambia drasticamente significa che è avvenuto un drift (nuovo deploy, cambio modello, modifica prompt) che merita investigazione.

Una variante che applico per casi specifici è il clustering gerarchico con scipy.cluster.hierarchy: produce un dendrogramma che permette di vedere visivamente come i cluster si raggruppano a vari livelli di granularità. Utile in fase di esplorazione iniziale quando non sai ancora quanti cluster aspettarti. Una volta stabilita la granularità giusta, passo a DBSCAN per il job batch notturno perché è molto più veloce su dataset grandi. Nel mio setup il job notturno gira in 8 minuti su 80K trace della settimana, output salvato in PostgreSQL come tabella error_clusters con colonne (cluster_id, week_start, size, model_dominant, root_cause_inferred, runbook_url).

Sezione 3 - Trace similarity con embedding e nearest neighbor

Il clustering raggruppa, ma quando arriva una nuova trace fallita la domanda operativa è "questo errore l'abbiamo già visto? Esiste un fix documentato?". Serve similarity search: dato un vector che rappresenta la nuova trace, trova le k trace storicamente più simili nel database. Se le top-10 sono tutte etichettate come "rerank overconfidence" e già risolte, non serve aprire un nuovo ticket di debug, basta applicare il fix esistente. Per generare l'embedding di una trace serializzo i suoi campi semantici come stringa strutturata e passo il tutto a sentence-transformers con il modello all-MiniLM-L6-v2 (lo stesso che già uso per RAG nel ml-service, zero overhead di deploy).

Per il nearest neighbor uso FAISS con IndexFlatIP (inner product equivalente a cosine su vettori normalizzati). 100K trace caricate in 200MB di RAM, query <5ms. Lo script seguente costruisce l'index, espone una funzione di query, e ranking finale con cross-encoder rerank per affinare i top-10 quando serve precisione massima.

# trace_similarity.py - similarity search fra trace via embedding
# Uso: from trace_similarity import find_similar_traces

import numpy as np
import faiss
from sentence_transformers import SentenceTransformer, CrossEncoder
from typing import Iterable

# Modelli condivisi con il pipeline RAG del ml-service portfolio
ENCODER = SentenceTransformer("all-MiniLM-L6-v2")
RERANKER = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

EMBEDDING_DIM = 384
NUMERIC_TAIL_DIM = 8  # latency, cost, tokens, error_count normalizzati


def serialize_trace(trace: dict) -> str:
    """Serializza la trace come stringa strutturata per encoding semantico.

    Formato: 'op:retrieve|model:nomic|topk:5||op:rerank|model:bge||...'
    Cattura la sequenza operazionale + parametri chiave.
    """
    parts = []
    for span in trace.get("spans", []):
        op = span.get("operation_name", "?")
        model = span.get("model", "n/a")
        extra = ""
        if "retrieval_top_k" in span:
            extra = f"|topk:{span['retrieval_top_k']}"
        if "tool_name" in span:
            extra = f"|tool:{span['tool_name']}"
        parts.append(f"op:{op}|model:{model}{extra}")

    return "||".join(parts)


def extract_numeric_features(trace: dict) -> np.ndarray:
    """Feature numeriche normalizzate da concatenare all'embedding."""
    spans = trace.get("spans", [])
    total_latency = sum(s.get("latency_ms", 0) for s in spans)
    total_cost = sum(s.get("cost_cents", 0) for s in spans)
    total_tokens = sum(s.get("tokens_in", 0) + s.get("tokens_out", 0) for s in spans)
    error_count = sum(1 for s in spans if s.get("error"))

    # Log-transform + min-max scaling fissi (non training-dependent)
    return np.array([
        np.log1p(total_latency) / 15,  # ~max 30s
        np.log1p(total_cost) / 10,     # ~max EUR 0.5
        np.log1p(total_tokens) / 12,   # ~max 100K token
        error_count / 10,              # ~max 10 errori
        len(spans) / 100,              # ~max 100 span
        # Padding a NUMERIC_TAIL_DIM
        0.0, 0.0, 0.0,
    ], dtype=np.float32)


def encode_trace(trace: dict) -> np.ndarray:
    """Genera vector finale 392d (384 embedding + 8 numeric)."""
    text = serialize_trace(trace)
    text_vec = ENCODER.encode(text, normalize_embeddings=True)
    num_vec = extract_numeric_features(trace)
    full_vec = np.concatenate([text_vec, num_vec])
    # Re-normalizza il vector totale per cosine via inner product
    return full_vec / np.linalg.norm(full_vec)


class TraceSimilarityIndex:
    """FAISS index in-memory per similarity search fra trace."""

    def __init__(self):
        self.index = faiss.IndexFlatIP(EMBEDDING_DIM + NUMERIC_TAIL_DIM)
        self.metadata: list[dict] = []

    def add(self, traces: Iterable[dict]) -> None:
        vectors = np.vstack([encode_trace(t) for t in traces]).astype(np.float32)
        self.index.add(vectors)
        for t in traces:
            self.metadata.append({
                "trace_id": t["trace_id"],
                "cluster_label": t.get("cluster_label"),
                "resolution": t.get("resolution"),  # es. "rerank_overconfidence_fix"
            })

    def query(self, trace: dict, k: int = 10) -> list[dict]:
        vec = encode_trace(trace).reshape(1, -1).astype(np.float32)
        scores, indices = self.index.search(vec, k)
        return [
            {**self.metadata[idx], "similarity": float(score)}
            for idx, score in zip(indices[0], scores[0])
            if idx != -1
        ]

    def query_with_rerank(self, trace: dict, k: int = 10, rerank_top: int = 3) -> list[dict]:
        """Top-k via FAISS, poi rerank via cross-encoder per precisione massima."""
        candidates = self.query(trace, k=k)
        new_text = serialize_trace(trace)
        pairs = [(new_text, serialize_trace_from_id(c["trace_id"])) for c in candidates]
        rerank_scores = RERANKER.predict(pairs)
        for c, s in zip(candidates, rerank_scores):
            c["rerank_score"] = float(s)
        return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)[:rerank_top]

Il workflow operativo nel ml-service è il seguente: ogni notte un job batch riprende le trace etichettate manualmente dal team (con campi cluster_label e resolution) e le indicizza in FAISS. A inference time, quando un alert scatta su una trace fallita, il sistema interroga l'index con k=10, se il top-3 ha rerank_score >0.85 e tutti hanno la stessa resolution, applica automaticamente il runbook documentato senza intervento umano. Risparmio sul mio progetto: il 40% degli incident viene gestito in modalità "auto-suggest fix" senza ticket aperto.

Un dettaglio cruciale è la pulizia continua del database di trace etichettate: senza disciplina il database degenera in mesi, contenendo trace obsolete con resolution che non è più applicabile (es. fix per bug Ollama 0.4.x non valido per 0.6.x). Mantengo una pipeline di "validazione resolution" notturna che per ogni trace etichettata vecchia di >90 giorni rilancia il replay framework e verifica che il fix proposto produca ancora output corretto. Se fallisce, marco la resolution come "stale" e l'esclude dalle auto-resolution future fino a re-validazione manuale. Questo evita il rischio peggiore: applicare automaticamente un fix che non funziona più e ritardare il debug reale di un incident urgente.

Per casi in cui il top-10 della similarity search non è omogeneo (es. 4 cluster diversi), il sistema NON auto-applica nulla ma genera un report di "ipotesi multiple" che il debug engineer riceve come primo input del ticket. Esempio: "trace simili 60% cluster_3 (rerank overconfidence), 30% cluster_7 (json decode), 10% noise. Ipotesi prevalente: rerank issue, controllare prima". Questo già restringe l'investigazione di partenza senza forzare una decisione automatica errata. Misura nel ml-service: dei 60% di incident NON auto-risolti, il 75% vengono comunque risolti seguendo l'ipotesi prevalente suggerita, riducendo MTTR anche quando l'auto-resolution non scatta.

Sezione 4 - Decision tree per root-cause analysis

Quando il diff e la similarity non bastano (caso veramente nuovo), serve un workflow strutturato che dato un set di evidenze (error type, latency, span con anomalie, content del completion) inferisce la categoria di root cause con un livello di confidence. Mi serve interpretabilita 100% (devo poter spiegare al team perché il sistema dice "hallucination da retrieval"), quindi NON uso DecisionTreeClassifier di scikit (black box trainata sui dati). Implemento invece un decision tree manuale come funzione Python pura con 30-40 if-else esplicite, ognuna con criteria check e contributo al confidence score.

# root_cause_analyzer.py - decision tree manuale per inferenza root-cause
# Output: (categoria, confidence 0-1, evidenze)

from typing import NamedTuple
import re


class RootCauseResult(NamedTuple):
    category: str
    confidence: float
    evidences: list[str]


def analyze_root_cause(trace: dict) -> RootCauseResult:
    """Decision tree a 5 livelli per inferenza root-cause.

    Livello 1: error_type del final span
    Livello 2: latency e resource check
    Livello 3: prompt template version diff
    Livello 4: hallucination check (entity vs context)
    Livello 5: retrieval quality check
    """
    spans = trace.get("spans", [])
    if not spans:
        return RootCauseResult("unknown", 0.0, ["trace senza span"])

    final_span = spans[-1]
    error_type = final_span.get("error_type", "")

    # === Livello 1: timeout / resource ===
    if "timeout" in error_type.lower():
        return _diagnose_timeout(trace)

    # === Livello 2: tool errors ===
    if "json" in error_type.lower() or "decode" in error_type.lower():
        return _diagnose_tool_output_malformed(trace)

    # === Livello 3: nessun errore esplicito ma flagged ===
    if trace.get("user_feedback") == "downvote":
        return _diagnose_quality_issue(trace)

    return RootCauseResult("unclassified", 0.3, [f"error_type={error_type}"])


def _diagnose_timeout(trace: dict) -> RootCauseResult:
    evidences = []
    confidence = 0.5

    # Check OOM upstream
    for span in trace["spans"]:
        if span.get("operation_name") == "ollama_generate":
            if span.get("latency_ms", 0) > 25000:
                evidences.append(f"ollama latency {span['latency_ms']}ms (>25s)")
                confidence += 0.2
            if span.get("gpu_memory_used_mb", 0) > 7500:
                evidences.append(f"GPU memory {span['gpu_memory_used_mb']}MB (saturata)")
                confidence += 0.2

    if confidence > 0.8:
        return RootCauseResult("ollama_oom_concurrent_request", confidence, evidences)
    return RootCauseResult("timeout_generic", confidence, evidences)


def _diagnose_tool_output_malformed(trace: dict) -> RootCauseResult:
    evidences = []
    confidence = 0.4

    # Check prompt template version recente
    for span in trace["spans"]:
        if span.get("operation_name") == "tool_call":
            template_version = span.get("prompt_template_version", "")
            if template_version.endswith(".0"):  # major version recente
                evidences.append(f"prompt template version {template_version} (cambiato di recente)")
                confidence += 0.4
            if "json" in span.get("output_format", "").lower():
                evidences.append("output format JSON ma parsing fallito")
                confidence += 0.2

    return RootCauseResult("tool_output_format_regression", confidence, evidences)


def _diagnose_quality_issue(trace: dict) -> RootCauseResult:
    """Hallucination vs retrieval failure decision."""
    evidences = []

    # Estrai context recuperato e completion finale
    context_text = ""
    completion_text = ""
    retrieval_score = None
    for span in trace["spans"]:
        if span.get("operation_name") == "retrieve":
            context_text = " ".join(d["content"] for d in span.get("documents", []))
            retrieval_score = span.get("cosine_score_top1")
        if span.get("operation_name") == "generate":
            completion_text = span.get("output", "")

    # Check entity in completion presente nel context
    entities = _extract_entities_simple(completion_text)
    entities_in_context = sum(1 for e in entities if e.lower() in context_text.lower())
    coverage = entities_in_context / max(len(entities), 1)

    evidences.append(f"entity coverage in context: {coverage:.0%} ({entities_in_context}/{len(entities)})")

    if coverage < 0.5 and retrieval_score and retrieval_score > 0.7:
        # Context buono, ma generator ha inventato => hallucination pure
        evidences.append(f"retrieval score {retrieval_score:.2f} ok, completion contiene entità assenti")
        return RootCauseResult("hallucination_generator", 0.85, evidences)

    if retrieval_score and retrieval_score < 0.5:
        # Retrieval debole, completion riflette context sbagliato
        evidences.append(f"retrieval cosine score {retrieval_score:.2f} (<0.5, sotto soglia)")
        return RootCauseResult("retrieval_failure_upstream", 0.9, evidences)

    return RootCauseResult("quality_issue_unclassified", 0.4, evidences)


def _extract_entities_simple(text: str) -> list[str]:
    """Named entity extraction veloce: capitalized words 3+ char.

    Per produzione sostituire con spaCy ner: nlp(text).ents
    """
    return list(set(re.findall(r"\b[A-Z][a-zA-Z]{2,}\b", text)))

Il vantaggio del decision tree manuale è che ogni decisione porta un'evidenza testuale che finisce nel report del ticket: il debug engineer apre il ticket e vede subito "category=retrieval_failure_upstream, confidence=0.9, evidenze: [retrieval score 0.41, entity coverage 25%]". Niente black box, niente "il modello dice così ma non so perché". Per il caso del bug Aprile sul recommend, il decision tree applicato post-mortem alle 50 trace flagged ha classificato 47/50 come "retrieval_failure_upstream" con confidence media 0.87, confermando che il sintomo (titolo inventato) era effetto e non causa.

Il versionamento del decision tree è essenziale: ogni modifica ai criteri va trattata come release software con changelog dettagliato. Quando aggiungo un nuovo branch (es. nuova categoria di errore emersa dal clustering settimanale), salvo la versione precedente, applico la nuova versione su un sample di 100 trace già etichettate manualmente, e verifico che la classificazione concordi con l'etichetta umana in >90% dei casi. Se concordanza scende sotto soglia, significa che il nuovo branch sta interferendo con la classificazione esistente e va riformulato. Mantengo storia delle versioni in git con tag (decision_tree_v1.3.2) per poter fare A/B test fra versioni diverse e capire impatto delle modifiche.

Una limitazione onesta del decision tree manuale: non scala oltre 60-80 if-else senza diventare ingestibile. Nel mio progetto sono attualmente a 42 branch, target massimo che mi do è 60. Oltre questa soglia consideriamo passare a sistemi più sofisticati come rule mining automatico (algoritmi tipo CN2 o RIPPER che apprendono regole interpretabili da dati etichettati) oppure weak supervision con framework tipo Snorkel. Ma per il volume di errori del ml-service portfolio (10K errori/settimana, 12-15 categorie ricorrenti) il decision tree manuale resta lo sweet spot: leggibile da chiunque del team, modificabile in 5 minuti, debuggabile linea per linea.

Sezione 5 - Span anomaly detector statistico

Detection in real-time di span che si discostano dal baseline storico, prima ancora che generino errori utente. Per ogni combinazione (operation_name, model) calcolo mean e std rolling 7 giorni di latency, tokens, cost, e a inference time confronto il valore corrente con z-score >3 come soglia anomalia. Le distribuzioni sono log-normali (long tail) quindi applico log() prima del z-score per evitare falsi positivi continui sulla coda. Il baseline è salvato in Redis come hash, ricalcolo full ogni Lunedi mattina e drift incrementale durante la settimana con exponential moving average alpha=0.1.

# span_anomaly_detector.py - detect span anomali in real-time
# Integrato nel ml-service portfolio come hook post-span

import numpy as np
import redis
import json
from datetime import datetime

REDIS = redis.Redis(host="localhost", port=6379, decode_responses=True)
ZSCORE_THRESHOLD = 3.0
EMA_ALPHA = 0.1


def baseline_key(operation: str, model: str) -> str:
    return f"span:baseline:{operation}:{model}"


def get_baseline(operation: str, model: str) -> dict | None:
    """Recupera baseline corrente per (operation, model). None se non disponibile."""
    raw = REDIS.hgetall(baseline_key(operation, model))
    if not raw or int(raw.get("count", 0)) < 100:
        return None  # campioni insufficienti per baseline affidabile
    return {
        "latency_log_mean": float(raw["latency_log_mean"]),
        "latency_log_std": float(raw["latency_log_std"]),
        "cost_log_mean": float(raw["cost_log_mean"]),
        "cost_log_std": float(raw["cost_log_std"]),
        "tokens_log_mean": float(raw["tokens_log_mean"]),
        "tokens_log_std": float(raw["tokens_log_std"]),
        "count": int(raw["count"]),
    }


def detect_span_anomaly(span: dict) -> dict:
    """Restituisce dict con flag anomalous + dettagli z-score per ogni metric."""
    operation = span.get("operation_name", "unknown")
    model = span.get("model", "unknown")
    baseline = get_baseline(operation, model)

    if not baseline:
        return {"anomalous": False, "reason": "baseline non disponibile"}

    anomalies = {}

    # Latency
    if "latency_ms" in span:
        z = (np.log1p(span["latency_ms"]) - baseline["latency_log_mean"]) / max(baseline["latency_log_std"], 0.01)
        if abs(z) > ZSCORE_THRESHOLD:
            anomalies["latency"] = {"value_ms": span["latency_ms"], "zscore": round(z, 2)}

    # Cost
    if "cost_cents" in span:
        z = (np.log1p(span["cost_cents"]) - baseline["cost_log_mean"]) / max(baseline["cost_log_std"], 0.01)
        if abs(z) > ZSCORE_THRESHOLD:
            anomalies["cost"] = {"value_cents": span["cost_cents"], "zscore": round(z, 2)}

    # Tokens
    total_tokens = span.get("tokens_in", 0) + span.get("tokens_out", 0)
    if total_tokens > 0:
        z = (np.log1p(total_tokens) - baseline["tokens_log_mean"]) / max(baseline["tokens_log_std"], 0.01)
        if abs(z) > ZSCORE_THRESHOLD:
            anomalies["tokens"] = {"value": total_tokens, "zscore": round(z, 2)}

    return {
        "anomalous": len(anomalies) > 0,
        "anomalies": anomalies,
        "baseline_count": baseline["count"],
    }


def update_baseline_incremental(span: dict) -> None:
    """Aggiorna baseline con EMA per drift incrementale."""
    operation = span.get("operation_name", "unknown")
    model = span.get("model", "unknown")
    key = baseline_key(operation, model)

    baseline = get_baseline(operation, model) or {
        "latency_log_mean": np.log1p(span.get("latency_ms", 0)),
        "latency_log_std": 0.5,
        "cost_log_mean": np.log1p(span.get("cost_cents", 0)),
        "cost_log_std": 0.5,
        "tokens_log_mean": np.log1p(span.get("tokens_in", 0) + span.get("tokens_out", 0)),
        "tokens_log_std": 0.5,
        "count": 0,
    }

    # EMA update mean (std lo ricalcolo full ogni lunedi via batch)
    new_latency = np.log1p(span.get("latency_ms", 0))
    baseline["latency_log_mean"] = EMA_ALPHA * new_latency + (1 - EMA_ALPHA) * baseline["latency_log_mean"]

    new_cost = np.log1p(span.get("cost_cents", 0))
    baseline["cost_log_mean"] = EMA_ALPHA * new_cost + (1 - EMA_ALPHA) * baseline["cost_log_mean"]

    new_tokens = np.log1p(span.get("tokens_in", 0) + span.get("tokens_out", 0))
    baseline["tokens_log_mean"] = EMA_ALPHA * new_tokens + (1 - EMA_ALPHA) * baseline["tokens_log_mean"]

    baseline["count"] += 1
    REDIS.hset(key, mapping={k: str(v) for k, v in baseline.items()})


# Hook integrato nello span finalize del tracing layer
def on_span_complete(span: dict) -> None:
    result = detect_span_anomaly(span)
    if result["anomalous"]:
        # Marca la trace come anomala per dashboard + alert
        REDIS.zadd(
            f"anomalous_spans:{datetime.utcnow().strftime('%Y-%m-%d')}",
            {span["span_id"]: datetime.utcnow().timestamp()},
        )
        # Log strutturato per Loki/Grafana
        print(json.dumps({"anomalous_span": span["span_id"], **result}))
    update_baseline_incremental(span)

Sul ml-service portfolio questo detector ha intercettato il bug del retrieve+rerank 3 giorni prima dei reclami utente: ha rilevato che il rerank_score nella media di una settimana stava salendo da 0.78 a 0.92 mentre il retrieval_score scendeva da 0.71 a 0.55 (drift). Lo span anomaly su delta(rerank-retrieval) ha sparato z-score 4.2, alert Slack, e una settimana dopo ho avuto contesto per agire prima dei downvote. La metrica derivata "delta fra score sequenziali" è uno dei custom anomaly più efficaci che ho aggiunto.

Una considerazione di tuning della soglia z-score: il valore 3 è standard statistico (corrisponde a ~0.27% di falsi positivi su distribuzione gaussiana pura), ma su distribuzioni log-trasformate ho misurato false positive rate intorno all'1% reale. Per ridurla applico il principio del multi-metric confirmation: marco lo span come "anomalous" solo se almeno 2 metriche su 3 (latency, cost, tokens) superano simultaneamente z=3. Singola metrica anomala diventa "warning" non "alert". Questa modifica ha portato false positive da ~50/giorno a ~8/giorno nel mio progetto, pur mantenendo la detection sui veri problemi (che tipicamente impattano più metriche correlate).

Per operation_name nuovi (appena introdotti dopo deploy di una feature) il sistema ha 0 baseline disponibili. Soluzione che applico: durante le prime 24 ore di traffico nuovo, il detector raccoglie dati senza emettere alert (warmup period). Dopo 24 ore se il sample size è >100 inizia a emettere alert, altrimenti estende warmup a 48 ore. Per evitare blind spot in questa finestra, monitoro le metriche in dashboard manualmente nelle prime ore post-deploy, e applico baseline globale (cross-operation, stesso model) come fallback temporaneo. Documentato come check obbligatorio nella checklist di deploy del ml-service.

Sezione 6 - Error replay framework deterministico

Quando hai identificato il pattern e ipotizzato un fix, devi testarlo prima di metterlo in produzione. Ricostruire manualmente lo state della trace originale (query utente, context retrieved, parametri modello) è tedioso e error-prone. Il pattern record-and-replay automatizza questo: ogni trace in produzione viene salvata come JSON completo (input + intermediate outputs + parametri), e il replay framework la ricostruisce localmente con seed fisso per determinismo. Ollama supporta seed parameter, quindi anche generation diventa riproducibile.

# trace_replay.py - record-and-replay deterministico per debug locale
# Uso: python trace_replay.py --replay /opt/portfolio/traces/2026-04-15/trace_xyz.json

import json
import os
from pathlib import Path
from datetime import datetime
import requests
import hashlib
import re

TRACE_STORAGE = Path("/opt/portfolio/traces")
RETENTION_DAYS = 14
PII_PATTERNS = [
    (r"\b[\w.-]+@[\w.-]+\.\w+\b", "[EMAIL]"),
    (r"\b\d{16}\b", "[CARD]"),
    (r"\b[A-Z]{6}\d{2}[A-Z]\d{2}[A-Z]\d{3}[A-Z]\b", "[CF]"),  # codice fiscale italiano
    (r"\+?\d{2,3}\s?\d{3}\s?\d{6,7}", "[PHONE]"),
]


def scrub_pii(text: str) -> str:
    """Sostituisce PII riconoscibili via regex con placeholder."""
    for pattern, replacement in PII_PATTERNS:
        text = re.sub(pattern, replacement, text)
    return text


def record_trace(trace: dict) -> str:
    """Salva trace completa su filesystem con PII scrubbing."""
    # Deep scrub su tutti i campi text
    scrubbed = json.loads(json.dumps(trace))
    for span in scrubbed.get("spans", []):
        for key in ["input", "output", "prompt", "completion"]:
            if key in span and isinstance(span[key], str):
                span[key] = scrub_pii(span[key])

    # Path: traces/YYYY-MM-DD/trace_id.json
    date_dir = TRACE_STORAGE / datetime.utcnow().strftime("%Y-%m-%d")
    date_dir.mkdir(parents=True, exist_ok=True)
    file_path = date_dir / f"{trace['trace_id']}.json"
    file_path.write_text(json.dumps(scrubbed, indent=2))

    return str(file_path)


def replay_trace(trace_path: str, ollama_url: str = "http://localhost:11434") -> dict:
    """Riproduce la trace localmente con seed fisso, mock retrieval, mock tool.

    Ritorna dict con confronto step-by-step: original vs replay.
    """
    trace = json.loads(Path(trace_path).read_text())
    replay_results = {"trace_id": trace["trace_id"], "steps": []}

    for i, span in enumerate(trace["spans"]):
        op = span["operation_name"]
        original_output = span.get("output", "")
        replay_output = None

        if op == "retrieve":
            # Mock: usa i documenti già salvati nella trace originale
            replay_output = span.get("documents", [])

        elif op == "generate":
            # Replay reale via Ollama con seed dalla trace
            payload = {
                "model": span.get("model", "llama3.1:8b"),
                "prompt": span.get("prompt", ""),
                "stream": False,
                "options": {
                    "temperature": span.get("temperature", 0.0),
                    "seed": span.get("seed", 42),
                    "num_ctx": span.get("num_ctx", 4096),
                },
            }
            response = requests.post(f"{ollama_url}/api/generate", json=payload, timeout=120)
            replay_output = response.json().get("response", "")

        elif op == "tool_call":
            # Mock: usa output cached dalla trace originale
            replay_output = span.get("output", "")

        # Confronto deterministico
        match = (replay_output == original_output) if isinstance(original_output, str) else (
            json.dumps(replay_output, sort_keys=True) == json.dumps(original_output, sort_keys=True)
        )
        replay_results["steps"].append({
            "span_index": i,
            "operation": op,
            "match": match,
            "diff_preview": _diff_preview(str(original_output), str(replay_output)) if not match else None,
        })

    replay_results["fully_reproducible"] = all(s["match"] for s in replay_results["steps"])
    return replay_results


def _diff_preview(a: str, b: str, max_chars: int = 200) -> str:
    """Anteprima compatta delle differenze fra due stringhe."""
    if len(a) > max_chars:
        a = a[:max_chars] + "..."
    if len(b) > max_chars:
        b = b[:max_chars] + "..."
    return f"original: {a}\nreplay:   {b}"


def cleanup_old_traces() -> int:
    """Rimuove trace più vecchie di RETENTION_DAYS giorni."""
    cutoff_ts = datetime.utcnow().timestamp() - RETENTION_DAYS * 86400
    removed = 0
    for date_dir in TRACE_STORAGE.iterdir():
        for trace_file in date_dir.glob("*.json"):
            if trace_file.stat().st_mtime < cutoff_ts:
                trace_file.unlink()
                removed += 1
    return removed


if __name__ == "__main__":
    import sys
    if sys.argv[1] == "--replay":
        result = replay_trace(sys.argv[2])
        print(json.dumps(result, indent=2))
    elif sys.argv[1] == "--cleanup":
        n = cleanup_old_traces()
        print(f"Rimosse {n} trace vecchie")

Volume reale sul ml-service: ~10K trace/giorno x 15KB media = 150MB/giorno = 2.1GB/settimana, ampiamente sostenibile sul VPS Hetzner CCX13. Cleanup notturno cron alle 03:00 mantiene rolling 14 giorni. Il replay si rivela utile soprattutto per validare fix prima di deploy: prendo le 50 trace fallite della categoria di errore, applico il fix proposto in branch locale, lancio replay batch, verifico che le 50 ora producano output corretti senza regressioni su altre 100 trace di controllo.

Una limitazione importante del replay determinismo è la non-perfetta riproducibilità di Ollama con seed. Anche con temperature=0 e seed fisso, ho misurato variazioni di output 1-2% delle volte tra replay successivi su modelli quantizzati q4 (probabilmente per differenze interne di scheduling GPU). Per casi che richiedono determinismo assoluto (es. regression test pre-deploy), uso modelli q8 o fp16 dove la riproducibilità sale al 99.5%. Documenta sempre il "determinism level" raggiunto nel report di replay: se sotto soglia il fix proposto va validato anche con A/B test live (rollout 5% traffico) prima del rollout completo, perché il replay offline da' solo evidenza necessaria ma non sufficiente.

Il PII scrubbing è obbligatorio per compliance GDPR sul mio progetto perché utenti europei. Le regex semplici intercettano l'80% dei casi (email, telefoni, codici fiscali, carte di credito) ma falliscono su nomi propri di persona, indirizzi parziali, identificativi business. Per il restante 20% applico passaggio aggiuntivo con microsoft/presidio che usa modelli NER lightweight per detection PII residua. Costo computazionale: ~50ms per trace, sostenibile in batch notturno ma non a runtime. Per i pochi casi in cui anche presidio non riconosce PII contestuali, accetto il rischio residuo e documento nella privacy policy che le trace possono contenere dati pseudonimizzati con retention 14 giorni e accesso ristretto a debug team.

Sezione 7 - Dashboard Grafana per error pattern

Tutti i tool precedenti generano dati strutturati (cluster_id, anomaly z-score, root_cause_category, replay_match) che vanno aggregati in dashboard operative. La mia configurazione su Grafana 11 con datasource Prometheus + Loki include 6 panel chiave per visibilità incident response. La definizione YAML del dashboard è versionata in repo, deploy via Grafana provisioning automatico.

# grafana/dashboards/llm-error-patterns.yaml
# Dashboard error pattern detection per ml-service portfolio

apiVersion: 1
providers:
  - name: 'llm-observability'
    folder: 'AI Observability'
    type: file
    options:
      path: /etc/grafana/dashboards/llm-observability

# Dashboard JSON inline (estratto dei panel principali)
dashboard:
  title: "LLM Error Patterns - ml-service portfolio"
  refresh: "30s"
  time:
    from: "now-24h"
    to: "now"

  panels:
    # Panel 1: Top 10 error categories ultime 24h
    - id: 1
      title: "Top 10 Error Categories (24h)"
      type: "barchart"
      targets:
        - expr: 'topk(10, sum by (root_cause_category) (rate(llm_errors_classified_total[24h])))'
          legendFormat: "{{root_cause_category}}"

    # Panel 2: Heatmap temporale cluster errori
    - id: 2
      title: "Error Cluster Frequency Heatmap"
      type: "heatmap"
      targets:
        - expr: 'sum by (cluster_id) (rate(llm_error_cluster_total[5m]))'
          format: "heatmap"

    # Panel 3: Span anomaly z-score distribution
    - id: 3
      title: "Span Anomaly Z-Score Distribution"
      type: "histogram"
      targets:
        - expr: 'histogram_quantile(0.95, sum by (operation_name, le) (rate(span_anomaly_zscore_bucket[1h])))'
          legendFormat: "p95 {{operation_name}}"

    # Panel 4: Replay reproducibility rate
    - id: 4
      title: "Replay Reproducibility Rate (% match)"
      type: "stat"
      targets:
        - expr: 'sum(rate(trace_replay_match_total[1h])) / sum(rate(trace_replay_total[1h])) * 100'

    # Panel 5: Similarity search auto-resolved rate
    - id: 5
      title: "Auto-Resolved via Similarity (% errori)"
      type: "gauge"
      targets:
        - expr: 'sum(rate(error_auto_resolved_via_similarity_total[1h])) / sum(rate(llm_errors_total[1h])) * 100'
      thresholds:
        - value: 30
          color: "yellow"
        - value: 50
          color: "green"

    # Panel 6: MTTR per error category
    - id: 6
      title: "MTTR per Error Category (minuti)"
      type: "table"
      targets:
        - expr: 'avg by (root_cause_category) (incident_resolution_duration_seconds / 60)'
          format: "table"

  # Annotation: deploy events per correlare error spike con release
  annotations:
    list:
      - name: "Deploy Events"
        datasource: "Loki"
        expr: '{service="deploy-tracker"} |= "deploy_completed"'
        iconColor: "rgba(0, 211, 255, 1)"
        tags: ["deploy"]

  # Variables: filtri dinamici
  templating:
    list:
      - name: "model"
        type: "query"
        query: "label_values(llm_errors_total, model)"
        multi: true
      - name: "cluster_id"
        type: "query"
        query: "label_values(llm_error_cluster_total, cluster_id)"
        multi: true

Il panel più utile in produzione è il numero 5 (Auto-Resolved via Similarity): misura la percentuale di errori che il sistema risolve automaticamente applicando un runbook già documentato per pattern simili. Sul ml-service portfolio sono al 42% di auto-resolution, target 60% entro Q3 2026 (richiede arricchire ulteriormente il database di trace etichettate). Il panel 6 (MTTR per categoria) è il più utile per priorità di lavoro: la categoria con MTTR più alto e frequenza significativa è la prossima da automatizzare con runbook + replay framework.

Le annotation di deploy sono il dettaglio che separa una dashboard utile da una dashboard che ti fa risparmiare ore di MTTR. Quando vedi visualmente sulla timeline che un error spike è iniziato esattamente 5 minuti dopo un deploy, la diagnosi è immediata e il rollback parte in 2 minuti. Senza annotation perdi 15-30 minuti a controllare se ci sono stati cambi recenti correlando manualmente git log con timeline metrics. Configurazione concreta nel mio script di deploy: ultima riga è un curl POST a /api/annotations di Grafana con tag service, version, commit_sha, deploy_user. Effort 5 minuti di setup, payback al primo incident deploy-related.

Una buona pratica che applico è di tenere TUTTE le dashboard in un repository git separato da quelli applicativi: il repo grafana-dashboards-portfolio contiene definizioni YAML versionate per dashboard, alert rules, datasource provisioning. Ogni modifica passa da pull request con review obbligatoria, evitando che debug ad-hoc rompano dashboard usate dall'on-call. Deploy automatico via Grafana provisioning sidecar che watcha il repo e ricarica configurazioni. Vantaggi: rollback in 30 secondi se una modifica rompe layout, history completa di chi ha cambiato cosa quando, possibilità di duplicare interi setup dashboard per nuovi servizi in 1 minuto.

Totale risparmio per singolo incident: 171 minuti (2 ore e 51 minuti). Su 30 incident/mese significa circa 85 ore mese, l'equivalente di mezza FTE liberata per lavoro a maggior valore (training di nuovi modelli, A/B test prompt, espansione tooling). Il ROI dell'investimento iniziale (~80 ore di sviluppo per costruire i 7 tool) si ammortizza in meno di 4 settimane.

FAQ

Trace diff: come isolo le differenze rilevanti fra una trace fallita e una successo quando i timestamp e gli ID sono sempre diversi?

Il trucco è definire una whitelist di campi semantici prima del diff e ignorare tutto il resto come rumore. Nel ml-service portfolio uso DeepDiff con il parametro exclude_paths per escludere span_id, trace_id, parent_id, start_time, end_time, latency_ms e qualsiasi UUID. Compare invece i campi semantici: operation_name, model, prompt_template_version, retrieval_top_k, tool_called, success_flag, error_type. Faccio anche normalizzazione preventiva su valori numerici: arrotondo cost_cents al decimo, latency in bucket 100ms, token count in bucket 50. Così due trace "praticamente uguali" producono diff vuoto e il pattern reale emerge. Per i contenuti testuali (prompt, completion) non comparo il testo letterale ma calcolo embedding e confronto cosine similarity: se >0.95 considero uguale. Output del diff è poi una mappa {field_path: (value_failed, value_success)} che vado a analizzare manualmente o passo a un classificatore.

Error pattern detection con clustering: DBSCAN o KMeans per raggruppare trace fallite simili?

DBSCAN vince in produzione per 3 motivi specifici al dominio LLM. Primo, non devi specificare K a priori: il numero di pattern di errore evolve nel tempo (oggi 5 categorie, domani 8 dopo deploy nuova feature), KMeans richiederebbe tuning continuo. Secondo, DBSCAN identifica naturalmente gli outlier come noise (cluster -1): i veri "casi unici" che meritano debug manuale individuale, mentre KMeans li forza in cluster esistenti distorcendo i centroidi. Terzo, la metrica di distanza per trace embedding è cosine, DBSCAN la supporta nativamente con metric='cosine'. Nel ml-service portfolio uso eps=0.3 e min_samples=5 su feature vector concatenato (embedding query 384d + one-hot model + log latency normalizzato + log cost). Risultato tipico su 10K trace fallite/settimana: 12-15 cluster significativi (>20 trace ciascuno) + 200-300 noise points. I cluster diventano "error categories" nella dashboard, ogni categoria ha runbook dedicato. KMeans lo uso solo per analisi esplorativa offline quando voglio K=5 fixed per slide presentazione.

Trace similarity embedding-based: quale modello uso per generare il vector di una trace LLM?

Nessun modello pretrained funziona out-of-the-box per "embedding di trace", va costruito a mano. La mia ricetta nel ml-service: serializzo la trace come stringa strutturata "op:retrieve|model:nomic-embed|topk:5||op:rerank|model:bge-reranker||op:generate|model:llama3.1:8b|tokens:512" poi la passo a sentence-transformers all-MiniLM-L6-v2 (lo stesso modello che già uso per RAG, zero overhead in deploy). Risultato: vector 384d che cattura la sequenza operazionale. Per query simili appare cluster naturale. Aggiungo concatenazione di feature numeriche normalizzate (latency, cost, error_count) come tail del vector, finale 392d. Per nearest neighbor uso FAISS IndexFlatIP (inner product = cosine su vettori normalizzati) con 100K trace caricate in 200MB RAM, query <5ms. Quando ho una trace fallita nuova, la embeddo e cerco i top-10 nearest: se 8/10 sono fallite e dello stesso cluster, ho identificato il pattern. Cross-encoder rerank sopra il top-10 per affinare ranking quando voglio confronto preciso.

Decision tree per root-cause: come passo dal diff fra trace alla causa reale senza false positivi?

Decision tree manuale strutturato a 5 livelli che applico nel ml-service. (1) Branch su error_type del span finale: se contiene "timeout" vai a sub-tree latency, se "JSONDecodeError" vai a tool-output-malformed, se nessun errore esplicito ma utente lamenta "risposta sbagliata" vai a hallucination-tree. (2) Per latency timeout controllo upstream span: se Ollama call >30s e GPU memory dump mostra OOM, root cause è modello troppo grande per request concurrent. (3) Per tool errors controllo input validation: se prompt template ha cambiato versione recente (diff su prompt_template_version), root cause è regression nel template. (4) Per hallucination uso il tool diff context vs completion: estraggo named entity dal completion (spaCy) e verifico presenza nel retrieved_context. Se entity assente in context = hallucination da generator. (5) Se entity presente in context ma context era sbagliato (cosine similarity context vs query <0.5) = retrieval failure, fix nel rerank. Implemento il tree come funzione Python pura con 30-40 if-else (no scikit DecisionTree training, mi serve interpretabilita 100%, niente black box). Aggiungo confidence score basato su quanti criteria matchano: alta confidence solo se tutti i check del branch sono positivi.

Span anomaly detection con statistiche: quale baseline uso per dichiarare uno span "anomalo"?

Baseline rolling window 7 giorni per operation_name + model combinato, con z-score > 3 come soglia. Implementazione concreta nel ml-service: ogni notte job batch (Python + Pandas) legge tutti gli span dell'ultima settimana raggruppa per (operation_name, model) e calcola mean/std di latency_ms, tokens_in, tokens_out, cost_cents. Salva in Redis come hash "baseline:retrieve:nomic-embed" = {latency_mean: 45, latency_std: 12, ...}. A inference time, dopo ogni span appena finito interrogo il baseline corrispondente, calcolo z-score (current - mean) / std per ogni metric, se uno >3 marco lo span anomalous=true e lo includo in alert e dashboard. Edge case importante: span con count <100 nella settimana non hanno baseline statisticamente affidabile, in quel caso uso baseline globale del operation_name (cross-model) come fallback. Per metriche log-normali (cost/latency) applico log() prima di z-score altrimenti la coda lunga genera falsi positivi continui. Ogni Lunedi mattina ricalcolo full baseline: durante la settimana drift incrementale aggiornato con exponential moving average alpha=0.1.

Error replay framework: come riproduco deterministicamente una trace fallita in produzione per fare debug locale?

Pattern record-and-replay che ho implementato nel ml-service. (1) Record: ogni trace in produzione viene serializzata in formato JSON che include input completi (query utente, system prompt, context retrieved, tool inputs/outputs, model parameters, seed se random) e salvata in S3-compatible storage o filesystem (nel mio caso /opt/portfolio/traces/YYYY-MM-DD/trace_id.json). Retention 14 giorni rolling. (2) Replay locale: script Python carica il JSON, ricostruisce il context (mock retrieval con docs salvati, mock tool con risposta cached), invoca il modello con stesso seed e stessi parametri. Se modello è deterministico (temperature=0, seed fixed) il replay riproduce esattamente l'errore originale. (3) Per modelli stocastici (temperature>0) impostazione seed fisso lo rende deterministico anche su Ollama. (4) Bonus: il replay può iterare su modifiche del prompt o context per testare fix prima di deploy. Salvataggio dimensione tipica: 5-50KB per trace, 10K trace/giorno = 250MB/giorno = 3.5GB/settimana, gestibile su VPS con cleanup notturno. Privacy: applico PII scrubbing sul JSON prima di salvarlo (regex email/phone/CF + LLM-based detection per nomi).

Esempio reale Federico: come hai debuggato l'hallucination del retrieve+rerank nel ml-service portfolio?

Caso concreto Aprile 2026 sul mio ml-service portfolio. Sintomo: utenti del tool /internal/ai/recommend riportavano raccomandazioni articolo "inventate" (titoli che non esistono nel database). Frequenza 5-8% delle query. Workflow di debug applicato. (1) Identificazione: ho preso 50 trace flagged dagli utenti via downvote button + 50 trace successo, ho calcolato embedding di tutte le 100 trace e fatto t-SNE plot, le fallite formavano un cluster denso separato. (2) Diff: DeepDiff fra 5 fallite vs 5 successe simili ha mostrato che nelle fallite il rerank score era >0.9 ma il context recuperato aveva cosine similarity con query <0.4 (rerank troppo confident su context sbagliato). (3) Root cause confermata: cross-encoder bge-reranker-base ha bias verso documenti lunghi perché max_length=512 troncava contesto, perdendo signal negativo. Documenti lunghi off-topic ricevevano score alto. (4) Fix: cambio modello a bge-reranker-large (max_length 1024) + soglia minima 0.7 di cosine retrieval prima di feed al reranker (filtro a monte). Ho creato regression test con le 50 trace fallite originali, post-fix tutte correttamente filtrate prima di generation. (5) Prevenzione: aggiunto span anomaly detector specifico per "rerank_score - retrieval_score" delta, alert se delta >0.5 (segno di rerank overconfidence). Tempo totale debug: 4 ore, di cui 30 minuti il primo trace_diff manuale e poi tooling automatizzato per il resto.

Conclusione

Il debug di trace LLM è il vero collo di bottiglia operativo dei sistemi AI in produzione: senza tooling dedicato, ogni incident assorbe 2-3 ore di un engineer senior, e con 30 incident/mese l'azienda paga il prezzo di una FTE solo per debug reattivo. I 7 tool costruiti in questo articolo (trace diff con DeepDiff, error pattern clustering DBSCAN, similarity search via embedding+FAISS, decision tree manuale interpretabile, span anomaly statistico con baseline rolling, replay framework deterministico, dashboard Grafana) abbattono il MTTR da ore a minuti, riducendo il tempo medio per incident da 3 ore a 9 minuti totali (171 minuti risparmiati per ticket). L'investimento iniziale di ~80 ore di sviluppo si ammortizza in meno di 4 settimane.

L'approccio chiave non è "comprare uno strumento commerciale magico" ma costruire un sistema modulare in cui ogni tool produce dati strutturati che alimentano il successivo: lo span anomaly detector marca le trace sospette, il clustering le aggrega in categorie, il similarity search trova casi storici simili già risolti, il decision tree inferisce root-cause con confidence interpretabile, il replay framework valida il fix proposto offline, la dashboard Grafana monitora il funnel end-to-end. Ogni componente è indipendente ma compositivo: puoi adottare il primo (trace diff) in 1 giorno e progressivamente aggiungere gli altri.

Il caso reale dell'hallucination retrieval+rerank del mio ml-service portfolio dimostra il pattern: senza tooling, sarebbero state probabilmente 2 settimane di analisi manuale prima di identificare che il problema era nel cross-encoder bge-reranker-base (non nel generator come sembrava). Con tooling sono state 4 ore concentrate in una sessione, fix in produzione il giorno dopo, regression test per prevenzione. Replicabile su qualunque stack LLM, da Ollama on-prem a OpenAI managed: il framework è agnostico al provider perché opera sul layer di trace, non sul modello.

Prossimo episodio: end-to-end workflow di instrumentation: setup completo Langfuse self-hosted per agente in sviluppo, criteri di evaluation pass per andare in produzione, deployment con cost SLA guarantee e monitoring integrato Grafana + Langfuse. L'esempio sarà il portfolio app agent reale con profiling completo cost/latency/quality.