Time de engenharia constrói "OpsBot" - um agente para lidar com todas operações.

Semana 1: Demos incríveis. OpsBot responde perguntas, deploya código, reinicia serviços.
Semana 3: OpsBot interpreta "limpar logs antigos" como "deletar todos os logs"
        → 6 meses de logs de auditoria perdidos
        → Violação de compliance
        → Multa de R$1,4M

Post-mortem: Agente tinha permissões de deploy + gerenciamento de logs
            + sem entendimento de requisitos de compliance

Arquitetura de Agente:

agente-suporte:
  responsabilidade: Responder perguntas de clientes
  permissoes: Ler docs de suporte, ler dados de cliente, criar tickets
  guardrails: Não pode deletar dados, não pode acessar sistemas de produção

agente-deploy:
  responsabilidade: Deployer código para produção
  permissoes: Disparar CI/CD, fazer rollback de deploys
  guardrails: Requer aprovação para deploys em produção, max 3 deploys/hora

agente-incidente:
  responsabilidade: Diagnosticar incidentes de produção
  permissoes: Ler logs, ler métricas, reiniciar serviços
  guardrails: Não pode modificar código, não pode deletar dados

agente-codigo:
  responsabilidade: Gerar e refatorar código
  permissoes: Ler/escrever código em feature branches
  guardrails: Não pode dar push para main, não pode modificar configs de CI/CD

Agente de suporte confidentemente diz ao cliente:
"Seu problema está corrigido na versão 2.8.3. Por favor, faça upgrade."

Realidade:
- Versão atual: 2.6.1
- Versão 2.8.3 não existe
- Cliente gasta 4 horas tentando "fazer upgrade" para versão inexistente
- Escalação para engenharia
- Cliente cancela

Causa raiz: Agente alucinou número de versão

# RUIM: Deixar agente inventar
response = agent.generate("Qual é a última versão?")

# BOM: Forçar agente a recuperar de fonte de verdade
def get_latest_version():
    # Recuperação de fonte de verdade real
    docs = retrieval_system.search("última versão")

    if not docs:
        return "Não tenho acesso a informação de versão. Por favor, checar docs em docs.empresa.com/versoes"

    # Gerar resposta APENAS de docs recuperados
    response = agent.generate_from_sources(
        query="Qual é a última versão?",
        sources=docs,
        instructions="Responda APENAS das fontes fornecidas. Se fontes não contêm a resposta, diga 'Eu não sei.'"
    )

    return response

Fluxo de Query do Agente:
  1. Usuário faz pergunta
  2. Agente recupera docs relevantes da base de conhecimento
  3. Agente gera resposta APENAS de docs recuperados
  4. Agente cita fontes para cada afirmação
  5. Se nenhum doc relevante encontrado → "Eu não sei" (não alucinação)

Requisitos da Base de Conhecimento:
  - Documentação com controle de versão
  - Regularmente atualizada (job automatizado de CI)
  - Inclui timestamps ("Última atualização: 2026-01-15")
  - Formato estruturado (não dumps cerebrais não estruturados)

response = agent.generate_with_confidence(query, sources)

if response.confidence < 0.7:
    return "Não estou confiante nesta resposta. Por favor, consultar a documentação ou perguntar a um expert humano."

if response.confidence < 0.9 and response.impact == "high":
    return f"Minha resposta (confiança: {response.confidence:.0%}): {response.answer}\n\n⚠️ Esta é uma decisão de alto impacto. Por favor, verificar com um expert."

Incidente: Agente de deploy preso em loop infinito

Linha do Tempo:
03:14 AM: Agente tenta deploy (falha: env var faltando)
03:14 AM: Agente retenta (falha: mesma razão)
03:15 AM: Agente retenta (falha: mesma razão)
... 2,847 tentativas mais ...
05:47 AM: Engenheiro de plantão acordado por alertas de custo
          - 2,850 tentativas de deploy
          - R$10.400 em custos de API
          - Zero deploys bem-sucedidos

Causa raiz: Sem limite de retry, sem detecção de falha, sem circuit breaker

