5 mars 2026

Récupération d'Erreurs pour Agents : 5 Patrons pour la Fiabilité en Production

Votre agent fonctionnait parfaitement en test. Puis en production, il a atteint une limite de débit à l’étape 3 d’un flux de travail en plusieurs étapes, a levé une exception non capturée et a laissé votre système dans un état indéfini. Aucun point de contrôle. Aucune nouvelle tentative. Aucun repli. Juste le silence—et un pipeline cassé à redémarrer manuellement.

La récupération d’erreurs pour agents est la différence entre une démo et un système en production. Cet article couvre cinq patrons utilisés dans les flux de travail agentiques en production : recul exponentiel, disjoncteurs de circuit, point de contrôle et reprise, stratégies de repli et files d’escalade. Chaque patron est implémenté avec le SDK Anthropic et fonctionne avec n’importe quel framework d’orchestration.

Prérequis : Vous devez être à l’aise avec Python et connaître le fonctionnement de l’API Claude.

Pourquoi les Agents Échouent Différemment du Logiciel Traditionnel

Les agents échouent de manière plus complexe que le logiciel traditionnel :

Progression partielle : Un agent complète les étapes 1–4 d’un flux de 8 étapes, puis échoue. Sans récupération, on perd toute progression.
État ambigu : L’agent a appelé un outil, mais la réponse était malformée. L’action a-t-elle eu lieu ? Faut-il réessayer ?
Défaillance en cascade : Un appel API lent bloque toute la boucle de raisonnement.
Défaillance silencieuse : Le LLM renvoie une réponse, mais elle ne suit pas le format attendu. L’analyseur en aval échoue silencieusement.

Patron 1 : Recul Exponentiel avec Gigue

Le recul exponentiel double le temps d’attente entre les tentatives. La gigue ajoute de l’aléatoire pour éviter que plusieurs agents réessaient simultanément.

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]:
    """
    Appelle l'API Claude avec recul exponentiel sur les erreurs transitoires.
    Réessaie sur les limites de débit et les erreurs serveur ;
    lève immédiatement sur les erreurs client.
    """
    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 débit. Nouvelle tentative dans {jitter:.1f}s (tentative {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"Erreur serveur {e.status_code}. Nouvelle tentative dans {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

Quand utiliser : Pour tout appel API dans une boucle agentique. Ce doit être votre enveloppe par défaut pour tous les appels LLM.

Quand NE PAS utiliser : Ne réessayez pas sur les erreurs de validation (400), les échecs d’authentification (401) ou les 404.

Patron 2 : Disjoncteur de Circuit

Un disjoncteur de circuit suit les taux de défaillance et arrête temporellement d’appeler un service défaillant. Il a trois états : Fermé (normal), Ouvert (défaillant) et Semi-ouvert (en récupération).

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("Le disjoncteur est OUVERT — service indisponible")
        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

Quand utiliser : Lorsque votre agent appelle des outils externes (APIs web, bases de données) qui pourraient être en panne pendant des périodes prolongées.

Patron 3 : Point de Contrôle et Reprise

Les agents de longue durée doivent pouvoir reprendre depuis la dernière étape réussie après une défaillance. Ce patron sérialise l’état de l’agent vers un stockage durable à chaque étape.

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)

Ce patron s’intègre naturellement avec MemorySaver et PostgresSaver de LangGraph. Voir Machines à États et Agents avec LangGraph.

Patron 4 : Stratégies de Repli

Certaines défaillances ne sont pas réessayables. Vous avez besoin d’un chemin alternatif qui maintient l’agent en mouvement.

from typing import Callable

def with_fallback(
    primary: Callable[[], str],
    fallback: Callable[[], str],
    error_types: tuple = (Exception,),
) -> str:
    """Essaie la fonction primaire ; repli vers la secondaire sur erreur."""
    try:
        return primary()
    except error_types as e:
        print(f"Principal échoué ({type(e).__name__}): {e}. Utilisation du repli.")
        return fallback()


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

    def fallback():
        # Repli : modèle plus rapide et léger
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{"role": "user", "content": question}],
        )
        return f"[Réponse de repli] {response.content[0].text}"

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

