Observabilidade de Agentes de IA: Traces, Logs e Alertas que Funcionam

Você construiu o agente. Testou localmente. Funcionou. Colocou em produção. Na primeira semana, descobriu que ele falha 12% das vezes — mas não sabe em quais cenários, por que, ou quanto isso custou.

Sem observabilidade, seu agente é uma caixa-preta que gera respostas e cobranças. Com observabilidade, você sabe exatamente onde a jornada quebrou, quanto custou cada passo, e recebe um alerta antes do cliente reclamar.

Este guia cobre:

• O que é observabilidade de agentes (e por que difere de observabilidade tradicional)
• Distributed tracing: rastreando cada passo de um agente multi-step
• Estruturação de logs: do print() ao JSON com contexto
• Métricas que importam: latência, custo, taxa de sucesso, qualidade
• Alertas que evitam incidentes, não só os reportam
• Stack prática: OpenTelemetry, Langfuse, e código real em Python

Por Que Observabilidade de Agentes É Diferente

Observabilidade tradicional monitora sistemas determinísticos: uma requisição entra, o código executa, uma resposta sai. Se algo falha, o stack trace aponta a linha exata.

Agentes de IA são sistemas não-determinísticos com três propriedades que quebram ferramentas tradicionais:

1. O mesmo input pode gerar outputs diferentes. Um agente que processa um email de suporte pode responder corretamente 9 vezes e inventar uma resposta na 10ª. O stack trace não mostra nada — o código executou perfeitamente, mas o modelo "alucinou".

2. A jornada é dinâmica. Um agente pode resolver uma tarefa em 3 steps ou em 15, dependendo do contexto. Não há um caminho fixo para instrumentar.

3. Cada step custa dinheiro. Um loop infinito de tool calls não é só um bug — é um buraco na carteira. Você precisa saber o custo acumulado durante a execução, não depois.

A boa notícia: OpenTelemetry foi estendido para LLMs em 2025-2026. As ferramentas existem. Você só precisa saber como usar.

O Que Você Precisa Observar

Antes de escolher ferramentas, defina o que importa. Para agentes de IA, existem 4 dimensões de observabilidade:

Nenhuma dimensão substitui outra. Traces mostram o caminho, logs mostram os detalhes, métricas mostram tendências, alertas te acordam.

Distributed Tracing para Agentes

Tracing tradicional rastreia uma requisição através de serviços: API → Auth → Database → Cache. Cada serviço é um "span" no trace.

Tracing de agentes rastreia uma intenção através de decisões: Input → Planning → Tool Call → Observation → Planning → Tool Call → ... → Output. Cada decisão do LLM é um span.

Implementando Tracing com OpenTelemetry

OpenTelemetry é o padrão aberto para observabilidade. Em 2025, a comunidade adicionou suporte semântico para LLMs — atributos como llm.model, llm.token_count, llm.tool.name.

Aqui está um decorator que instrumenta qualquer função de agente:

import functools
import time
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource, SERVICE_NAME

# Configurar tracer provider
resource = Resource.create({SERVICE_NAME: "agente-suporte"})
provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("agente.suporte")


