5 mar 2026

Debug e Osservabilità nei Sistemi di Agenti Autonomi

Un agente autonomo che fallisce silenziosamente è peggio di nessun agente. Quando una funzione tradizionale lancia un’eccezione, si ottiene uno stack trace. Quando un agente prende la strada sbagliata attraverso venti chiamate a strumenti e tre invocazioni del modello, si ottiene una risposta sbagliata — senza spiegazioni ovvie.

Il debug degli agenti richiede un modello mentale diverso. Il sistema non sta eseguendo un percorso deterministico; sta prendendo una serie di decisioni. L’osservabilità significa catturare queste decisioni — non solo input e output, ma il ragionamento che le collega.

Perché il Debug Tradizionale Fallisce per gli Agenti

Il logging standard cattura cosa è successo. L’osservabilità degli agenti richiede di catturare il perché — cosa ha concluso il modello, quale strumento ha scelto e perché, e da quale stato intermedio stava lavorando.

Le modalità di fallimento sono anche diverse:

Allucinazione silenziosa: L’agente produce con sicurezza una risposta sbagliata senza segnalare incertezza.
Deriva delle decisioni: Ogni passo sembra ragionevole localmente, ma la sequenza si allontana dall’obiettivo.
Uso improprio degli strumenti: L’agente chiama lo strumento giusto con parametri sottilmente errati.
Loop infiniti: L’agente rimane bloccato a riprovare un approccio fallito.
Avvelenamento del contesto: Un output errato in una fase iniziale corrompe tutto il ragionamento successivo.

Nessuno di questi produce un’eccezione. Producono comportamento errato che è visibile solo quando si ricostruisce il trace di esecuzione completo.

Logging Strutturato per le Decisioni degli Agenti

Il primo passo è avvolgere ogni interazione dell’agente in log strutturati. Non loggare le risposte grezze dell’API — logga eventi semantici.

import json
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Any
import anthropic

client = anthropic.Anthropic()

@dataclass
class AgentEvent:
    trace_id: str
    step: int
    event_type: str      # "llm_call", "tool_call", "tool_result", "decision", "error"
    model: str | None
    input_tokens: int | None
    output_tokens: int | None
    latency_ms: float | None
    content: dict[str, Any]
    timestamp: float

def log_event(event: AgentEvent):
    print(json.dumps(asdict(event)))  # Sostituisci con la tua destinazione di log

class TracedAgent:
    def __init__(self, trace_id: str | None = None):
        self.trace_id = trace_id or str(uuid.uuid4())
        self.step = 0
        self.tools = []

    def add_tool(self, name: str, description: str, input_schema: dict):
        self.tools.append({
            "name": name,
            "description": description,
            "input_schema": input_schema
        })

    def call(self, messages: list[dict], system: str = "") -> str:
        self.step += 1
        start = time.monotonic()

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4096,
            system=system,
            tools=self.tools,
            messages=messages
        )

        latency_ms = (time.monotonic() - start) * 1000

        log_event(AgentEvent(
            trace_id=self.trace_id,
            step=self.step,
            event_type="llm_call",
            model="claude-opus-4-6",
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            latency_ms=latency_ms,
            content={
                "stop_reason": response.stop_reason,
                "text_blocks": [b.text for b in response.content if b.type == "text"],
                "tool_calls": [
                    {"name": b.name, "input": b.input}
                    for b in response.content if b.type == "tool_use"
                ]
            },
            timestamp=time.time()
        ))

        return response

Costruire un Trace Completo

Una singola riga di log non basta — hai bisogno del trace di esecuzione completo che collega ogni decisione al suo risultato:

from typing import Callable

def run_traced_agent(
    task: str,
    tools: dict[str, Callable],
    tool_schemas: list[dict],
    system: str,
    max_steps: int = 20,
) -> dict:
    agent = TracedAgent()
    for schema in tool_schemas:
        agent.add_tool(**schema)

    messages = [{"role": "user", "content": task}]
    trace = {"trace_id": agent.trace_id, "task": task, "steps": []}
    step_count = 0

    while step_count < max_steps:
        step_count += 1
        response = agent.call(messages, system=system)

        step_record = {
            "step": step_count,
            "stop_reason": response.stop_reason,
            "model_output": [],
            "tool_results": []
        }

        if response.stop_reason == "end_turn":
            for block in response.content:
                if block.type == "text":
                    step_record["model_output"].append(block.text)
            trace["steps"].append(step_record)
            trace["final_answer"] = step_record["model_output"][-1] if step_record["model_output"] else ""
            break

        tool_results = []
        for block in response.content:
            if block.type == "tool_use":
                step_record["model_output"].append({
                    "tool": block.name,
                    "input": block.input
                })

                tool_fn = tools.get(block.name)
                if not tool_fn:
                    result = f"Errore: strumento sconosciuto '{block.name}'"
                else:
                    try:
                        result = tool_fn(**block.input)
                    except Exception as e:
                        result = f"Errore strumento: {e}"

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result)
                })
                step_record["tool_results"].append({
                    "tool": block.name,
                    "result_preview": str(result)[:200]
                })

        trace["steps"].append(step_record)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})

    else:
        trace["error"] = f"superato il numero massimo di passi ({max_steps})"

    return trace

Rilevamento dei Loop

I loop infiniti sono un modo di fallimento comune. Rilevali prendendo l’impronta digitale del pattern di invocazione degli strumenti di ogni chiamata LLM:

def detect_loop(trace: dict, window: int = 4) -> bool:
    steps = trace["steps"]
    if len(steps) < window:
        return False

    def step_signature(step: dict) -> str:
        tools_called = sorted(
            t["tool"] if isinstance(t, dict) else t
            for t in step.get("model_output", [])
            if isinstance(t, dict) and "tool" in t
        )
        return "|".join(tools_called)

    recent = [step_signature(s) for s in steps[-window:]]
    if len(set(recent)) == 1 and recent[0]:
        return True

    if len(steps) >= 4:
        pattern = [step_signature(s) for s in steps[-4:]]
        if pattern[0] == pattern[2] and pattern[1] == pattern[3]:
            return True

    return False

Metriche da Monitorare in Produzione

from collections import Counter

def compute_trace_metrics(trace: dict) -> dict:
    steps = trace["steps"]
    errors = [s for s in steps if "error" in s]

    tool_calls_by_name: Counter = Counter()
    for step in steps:
        for output in step.get("model_output", []):
            if isinstance(output, dict) and "tool" in output:
                tool_calls_by_name[output["tool"]] += 1

    return {
        "trace_id": trace["trace_id"],
        "total_steps": len(steps),
        "error_steps": len(errors),
        "tool_call_distribution": dict(tool_calls_by_name),
        "completed": "final_answer" in trace,
        "loop_detected": detect_loop(trace),
    }

Segnali chiave per gli alert:

Tasso di loop > 5% — l’agente si blocca
Tasso di errore per strumento > soglia — uno strumento è rotto
Media passi in tendenza crescente — i compiti diventano più difficili
Latenza p99 in picco — un endpoint del modello è lento

Integrazione OpenTelemetry

Per i team che già usano OpenTelemetry, emetti i trace degli agenti come span:

from opentelemetry import trace as otel_trace

tracer = otel_trace.get_tracer("agent")

def run_with_otel(task: str, tools: dict, tool_schemas: list, system: str):
    with tracer.start_as_current_span("agent.run") as root_span:
        root_span.set_attribute("agent.task", task[:200])

        agent = TracedAgent()
        for schema in tool_schemas:
            agent.add_tool(**schema)

        messages = [{"role": "user", "content": task}]

        for step in range(20):
            with tracer.start_as_current_span(f"agent.step.{step}") as step_span:
                response = agent.call(messages, system=system)
                step_span.set_attribute("llm.stop_reason", response.stop_reason)
                step_span.set_attribute("llm.input_tokens", response.usage.input_tokens)

                if response.stop_reason == "end_turn":
                    break

Redazione dei Dati Personali nei Log

I log degli agenti spesso contengono dati sensibili. Prima di emetterli a qualsiasi sistema esterno, eseguire la redazione:

import re

PII_PATTERNS = [
    (re.compile(r'\b[\w.+-]+@[\w-]+\.[a-z]{2,}\b'), '[EMAIL]'),
    (re.compile(r'\b\+?39[-\s]?\d{2,3}[-\s]?\d{6,8}\b'), '[TELEFONO]'),
    (re.compile(r'\bsk-[a-zA-Z0-9]{20,}\b'), '[CHIAVE_API]'),
]

def redact(text: str) -> str:
    for pattern, replacement in PII_PATTERNS:
        text = pattern.sub(replacement, text)
    return text

Le Tre Metriche Più Importanti

Tasso di completamento dei compiti — quale frazione delle esecuzioni raggiunge final_answer vs. max_steps o un errore. Stabilire una baseline per tipo di compito.

Costo in token per compito — somma di input_tokens + output_tokens su tutti i passi. Un aumento del costo del 20% senza cambiamento nel tasso di completamento segnala tipicamente una degradazione del prompt.

Tasso di errore degli strumenti — error_steps / total_steps. I picchi in questa metrica puntano direttamente a uno strumento o API rotti.

L’osservabilità nei sistemi di agenti non è opzionale — è la differenza tra un sistema che puoi iterare e uno che puoi solo riavviare quando si rompe. Inizia con eventi strutturati e ID di trace. Aggiungi il rilevamento dei loop. Pubblica le metriche. L’investimento si ripaga la prima volta che hai un guasto in produzione e puoi ricostruire esattamente cosa è successo invece di indovinare.