Como saber se seu agente autônomo está fazendo o certo? Observabilidade em produção

Por que métricas tradicionais não bastam para agentes?

Um microserviço tem inputs determinísticos, outputs previsíveis e um grafo de chamadas que você pode instrumentar com distributed tracing convencional. Um agente autônomo recebe uma instrução de alto nível ("reconcilie o inventário com a nota fiscal") e decide, em runtime, quais ferramentas chamar, em que ordem e quantas vezes. O trace não é uma DAG fixa; é uma árvore de decisões que pode variar a cada execução.

Além disso, o custo não é linear: cada token enviado ao modelo tem preço, e uma única ação do agente pode gerar dezenas de chamadas internas. Você não está pagando por requests; está pagando por cadeias de raciocínio. Métricas como latência p99 e throughput por segundo perdem relevância. O que importa é custo por tarefa concluída e taxa de conclusão bem-sucedida.

flowchart LR
    subgraph Core[" "]
        Agent["Agente Autônomo"]
        LLM["LLM"]
        Guardrail{"Guardrail"}
    end
    subgraph Tools["Ferramentas"]
        API["API Calls"]
        DB["Database"]
        FS["File System"]
    end
    subgraph Observability["Observabilidade"]
        Trace["Tracing"]
        Metrics["Metrics"]
        Logs["Logs"]
    end
    KillSwitch(("Kill Switch"))
    LLM --> Agent
    Agent --> Guardrail
    Guardrail -->|Permitido| API
    Guardrail -->|Permitido| DB
    Guardrail -->|Permitido| FS
    Guardrail -.->|Bloqueado| KillSwitch
    Agent -->|executa| API
    Agent -->|executa| DB
    Agent -->|executa| FS
    Agent -->|telemetry| Trace
    Agent -->|telemetry| Metrics
    Agent -->|telemetry| Logs
    Agent -.->|emergência| KillSwitch

Como fazer tracing de chamadas de ferramentas?

A base é instrumentar cada ferramenta com contexto de trace. Se você usa LangChain, a integração com LangSmith ou Lilypics já gera traces automaticamente, mas você vai querer controle mais fino para produção.

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource

provider = TracerProvider(
    resource=Resource.create({
        "service.name": "inventory-reconciler-agent",
        "agent.version": os.getenv("AGENT_VERSION"),
        "deployment.environment": os.getenv("ENV"),
    })
)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)

class ObservedTool:
    def __init__(self, name: str, base_tool):
        self.name = name
        self.base_tool = base_tool
        
    def __call__(self, *args, **kwargs):
        with tracer.start_as_current_span(
            f"tool.{self.name}",
            kind=trace.SpanKind.CLIENT,
            attributes={
                "tool.name": self.name,
                "agent.task_id": context_var.get("task_id"),
                "agent.attempt": context_var.get("attempt", 1),
            }
        ) as span:
            inicio = time.time()
            try:
                result = self.base_tool(*args, **kwargs)
                span.set_status(trace.StatusCode.OK)
                span.set_attribute("tool.success", True)
                return result
            except Exception as e:
                span.set_status(trace.StatusCode.ERROR, str(e))
                span.set_attribute("tool.success", False)
                span.record_exception(e)
                raise
            finally:
                span.set_attribute("tool.duration_ms", (time.time() - inicio) * 1000)

Isso gera spans aninhados que você consegue visualizar como waterfall no Grafana ou Jaeger. O span de cada ferramenta carrega task_id, attempt e duration_ms. Com isso, você vê exatamente onde o agente gastou tempo e onde falhou.

Quais métricas importam de verdade?

Organize em três camadas:

Calcule custo por execução acumulando input_tokens * preço_input + output_tokens * preço_output para cada span. Agrupe por task_type para descobrir que tipo de tarefa está estourando o budget.

Como detectar loops e ações repetidas?

Loop detection é o guardrail comportamental mais crítico. Existem duas abordagens complementares:

1. Detecção por hash de estado

Gere um hash do estado interno do agente (últimas N ações + contexto) e monitore se o mesmo hash aparece N vezes em janela curta.

import hashlib
from collections import deque

class LoopDetector:
    def __init__(self, window_size=10, threshold=3):
        self.history = deque(maxlen=window_size)
        self.threshold = threshold
        
    def register_action(self, action: dict) -> bool:
        state_hash = hashlib.sha256(
            json.dumps(action, sort_keys=True).encode()
        ).hexdigest()[:16]
        
        occurrences = sum(1 for h in self.history if h == state_hash)
        self.history.append(state_hash)
        
        if occurrences >= self.threshold:
            metrics.increment("agent.loop_detected", tags={
                "action_type": action.get("type"),
                "occurrence": occurrences + 1
            })
            return True
        return False

