5 de mar. de 2026

Recuperação de Erros em Agentes: 5 Padrões para Confiabilidade em Produção

Seu agente funcionava perfeitamente nos testes. Então, em produção, ele atingiu um limite de taxa no passo 3 de um fluxo de trabalho de múltiplos passos, lançou uma exceção não capturada e deixou seu sistema em um estado indefinido. Sem ponto de controle. Sem nova tentativa. Sem alternativa. Apenas silêncio—e um pipeline quebrado para reiniciar manualmente.

A recuperação de erros em agentes é a diferença entre uma demonstração e um sistema em produção. Este artigo cobre cinco padrões usados em fluxos de trabalho agênticos de produção: recuo exponencial, disjuntores de circuito, ponto de controle e retomada, estratégias de fallback e filas de escalonamento. Cada padrão está implementado com o SDK da Anthropic e funciona com qualquer framework de orquestração.

Pré-requisitos: Você deve estar familiarizado com Python e com o funcionamento da API Claude.

Por Que Agentes Falham Diferente do Software Tradicional

Agentes falham de formas mais complexas:

Progresso parcial: Um agente completa os passos 1–4 de um fluxo de 8 passos, depois falha. Sem recuperação, perde-se todo o progresso.
Estado ambíguo: O agente chamou uma ferramenta, mas a resposta estava malformada. A ação ocorreu? Deve-se tentar novamente?
Falha em cascata: Uma chamada API lenta paralisa todo o loop de raciocínio.
Falha silenciosa: O LLM retorna uma resposta, mas não segue o formato esperado. O parser downstream falha silenciosamente.

Padrão 1: Recuo Exponencial com Variação Aleatória

O recuo exponencial dobra o tempo de espera entre as tentativas. A variação aleatória (jitter) adiciona aleatoriedade para evitar que múltiplos agentes tentem novamente simultaneamente.

import anthropic
import time
import random
from typing import Optional

client = anthropic.Anthropic()

def call_with_backoff(
    messages: list,
    model: str = "claude-sonnet-4-6",
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
) -> Optional[anthropic.types.Message]:
    """
    Chama a API Claude com recuo exponencial em erros transitórios.
    Tenta novamente em limites de taxa e erros de servidor;
    lança imediatamente em erros do cliente.
    """
    for attempt in range(max_retries):
        try:
            return client.messages.create(
                model=model,
                max_tokens=1024,
                messages=messages,
            )
        except anthropic.RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay)
            print(f"Limite de taxa. Tentando novamente em {jitter:.1f}s (tentativa {attempt + 1}/{max_retries})")
            time.sleep(jitter)
        except anthropic.APIStatusError as e:
            if e.status_code < 500:
                raise
            if attempt == max_retries - 1:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay)
            print(f"Erro do servidor {e.status_code}. Tentando novamente em {jitter:.1f}s")
            time.sleep(jitter)
        except anthropic.APIConnectionError:
            if attempt == max_retries - 1:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            time.sleep(delay)
    return None

Quando usar: Em qualquer chamada API dentro de um loop agêntico. Este deve ser seu wrapper padrão para todas as chamadas LLM.

Padrão 2: Disjuntor de Circuito

Um disjuntor de circuito rastreia as taxas de falha e para temporariamente de chamar um serviço que está falhando. Tem três estados: Fechado (normal), Aberto (falhando) e Semi-aberto (recuperando).

import time
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, TypeVar, Any

T = TypeVar("T")

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: float = 60.0
    success_threshold: int = 2

    _state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    _failure_count: int = field(default=0, init=False)
    _success_count: int = field(default=0, init=False)
    _last_failure_time: float = field(default=0.0, init=False)

    @property
    def state(self) -> CircuitState:
        if self._state == CircuitState.OPEN:
            if time.time() - self._last_failure_time > self.recovery_timeout:
                self._state = CircuitState.HALF_OPEN
                self._success_count = 0
        return self._state

    def call(self, func: Callable[..., T], *args: Any, **kwargs: Any) -> T:
        if self.state == CircuitState.OPEN:
            raise RuntimeError("Disjuntor ABERTO — serviço indisponível")
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception:
            self._on_failure()
            raise

    def _on_success(self) -> None:
        self._failure_count = 0
        if self._state == CircuitState.HALF_OPEN:
            self._success_count += 1
            if self._success_count >= self.success_threshold:
                self._state = CircuitState.CLOSED

    def _on_failure(self) -> None:
        self._failure_count += 1
        self._last_failure_time = time.time()
        if self._failure_count >= self.failure_threshold:
            self._state = CircuitState.OPEN

Padrão 3: Ponto de Controle e Retomada

Agentes de longa duração precisam retomar a partir do último passo bem-sucedido após uma falha. Este padrão serializa o estado do agente em armazenamento durável a cada passo.

import json
import os
from dataclasses import dataclass, asdict, field
from typing import Optional