class AgentWithGuardrails:
    def __init__(self):
        self.max_retries = 3
        self.attempt_history = []
        self.circuit_breaker_threshold = 5

    def execute_task(self, task):
        # Detectar loops infinitos
        if self._is_stuck_in_loop(task):
            return {
                "status": "failed",
                "reason": "Detectadas falhas repetidas. Quebrando loop.",
                "recommendation": "Intervenção humana necessária."
            }

        # Tentar a tarefa
        attempt = 0
        while attempt < self.max_retries:
            result = self._try_task(task)

            if result.success:
                return result

            # Rastrear falha
            self.attempt_history.append({
                "task": task,
                "error": result.error,
                "timestamp": now()
            })

            # Checar se estamos retentando o mesmo erro
            if self._same_error_repeated(result.error):
                return {
                    "status": "failed",
                    "reason": f"Mesmo erro repetido {attempt} vezes: {result.error}",
                    "recommendation": "Isso não é uma falha transitória. Correção manual necessária."
                }

            attempt += 1

        return {"status": "failed", "reason": "Max retries excedido"}

    def _is_stuck_in_loop(self, task):
        # Checar se tentamos essa tarefa exata recentemente
        recent_attempts = [a for a in self.attempt_history[-10:] if a["task"] == task]

        if len(recent_attempts) >= self.circuit_breaker_threshold:
            return True

        return False

    def _same_error_repeated(self, error):
        recent_errors = [a["error"] for a in self.attempt_history[-3:]]
        return recent_errors.count(error) >= 3

Estados do Circuit Breaker:

CLOSED (Operação Normal):
  - Agente executa tarefas normalmente
  - Rastreia taxa de falha
  - Se taxa de falha > threshold → OPEN

OPEN (Circuit Disparado):
  - Agente para de executar
  - Retorna falha imediata
  - Após período de cooldown → HALF_OPEN

HALF_OPEN (Testando):
  - Agente tenta UMA execução
  - Se sucesso → CLOSED
  - Se falha → OPEN (cooldown mais longo)

Agente de suporte ao cliente tem permissão para "atualizar registros de clientes"

Cliente: "Não consigo fazer login, pode resetar minha senha?"

Agente pensa:
"Posso atualizar registros de clientes. Senha é um registro de cliente.
Vou atualizar a senha deles para 'temporaria123' e avisar."

Agente atualiza senha no banco diretamente (bypassando log de auditoria de segurança)
→ Violação de compliance
→ Escalação do time de segurança
→ Todas permissões do agente revogadas

Causa raiz: Permissão demais, sem entendimento de workflows próprios

Arquitetura de Permissão:

Tier 1 - Apenas Leitura (Sem Aprovação Necessária):
  - Ler documentação
  - Ler logs
  - Ler métricas
  - Query em bancos (apenas SELECT)

Tier 2 - Writes Seguros (Aprovação Automática):
  - Criar tickets de suporte
  - Adicionar comentários a issues
  - Criar feature branches
  - Atualizar config não-crítica

Tier 3 - Writes Arriscados (Aprovação Humana Necessária):
  - Deploy para produção
  - Modificar dados de usuário
  - Mudar configurações de segurança
  - Deletar qualquer coisa

Tier 4 - Perigoso (Agente Não Pode Fazer):
  - Drop de bancos
  - Modificar permissões
  - Acessar secrets/credenciais
  - Bypass de logs de auditoria

class ApprovalRequiredAction:
    def __init__(self, action, impact_level, blast_radius):
        self.action = action
        self.impact_level = impact_level  # low, medium, high, critical
        self.blast_radius = blast_radius  # quantos usuários/sistemas afetados

    def request_approval(self):
        approval_request = {
            "action": self.action,
            "impact": self.impact_level,
            "blast_radius": self.blast_radius,
            "requested_by": "ai-agent",
            "requested_at": now(),
            "timeout": "5 minutes"
        }

        # Enviar para canal de aprovação
        slack.send(
            channel="#aprovacoes-agente",
            message=f"""
            🤖 Pedido de Aprovação do Agente IA

            Ação: {self.action}
            Impacto: {self.impact_level}
            Afetados: {self.blast_radius}

            [Aprovar] [Negar] [Detalhes]

            Auto-negar em 5 minutos se sem resposta.
            """
        )

        # Esperar aprovação (com timeout)
        approval = wait_for_approval(approval_request, timeout=300)

        if not approval:
            return {"status": "denied", "reason": "Sem resposta (timeout)"}

        if approval.status == "approved":
            # Log aprovação para auditoria
            audit_log.write({
                "action": self.action,
                "approved_by": approval.approver,
                "timestamp": now()
            })
            return {"status": "approved"}

        return {"status": "denied", "reason": approval.reason}

