Observability agent: tracing span-per-thought con OpenTelemetry

Observability agent: tracing span-per-thought con OpenTelemetry

Il tuo agent ha fallito dopo 47 tool call. Il log dice AssertionError: unexpected output format. Benvenuto nel problema che affligge il 90% dei team che mettono agent in produzione: sai che qualcosa è andato storto, ma non sai a quale step e perché.

Il logging tradizionale - quello che usi su una REST API - non è progettato per sistemi non deterministici che decidono autonomamente il proprio percorso di esecuzione. Un agent ReAct con 47 iterazioni produce un flusso causale ramificato: ogni thought genera una decisione, ogni decisione chiama un tool, ogni tool produce output che modifica il contesto del thought successivo. In questo albero di dipendenze, un log lineare è inutile quanto leggere un romanzo partendo dall'ultimo capitolo.

In questo episodio costruiamo observability granulare con OpenTelemetry custom spans: 1 span per thought, 1 per tool call, 1 per reflection. Confrontiamo LangSmith, Phoenix, Helicone e Langfuse su criteri tecnici concreti. Mostriamo come ho applicato questi pattern al ml-service del mio portfolio - riducendo il tempo di root-cause analysis da 40 minuti a meno di 5 secondi su un agent da 47 step.

Sezione 1 - Perché il trace lineare non basta per un agent

Considera questo scenario reale: un agent di analisi documentale riceve una query complessa. Decide di dividere il problema in 3 sotto-query (thought 1), chiama un tool di ricerca vettoriale per ciascuna (tool call 2, 3, 4), riceve risultati parziali, decide che uno dei risultati è insufficiente (thought 5), riformula la query e richiama il tool (tool call 6), rivaluta i tre risultati insieme (reflection 7), e produce la risposta finale.

Con logging standard, il tuo output è una sequenza piatta di righe con timestamp. Non vedi che la tool call 6 è figlia del thought 5, che è stato generato perché il risultato della tool call 4 aveva un confidence_score sotto soglia. Quando l'agente fallisce alla tool call 41 di 47, hai 41 righe di log da leggere in sequenza sperando di trovare il contesto corretto.

Con span gerarchici OTel, hai un albero: apri il nodo corrispondente al thought che ha generato la tool call 41, vedi immediatamente l'input e l'output di quel thought, quali attributi ha scritto sulla trace, e perché il tool successivo ha prodotto un errore. Il tempo di orientamento passa da minuti a secondi.

Il modello mentale corretto è questo: ogni trace è un'istanza di esecuzione completa dell'agent, dall'input alla risposta finale. All'interno di ogni trace, gli span sono le operazioni atomiche, annidati gerarchicamente a rispecchiare il flusso decisionale. Un thought span è padre delle tool call che ha generato. Una reflection span è figlia del thought che l'ha attivata.

Sezione 2 - Inizializzazione OTel Python SDK per agent

OpenTelemetry Python SDK v1.40+ è production-stable. Configurarlo correttamente per un agent richiede alcune scelte non ovvie: il BatchSpanProcessor invece del SimpleSpanProcessor (evita di bloccare il thread dell'agent durante l'esportazione), un resource che identifica il servizio, e un exporter compatibile con il backend che userai.

# agent_telemetry.py - inizializzazione OTel per agent service
# Installa: pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-grpc

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider, SpanLimits
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
import os

