5 de mar. de 2026

Depuração e Observabilidade em Sistemas de Agentes Autônomos

Um agente autônomo que falha silenciosamente é pior do que nenhum agente. Quando uma função tradicional lança uma exceção, você obtém um stack trace. Quando um agente toma um caminho errado ao longo de vinte chamadas de ferramentas e três invocações do modelo, você obtém uma resposta incorreta — e sem explicação óbvia.

Depurar agentes requer um modelo mental diferente. O sistema não está executando um caminho determinístico; está tomando uma série de decisões. Observabilidade significa capturar essas decisões — não apenas entradas e saídas, mas o raciocínio que as conecta.

Por Que a Depuração Tradicional Falha para Agentes

O logging padrão captura o que aconteceu. A observabilidade de agentes exige capturar o porquê — o que o modelo concluiu, qual ferramenta ele escolheu e por quê, e de qual estado intermediário ele estava trabalhando.

Os modos de falha também são diferentes:

Alucinação silenciosa: O agente produz uma resposta incorreta com confiança sem sinalizar incerteza.
Deriva de decisões: Cada passo parece razoável localmente, mas a sequência se afasta do objetivo.
Mau uso de ferramentas: O agente chama a ferramenta certa com parâmetros sutilmente incorretos.
Loops infinitos: O agente fica preso recomençando uma abordagem que falha.
Envenenamento de contexto: Uma saída ruim em um passo inicial corrompe todo o raciocínio subsequente.

Nenhum desses produz uma exceção. Eles produzem comportamento incorreto que só é visível quando você reconstrói o trace de execução completo.

Logging Estruturado para Decisões de Agentes

O primeiro passo é envolver cada interação do agente em logs estruturados. Não registre respostas brutas da API — registre 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)))  # Substitua pelo seu 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

Construindo um Trace Completo

Uma única linha de log não é suficiente — você precisa do trace de execução completo que conecta cada decisão ao seu resultado:

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"Erro: ferramenta desconhecida '{block.name}'"
                else:
                    try:
                        result = tool_fn(**block.input)
                    except Exception as e:
                        result = f"Erro de ferramenta: {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 passos excedido ({max_steps})"

    return trace

Detecção de Loops

Loops infinitos são um modo de falha comum. Detecte-os tirando a impressão digital do padrão de invocação de ferramentas de cada chamada ao 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

Métricas para Acompanhar em Produção

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),
    }

Sinais principais para alertas:

Taxa de loops > 5% das execuções — o agente está travando
Taxa de erro por ferramenta > limite — uma ferramenta está quebrada
Média de passos em tendência de alta — as tarefas estão ficando mais difíceis
Latência p99 em pico — um endpoint do modelo está lento

Integração com OpenTelemetry

Para equipes que já usam OpenTelemetry, emita os traces do 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)

                if response.stop_reason == "end_turn":
                    break

Redação de PII nos Logs

Os logs de agentes frequentemente contêm dados sensíveis. Antes de emiti-los para qualquer sistema externo, faça a redação:

import re

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

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

As Três Métricas Mais Importantes

Taxa de conclusão de tarefas — qual fração das execuções atinge final_answer vs. max_steps ou um erro. Estabeleça uma baseline por tipo de tarefa.

Custo em tokens por tarefa — soma de input_tokens + output_tokens em todos os passos. Um aumento de 20% no custo sem mudança na taxa de conclusão geralmente sinaliza degradação do prompt.

Taxa de erro de ferramentas — error_steps / total_steps. Picos nesta métrica apontam diretamente para uma ferramenta ou API quebrada.

A observabilidade em sistemas de agentes não é opcional — é a diferença entre um sistema que você pode iterar e um que só pode reiniciar quando quebra. Comece com eventos estruturados e IDs de trace. Adicione detecção de loops. Publique métricas. O investimento se paga na primeira vez que você tem uma falha em produção e consegue reconstruir exatamente o que aconteceu.