Intentos en capas para agentes de IA en producción

Reintentos en capas para agentes de IA en producción: retraso en las llamadas a LLM, recuperación de errores en herramientas, puntos de control y flujos de trabajo duraderos, para que las APIs inestables y los cierres repentinos no interrumpan las ejecuciones.

Editor: REACTAPP.TOPAñadido a este sitio 26 sept 202610 min de lectura2377 palabras

Fuentes: Building Retries in Agents: How to Build AI Agents That Survive Failures

Reintentos de los agentes que sobreviven a fallos reales

Un agente sencillo con dos herramientas y un bucle while parece funcionar bien hasta que se produce el primer límite de velocidad, una API inestable o un fallo en medio del plan. Los agentes en producción necesitan resiliencia en capas, no solo un try/except alrededor del LLM.

Los fallos no son todos iguales

Clasifique antes de reintentar:

Errores transitorios del LLM/API (429, 503, tiempos de espera)
Errores en las herramientas que el modelo puede corregir (argumentos incorrectos, campos faltantes)
Atascos lógicos (plan incorrecto) que requieren replanificación
Fallas en el proceso o la infraestructura que provocan la pérdida del estado en memoria

Los reintentos ciegos en la categoría incorrecta desperdician dinero y agravan las acciones negativas.

Capa 1: Reintentar la llamada al LLM

Las solicitudes de completación idempotentes utilizan un retroceso exponencial con variabilidad aleatoria:

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import anthropic

llm_retry = retry(
    stop=stop_after_attempt(2),           # 2 total attempts (1 retry)
    wait=wait_exponential(multiplier=2, min=4, max=60),
    retry=retry_if_exception_type((
        anthropic.RateLimitError,          # 429
        anthropic.APITimeoutError,         # Timeout
        anthropic.APIConnectionError,      # Network errors
        anthropic.InternalServerError,     # 500s (often transient)
    )),
)

@llm_retry
async def call_llm(messages, tools):
    response = client.messages.create(
        model="claude-sonnet-4-5-20250929",
        messages=messages,
        tools=tools,
    )
    return response

Límite de intentos; se muestra un error tipado cuando se agota el presupuesto.

Capa 2: Dejar que el modelo maneje los errores de las herramientas

Registre los fallos de las herramientas como observaciones y no como caídas graves, para que el agente pueda corregir los argumentos:

# Inside the agentic loop
for block in response.content:
    if block.type == "tool_use":
        try:
            result = run_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result),
            })
        except Exception as e:
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": f"Error: {str(e)}",
                "is_error": True,
            })

Los bucles al estilo Claude Code tratan stderr/stdout como contexto para el siguiente paso de razonamiento.

@agent.tool
async def process_document(ctx, file_path: str) -> dict:
    """Process a document through the enrichment pipeline."""
    try:
        result = await run_enrichment(file_path)

        if result.status == "error":
            raise RuntimeError(f"Processing failed: {result.error}")

        return {
            "status": "success",
            "records_processed": result.count,
            "output_path": result.output_path,
        }
    except Exception as e:
        return {
            "status": "error",
            "error": str(e),
            "traceback": traceback.format_exc(),
        }

Nivel 3: Resiliencia a nivel de agente

Creación progresiva de puntos de control

Guarde los pasos del plan, los resultados de las herramientas y fragmentos de la conversación para que al reiniciar se continúe en lugar de tener que volver a realizar el trabajo ya realizado:

async def execute_with_resilience(agent, prompt, deps, max_retries=3):
    """Execute agent with checkpoint-based retry."""
    message_history = []

    for attempt in range(max_retries):
        try:
            result = await agent.run(
                prompt,
                deps=deps,
                message_history=message_history,  # Resume from last checkpoint
            )
            return result

        except (RateLimitError, APITimeoutError, APIConnectionError) as e:
            if attempt == max_retries - 1:
                raise  # Final attempt failed

            # Checkpoint: save what we have so far
            # How you access partial messages depends on your framework.
            # PydanticAI exposes all_messages() on the result; other frameworks
            # may store it differently. The key is: grab whatever history exists.
            if hasattr(e, '_partial_messages'):
                message_history = e._partial_messages

            # Also save context that tools have accumulated
            save_checkpoint({
                "message_history": message_history,
                "context_fields": deps.get_context_snapshot(),
            })

            wait_time = min(2 ** attempt * 5, 60)
            logger.warning(
                f"Agent run failed (attempt {attempt + 1}/{max_retries}), "
                f"retrying in {wait_time}s with {len(message_history)} messages"
            )
            await asyncio.sleep(wait_time)