Patron 5 : File d’Escalade

Certaines défaillances ne peuvent genuinement pas être résolues automatiquement. Une file d’escalade capture les tâches échouées avec suffisamment de contexte pour qu’un humain puisse les résoudre.

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"[ESCALADE] Tâche {task_id}: {reason.value}")
    return record

Quand utiliser : Pour les tâches impliquant des actions irréversibles ou des décisions à faible confiance. Voir Patrons Multi-Agents pour un agent superviseur qui résout les escalades.

Cas d’Usage Réels

Pipeline de Traitement de Documents

Un agent qui extrait des données de PDFs, les valide et écrit des enregistrements dans une base de données. Solution : point de contrôle et reprise par document ; file d’escalade pour les documents qui échouent ; disjoncteur sur la connexion à la base de données.

Agent de Support Client

Un agent qui lit les tickets de support, les catégorise et les route. Solution : escalade à faible confiance vers des agents humains ; repli vers un modèle de classification plus simple en cas d’expiration.

Agent de Recherche et Synthèse

Un agent qui interroge plusieurs APIs et synthétise les résultats. Solution : repli d’outil entre les sources de données ; recul exponentiel sur les APIs limitées en débit ; résultats partiels avec une note claire.

Erreurs Courantes

Erreur 1 : Réessayer des Erreurs Non Réessayables

La solution : Classifiez les erreurs avant de réessayer. Ne réessayez que les erreurs transitoires.

# Incorrect : capture tout
except Exception:
    retry()

# Correct : réessaie uniquement les erreurs transitoires
except (anthropic.RateLimitError, anthropic.APIConnectionError):
    retry()
except anthropic.APIStatusError as e:
    if e.status_code >= 500:
        retry()
    else:
        raise

Erreur 2 : Perdre l’État lors d’une Nouvelle Tentative

La solution : Préservez l’historique des messages entre les tentatives. Ne réessayez que l’étape spécifique échouée.

Erreur 3 : Boucles de Nouvelle Tentative Non Bornées

La solution : Définissez toujours un max_retries. Après épuisement, escaladez ou échouez rapidement.

Tests de Récupération d’Erreurs

import pytest
from unittest.mock import patch, MagicMock

def test_backoff_retries_on_rate_limit():
    """Vérifie que l'agent réessaie jusqu'à max_retries sur les erreurs de limite de débit."""
    call_count = 0

    def mock_create(**kwargs):
        nonlocal call_count
        call_count += 1
        if call_count < 3:
            raise anthropic.RateLimitError("Limite atteinte", response=MagicMock(), body={})
        return MagicMock(content=[MagicMock(text="Succès")])

    with patch.object(client.messages, "create", side_effect=mock_create):
        with patch("time.sleep"):
            result = call_with_backoff(
                [{"role": "user", "content": "test"}],
                max_retries=5,
            )
    assert result.content[0].text == "Succès"
    assert call_count == 3

Liste de Vérification pour la Production

Tous les appels API enveloppés avec recul exponentiel (base 1s, max 60s, gigue activée)
Disjoncteurs sur chaque dépendance externe
Points de contrôle sauvegardés dans un stockage durable après chaque étape significative
Chemins de repli testés et marqués dans les métadonnées de réponse
File d’escalade surveillée et revue au moins quotidiennement
Comptages de tentatives bornés (max_retries ≤ 5)
Classification d’erreurs revue : 4xx ≠ 5xx
Tests pour chaque chemin de récupération avec défaillances injectées

Prochaines Étapes

Commencez par le recul — Enveloppez chaque appel API dans call_with_backoff.
Ajoutez des points de contrôle à tout flux de plus de 2–3 étapes.
Construisez une file d’escalade pour les tâches impliquant des actions irréversibles.
Écrivez des tests d’injection de défaillances — Testez vos chemins de récupération en déclenchant délibérément chaque type d’erreur.

Guides connexes :