Essais en couches pour les agents IA en production

Tentatives répétées en plusieurs niveaux pour les agents IA en production : retards dans les appels aux LLM, récupération en cas d’erreur des outils, points de contrôle et workflows durables, afin que les APIs instables et les pannes ne mettent pas fin aux exécutions.

Éditeur : REACTAPP.TOPAjouté sur ce site 26 sept. 202610 min de lecture2377 mots

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

Re tentatives des agents qui survivent aux véritables pannes

try/except autour du LLM.

Les pannes ne sont pas toutes identiques

Erreurs temporaires du LLM/API (429, 503, temps d’attente dépassés)
Erreurs des outils que le modèle peut corriger (arguments incorrects, champs manquants)
Blocages logiques (plan erroné) nécessitant une nouvelle planification
Pannes du processus ou de l’infrastructure entraînant la perte des données en mémoire

Couche 1 — Re tenter l’appel au LLM

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

Couche 2 — Laisser le modèle gérer les erreurs des outils

Rapportez les échecs des outils en tant qu’observations et non comme des plantages totaux, afin que l’agent puisse corriger les arguments utilisés :

# 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,
            })

Les boucles de style Claude Code traitent stderr/stdout comme du contexte pour la réflexion suivante.

@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(),
        }

Niveau 3 — Résilience au niveau de l’agent

Vérification progressive des points d’état

Conservez les étapes du plan, les résultats des outils ainsi que les extraits de conversation afin qu’un redémarrage permette de reprendre là où on en était, sans avoir à refaire du travail déjà effectué :

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)

Faites un point d’état surtout après des effets secondaires réussis de l’outil.

Niveau 4 — Plantages d’infrastructure et Temporal

Les workflows durables se révèlent utiles lorsque les étapes doivent survivre à la fin de vie du processus d’exécution :

# 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}")

Où Temporal est utile

Approbations longues, intervention humaine, chaînes d’outils s’étendant sur plusieurs heures.

Où cela devient complexe

Les agents de chat simples n’ont pas forcément besoin d’un moteur de workflow ; le coût opérationnel peut dépasser le risque associé.

Quand l’utiliser

Utilisez Temporal (ou un outil similaire) lorsque les effets secondaires doivent se produire exactement une fois et survivre à la fin du processus. Évitez-le pour les agents de demande/réponse courts avec des tentatives de réessai peu coûteuses.

Un modèle de production plus léger

De nombreuses équipes combinent : des tentatives de réessai avec LLM + des observations des erreurs des outils + des points de contrôle Redis/DB + un superviseur qui reprend le travail à partir du dernier étape terminée — sans une orchestration complète du flux de travail dès le début.

from temporalio.common import RetryPolicy

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

Liste de contrôle

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
)

Délai d’attente en cas d’erreurs temporaires du LLM
Les erreurs des outils sous forme de messages, et non pas d’arrêts silencieux
Points de contrôle après modification des outils
Délais d’expiration et budgets maximal de pas
Clés d’idempotence pour les écritures
Métriques : taux de réessai, latence des étapes, taux d’abandon

À venir

Ajoutez des tests de chaos : arrêtez le processus au milieu d’une utilisation d’outil, injectez des réponses 429, et vérifiez le comportement de reprise. La résilience est une fonctionnalité du produit pour les agents, et non quelque chose ajouté en dernier recours.

Budget maximal de pas et de dollars par exécution ; les interrupteurs d’arrêt se trouvent à côté des boutons de tentative répétée.

Budget maximal de pas et de dollars par exécution ; les interrupteurs d’arrêt se trouvent à côté des boutons de tentative.

← Tous les articles Envoyer le brief