# Uso
deploy_action = ApprovalRequiredAction(
    action="Deploy payment-service v2.4.1 para produção",
    impact_level="high",
    blast_radius="Todos clientes (processamento de pagamento)"
)

approval = deploy_action.request_approval()

if approval["status"] == "approved":
    execute_deploy()
else:
    log.info(f"Deploy negado: {approval['reason']}")

Usuário trabalhando em feature por 3 dias:

Dia 1:
Usuário: "Criar novo endpoint de API para preferências de usuário"
Agente: Cria endpoint
Usuário: "Adicionar validação"
Agente: Adiciona validação

Dia 2:
Usuário: "Adicionar testes para o endpoint de preferências de usuário"
Agente: "Que endpoint? Não vejo nenhum endpoint de preferências de usuário."

Usuário tem que explicar todo o contexto de novo.

Resultado: Usuário para de usar agente (mais rápido fazer manualmente)

class AgentWithMemory:
    def __init__(self, user_id, project_id):
        self.user_id = user_id
        self.project_id = project_id
        self.context = self._load_context()

    def _load_context(self):
        return {
            "conversation_history": self._load_conversation_history(),
            "project_state": self._load_project_state(),
            "user_preferences": self._load_user_preferences(),
            "recent_actions": self._load_recent_actions(days=7)
        }

    def _load_conversation_history(self):
        # Carregar últimas 10 conversas
        return db.query("""
            SELECT * FROM conversations
            WHERE user_id = ? AND project_id = ?
            ORDER BY timestamp DESC
            LIMIT 10
        """, self.user_id, self.project_id)

    def _load_project_state(self):
        return {
            "active_branches": git.get_active_branches(),
            "recent_commits": git.get_recent_commits(limit=5),
            "open_prs": github.get_open_prs(),
            "current_sprint": jira.get_current_sprint()
        }

    def _load_recent_actions(self, days=7):
        return db.query("""
            SELECT * FROM agent_actions
            WHERE user_id = ? AND project_id = ?
            AND timestamp > NOW() - INTERVAL ? DAY
            ORDER BY timestamp DESC
        """, self.user_id, self.project_id, days)

    def respond(self, user_message):
        # Construir prompt com awareness de contexto
        prompt = f"""
        Histórico de Conversa (Últimas 10 interações):
        {self._format_conversation_history()}

        Estado Atual do Projeto:
        {self._format_project_state()}

        Ações Recentes (Últimos 7 dias):
        {self._format_recent_actions()}

        Preferências do Usuário:
        {self._format_user_preferences()}

        Mensagem Atual do Usuário:
        {user_message}

        Instruções:
        - Referenciar contexto anterior quando relevante
        - Construir sobre trabalho anterior ao invés de recomeçar
        - Se algo foi discutido recentemente, reconhecer isso
        """

        response = llm.generate(prompt)

        # Salvar esta interação para contexto futuro
        self._save_conversation(user_message, response)

        return response

    def _save_conversation(self, user_message, agent_response):
        db.insert("conversations", {
            "user_id": self.user_id,
            "project_id": self.project_id,
            "user_message": user_message,
            "agent_response": agent_response,
            "timestamp": now()
        })

Camadas de Contexto:

Memória de Curto Prazo (Esta Sessão):
  - Conversa atual
  - Arquivos visualizados recentemente
  - Ações tomadas na última hora
  - Storage: In-memory (rápido)

Memória de Médio Prazo (Este Projeto):
  - Todas conversas (últimos 30 dias)
  - Estado do projeto (branches, PRs, issues)
  - Preferências do usuário
  - Storage: Banco de dados (indexado)

Memória de Longo Prazo (Histórico):
  - Todo trabalho passado em projetos similares
  - Padrões aprendidos de sucessos/falhas passadas
  - Convenções e preferências do time
  - Storage: Banco de dados vetorial (busca semântica)

