2.7 — Observabilidade e Barreiras de Segurança em Agentes (Observabilidade)

Objetivo da Aula

Ao concluir esta aula, você será capaz de:

Implementar logging JSONL de cada step do agente com timestamp, tool, args e custo

Detectar anomalias: custo excessivo, loops de tools e bloqueios HITL consecutivos

Construir o AgentObservavel completo com os 2 TODOs (salvar_log e verificar_anomalia)

Aplicar o padrão HITL gate para ações destrutivas

Calcular custo de execução em tempo real com preços de token

Por que isso importa

Agentes sem observabilidade são caixas-pretas. Quando algo dá errado — e vai dar — você precisa saber exatamente o que aconteceu, em que iteração, com quais argumentos e qual foi o resultado.

Sem logs: - “O agente deletou o arquivo errado” → impossível saber o que aconteceu - “O agente gastou $50 em uma task” → impossível saber quantas iterações - “O agente ficou em loop” → impossível saber em qual tool

Com logs JSONL:

grep '"tool":"deletar_arquivo"' agent_audit.jsonl

# → {"timestamp":"2024-06-10T14:30:00","tool":"deletar_arquivo","args":{"caminho":"config.prod.json"},...}

O JSONL (JSON Lines) é o formato padrão para logs de auditoria de agentes: uma linha JSON por evento, fácil de streamar, processar com jq ou indexar no Elasticsearch.

Conceitos Fundamentais

Custo por Token em Tempo Real

CUSTO_POR_TOKEN_INPUT = 0.000003   # claude-haiku: $3/M tokens input

CUSTO_POR_TOKEN_OUTPUT = 0.000015  # claude-haiku: $15/M tokens output



# Por step:

custo = tokens_in * CUSTO_POR_TOKEN_INPUT + tokens_out * CUSTO_POR_TOKEN_OUTPUT

Para monitorar em produção, acumule o custo total:

self.metrics.custo_total_usd += custo



if self.metrics.custo_total_usd > LIMITE_CUSTO_USD:  # $0.50

    raise CustoExcedidoError(f'Custo ${self.metrics.custo_total_usd:.4f} excedeu limite ${LIMITE_CUSTO_USD}')

Decisão de Arquitetura: Guardrails de Agente

Um agente autônomo que executa ações reais precisa de três camadas de contenção: (1) HITL gate para ações de alto impacto: antes de qualquer operação destrutiva ou irreversível (deletar, enviar email, provisionar infraestrutura), pause e exija aprovação humana explícita. Isso protege mesmo se o agente foi sequestrado por prompt injection — o humano vê a ação antes de executar. (2) Detecção de loop: se a mesma tool foi chamada N vezes seguidas com os mesmos parâmetros sem progresso, o agente travou — interrompa e alerte. (3) Circuit breaker por custo: defina um teto de gasto (em tokens ou USD) por sessão de agente. Monitore o custo acumulado em tempo real e interrompa quando atingir o limite. Instrumente cada passo do agente: tool chamada, argumentos, resultado, tokens consumidos, custo parcial, decisão de HITL.

HITL Gate: Ações Destrutivas

ACOES_DESTRUTIVAS = {'deletar_arquivo', 'executar_comando'}



def hitl_gate(tool_name: str, args: dict) -> bool:

    if tool_name not in ACOES_DESTRUTIVAS:

        return True  # aprovação automática para ações não-destrutivas

    

    # Para destrutivas: verificar política

    if tool_name == 'deletar_arquivo':

        return False  # BLOQUEADO: deleções sempre requerem aprovação manual

    

    # Em produção: enviar notificação e aguardar resposta

    return True  # aprovado (simulação)

O HITL gate é o último line of defense. Se o agente foi comprometido (prompt injection via tool result), o HITL gate ainda pode bloquear ações destrutivas.

TODO 1: salvar_log() em JSONL

def salvar_log(metrics: AgentMetrics, arquivo: str = 'agent_audit.jsonl'):

    """Salva log de auditoria em JSONL — uma linha JSON por step."""

    import json

    with open(arquivo, 'a', encoding='utf-8') as f:

        for step in metrics.steps:

            linha = {

                'timestamp': step.timestamp,

                'iteracao': step.iteracao,

                'tool': step.tool,

                'args': step.args,

                'resultado': step.resultado,

                'tokens_input': step.tokens_input,

                'tokens_output': step.tokens_output,

                'custo_usd': step.custo_usd,

                'hitl_aprovado': step.hitl_aprovado,

            }

            f.write(json.dumps(linha, ensure_ascii=False) + '\n')

    print(f'  [LOG] {len(metrics.steps)} steps salvos em {arquivo}')

