Recuperación de Errores en Agentes: 5 Patrones para Fiabilidad en Producción

Tu agente funcionó perfectamente en las pruebas. Luego, en producción, alcanzó un límite de velocidad en el paso 3 de un flujo de trabajo de múltiples pasos, lanzó una excepción no capturada y dejó el sistema en un estado indefinido. Sin punto de control. Sin reintento. Sin alternativa. Solo silencio—y un pipeline roto que hay que reiniciar manualmente.

La recuperación de errores en agentes es la diferencia entre una demo y un sistema en producción. Este artículo cubre cinco patrones utilizados en flujos de trabajo agénticos de producción: retroceso exponencial, disyuntores de circuito, punto de control y reanudación, estrategias de respaldo y colas de escalado. Cada patrón está implementado con el SDK de Anthropic y funciona con cualquier framework de orquestación.

Requisitos previos: Debes estar familiarizado con Python y con el funcionamiento de la API de Claude. No se requiere experiencia previa con sistemas tolerantes a fallos.

Por Qué los Agentes Fallan de Forma Diferente al Software Tradicional

El software tradicional falla en límites claros: una consulta de base de datos devuelve un error, una llamada HTTP agota el tiempo, un archivo no se encuentra. Gestionas la excepción y continúas.

Los agentes fallan de formas más complejas:

• Progreso parcial: Un agente completa los pasos 1–4 de un flujo de 8 pasos, luego falla. Sin recuperación, se pierde todo el progreso o se repite trabajo.
• Estado ambiguo: El agente llamó a una herramienta, pero la respuesta estaba malformada. ¿Ocurrió la acción? ¿Se debe reintentar?
• Fallo en cascada: Una llamada API lenta paraliza todo el bucle de razonamiento. El agente no falla directamente—simplemente se bloquea.
• Fallo silencioso: El LLM devuelve una respuesta, pero no sigue el formato esperado. El analizador posterior falla silenciosamente.

Estos modos de fallo requieren patrones más allá de simples bloques try/except.

Patrón 1: Retroceso Exponencial con Variación Aleatoria

El fallo más común en un agente es un error transitorio: límites de velocidad, interrupciones de red, indisponibilidad temporal del servicio. La solución es reintentar—pero los reintentos ingenuos empeoran la limitación de velocidad al enviar ráfagas de solicitudes.