def init_agent_tracer(
    service_name: str = "ml-service-agent",
    otlp_endpoint: str = "http://localhost:4317",
) -> trace.Tracer:
    """
    Inizializza TracerProvider con BatchSpanProcessor.
    BatchSpanProcessor: esporta in background, non blocca l'agent.
    SpanLimits: previene memory leak su agent con 100+ attributi per span.
    """
    resource = Resource.create({
        "service.name": service_name,
        "service.version": os.getenv("APP_VERSION", "dev"),
        "deployment.environment": os.getenv("ENV", "local"),
    })

    # Limiti espliciti: agent ad alta iterazione rischiano span con troppi attributi
    span_limits = SpanLimits(
        max_events=256,        # eventi per span (es. log di ogni token streaming)
        max_attributes=128,    # attributi per span
        max_links=32,          # link cross-trace (es. session correlati)
    )

    provider = TracerProvider(resource=resource, span_limits=span_limits)

    # Exporter OTLP gRPC: compatibile con Phoenix, Langfuse, Jaeger, Grafana Tempo
    exporter = OTLPSpanExporter(
        endpoint=otlp_endpoint,
        insecure=True,  # TLS in produzione: insecure=False + credentials
    )

    # BatchSpanProcessor: flush asincrono ogni 5s o quando buffer raggiunge 512 span
    provider.add_span_processor(
        BatchSpanProcessor(
            exporter,
            max_export_batch_size=512,
            export_timeout_millis=5000,
            schedule_delay_millis=5000,
        )
    )

    trace.set_tracer_provider(provider)
    return trace.get_tracer(service_name)


# Istanza globale - usata da tutti i moduli agent
AGENT_TRACER = init_agent_tracer(
    otlp_endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
)

Il punto critico qui è BatchSpanProcessor. In un agent con 47 step, ogni step aspetta la risposta LLM: i tempi di attesa dominano su tutto. Il SimpleSpanProcessor invece flush ogni span in modo sincrono, aggiungendo latency a ogni operazione. Con Batch, i trace vengono accumulati e inviati al backend ogni 5 secondi in background, lasciando il thread dell'agent libero.

Sezione 3 - Span per thought: decorator pattern

Il pattern più ergonomico per aggiungere span a ogni thought dell'agent è un decorator funzionale. Questo approccio ha un vantaggio fondamentale: aggiungi observability senza modificare la logica dell'agent, rispettando il principio di separazione delle responsabilità.

# agent_spans.py - decoratori per span gerarchici thought/tool/reflection

import functools
import time
from typing import Callable, Any, Optional
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

from agent_telemetry import AGENT_TRACER