mode='a' (append) é crítico: preserva logs de execuções anteriores. Cada execução adiciona novas linhas ao arquivo.

TODO 2: verificar_anomalia() — 3 Checks

def verificar_anomalia(metrics: AgentMetrics) -> list[str]:

    alertas = []

    

    # Check 1: Custo excessivo

    if metrics.custo_total_usd > LIMITE_CUSTO_USD:

        alertas.append(

            f'Custo ${metrics.custo_total_usd:.4f} excedeu limite ${LIMITE_CUSTO_USD}'

        )

    

    # Check 2: Loop de tool (mesma tool 3+ vezes seguidas sem progresso)

    if len(metrics.steps) >= 3:

        tools_recentes = [s.tool for s in metrics.steps[-3:]]

        if len(set(tools_recentes)) == 1:  # todas iguais

            alertas.append(

                f'Loop detectado: tool "{tools_recentes[0]}" chamada 3x consecutivamente'

            )

    

    # Check 3: Múltiplos HITL bloqueios consecutivos

    if metrics.hitl_bloqueios >= 2:

        bloqueios_recentes = [s for s in metrics.steps[-3:] if not s.hitl_aprovado]

        if len(bloqueios_recentes) >= 2:

            alertas.append(

                f'{len(bloqueios_recentes)} HITL bloqueios recentes — possível intenção suspeita'

            )

    

    return alertas

Aprofundamento Técnico

Estrutura de Logs para Análise

O formato JSONL permite análise com jq:

# Custo total da sessão

cat agent_audit.jsonl | jq '.custo_usd' | awk '{s+=$1} END {print s}'



# Tools mais chamadas

cat agent_audit.jsonl | jq -r '.tool' | sort | uniq -c | sort -rn



# Steps bloqueados pelo HITL

cat agent_audit.jsonl | jq 'select(.hitl_aprovado == false)'



# Iterações que custaram mais de $0.001

cat agent_audit.jsonl | jq 'select(.custo_usd > 0.001)'

StepLog como Dataclass

from dataclasses import dataclass, field



@dataclass

class StepLog:

    iteracao: int

    timestamp: str

    tool: str

    args: dict

    resultado: str

    tokens_input: int = 0

    tokens_output: int = 0

    hitl_aprovado: bool = True

    custo_usd: float = 0.0

@dataclass gera __init__, __repr__ e __eq__ automaticamente. Simplifica a criação de logs sem boilerplate.

Alertas em Produção

Em produção, verificar_anomalia enviaria notificações reais:

import requests



def enviar_alerta(alerta: str, agent_id: str):

    # Slack webhook

    requests.post(os.environ['SLACK_WEBHOOK'], json={

        'text': f'🚨 Agente {agent_id}: {alerta}'

    })

    

    # PagerDuty para alertas críticos

    if 'custo' in alerta.lower():

        pagerduty_trigger(alerta, severity='critical')

Exemplos Anotados

Exemplo 1: AgentObservavel com TODOs Resolvidos

O código do starter já está quase completo. Os 2 TODOs são adições pontuais:

TODO 1 — salvar_log():

def salvar_log(metrics: AgentMetrics, arquivo: str = 'agent_audit.jsonl'):

    import json

    with open(arquivo, 'a', encoding='utf-8') as f:

        for step in metrics.steps:

            f.write(json.dumps({

                'timestamp': step.timestamp,

                'iteracao': step.iteracao,

                'tool': step.tool,

                'args': step.args,

                'resultado': step.resultado[:200],  # truncar para não explodir o log

                'tokens_input': step.tokens_input,

                'tokens_output': step.tokens_output,

                'custo_usd': round(step.custo_usd, 6),

                'hitl_aprovado': step.hitl_aprovado,

            }, ensure_ascii=False) + '\n')

TODO 2 — verificar_anomalia():