El retroceso exponencial duplica el tiempo de espera entre reintentos. La variación aleatoria (jitter) añade aleatoriedad para evitar que múltiples agentes reintenten simultáneamente (el problema de la “manada atronadora”).

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]:
    """
    Llama a la API de Claude con retroceso exponencial en errores transitorios.
    Reintenta en límites de velocidad y errores del servidor; lanza inmediatamente en errores del 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"Límite de velocidad. Reintentando en {jitter:.1f}s (intento {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"Error del servidor {e.status_code}. Reintentando en {jitter:.1f}s")
            time.sleep(jitter)
        except anthropic.APIConnectionError as e:
            if attempt == max_retries - 1:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            print(f"Error de conexión. Reintentando en {delay:.1f}s")
            time.sleep(delay)
    return None

Cuándo usar: En cualquier llamada API dentro de un bucle agéntico. Este debe ser tu envoltorio predeterminado para todas las llamadas LLM.

Cuándo NO usar: No reintentes en errores de validación (400), fallos de autenticación (401) o 404. Estos no tendrán éxito al reintentar—requieren correcciones de código.

Patrón 2: Disyuntor de Circuito

El retroceso exponencial gestiona fallos transitorios breves. Pero si un servicio externo está caído durante 20 minutos, no quieres que tu agente reintente cada pocos segundos durante todo ese tiempo. Un disyuntor de circuito rastrea las tasas de fallos y detiene temporalmente las llamadas a un servicio que está fallando.

El disyuntor tiene tres estados:

• Cerrado (normal): Las solicitudes pasan.
• Abierto (fallando): Las solicitudes fallan inmediatamente sin llamar al servicio.
• Semiabierto (recuperándose): Se permite una solicitud de prueba; si tiene éxito, el circuito se cierra.

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("El disyuntor está ABIERTO — servicio no disponible")
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            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

Cuándo usar: Cuando tu agente llama a herramientas externas (APIs web, bases de datos, servicios de terceros) que podrían estar inactivos durante períodos prolongados.

Patrón 3: Punto de Control y Reanudación

Los agentes de larga duración que completan trabajo en etapas necesitan poder reanudar desde el último paso exitoso tras un fallo. Sin puntos de control, un fallo en el paso 7 de 10 significa volver a ejecutar los pasos 1–6 — desperdiciando tiempo y dinero.

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 patrón se integra de forma natural con los checkpointers integrados MemorySaver y PostgresSaver de LangGraph. Consulta Máquinas de Estado y Agentes: Construyendo Flujos de Trabajo Fiables con LangGraph para el enfoque a nivel de framework.

Patrón 4: Estrategias de Respaldo

Algunos fallos no son reintentables. La API está caída. La herramienta devolvió un resultado inutilizable. El modelo se negó a responder. Para estos casos, necesitas un respaldo: un camino alternativo que mantenga al agente avanzando.

from typing import Callable

def with_fallback(
    primary: Callable[[], str],
    fallback: Callable[[], str],
    error_types: tuple = (Exception,),
) -> str:
    """
    Intenta la función primaria; recurre a la secundaria en errores especificados.
    """
    try:
        return primary()
    except error_types as e:
        print(f"Primario falló ({type(e).__name__}): {e}. Usando respaldo.")
        return fallback()


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

    def fallback():
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{"role": "user", "content": question}],
        )
        return f"[Respuesta de respaldo] {response.content[0].text}"

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

Cuándo usar: Cuando tienes una jerarquía clara de primario/secundario y los resultados parciales son aceptables.

Patrón 5: Cola de Escalado

Algunos fallos genuinamente no pueden resolverse automáticamente. El modelo está atascado en un bucle. La tarea requiere información que solo un humano tiene. La acción es irreversible y la confianza es baja. Para estos casos, necesitas un camino de escalado estructurado.

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"[ESCALADO] Tarea {task_id}: {reason.value}")
    return record

Cuándo usar: Para tareas que involucran acciones irreversibles, decisiones de baja confianza, o cualquier caso donde equivocarse tiene consecuencias reales. Consulta Patrones Multi-Agente para un agente supervisor que lee y resuelve colas de escalado.

Casos de Uso del Mundo Real

Pipeline de Procesamiento de Documentos

Un agente que extrae datos de PDFs, los valida contra un esquema y escribe registros en una base de datos. Solución: punto de control y reanudación para cada documento; cola de escalado para documentos que fallan repetidamente; disyuntor en la conexión de base de datos.

Agente de Soporte al Cliente

Un agente que lee tickets de soporte, los categoriza, redacta respuestas y los enruta. Solución: escalado de baja confianza a agentes humanos; respaldo a un modelo de clasificación más simple cuando el principal agota el tiempo.

Agente de Investigación y Síntesis

Un agente que consulta múltiples APIs, sintetiza hallazgos y escribe un informe. Solución: respaldo de herramienta entre fuentes de datos; retroceso exponencial en APIs con límite de velocidad; resultados parciales con una nota clara de “datos no disponibles”.

Errores Comunes

Error 1: Reintentar Errores No Reintentables

El problema: Tu bucle de reintentos captura todas las excepciones, incluyendo errores de validación y fallos de autenticación.

La solución: Clasifica los errores antes de reintentar. Solo reintenta errores transitorios (5xx, 429, tiempos de espera de conexión).

# Incorrecto: captura todo
except Exception:
    retry()

# Correcto: solo reintenta errores reintentables
except (anthropic.RateLimitError, anthropic.APIConnectionError):
    retry()
except anthropic.APIStatusError as e:
    if e.status_code >= 500:
        retry()
    else:
        raise

Error 2: Perder el Estado al Reintentar

El problema: Reinicias el bucle del agente pero reseteas los messages cada vez.

La solución: Preserva el historial de mensajes entre reintentos. Solo reintenta el paso específico fallido.

Error 3: Bucles de Reintento Ilimitados

El problema: Sin límite de max_retries. Un agente se queda atascado reintentando indefinidamente.

La solución: Siempre establece un número máximo de reintentos. Después de agotarlos, escala o falla rápidamente.

Lista de Verificación para Producción

Antes de desplegar un agente en producción, verifica:

• Todas las llamadas API envueltas con retroceso exponencial (base 1s, máximo 60s, variación aleatoria activada)
• Disyuntores en cada dependencia externa
• Puntos de control guardados en almacenamiento duradero después de cada paso significativo
• Rutas de respaldo probadas y marcadas en los metadatos de respuesta
• Cola de escalado monitoreada y revisada al menos diariamente
• Conteos de reintentos acotados (max_retries ≤ 5)
• Pruebas para cada ruta de recuperación con fallos inyectados

Próximos Pasos

1. Comienza con el retroceso — Envuelve cada llamada API en call_with_backoff como línea base.
2. Añade puntos de control a cualquier flujo de trabajo de más de 2–3 pasos.
3. Construye una cola de escalado para tareas que involucran acciones irreversibles.
4. Escribe pruebas de inyección de fallos — Prueba tus rutas de recuperación desencadenando deliberadamente cada tipo de error.

Guías relacionadas:

• Máquinas de Estado y Agentes: Construyendo Flujos de Trabajo Fiables con LangGraph
• Patrones Multi-Agente
• Depuración y Observabilidad para Agentes de IA

Artículos Relacionados

• Patrones de Uso de Herramientas: Interfaces Agente-Herramienta Confiables
• Sistemas de Memoria para Agentes: Contexto Persistente para tu IA