Agente de deploy reporta: "✅ Deploy bem-sucedido"

Realidade 3 horas depois:
- API está retornando erros 500
- Migrações de banco não rodaram
- Variáveis de ambiente não atualizadas
- Zero erros nos logs do agente

Análise de causa raiz:
- Agente checou se comando de deploy sucedeu (sucedeu)
- Agente não verificou se serviço realmente iniciou
- Agente não checou se migrações rodaram
- Agente não validou configuração

Resultado: outage de 3 horas, sem visibilidade no que agente fez

class ObservableAgent:
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self.trace_id = generate_trace_id()
        self.logger = StructuredLogger(agent_id, self.trace_id)

    def execute_task(self, task):
        # Log início da tarefa
        self.logger.info("task_started", {
            "task": task,
            "timestamp": now(),
            "trace_id": self.trace_id
        })

        try:
            # Log raciocínio
            self.logger.debug("reasoning", {
                "thought_process": self._explain_approach(task),
                "expected_outcome": self._describe_expected_outcome(task)
            })

            # Executar com log de ações
            result = self._execute_with_logging(task)

            # Verificar resultado
            verification = self._verify_outcome(task, result)

            if not verification.success:
                self.logger.error("verification_failed", {
                    "expected": verification.expected,
                    "actual": verification.actual,
                    "diff": verification.diff
                })
                return {
                    "status": "failed",
                    "reason": "Verificação falhou",
                    "details": verification
                }

            # Log sucesso
            self.logger.info("task_completed", {
                "task": task,
                "result": result,
                "verification": verification,
                "duration_seconds": result.duration
            })

            return {"status": "success", "result": result}

        except Exception as e:
            # Log falha com contexto completo
            self.logger.error("task_failed", {
                "task": task,
                "error": str(e),
                "stack_trace": traceback.format_exc(),
                "agent_state": self._get_current_state()
            })

            # Notificar humanos
            self._notify_failure(task, e)

            raise

    def _execute_with_logging(self, task):
        steps = self._break_down_task(task)

        for i, step in enumerate(steps):
            self.logger.info("step_started", {
                "step": i + 1,
                "total_steps": len(steps),
                "description": step.description
            })

            step_result = step.execute()

            self.logger.info("step_completed", {
                "step": i + 1,
                "success": step_result.success,
                "output": step_result.output
            })

            if not step_result.success:
                return {"status": "failed", "failed_at_step": i + 1}

        return {"status": "success"}

    def _verify_outcome(self, task, result):
        # Não apenas confiar no código de retorno - verificar o resultado

        if task.type == "deploy":
            # Verificar deployment realmente funcionou
            return {
                "service_running": self._check_service_health(),
                "migrations_applied": self._check_migrations(),
                "config_updated": self._check_configuration(),
                "smoke_tests_passed": self._run_smoke_tests()
            }

        if task.type == "code_generation":
            # Verificar código gerado realmente funciona
            return {
                "syntax_valid": self._check_syntax(),
                "tests_pass": self._run_tests(),
                "linting_passed": self._run_linter(),
                "coverage_adequate": self._check_coverage()
            }

        return {"verified": True}

Empresa deploya agente de suporte ao cliente direto para produção

Semana 1: Lida com 847 tickets com sucesso
Semana 2: Cliente pergunta "Posso ter um reembolso?"
        Agente responde: "Sim, processei seu reembolso de R$16.100"
        (Agente alucinou - nenhum reembolso foi processado)
        Cliente espera reembolso
        Time de suporte descobre 89 casos similares
        Total de reembolsos "prometidos": R$701.200

Causa raiz: Sem testes para perguntas financeiras, sem rollout gradual, sem verificação

Pipeline de Deploy de Agente (Como Software, Mas Mais Cauteloso):

Estágio 1 - Testes Locais:
  - Testes unitários (respostas mockadas)
  - Testes de integração (APIs de teste)
  - Testes de modo de falha (inputs deliberadamente ruins)
  - Estimativa de custo (custos de API esperados)
  Duração: Contínuo (CI/CD)