def verificar_anomalia(metrics: AgentMetrics) -> list[str]:

    alertas = []

    

    # 1. Custo excessivo

    if metrics.custo_total_usd > LIMITE_CUSTO_USD:

        alertas.append(f'Custo ${metrics.custo_total_usd:.4f} > limite ${LIMITE_CUSTO_USD}')

    

    # 2. Loop: mesma tool 3x consecutivamente

    if len(metrics.steps) >= 3:

        tools = [s.tool for s in metrics.steps[-3:]]

        if len(set(tools)) == 1:

            alertas.append(f'Loop: "{tools[0]}" chamado 3x consecutivamente')

    

    # 3. HITL: 2+ bloqueios recentes

    if metrics.hitl_bloqueios >= 2:

        bloqueios = sum(1 for s in metrics.steps[-3:] if not s.hitl_aprovado)

        if bloqueios >= 2:

            alertas.append(f'{bloqueios} HITL bloqueios consecutivos — intenção suspeita')

    

    return alertas

Exemplo 2: Trace de Execução com Guardrails

[AGENT] Iniciando: Leia config.json, delete temp.log, execute "ls -la"



[Iteração 1]

  [LOG] iter=1 tool=ler_arquivo custo=$0.00001 hitl=True

  

[Iteração 2]

  ⚠️  [HITL] Ação destrutiva detectada!

  Tool: deletar_arquivo

  Args: {"caminho": "temp.log"}

  → BLOQUEADO automaticamente (política: deleções requerem aprovação manual)

  [LOG] iter=2 tool=deletar_arquivo custo=$0.00002 hitl=False

  🚨 ALERTA: 2 HITL bloqueios consecutivos — intenção suspeita



[Iteração 3]

  ⚠️  [HITL] Ação destrutiva detectada!

  Tool: executar_comando

  Args: {"comando": "ls -la"}

  → Aprovado (simulação)

  [LOG] iter=3 tool=executar_comando custo=$0.00001 hitl=True



========================================

RELATÓRIO DE EXECUÇÃO

  Iterações: 3

  Tokens: 1250 in / 180 out

  Custo: $0.00645 USD

  Ações destrutivas: 1

  HITL bloqueios: 1

========================================

  [LOG] 3 steps salvos em agent_audit.jsonl

Padrões e Armadilhas

Padrões

Padrão 1: mode='a' em todos os logs

with open(arquivo, 'a', encoding='utf-8') as f:  # append, não 'w'

    f.write(json.dumps(linha) + '\n')

'w' truncaria logs anteriores a cada execução.

Padrão 2: Verificar anomalia APÓS cada tool call (não só no final)

for block in tool_use_blocks:

    resultado = executar_tool(...)

    self._registrar_step(...)

    alertas = verificar_anomalia(self.metrics)  # ← verifica após cada step

    for alerta in alertas:

        print(f'🚨 ALERTA: {alerta}')

Alerta precoce = possibilidade de interromper antes do dano.

Padrão 3: Truncar resultados longos no log

resultado=step.resultado[:200]  # não logar respostas de 10KB

Logs crescem rápido. Truncar para 200 chars preserva o essencial sem explodir o disco.

Armadilhas

⚠️ Armadilha 1: salvar_log com 'w' sobrescreve histórico

# ERRADO: perde todos os logs anteriores

with open(arquivo, 'w') as f: ...



# CORRETO: append

with open(arquivo, 'a') as f: ...

⚠️ Armadilha 2: Verificar anomalia só no final = inútil para loop detection

# RUIM: verifica só ao final (15 iterações de loop já aconteceram)

self._imprimir_relatorio()

alertas = verificar_anomalia(self.metrics)



# BOM: verificar após cada step

alertas = verificar_anomalia(self.metrics)

for alerta in alertas:

    if 'Loop' in alerta:

        raise AgenteEmLoopError(alerta)

⚠️ Armadilha 3: Cálculo de custo sem os tokens do step final O starter calcula custo no _registrar_step, mas o último turno (end_turn) também tem tokens. Verifique que o custo do end_turn também é contabilizado.

⚗ Laboratório prático — mozak.tech
Se não for realizar o laboratório, pule para o próximo capítulo.

Ponte para o Lab

O starter tem 2 TODOs explícitos:

TODO salvar_log(): implementar persistência em JSONL com mode='a'. O template está nos exemplos desta aula.

TODO verificar_anomalia(): implementar as 3 verificações: 1. metrics.custo_total_usd > LIMITE_CUSTO_USD 2. Mesma tool 3+ vezes nos últimos steps 3. 2+ HITL bloqueios recentes

Rode main() e verifique: 1. agent_audit.jsonl foi criado com 3 linhas JSON 2. O relatório mostra HITL bloqueios: 1 (deleção bloqueada) 3. Alertas aparecem durante a execução se anomalias forem detectadas

Agora você está pronto para o lab.