2. Detecção por sequência de ferramentas

Monitore o padrão de chamadas: se API_call → API_call → API_call se repete 5 vezes sem progressão (sem mudança de estado relevante), é loop. Registre o span como "loop_detected": true e dispare alert.

O que são guardrails e quando você precisa deles?

Guardrails são restrições que impedem o agente de executar ações fora de um espectro seguro. Diferente de validação pós-execução, guardrails avaliam antes de cada ação.

Exemplos concretos:

• Verificação de schema: a saída de uma ferramenta está no formato esperado?
• Validação de limites: o agente não pode chamar a API de pagamento mais de 3 vezes por tarefa.
• Bloqueio de dados sensíveis: PII detectado na saída? Interrompa e marque para revisão.
• Rate limiting por role: agentes com privilégio de escrita precisam de confirmação explícita.

def guardrail_tool(tool_name: str, params: dict, context: AgentContext) -> ToolResult:
    # Limite de chamadas por ferramenta por tarefa
    call_count = context.get_metric(f"calls.{tool_name}")
    if call_count >= MAX_CALLS_PER_TOOL:
        return ToolResult(
            blocked=True,
            reason=f"Rate limit: {tool_name} called {call_count} times, max is {MAX_CALLS_PER_TOOL}"
        )
    
    # Verificação de PII em outputs
    if tool_name in PII_EXPOSING_TOOLS:
        if contains_pii(params) or contains_pii(context.last_output):
            context.flag_for_review("pii_detected")
            return ToolResult(blocked=True, reason="PII detected, flagged for review")
    
    return ToolResult(blocked=False)

Kill switch: como implementar sem criar Single Point of Failure?

O kill switch precisa ser rápido, confiável e auditável. Não é um botão no dashboard que o developer clica e torce para o agente respeitar.

Arquitetura recomendada:

1. Layer de proxy: todas as chamadas do agente passam por um proxy que avalia kill switch status. Se kill_switch_active=true, retorna erro padronizado sem executar.
2. Storage distribuído: o flag deve estar em Redis ou etcd com TTL curto (30s). Não em banco relacional que pode ficar indisponível.
3. Atomicidade: ativa e desativa devem ser operações atômicas com log de auditoria (quem, quando, motivo).
4. Fallback: se o proxy perder conexão com o storage, assuma kill_switch_active=true. Fail-secure, não fail-open.

curl -X POST https://agent-gateway.internal/killswitch \
  -H "Authorization: Bearer $KILLSWITCH_TOKEN" \
  -d '{
    "active": true,
    "reason": "anomalous_cost_detected",
    "affected_agents": ["inventory-reconciler-v2"],
    "expires_at": "2024-01-15T14:30:00Z"
  }'

Configure alertas que auto-ativam o kill switch: custo por hora > 2x média histórica, ou taxa de erro > 30% em janela de 5 minutos.

Checklist operacional para produção

• Tracing instrumentado em todas as ferramentas com task_id e attempt propagados.
• Métricas de custo por tarefa calculadas e expostas no dashboard (não apenas custo por API call).
• Loop detector ativo com threshold configurável por tipo de agente.
• Guardrails com validação de schema e rate limiting por ferramenta.
• Kill switch com storage distribuído e ativação automática por alerta.
• Log de auditoria completo: todas as ações do agente com timestamp, contexto e resultado.
• Runbook documentado para cada tipo de intervenção (loop detectado, custo estourado, PII flag).

O tradeoff que você precisa aceitar

Observabilidade tem custo. Cada span gerado, cada métrica calculada, cada log escrito consome recursos. Para agentes simples, você pode começar com tracing básico e métricas de resultado. Para agentes que tocam sistemas críticos (pagamentos, dados de clientes, infraestrutura), o nível de instrumentação mostrado aqui não é overkill; é o mínimo para dormir tranquilo.

A decisão real não é se você vai instrumentar, mas onde traçar a linha entre visibilidade e overhead. Comece pelo que te acorda às 3 da manhã, ajuste a granularidade conforme o regime de risco do agente.

Se você está operando agentes autônomos sem visibilidade clara do que eles fazem, o risco está escondido. Em 30 minutos, podemos mapear onde estão os pontos cegos e construir um plano de observabilidade que escala.

Fale com um especialista da Eficify