def thought_span(thought_name: str, capture_output: bool = True):
    """
    Decorator per wrappare un thought dell'agent in uno span OTel.
    Applica GenAI semantic conventions: gen_ai.operation.name = 'invoke_agent'.
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            # Nome span secondo convenzione OTel GenAI: "invoke_agent {agent_name}"
            span_name = f"agent.thought.{thought_name}"

            with AGENT_TRACER.start_as_current_span(span_name) as span:
                # Attributi GenAI semantic conventions ufficiali
                span.set_attribute("gen_ai.operation.name", "invoke_agent")
                span.set_attribute("gen_ai.agent.name", thought_name)
                span.set_attribute("agent.thought.start_time", time.time())

                # Cattura input del thought se disponibile nei kwargs
                if "context" in kwargs and kwargs["context"]:
                    ctx = kwargs["context"]
                    span.set_attribute(
                        "agent.thought.context_length",
                        len(str(ctx)) if ctx else 0
                    )
                    span.set_attribute(
                        "agent.thought.iteration",
                        kwargs.get("iteration", 0)
                    )

                try:
                    result = await func(*args, **kwargs)

                    # Cattura output: attenzione a PII - in produzione usa event log
                    if capture_output and result:
                        span.set_attribute(
                            "agent.thought.output_length",
                            len(str(result))
                        )
                        span.set_attribute(
                            "agent.thought.decision",
                            str(result.get("decision", "unknown"))[:256]
                            if isinstance(result, dict) else str(result)[:256]
                        )

                    span.set_status(Status(StatusCode.OK))
                    return result

                except Exception as exc:
                    # Span in stato ERROR: visibile come rosso nel trace viewer
                    span.set_status(
                        Status(StatusCode.ERROR, description=str(exc))
                    )
                    span.record_exception(exc)
                    raise

        return wrapper
    return decorator

Il decorator thought_span cattura automaticamente il duration del thought (OTel misura start/end dello span), la lunghezza del contesto in input, l'output decisionale troncato a 256 caratteri, e qualsiasi eccezione con stack trace completo. Nota l'avvertenza su PII: in produzione, non scrivere mai il testo grezzo delle richieste utente come attributo di span - usa invece span.add_event() con un filtro di sanitizzazione, oppure configura un sampler che esclude quei campi.

Sezione 4 - Span per tool call: gerarchia e propagazione del contesto

Le tool call sono i punti più critici da tracciare. Sono qui che gli agent falliscono più spesso: tool con output formato errato, timeout, rate limit, o semplicemente una risposta che l'agent non sa interpretare. Ogni tool call deve avere uno span figlio dello span thought che l'ha generata - OTel propaga automaticamente il contesto parent-child purché tu usi start_as_current_span all'interno dello span thought aperto.

# tool_spans.py - span per tool call con gerarchia parent-child automatica

import time
import json
from typing import Any, Optional
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

from agent_telemetry import AGENT_TRACER


class ToolCallTracer:
    """
    Context manager per tracciare tool call come span figlio del thought corrente.
    Usa con: async with ToolCallTracer("search_documents", input_params) as tc:
    """

    def __init__(
        self,
        tool_name: str,
        input_params: dict,
        tool_version: str = "1.0",
    ):
        self.tool_name = tool_name
        self.input_params = input_params
        self.tool_version = tool_version
        self.span: Optional[trace.Span] = None
        self._start_time = 0.0

    async def __aenter__(self) -> "ToolCallTracer":
        # span_name segue convenzione OTel: "execute_tool {tool_name}"
        self.span = AGENT_TRACER.start_span(
            f"agent.tool.{self.tool_name}",
            # Eredita automaticamente il context del thought span corrente
            context=trace.get_current_span().get_span_context(),
        )
        self._start_time = time.time()

        # Attributi GenAI: tool call semantics
        self.span.set_attribute("gen_ai.operation.name", "execute_tool")
        self.span.set_attribute("gen_ai.tool.name", self.tool_name)
        self.span.set_attribute("gen_ai.tool.version", self.tool_version)

        # Input params serializzati - tronca a 1024 char per evitare span giganti
        input_str = json.dumps(self.input_params, default=str)[:1024]
        self.span.set_attribute("agent.tool.input", input_str)
        self.span.set_attribute(
            "agent.tool.input_param_count", len(self.input_params)
        )

        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> bool:
        duration_ms = (time.time() - self._start_time) * 1000
        self.span.set_attribute("agent.tool.duration_ms", round(duration_ms, 2))

        if exc_type is not None:
            self.span.set_status(
                Status(StatusCode.ERROR, description=str(exc_val))
            )
            self.span.record_exception(exc_val)
            self.span.set_attribute("agent.tool.success", False)
        else:
            self.span.set_status(Status(StatusCode.OK))
            self.span.set_attribute("agent.tool.success", True)

        self.span.end()
        # Non sopprimere l'eccezione: l'agent la gestisce a livello superiore
        return False

    def set_output(self, output: Any, output_type: str = "json") -> None:
        """Chiama dopo l'esecuzione del tool per registrare l'output."""
        if self.span is None:
            return
        output_str = json.dumps(output, default=str)[:2048]
        self.span.set_attribute("agent.tool.output", output_str)
        self.span.set_attribute("agent.tool.output_type", output_type)
        self.span.set_attribute(
            "agent.tool.output_length", len(str(output))
        )


# Esempio di utilizzo all'interno di un thought span
@thought_span("analyze_query")
async def analyze_query_thought(query: str, context: dict, iteration: int = 0):
    """Thought che decide quale tool usare e lo esegue."""

    # Qui siamo dentro uno span "agent.thought.analyze_query"
    search_params = {"query": query, "top_k": 5, "filter": context.get("filter")}

    # Tool call: span figlio automaticamente del thought span corrente
    async with ToolCallTracer("search_documents", search_params) as tc:
        # Chiamata reale al tool
        results = await search_documents_tool(query=query, **search_params)
        tc.set_output(results, output_type="list")

    return {"decision": "use_results", "results": results}

Il punto fondamentale nella gerarchia degli span è la propagazione automatica del contesto. Quando ToolCallTracer.__aenter__ chiama trace.get_current_span(), OTel restituisce lo span thought attivo nel thread corrente. Il tool call span viene così collegato automaticamente come figlio. Nel trace viewer vedrai l'albero completo: ogni thought con i suoi figli tool call, indentati e colorati per stato (verde=OK, rosso=ERROR).

Sezione 5 - Span per reflection: rilevare loop prima che costino

Le reflection sono il momento in cui l'agent valuta la qualità del proprio output e decide se iterate. Sono anche il punto dove un agent può entrare in loop infiniti - vedi EP04 su debugging loop per approfondire. Tracciare ogni reflection con span separato ti permette di rilevare questi pattern direttamente sul trace: se vedi 15 span reflection consecutivi sullo stesso parent thought, hai trovato il tuo loop.

# reflection_spans.py - span per reflection con rilevamento loop

from dataclasses import dataclass, field
from typing import List
import time
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

from agent_telemetry import AGENT_TRACER


@dataclass
class ReflectionContext:
    """Stato della reflection corrente - immutabile per istanza."""
    thought_span_id: str
    iteration: int
    previous_scores: List[float] = field(default_factory=list)


def reflection_span(func):
    """
    Decorator per span di reflection.
    Aggiunge rilevamento automatico di loop tramite confronto score.
    """
    @functools.wraps(func)
    async def wrapper(
        ctx: ReflectionContext, output: str, *args, **kwargs
    ):
        span_name = f"agent.reflection.iter_{ctx.iteration}"

        with AGENT_TRACER.start_as_current_span(span_name) as span:
            span.set_attribute("agent.reflection.iteration", ctx.iteration)
            span.set_attribute(
                "agent.reflection.parent_thought", ctx.thought_span_id
            )
            span.set_attribute(
                "agent.reflection.output_length", len(output)
            )

            result = await func(ctx, output, *args, **kwargs)

            # Salva quality score per rilevamento loop
            quality_score = result.get("quality_score", 0.0)
            span.set_attribute(
                "agent.reflection.quality_score", quality_score
            )

            # Rilevamento loop: score non migliora dopo 3 iterazioni
            if len(ctx.previous_scores) >= 3:
                recent = ctx.previous_scores[-3:]
                improvement = quality_score - min(recent)
                span.set_attribute(
                    "agent.reflection.score_improvement", improvement
                )

                if improvement < 0.02:
                    # Loop rilevato - evento sul span per alert immediato
                    span.add_event(
                        "potential_reflection_loop_detected",
                        attributes={
                            "recent_scores": str(recent),
                            "current_score": quality_score,
                            "recommendation": "force_terminate",
                        }
                    )
                    span.set_attribute("agent.reflection.loop_suspected", True)
                    result["should_terminate"] = True

            span.set_status(Status(StatusCode.OK))
            return result

    return wrapper

L'aggiunta di un span.add_event() per loop sospetti è particolarmente utile: gli eventi OTel sono ricercabili nella maggior parte dei backend di observability. Puoi creare un alert su Phoenix o Langfuse che si attiva ogni volta che compare un evento con nome potential_reflection_loop_detected, notificandoti in tempo reale invece di scoprirlo durante il post-mortem.

Sezione 6 - Caso reale: ml-service portfolio, agent 47 step

Nel ml-service del mio portfolio, gestisco un agent di raccomandazione articoli che funziona su query complesse degli utenti. L'agent usa un pattern ibrido: embedding similarity + reranking + riflessione sulla coerenza della risposta. Senza tracing, un failure silenzioso (il reranker restituiva risultati non pertinenti per query in lingue non-IT) ci metteva 40 minuti a diagnosticare perché dovevamo leggere l'intera sequenza di log.

Dopo aver integrato OTel con span per ogni step, il root-cause di quel failure specifico è diventato visibile in 5 secondi: aprendo il trace dello step 31 di 47 nel viewer Phoenix locale, vedevamo immediatamente che il tool call rerank_documents aveva agent.tool.output_length = 0 - output vuoto - mentre l'input aveva agent.tool.input_param_count = 3 con un parametro language="de" non supportato dal modello di reranking installato.

# ml_service/agent/recommend_agent.py - integrazione reale FastAPI + OTel
# Semplificato per chiarezza

from fastapi import APIRouter, Depends
from opentelemetry import trace
import time

from agent_telemetry import AGENT_TRACER
from agent_spans import thought_span
from tool_spans import ToolCallTracer

router = APIRouter(prefix="/internal/ai")


@router.post("/recommend")
async def recommend_articles(request: RecommendRequest):
    """
    Endpoint raccomandazione con tracing granulare.
    Ogni chiamata genera 1 trace root con N span figlio.
    """
    # Span root: rappresenta l'intera esecuzione dell'agent per questa request
    with AGENT_TRACER.start_as_current_span("agent.recommend.root") as root_span:
        root_span.set_attribute("gen_ai.operation.name", "invoke_agent")
        root_span.set_attribute("gen_ai.agent.name", "recommend_agent")
        root_span.set_attribute("agent.request.query_length", len(request.query))
        root_span.set_attribute(
            "agent.request.language", request.language or "it"
        )
        root_span.set_attribute(
            "gen_ai.conversation.id", request.session_id or "anonymous"
        )

        start_total = time.time()
        iteration = 0

        try:
            # Step 1: embedding della query utente
            async with ToolCallTracer(
                "embed_query",
                {"query": request.query, "model": "all-MiniLM-L6-v2"}
            ) as tc:
                embedding = await embed_query(request.query)
                tc.set_output(
                    {"embedding_dim": len(embedding), "norm": float(sum(x**2 for x in embedding)**0.5)}
                )

            # Step 2: retrieval vettoriale (RediSearch HNSW)
            async with ToolCallTracer(
                "vector_search",
                {"top_k": 20, "language": request.language}
            ) as tc:
                candidates = await vector_search(embedding, top_k=20)
                tc.set_output({"candidate_count": len(candidates)})

            # Step 3-N: iterazione thought + rerank + reflection
            results = await iterative_rerank_thought(
                query=request.query,
                candidates=candidates,
                language=request.language,
                max_iterations=5,
            )

            total_ms = (time.time() - start_total) * 1000
            root_span.set_attribute("agent.total_duration_ms", round(total_ms, 2))
            root_span.set_attribute(
                "agent.final_result_count", len(results)
            )

            return {"recommendations": results, "trace_id": get_current_trace_id()}

        except Exception as exc:
            root_span.set_status(
                trace.Status(trace.StatusCode.ERROR, str(exc))
            )
            root_span.record_exception(exc)
            raise


def get_current_trace_id() -> str:
    """Restituisce il trace ID corrente per correlazione nei log applicativi."""
    ctx = trace.get_current_span().get_span_context()
    if ctx and ctx.is_valid:
        return format(ctx.trace_id, "032x")
    return "no-trace"

Includere il trace_id nella risposta è una pratica che consiglio fortemente: quando un utente riporta un errore, puoi chiedergli di copiarti il trace_id dalla risposta e trovare immediatamente l'intera esecuzione nel backend observability. Nessuna correlazione manuale su timestamp o log.

Sezione 7 - LangSmith vs Phoenix vs Helicone vs Langfuse: confronto tecnico

Quattro strumenti dominano il mercato dell'observability per LLM agent. Ognuno ha un punto di forza primario e compromessi diversi. La scelta dipende dal tuo contesto: stack, budget, requisiti di privacy, e se usi LangChain o un agent custom.

Per il ml-service del mio portfolio ho scelto Phoenix OSS self-hosted su Docker. I motivi: zero costo (Ollama on-prem, no cloud), nessun dato utente che esce dall'infrastruttura, supporto nativo OTel GenAI semantic conventions, e un'interfaccia di trace viewer che rispecchia esattamente la struttura ad albero degli span. Langfuse è l'alternativa più solida se hai bisogno di multi-tenancy e ClickHouse per volumi elevati.

Helicone è interessante per prototipazione rapida - letteralmente cambi la base URL della chiamata LLM e hai il logging - ma non è adatto per agent con logica custom perché traccia solo le singole LLM call, non la gerarchia thought/tool/reflection. LangSmith rimane il re per chi usa LangChain, ma il lock-in è reale e il pricing scala velocemente.

Sezione 8 - Phoenix locale: setup in 10 minuti

Phoenix OSS è la scelta consigliata per chi vuole tracing granulare agent senza dipendenze cloud. Funziona con Docker Compose, espone un'interfaccia web locale su porta 6006, e accetta span OTel via OTLP su porta 4317.

# docker-compose.phoenix.yml - Phoenix OSS locale per agent tracing
version: "3.9"

services:
  phoenix:
    image: arizephoenix/phoenix:latest
    container_name: agent-phoenix
    ports:
      - "6006:6006"   # UI web
      - "4317:4317"   # OTLP gRPC receiver
      - "4318:4318"   # OTLP HTTP receiver (alternativa)
    volumes:
      - phoenix_data:/mnt/data  # Persistenza trace su disco
    environment:
      - PHOENIX_WORKING_DIR=/mnt/data
      - PHOENIX_PORT=6006
    restart: unless-stopped

volumes:
  phoenix_data:
    driver: local

# Avvia Phoenix e verifica
docker compose -f docker-compose.phoenix.yml up -d

# Verifica che l'OTLP receiver sia attivo
curl -s http://localhost:6006/healthz | python3 -m json.tool

# Nel tuo agent Python, punta l'exporter a localhost:4317
export OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:4317"

# Apri la UI per vedere i trace
# http://localhost:6006

Sezione 9 - Storage strategy: cold vs hot per trace a 100+ step

Un agent con 100 step e 50 attributi per span produce circa 150KB di dati per singolo trace. Moltiplica per 1000 run/giorno e hai 150MB di trace al giorno. Dopo 30 giorni, 4.5GB. Dopo 1 anno, oltre 50GB. Senza una storage strategy, il tuo backend di observability diventa lento e costoso.

La strategia più pragmatica per team piccoli e medi è sampling + hot storage: tieni il 100% dei trace in errore (failure sono rari ma critici), il 10-20% dei trace di successo campionati casualmente, e spedisci tutto il resto su cold storage compresso. Questo riduce il volume di circa 85% mantenendo la capacità di debug su tutti i failure.

# sampling_strategy.py - sampler OTel per agent con priorità ai failure

from opentelemetry.sdk.trace.sampling import (
    Sampler, SamplingResult, Decision, ALWAYS_ON, ALWAYS_OFF
)
from opentelemetry.trace import SpanKind
from opentelemetry.context import Context
import random


class AgentFailurePrioritySampler(Sampler):
    """
    Sampler custom: 100% failure, X% success.
    Si applica a livello di trace root: o tutto il trace o niente.
    """

    def __init__(self, success_sample_rate: float = 0.15):
        # 15% dei run riusciti, 100% dei failure
        self.success_rate = success_sample_rate

    def should_sample(
        self,
        parent_context: Context,
        trace_id: int,
        name: str,
        kind: SpanKind,
        attributes: dict,
        links: list,
    ) -> SamplingResult:
        # Failure: campiona sempre (attributo scritto dallo span root al termine)
        # Nota: al momento dello start non sappiamo ancora se fallirà.
        # Strategia: campiona basandosi sul nome span e su attributi di request.
        # I failure vengono "promossi" a posteriori tramite tail-based sampling.

        # Per ora: campiona tutto (tail sampler gestisce la retention)
        if random.random() < self.success_rate:
            return SamplingResult(Decision.RECORD_AND_SAMPLE)
        else:
            # Registra ma non esporta: disponibile localmente per debug immediato
            return SamplingResult(Decision.RECORD_ONLY)

    def get_description(self) -> str:
        return f"AgentFailurePriority(success_rate={self.success_rate})"

Sezione 10 - Query pattern per incident: root-cause in 5 minuti

Avere i trace è necessario ma non sufficiente. Devi sapere come interrogarli quando un incident ti sveglia alle 3 di notte. Ecco il workflow che uso sul portfolio con Phoenix locale.

# trace_query.py - query programmatica su Phoenix per root-cause analysis
# Richiede: pip install arize-phoenix-client

import phoenix as px
from phoenix.client import Client
from datetime import datetime, timedelta


def find_failed_traces_last_hour(project_name: str = "ml-service-agent"):
    """
    Recupera tutti i trace in errore dell'ultima ora.
    Utile come primo step di incident response.
    """
    client = Client()  # Connette a http://localhost:6006 di default

    # Filtra per span root in ERROR nell'ultima ora
    now = datetime.utcnow()
    one_hour_ago = now - timedelta(hours=1)

    traces = client.get_traces(
        project_name=project_name,
        filters=[
            {"attribute": "span.status_code", "value": "ERROR"},
            {"attribute": "span.start_time", "operator": ">=", "value": one_hour_ago.isoformat()},
        ],
        limit=50,
    )

    # Per ogni trace in errore, stampa il breadcrumb dell'errore
    for trace in traces:
        error_spans = [s for s in trace.spans if s.status_code == "ERROR"]
        if error_spans:
            first_error = error_spans[0]
            print(f"Trace: {trace.trace_id[:16]}...")
            print(f"  Fallito a: {first_error.name}")
            print(f"  Iterazione: {first_error.attributes.get('agent.thought.iteration', 'N/A')}")
            print(f"  Tool: {first_error.attributes.get('gen_ai.tool.name', 'N/A')}")
            print(f"  Errore: {first_error.status_message}")
            print(f"  Durata totale: {trace.duration_ms:.0f}ms")
            print()

    return traces


def compare_failure_pattern(trace_id: str, project_name: str = "ml-service-agent"):
    """
    Analisi dettagliata di un singolo trace.
    Stampa tutti gli span con attributi chiave per root-cause rapido.
    """
    client = Client()
    trace = client.get_trace(project_name=project_name, trace_id=trace_id)

    print(f"Trace {trace_id} - {len(trace.spans)} span totali")
    print("=" * 60)

    for i, span in enumerate(sorted(trace.spans, key=lambda s: s.start_time)):
        indent = "  " * span.depth
        status_icon = "OK" if span.status_code == "OK" else "ERR"
        print(
            f"{indent}[{status_icon}] {i+1:02d}. {span.name} "
            f"({span.duration_ms:.0f}ms)"
        )
        if span.status_code == "ERROR":
            print(f"{indent}       ^ ERRORE: {span.status_message}")
            # Attributi chiave per il debug
            for key in ["gen_ai.tool.name", "agent.tool.input", "agent.thought.decision"]:
                val = span.attributes.get(key)
                if val:
                    print(f"{indent}       {key}: {str(val)[:80]}")

Questa coppia di funzioni - find_failed_traces_last_hour e compare_failure_pattern - è la base del mio runbook di incident response. Nella pratica: primo, trovo tutti i trace in errore dell'ultima ora con 3 righe di output per trace. Secondo, prendo il trace ID del caso più rappresentativo e lo passo a compare_failure_pattern che stampa l'albero completo. Il failure è sempre visibile entro i primi 10 righe di output: lo span in errore ha il prefisso [ERR] e i suoi attributi chiave sono stampati immediatamente sotto.

FAQ

Perché il logging standard non basta per un LLM agent?

Un agent esegue step non deterministici in sequenza non prevedibile. Il logging lineare non cattura la gerarchia causa-effetto tra thought, tool call e reflection. Con span gerarchici OpenTelemetry vedi esattamente quale step ha introdotto l'errore e come si è propagato nel flusso successivo.

Qual è la differenza tra un trace e uno span nell'agent observability?

Un trace rappresenta l'intera esecuzione dell'agent dall'input all'output finale - è il "film" completo. Uno span è un'operazione atomica all'interno del trace: un thought, una tool call, una reflection - sono i "fotogrammi". Gli span sono annidati gerarchicamente e formano un albero che rispecchia il flusso decisionale dell'agent. I trace a loro volta possono essere correlati tra sessioni diverse tramite gen_ai.conversation.id.

Posso usare OpenTelemetry con Ollama locale e senza cloud?

Sì, è esattamente la configurazione che uso nel portfolio. OpenTelemetry funziona indipendentemente dal provider LLM. Phoenix OSS gira su Docker Compose locale, riceve span tramite OTLP gRPC su porta 4317, e non trasmette nulla all'esterno. L'unico requisito è che il tuo agent Python abbia opentelemetry-sdk installato e l'exporter puntato a localhost:4317.

Quanto overhead introduce il tracing granulare su un agent in produzione?

Con BatchSpanProcessor l'overhead è tipicamente inferiore a 2ms per span, eseguito in background. Su un agent con 50 step, l'impatto totale sul latency è sotto i 100ms - accettabile rispetto a decine di secondi di attesa LLM. In produzione ad alto volume si applica sampling: 100% sui failure, 10-20% sui run di successo, riducendo il volume di dati senza perdere copertura sugli errori critici.

Cos'è la storage strategy cold/hot per i trace agent?

Hot storage mantiene i trace recenti (7-30 giorni) in un backend veloce come ClickHouse o Elasticsearch per query rapide durante incident response - latency sotto i 100ms. Cold storage archivia trace più vecchi in object storage compresso in formato Parquet (MinIO self-hosted per privacy), accessibile per analisi aggregate ma con latency di 10-60 secondi. La regola pratica: incident response richiede hot, analisi longitudinale e training dataset richiedono cold.

LangSmith funziona con agent non LangChain?

Sì. LangSmith ora accetta trace OpenTelemetry standard via OTLP endpoint, quindi puoi inviare span da qualsiasi agent Python - anche scritto con codice custom, FastAPI, o un framework proprietario - senza usare LangChain. L'integrazione nativa LangChain offre però più attributi auto-estratti e una UI ottimizzata per i pattern LangGraph. Se non usi LangChain, Phoenix o Langfuse offrono un'esperienza migliore con agent OTLP custom.

Come traccio una reflection loop senza creare span infiniti?

Ogni reflection è uno span figlio del thought che l'ha generata. Si imposta un max_reflections hard limit (tipicamente 5-7) nel codice dell'agent, non nel sistema di tracing. Il trace mostra chiaramente se l'agent è in loop: vedi N span reflection sullo stesso parent thought con agent.reflection.quality_score stazionario. L'aggiunta di un span.add_event("potential_reflection_loop_detected") - come mostrato nel codice - permette alert automatici sul backend observability.

Conclusioni

Costruire observability granulare per LLM agent non è opzionale in produzione: è la differenza tra 5 secondi e 40 minuti di root-cause analysis. Il modello span-per-thought, span-per-tool-call, span-per-reflection con OTel ti dà un albero causale completo di ogni esecuzione dell'agent. Le GenAI semantic conventions ufficiali OTel assicurano che i tuoi span siano compatibili con qualsiasi backend - Phoenix, Langfuse, LangSmith - senza lock-in.

La scelta del backend dipende dai tuoi vincoli: Phoenix OSS per privacy e costo zero, Langfuse per multi-tenancy e alta scalabilità con ClickHouse, LangSmith per deep integration LangChain, Helicone per prototipazione rapida single-provider. In ogni caso, non aspettare di avere il primo failure in produzione: il costo di aggiungere OTel a un agent esistente è basso (mezza giornata di lavoro), il costo di un incident senza trace è molto più alto.

Prossimo episodio: EP03 - Cost optimization agent: 7 tecniche per ridurre il token spend fino all'80% senza perdere quality. Dall'Anthropic prompt caching al routing model-per-task, con benchmark before/after su dati reali.

Se ti interessa come questi pattern di AI engineering si traducono in workflow quotidiano su un codebase reale, ho documentato il mio setup (skill design + planning fasato + sub-agent orchestration) in Claude Code Workflow: Superpowers + GSD framework.