Especialmente, cree un punto de control después de que las herramientas generen efectos secundarios exitosos.

Nivel 4: Caídas de la infraestructura y Temporal

Los flujos de trabajo duraderos son útiles cuando los pasos deben sobrevivir a la interrupción del proceso:

# After each successful tool call, save a checkpoint
async def on_tool_complete(tool_name, result, current_messages):
    """Called after each tool completes successfully."""
    save_checkpoint({
        "message_history": current_messages,
        "last_completed_tool": tool_name,
        "context_fields": get_current_context(),
    })
    logger.info(f"Checkpoint saved after {tool_name}")

Dónde ayuda Temporal

Aprobaciones de larga duración, procesos con intervención humana y cadenas de herramientas que duran varias horas.

Dónde resulta complicado

Los agentes de chat simples pueden no necesitar un motor de flujos de trabajo; el costo operativo podría superar al riesgo.

Cuándo usarlo

Utilice Temporal (o similar) cuando los efectos secundarios deben producirse exactamente una vez y el proceso debe seguir funcionando después de finalizar. Úselo solo en casos de agentes de solicitud/respuesta cortos con posibilidades económicas de reintentos.

Un patrón de producción más ligero

Muchos equipos combinan: reintentos de LLM + observación de errores en las herramientas + puntos de control en Redis/DB + un supervisor que reanuda desde el último paso completado, sin necesidad de una orquestación completa del flujo de trabajo desde el principio.

from temporalio.common import RetryPolicy

TOOL_RETRY_POLICY = RetryPolicy(
    maximum_attempts=2,
    initial_interval=timedelta(seconds=5),
    backoff_coefficient=2.0,
)

Lista de verificación

SandboxRestrictions.default.with_passthrough_modules(
    "anthropic", "httpx", "httpcore", "os", "anyio",
    "anyio._backends", "anyio._backends._asyncio", "threading",
    "pydantic_ai", "pydantic_ai.agent",
    "pydantic_ai.durable_exec", "pydantic_ai.durable_exec.temporal",
    "pydantic_graph", "pydantic_graph.beta", "fitz",  # ... and more
)

Retrasos en caso de errores transitorios del LLM
Errores en las herramientas como mensajes, no como cierres silenciosos
Puntos de control después de modificar las herramientas
Tiempo de espera y límites máximos de pasos
Claves de idempotencia en las operaciones de escritura
Métricas: tasa de reintentos, latencia por paso, tasa de abandono

Siguiente paso

Agregue pruebas de caos: cierre del proceso durante el uso de una herramienta, inserción de respuestas 429 y verificación del comportamiento de reanudación. La resiliencia es una característica clave del producto para los agentes, no algo que se considere posteriormente.

Límite de pasos y límite de dólares por ejecución; los interruptores de apagado deben estar junto a las perillas de reintentos.

Lecturas relacionadas

Fortalecimiento de grafos agentes: contratos de estado, aislamiento y puertas HITL — Fiabilidad en producción para LangGraph: estado plano de TypedDict, nodos de uso único, aristas de reintentos/errores, revisión humana y capacidades de observabilidad.
Estructurar un prompt para un sistema de agentes AI antes de que se convierta en una segunda base de datos — Particionar la identidad, el comportamiento, las herramientas, los principios y las restricciones para que los agentes en producción sigan siendo mantenibles, y mantener la lógica determinista en el código de la aplicación en lugar del prompt.

← Todos los artículos Enviar el brief