@dataclass
class AgentCheckpoint:
    task_id: str
    current_step: int
    completed_steps: list[str] = field(default_factory=list)
    results: dict = field(default_factory=dict)
    messages: list[dict] = field(default_factory=list)

    def save(self, directory: str = ".checkpoints") -> None:
        os.makedirs(directory, exist_ok=True)
        path = os.path.join(directory, f"{self.task_id}.json")
        with open(path, "w") as f:
            json.dump(asdict(self), f, indent=2)

    @classmethod
    def load(cls, task_id: str, directory: str = ".checkpoints") -> Optional["AgentCheckpoint"]:
        path = os.path.join(directory, f"{task_id}.json")
        if not os.path.exists(path):
            return None
        with open(path) as f:
            data = json.load(f)
        return cls(**data)

    def clear(self, directory: str = ".checkpoints") -> None:
        path = os.path.join(directory, f"{self.task_id}.json")
        if os.path.exists(path):
            os.remove(path)

Este padrão se integra naturalmente com os checkpointers MemorySaver e PostgresSaver do LangGraph. Veja Máquinas de Estado e Agentes com LangGraph.

Padrão 4: Estratégias de Fallback

Algumas falhas não são passíveis de nova tentativa. Você precisa de um caminho alternativo que mantenha o agente em movimento.

from typing import Callable

def with_fallback(
    primary: Callable[[], str],
    fallback: Callable[[], str],
    error_types: tuple = (Exception,),
) -> str:
    """Tenta a função primária; recorre à secundária em erros especificados."""
    try:
        return primary()
    except error_types as e:
        print(f"Primário falhou ({type(e).__name__}): {e}. Usando fallback.")
        return fallback()


def answer_question(question: str) -> str:
    def primary():
        # Modelo primário: o mais capaz
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": question}],
        )
        return response.content[0].text

    def fallback():
        # Fallback: modelo mais rápido e leve
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{"role": "user", "content": question}],
        )
        return f"[Resposta de fallback] {response.content[0].text}"

    return with_fallback(
        primary, fallback,
        error_types=(anthropic.RateLimitError, anthropic.APIStatusError),
    )

Padrão 5: Fila de Escalonamento

Algumas falhas genuinamente não podem ser resolvidas automaticamente. Uma fila de escalonamento captura tarefas com falha com contexto suficiente para que um humano as resolva.

import json
import uuid
from datetime import datetime
from dataclasses import dataclass, asdict
from enum import Enum

class EscalationReason(Enum):
    MAX_RETRIES_EXCEEDED = "max_retries_exceeded"
    AMBIGUOUS_STATE = "ambiguous_state"
    LOW_CONFIDENCE = "low_confidence"
    REQUIRES_HUMAN = "requires_human"
    IRREVERSIBLE_ACTION = "irreversible_action"

@dataclass
class EscalationRecord:
    id: str
    task_id: str
    reason: str
    error_message: str
    agent_state: dict
    context: str
    timestamp: str
    resolved: bool = False

    def save(self, queue_file: str = "escalation_queue.jsonl") -> None:
        with open(queue_file, "a") as f:
            f.write(json.dumps(asdict(self)) + "\n")


def escalate(
    task_id: str,
    reason: EscalationReason,
    error_message: str,
    agent_state: dict,
    context: str = "",
) -> EscalationRecord:
    record = EscalationRecord(
        id=str(uuid.uuid4()),
        task_id=task_id,
        reason=reason.value,
        error_message=error_message,
        agent_state=agent_state,
        context=context,
        timestamp=datetime.utcnow().isoformat(),
    )
    record.save()
    print(f"[ESCALONADO] Tarefa {task_id}: {reason.value}")
    return record

Quando usar: Para tarefas com ações irreversíveis ou decisões de baixa confiança. Veja Padrões Multi-Agente para um agente supervisor.

Erros Comuns

Erro 1: Tentar Novamente Erros Não Tentáveis

A solução: Classifique os erros antes de tentar novamente. Apenas tente novamente erros transitórios.

# Errado: captura tudo
except Exception:
    retry()

# Correto: apenas erros tentáveis
except (anthropic.RateLimitError, anthropic.APIConnectionError):
    retry()
except anthropic.APIStatusError as e:
    if e.status_code >= 500:
        retry()
    else:
        raise

Erro 2: Perder o Estado ao Tentar Novamente

A solução: Preserve o histórico de mensagens entre as tentativas. Tente novamente apenas o passo específico que falhou.

Erro 3: Loops de Nova Tentativa Ilimitados

A solução: Sempre defina um max_retries. Após esgotá-las, escalone ou falhe rapidamente.

Lista de Verificação para Produção

Todas as chamadas API envolvidas com recuo exponencial (base 1s, máx 60s, jitter ativado)
Disjuntores em cada dependência externa
Pontos de controle salvos em armazenamento durável após cada passo significativo
Caminhos de fallback testados e marcados nos metadados de resposta
Fila de escalonamento monitorada e revisada pelo menos diariamente
Contagens de tentativas limitadas (max_retries ≤ 5)

Próximos Passos

Comece com recuo — Envolva cada chamada API em call_with_backoff.
Adicione pontos de controle a qualquer fluxo com mais de 2–3 passos.
Construa uma fila de escalonamento para tarefas com ações irreversíveis.
Escreva testes de injeção de falhas — Teste seus caminhos de recuperação.

Guias relacionados: