5 mar 2026

Depuración y Observabilidad en Sistemas de Agentes Autónomos

Un agente autónomo que falla silenciosamente es peor que ningún agente. Cuando una función tradicional lanza una excepción, obtienes un stack trace. Cuando un agente toma un camino equivocado a lo largo de veinte llamadas a herramientas y tres invocaciones al modelo, obtienes una respuesta incorrecta — y sin explicación obvia.

Depurar agentes requiere un modelo mental diferente. El sistema no ejecuta un camino determinista; toma una serie de decisiones. La observabilidad significa capturar esas decisiones — no solo entradas y salidas, sino el razonamiento que las conecta.

Por Qué el Debugging Tradicional Falla para Agentes

El logging estándar captura lo que sucedió. La observabilidad de agentes requiere capturar el por qué — qué concluyó el modelo, qué herramienta eligió y por qué, y qué estado intermedio estaba procesando.

Los modos de fallo también son diferentes:

Alucinación silenciosa: El agente produce una respuesta incorrecta con confianza sin señalar incertidumbre.
Deriva de decisiones: Cada paso parece razonable localmente, pero la secuencia se aleja del objetivo.
Mal uso de herramientas: El agente llama a la herramienta correcta con parámetros sutilmente incorrectos.
Bucles infinitos: El agente queda atrapado reintentando un enfoque fallido.
Envenenamiento de contexto: Una salida incorrecta en un paso temprano corrompe todo el razonamiento posterior.

Ninguno de estos produce una excepción. Producen comportamiento incorrecto que solo es visible cuando reconstruyes el trace de ejecución completo.

Logging Estructurado para Decisiones de Agentes

El primer paso es envolver cada interacción del agente en logs estructurados. No registres respuestas brutas de la API — registra eventos semánticos.

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)))  # Reemplaza con tu destino de logs

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

Cada llamada al LLM emite ahora un evento estructurado con el trace ID, número de paso, conteos de tokens, latencia y la decisión del modelo. Estos eventos son el material bruto para todo lo demás.

Construyendo un Trace Completo

Una sola línea de log no es suficiente — necesitas el trace de ejecución completo que conecta cada decisión con su resultado. Construye un acumulador de traces que registre el bucle completo del agente:

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"Error: herramienta desconocida '{block.name}'"
                else:
                    t_start = time.monotonic()
                    try:
                        result = tool_fn(**block.input)
                    except Exception as e:
                        result = f"Error de herramienta: {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"máximo de pasos excedido ({max_steps})"

    return trace

Detección de Bucles

Los bucles infinitos son un modo de fallo común. Detéctalos tomando la huella digital del patrón de invocación de herramientas de cada llamada al LLM:

from collections import Counter

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

Métricas a Seguir en Producción

Una vez que tienes logs estructurados, deriva métricas agregadas por agente y por tipo de tarea:

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),
        "error_messages": [s.get("error") for s in errors],
    }

Señales clave para alertar:

Tasa de bucles > 5% de ejecuciones — el agente se está atascando
Tasa de error por herramienta > umbral — una herramienta está rota
Promedio de pasos en tendencia ascendente — las tareas se están volviendo más difíciles
Latencia p99 en pico — un endpoint del modelo está lento

Integración con OpenTelemetry

Para equipos que ya usan OpenTelemetry, emite los traces del agente como spans:

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)
                step_span.set_attribute("llm.output_tokens", response.usage.output_tokens)

                if response.stop_reason == "end_turn":
                    break

Redacción de PII en Logs

Los logs de agentes a menudo contienen datos sensibles. Antes de emitirlos a cualquier sistema externo, redáctalos:

import re

PII_PATTERNS = [
    (re.compile(r'\b[\w.+-]+@[\w-]+\.[a-z]{2,}\b'), '[EMAIL]'),
    (re.compile(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b'), '[TELÉFONO]'),
    (re.compile(r'\bsk-[a-zA-Z0-9]{20,}\b'), '[API_KEY]'),
]

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

Las Tres Métricas Más Importantes

Tasa de completitud de tareas — qué fracción de ejecuciones llega a final_answer vs. alcanzar max_steps o un error. Establece una línea base por tipo de tarea.

Costo en tokens por tarea — suma input_tokens + output_tokens en todos los pasos. Rastrear esto con el tiempo. Un aumento del 20% en costo sin cambio en la tasa de completitud generalmente señala degradación del prompt.

Tasa de error de herramientas — error_steps / total_steps. Los picos en esta métrica apuntan directamente a una herramienta o API rota.

La observabilidad en sistemas de agentes no es opcional — es la diferencia entre un sistema que puedes iterar y uno que solo puedes reiniciar cuando se rompe. Empieza con eventos estructurados e IDs de trace. Agrega detección de bucles. Publica métricas. La inversión se paga la primera vez que tienes un fallo en producción y puedes reconstruir exactamente qué sucedió.