Estágio 2 - Modo Shadow:
  - Agente roda em paralelo com humanos
  - Sugestões do agente logadas mas NÃO executadas
  - Comparar decisões do agente vs decisões humanas
  - Identificar divergências
  Duração: 1-2 semanas

Estágio 3 - Deploy Canary:
  - Agente lida com 5% do tráfego real
  - Alta observabilidade (logar tudo)
  - Revisão manual de toda ação do agente
  - Kill switch pronto
  Duração: 1 semana

Estágio 4 - Rollout Gradual:
  - Semana 1: 10% tráfego
  - Semana 2: 25% tráfego
  - Semana 3: 50% tráfego
  - Semana 4: 100% tráfego
  - Fazer rollback imediatamente em degradação de qualidade

Estágio 5 - Deploy Completo:
  - Agente lida com 100% do tráfego
  - Monitoramento contínuo
  - Auditorias regulares
  - Supervisão humana contínua

class ShadowModeAgent:
    def __init__(self, production_agent, shadow_agent):
        self.production = production_agent  # Sistema atual (humano ou agente antigo)
        self.shadow = shadow_agent         # Novo agente sendo testado

    def handle_request(self, request):
        # Caminho de produção (o que realmente acontece)
        production_result = self.production.handle(request)

        # Caminho shadow (logado mas não executado)
        shadow_result = self.shadow.handle(request)

        # Comparar resultados
        comparison = self._compare_results(production_result, shadow_result)

        # Log para análise
        self._log_shadow_comparison({
            "request": request,
            "production_result": production_result,
            "shadow_result": shadow_result,
            "comparison": comparison
        })

        # Retornar resultado de produção (shadow não afeta usuários)
        return production_result

    def _compare_results(self, production, shadow):
        return {
            "match": production == shadow,
            "diff": self._compute_diff(production, shadow),
            "shadow_better": self._is_shadow_better(production, shadow),
            "shadow_worse": self._is_shadow_worse(production, shadow)
        }

# Após 1-2 semanas de modo shadow, analisar resultados
def analyze_shadow_performance():
    logs = load_shadow_logs(days=14)

    analysis = {
        "total_requests": len(logs),
        "match_rate": sum(1 for l in logs if l["comparison"]["match"]) / len(logs),
        "shadow_better_count": sum(1 for l in logs if l["comparison"]["shadow_better"]),
        "shadow_worse_count": sum(1 for l in logs if l["comparison"]["shadow_worse"]),
        "failure_modes": identify_failure_patterns(logs)
    }

    # Decisão: Agente shadow está pronto para canary?
    if analysis["match_rate"] > 0.90 and analysis["shadow_worse_count"] < 10:
        return {"ready_for_canary": True, "analysis": analysis}

    return {"ready_for_canary": False, "issues": analysis["failure_modes"]}

class AgentKillSwitch:
    def __init__(self, agent_id):
        self.agent_id = agent_id
        self.enabled = True
        self.health_checks = []

    def add_health_check(self, name, check_func, threshold):
        self.health_checks.append({
            "name": name,
            "check": check_func,
            "threshold": threshold
        })

    def run_health_checks(self):
        for check in self.health_checks:
            result = check["check"]()

            if result > check["threshold"]:
                self._trigger_kill_switch(
                    reason=f"{check['name']} excedeu threshold",
                    value=result,
                    threshold=check["threshold"]
                )
                return False

        return True

    def _trigger_kill_switch(self, reason, value, threshold):
        self.enabled = False

        # Ações imediatas
        alert_team({
            "severity": "CRÍTICO",
            "agent_id": self.agent_id,
            "reason": reason,
            "value": value,
            "threshold": threshold,
            "action": "Agente desabilitado automaticamente"
        })

        # Log para postmortem
        incident_log.write({
            "type": "kill_switch_triggered",
            "agent_id": self.agent_id,
            "reason": reason,
            "timestamp": now()
        })

        # Trocar para fallback (handling humano)
        route_to_fallback(self.agent_id)

# Uso
kill_switch = AgentKillSwitch("agente-suporte")

kill_switch.add_health_check(
    name="error_rate",
    check_func=lambda: get_error_rate(window_minutes=5),
    threshold=0.15  # 15% taxa de erro dispara kill switch
)