def trace_agent_step(step_name, step_type="planning"):
    """Decorator que cria um span OpenTelemetry para cada step do agente."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            with tracer.start_as_current_span(
                step_name,
                attributes={
                    "agent.step_type": step_type,
                    "agent.function": func.__name__,
                }
            ) as span:
                start = time.time()
                try:
                    result = func(*args, **kwargs)
                    span.set_attribute("agent.success", True)
                    return result
                except Exception as e:
                    span.set_attribute("agent.success", False)
                    span.set_attribute("error.message", str(e))
                    span.record_exception(e)
                    raise
                finally:
                    span.set_attribute("agent.duration_ms", (time.time() - start) * 1000)
        return wrapper
    return decorator


class AgenteSuporte:
    def __init__(self, llm_client, tools):
        self.llm = llm_client
        self.tools = tools
    
    @trace_agent_step("planning", step_type="planning")
    def plan(self, query, context):
        """LLM decide qual tool usar."""
        prompt = f"Query: {query}\nContexto: {context}\nDecida a próxima ação."
        response = self.llm.complete(prompt)
        return self._parse_action(response)
    
    @trace_agent_step("tool_call", step_type="tool")
    def execute_tool(self, tool_name, params):
        """Executa uma tool e retorna o resultado."""
        tool = self.tools.get(tool_name)
        if not tool:
            raise ValueError(f"Tool desconhecida: {tool_name}")
        return tool(**params)
    
    @trace_agent_step("generate_response", step_type="output")
    def generate_response(self, query, observations):
        """Gera resposta final baseada nas observações."""
        prompt = f"Query: {query}\nDados: {observations}\nResponda ao usuário."
        return self.llm.complete(prompt)
    
    def run(self, query):
        with tracer.start_as_current_span(
            "agent.run",
            attributes={"agent.query": query}
        ) as root_span:
            context = ""
            observations = []
            max_steps = 10
            
            for step in range(max_steps):
                action = self.plan(query, context)
                
                if action["type"] == "final":
                    response = self.generate_response(query, observations)
                    root_span.set_attribute("agent.steps", step + 1)
                    root_span.set_attribute("agent.success", True)
                    return response
                
                result = self.execute_tool(action["tool"], action["params"])
                observations.append(result)
                context += f"\nStep {step}: {result}"
            
            root_span.set_attribute("agent.success", False)
            root_span.set_attribute("agent.error", "max_steps_exceeded")
            raise RuntimeError("Agente excedeu número máximo de steps")

O que esse código faz:

• Cada plan, execute_tool e generate_response gera um span no trace
• O span raiz agent.run agrupa toda a jornada
• Atributos como agent.step_type, agent.duration_ms, agent.success permitem filtrar e analisar
• Se algo falha, a exceção é registrada no span com stack trace

Nota: O endpoint OTLP (localhost:4317) pode ser substituído por qualquer backend: Jaeger, Zipkin, Grafana Tempo, ou serviços como Langfuse e LangSmith.

Logs Estruturados: Do Print ao JSON

Logs de print() são inúteis em produção. Você precisa de logs estruturados — cada log é um objeto JSON com campos consistentes que podem ser filtrados, agregados e correlacionados.

O Que Logar em Cada Step

Implementação com structlog

import structlog
import logging
import sys

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.dict_tracebacks,
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

logger = structlog.get_logger()


def log_step(logger, step_name, step_num, model, input_tokens, output_tokens, cost, tool_name=None, duration_ms=None, success=True, error=None):
    """Loga um step do agente com todos os campos relevantes."""
    event = {
        "event": "agent_step",
        "agent.step": step_name,
        "agent.step_num": step_num,
        "llm.model": model,
        "llm.input_tokens": input_tokens,
        "llm.output_tokens": output_tokens,
        "llm.cost_usd": cost,
        "agent.success": success,
    }
    
    if tool_name:
        event["tool.name"] = tool_name
        event["tool.duration_ms"] = duration_ms
    
    if error:
        event["error.message"] = str(error)
        event["error.type"] = type(error).__name__
        logger.error("Agent step failed", **event)
    else:
        logger.info("Agent step completed", **event)

Correlação Trace + Log

O campo trace_id é o elo de ligação. Quando você investiga um incidente:

1. O alerta dispara com trace_id: abc123
2. Você abre o trace no Jaeger/Langfuse e vê o caminho
3. Você busca logs com trace_id=abc123 e vê os detalhes de cada step
4. Você sabe exatamente o que aconteceu, em que ordem, e quanto custou

Métricas que Importam

Métricas são números agregados ao longo do tempo. Elas mostram tendências, não detalhes.

As 6 Métricas Essenciais

1. Taxa de Sucesso

Percentual de jornadas que completam sem erro. "Sucesso" significa: o agente retornou uma resposta e não excedeu max_steps.

taxa_sucesso = (jornadas_completadas / total_jornadas) * 100

Por que importa: Se cair abaixo de 90%, seus usuários estão vendo erros. Se cair abaixo de 80%, o agente está quebrado.

2. Latência (P50, P95, P99)

Tempo de resposta. P50 é a mediana — metade das jornadas é mais rápida. P95 é o pior caso aceitável — 95% das jornadas terminam antes disso.

Por que importa: Usuários não toleram latência. P95 > 5s significa que 1 em 20 usuários espera mais de 5 segundos.

3. Custo por Jornada

Custo médio de uma execução completa, somando todos os calls de LLM.

custo_jornada = sum(step["llm.cost_usd"] for step in jornada)

Por que importa: Um agente que custa $0.01 por jornada e processa 100k jornadas/dia custa $1.000/dia. Se o custo dobrar, você precisa saber por quê.

4. Qualidade (Human-in-the-Loop)

Avaliação humana de uma amostra das respostas. Escala de 1-5 ou binária (bom/ruim).

Por que importa: Métricas técnicas não capturam "alucinações" ou respostas tecnicamente corretas mas inúteis. Só humanos avaliam qualidade.

5. Taxa de Erro de Tool

Percentual de tool calls que falham (timeout, exception, resposta inválida).

tool_error_rate = (tool_calls_falhas / total_tool_calls) * 100

Por que importa: Tool errors frequentes indicam problemas de infraestrutura — database lento, API instável, ou parâmetros mal formados.

6. Steps por Jornada

Número médio de steps (planning + tool calls) que o agente executa.

Por que importa: Steps excessivos indicam loops, indecisão, ou falta de clareza no prompt. Também correlaciona com custo — cada step é um call de LLM.

Exportando Métricas com OpenTelemetry

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter

metric_reader = PeriodicExportingMetricReader(OTLPMetricExporter(endpoint="http://localhost:4317"))
metrics.set_meter_provider(MeterProvider(resource=resource, metric_readers=[metric_reader]))
meter = metrics.get_meter("agente.suporte")

# Criar métricas
success_counter = meter.create_counter("agent.journeys.success", description="Jornadas completadas com sucesso")
latency_histogram = meter.create_histogram("agent.journey.duration", description="Duração da jornada em ms", unit="ms")
cost_histogram = meter.create_histogram("agent.journey.cost", description="Custo da jornada em USD", unit="USD")
steps_histogram = meter.create_histogram("agent.journey.steps", description="Número de steps por jornada")

# Usar no código
def on_journey_complete(success, duration_ms, cost_usd, steps):
    success_counter.add(1, {"success": str(success)})
    latency_histogram.record(duration_ms)
    cost_histogram.record(cost_usd)
    steps_histogram.record(steps)

Alertas que Evitam Incidentes

Alertas ruins acordam você às 3h para algo que não importa. Alertas bons te avisam antes do problema afetar usuários.

Regras de Alerta para Agentes

Implementando Alertas com Prometheus + Alertmanager

# prometheus_rules.yml
groups:
  - name: agent_alerts
    rules:
      - alert: AgentSuccessRateLow
        expr: rate(agent_journeys_success_total{success="true"}[10m]) / rate(agent_journeys_success_total[10m]) < 0.9
        for: 2m
        labels:
          severity: p1
        annotations:
          summary: "Taxa de sucesso do agente abaixo de 90%"
          description: "O agente {{ $labels.service }} tem taxa de sucesso de {{ $value | humanizePercentage }}"
      
      - alert: AgentLatencyHigh
        expr: histogram_quantile(0.95, rate(agent_journey_duration_bucket[5m])) > 5000
        for: 5m
        labels:
          severity: p2
        annotations:
          summary: "Latência P95 do agente acima de 5s"
          description: "P95 latência: {{ $value }}ms"
      
      - alert: AgentCostSpike
        expr: avg_over_time(agent_journey_cost_bucket[15m]) > 2 * avg_over_time(agent_journey_cost_bucket[1h])
        for: 5m
        labels:
          severity: p2
        annotations:
          summary: "Custo do agente duplicou"
          description: "Custo médio por jornada está 2x acima do baseline"
      
      - alert: AgentLoopDetected
        expr: agent_journey_steps_bucket{le="+Inf"} > 20
        for: 1m
        labels:
          severity: p2
        annotations:
          summary: "Loop detectado no agente"
          description: "Uma jornada executou mais de 20 steps"

Kill Switch

Todo agente em produção precisa de um kill switch — um circuit breaker que para o agente se algo der muito errado.

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "HALF_OPEN"
            else:
                raise RuntimeError("Circuit breaker OPEN — agente desabilitado")
        
        try:
            result = func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failures = 0
            return result
        except Exception as e:
            self.failures += 1
            self.last_failure_time = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "OPEN"
            raise

Stack Completa: Do Zero ao Dashboard

Aqui está uma stack que funciona em produção e não custa uma fortuna:

Arquitetura de Referência

Setup Mínimo Viable

Se você quer começar hoje com o mínimo:

1. Instrumente seu agente com o decorator trace_agent_step acima
2. Use Langfuse (grátis até 10k traces/mês) — ele já tem OTel collector, traces, logs, métricas e dashboard
3. Configure 3 alertas: taxa de sucesso < 90%, latência P95 > 5s, custo > 2x baseline
4. Adicione kill switch no circuit breaker

Quando escalar, migre para a stack open source (Jaeger + Loki + Prometheus + Grafana) ou pague por Datadog/LangSmith.

Checklist de Observabilidade

Antes de colocar qualquer agente em produção, verifique:

• [ ] Cada step do agente gera um span com step_type, duration_ms, success
• [ ] Cada jornada tem um trace_id que correlaciona todos os spans
• [ ] Logs são JSON estruturados com trace_id, span_id, timestamp
• [ ] Métricas de taxa de sucesso, latência, custo e steps estão exportadas
• [ ] Alertas configurados para taxa de sucesso, latência, custo e tool errors
• [ ] Kill switch / circuit breaker implementado
• [ ] Dashboard mostra traces, logs e métricas em um só lugar
• [ ] Runbook documenta como investigar cada tipo de alerta
• [ ] Teste de carga valida comportamento sob stress
• [ ] Rollback plan definido se métricas degradarem após deploy

Conclusão

Observabilidade de agentes não é um luxo — é infraestrutura. Sem ela, você está voando no escuro em um avião que pode decidir sozinho para onde ir.

As quatro dimensões (traces, logs, métricas, alertas) se reforçam mutuamente. Traces mostram o caminho, logs mostram os detalhes, métricas mostram tendências, alertas te acordam. Nenhuma substitui outra.

Comece com o decorator trace_agent_step e o structlog. Adicione métricas. Configure 3 alertas básicos. Depois evolua para a stack completa. O importante é ter visibilidade desde o primeiro deploy — não depois do primeiro incidente.

Se você já leu o guia de segurança, este é o próximo passo natural: segurança protege contra ataques, observabilidade te mostra quando algo deu errado — ataque ou não.

Publicado em andrecosta.ia.br.