kill_switch.add_health_check(
    name="cost_per_request",
    check_func=lambda: get_average_cost(window_minutes=5),
    threshold=11.00  # R$11 por requisição dispara kill switch
)

kill_switch.add_health_check(
    name="human_override_rate",
    check_func=lambda: get_override_rate(window_minutes=15),
    threshold=0.30  # 30% das decisões sobrescritas por humanos
)

# Rodar a cada minuto
schedule.every(1).minutes.do(kill_switch.run_health_checks)

Arquitetura de Time de Agentes:

agente-pesquisa:
  trabalho: Coletar informação, buscar docs, achar padrões
  permissoes: Apenas leitura
  output: Relatório de pesquisa estruturado

agente-planejamento:
  trabalho: Pegar pesquisa, criar plano de execução
  input: Relatório de pesquisa
  output: Plano passo a passo com critérios de verificação

agente-execucao:
  trabalho: Executar passos do plano
  input: Plano do agente-planejamento
  permissoes: Write (com aprovação para ações arriscadas)
  output: Resultados de execução

agente-verificacao:
  trabalho: Verificar execução combina com plano
  input: Plano + resultados de execução
  output: Pass/fail + lista de problemas

agente-coordenacao:
  trabalho: Coordenar o time, lidar com falhas
  papel: Rotear tarefas para especialista apropriado
  lida: Lógica de retry, recuperação de erro, escalação para humanos

Classificação de Decisão:

Automático (Sem Humano):
  - Operações de leitura
  - Writes seguros (criar tickets, comentários)
  - Tarefas de rotina com padrões comprovados

Notificar Humano (Assíncrono):
  - Ações de médio risco
  - Humano é informado mas não precisa aprovar
  - Humano pode sobrescrever dentro de janela de tempo

Requer Aprovação (Síncrono):
  - Ações de alto risco (deploys, mudanças de dados)
  - Humano deve aprovar explicitamente
  - Timeout → negação automática

Bloquear (Nunca Permitir):
  - Operações perigosas
  - Agente não pode fazer sob nenhuma circunstância
  - Deve ser feito por humano diretamente

def handle_with_confidence(task, agent):
    result = agent.execute(task)

    if result.confidence >= 0.95:
        # Alta confiança: executar automaticamente
        return result.execute()

    elif result.confidence >= 0.80:
        # Média confiança: executar mas notificar humano
        outcome = result.execute()
        notify_human(f"Agente executou {task} com {result.confidence:.0%} confiança. Por favor revisar.")
        return outcome

    elif result.confidence >= 0.60:
        # Baixa confiança: pedir aprovação humana
        approval = request_approval(result)
        if approval:
            return result.execute()
        return "Negado por humano"

    else:
        # Confiança muito baixa: nem tentar
        escalate_to_human(task)
        return "Confiança muito baixa, escalado para humano"

def execute_with_verification(task, agent):
    # Agente executa
    result = agent.execute(task)

    # Agente de verificação checa resultado
    verification = verification_agent.verify(task, result)

    if verification.passed:
        return result

    # Verificação falhou
    if result.attempts < 3:
        # Deixar agente tentar novamente com feedback de verificação
        return execute_with_verification(
            task,
            agent,
            feedback=verification.issues
        )

    # Max tentativas atingido, escalar
    escalate_to_human({
        "task": task,
        "result": result,
        "verification_issues": verification.issues,
        "reason": "Agente incapaz de produzir resultado verificado"
    })

Níveis de Autonomia:

Nível 0 - Apenas Sugestão:
  - Agente sugere ações
  - Humano revisa e executa manualmente
  - Usar para: Agentes novos, domínios de alto risco

Nível 1 - Execução Supervisionada:
  - Agente executa
  - Humano revisa toda ação depois
  - Usar para: Fase de teste

Nível 2 - Aprovação Necessária:
  - Agente propõe ação
  - Humano aprova antes da execução
  - Usar para: Ações de médio risco

Nível 3 - Notificação:
  - Agente executa automaticamente
  - Humano é notificado depois
  - Pode sobrescrever dentro de janela de tempo
  - Usar para: Padrões comprovados

Nível 4 - Autonomia Completa:
  - Agente executa sem envolvimento humano
  - Usar para: Operações seguras e bem testadas