Récupération d'erreurs dans les systèmes multi-agents IA

Dans un système à agent unique, l'échec est simple : l'agent échoue, vous relancez. Dans les systèmes multi-agents, l'échec est un problème de graphe. Et c'est là que la plupart des implémentations d'entreprise échouent.

Ce guide vous montre comment concevoir des systèmes multi-agents résilients qui récupèrent automatiquement des pannes.

Le problème des pannes en cascade

Imaginez un workflow classique de traitement de documents :

Agent A (Extraction) → Agent B (Validation) → Agent C (Enrichissement) → Agent D (Rapport)

Si l'Agent B timeout, voici ce qui se passe :

• Agent A : Succès
• Agent B : Timeout (dépend de A)
• Agent C : Ignoré (dépend de B)
• Agent D : Données partielles (dépend de C)

Un seul timeout propage l'échec à travers tout le graphe. C'est le problème des pannes en cascade.

Selon une étude de Google sur leurs systèmes de production, 73% des incidents majeurs impliquent des pannes en cascade. La complexité augmente exponentiellement avec le nombre d'agents. Avec 5 agents ayant chacun 95% de fiabilité individuelle, la fiabilité globale du système tombe à 77%.

Pourquoi les systèmes multi-agents sont particulièrement vulnérables

Les systèmes multi-agents présentent des caractéristiques qui amplifient les risques de panne. Premièrement, les dépendances sont souvent implicites. Un agent peut dépendre d'un autre sans que cette dépendance soit explicitement déclarée dans le code. Deuxièmement, les timeouts se propagent. Un agent lent peut bloquer tous les agents en aval, créant un effet domino. Troisièmement, l'état partagé complique la reprise. Quand plusieurs agents modifient un même état, revenir en arrière devient un défi de coordination.

Ces vulnérabilités ne sont pas insurmontables, mais elles nécessitent une approche architecturale délibérée. Les patterns que nous présentons ci-dessous ont été testés en production sur des systèmes traitant des milliers de requêtes par heure.

Les 5 patterns de récupération d'erreurs

Après avoir déployé des systèmes d'agents IA pour plusieurs entreprises, nous avons identifié 5 patterns essentiels.

Pattern 1 : Retry avec backoff exponentiel

Le plus simple, mais souvent mal implémenté. Le backoff exponentiel évite de surcharger un service déjà en difficulté.

import asyncio
from typing import Callable, TypeVar

T = TypeVar('T')

async def retry_with_backoff(
    func: Callable[[], T],
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> T:
    for attempt in range(max_retries):
        try:
            return await func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            await asyncio.sleep(delay)

Quand l'utiliser : Erreurs transitoires (timeouts réseau, rate limits).

Quand l'éviter : Erreurs de logique métier qui ne se résoudront pas avec le temps.

Pattern 2 : Circuit Breaker

Quand un service échoue répétitivement, le circuit breaker "ouvre" le circuit et retourne immédiatement une erreur. Cela évite de gaspiller des ressources sur un service défaillant.

from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # Normal
    OPEN = "open"          # Bloqué
    HALF_OPEN = "half_open"  # Test

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: timedelta = timedelta(seconds=30)

    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    last_failure: datetime | None = None

    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            if datetime.now() - self.last_failure > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                return True
            return False
        return True  # HALF_OPEN permet un essai

    def record_success(self):
        self.failure_count = 0
        self.state = CircuitState.CLOSED

    def record_failure(self):
        self.failure_count += 1
        self.last_failure = datetime.now()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

Quand l'utiliser : Appels à des services externes (APIs tierces, bases de données).

Conseil : Configurez des seuils différents par criticité de service.

Pattern 3 : Checkpointing et reprise

Pour les workflows longs, sauvegardez l'état à chaque étape. En cas d'échec, reprenez depuis le dernier checkpoint au lieu de tout recommencer.

@dataclass
class WorkflowCheckpoint:
    workflow_id: str
    current_step: str
    state: dict
    completed_steps: list[str]
    timestamp: datetime

class CheckpointedWorkflow:
    def __init__(self, storage: CheckpointStorage):
        self.storage = storage

    async def execute(self, workflow_id: str, steps: list[Step]):
        checkpoint = await self.storage.load(workflow_id)

        start_index = 0
        state = {}

        if checkpoint:
            # Reprendre depuis le dernier checkpoint
            start_index = self.find_step_index(
                steps, checkpoint.current_step
            )
            state = checkpoint.state

        for i, step in enumerate(steps[start_index:], start_index):
            try:
                state = await step.execute(state)
                await self.storage.save(WorkflowCheckpoint(
                    workflow_id=workflow_id,
                    current_step=steps[i+1].name if i+1 < len(steps) else "done",
                    state=state,
                    completed_steps=[s.name for s in steps[:i+1]],
                    timestamp=datetime.now()
                ))
            except Exception as e:
                # L'état est sauvegardé, on peut reprendre plus tard
                raise WorkflowPausedException(workflow_id, step.name, e)

Quand l'utiliser : Workflows de plus de 5 minutes ou impliquant des appels LLM coûteux.

Conseil : Utilisez un stockage persistant (Redis, PostgreSQL) pour survivre aux redémarrages.

Pattern 4 : Dégradation gracieuse

Quand un agent non-critique échoue, continuez avec des données partielles plutôt que de bloquer tout le workflow.

@dataclass
class AgentResult:
    success: bool
    data: dict | None
    error: str | None
    is_critical: bool

async def execute_with_graceful_degradation(
    agents: list[Agent],
    critical_agents: set[str]
) -> dict:
    results = {}

    for agent in agents:
        try:
            result = await agent.execute()
            results[agent.name] = AgentResult(
                success=True,
                data=result,
                error=None,
                is_critical=agent.name in critical_agents
            )
        except Exception as e:
            if agent.name in critical_agents:
                # Agent critique : on arrête tout
                raise CriticalAgentFailure(agent.name, e)

            # Agent non-critique : on continue avec valeur par défaut
            results[agent.name] = AgentResult(
                success=False,
                data=agent.default_value,
                error=str(e),
                is_critical=False
            )

    return results

Exemple concret : Dans un système d'analyse de CV, l'extraction des compétences techniques est critique, mais l'analyse de sentiment des recommandations est optionnelle.

Ce pattern est particulièrement puissant pour les systèmes qui doivent fournir une réponse même en cas de défaillance partielle. L'utilisateur préfère généralement une réponse incomplète à pas de réponse du tout, surtout si le système indique clairement quelles informations manquent.

Pour implémenter la dégradation gracieuse efficacement, documentez pour chaque agent sa criticité et sa valeur par défaut. Les valeurs par défaut doivent être des valeurs neutres qui ne fausseront pas les décisions en aval. Par exemple, une valeur nulle ou une liste vide sont préférables à des valeurs inventées.

Pattern 5 : Orchestration avec compensation

Pour les workflows transactionnels, chaque étape doit avoir une action de compensation qui annule ses effets.

@dataclass
class CompensatableStep:
    name: str
    execute: Callable
    compensate: Callable

async def saga_orchestrator(steps: list[CompensatableStep], context: dict):
    completed = []

    try:
        for step in steps:
            result = await step.execute(context)
            context.update(result)
            completed.append(step)
    except Exception as e:
        # Compensation en ordre inverse
        for step in reversed(completed):
            try:
                await step.compensate(context)
            except Exception as comp_error:
                # Log mais continue la compensation
                logger.error(f"Compensation failed for {step.name}: {comp_error}")
        raise SagaFailedException(e, completed)

    return context

Exemple : Un workflow de création de compte client qui crée un utilisateur, un abonnement, et envoie un email. Si l'email échoue, on doit supprimer l'abonnement et l'utilisateur.

Cas d'étude : système de traitement de factures

Pour illustrer ces patterns en contexte réel, prenons l'exemple d'un système de traitement automatisé de factures que nous avons déployé pour un client du secteur logistique.

Le système comprend 4 agents : extraction OCR, validation des montants, rapprochement avec les bons de commande, et génération des écritures comptables. Avant l'implémentation des patterns de récupération, le taux de succès end-to-end était de 78%. Les échecs provenaient principalement de timeouts OCR sur les factures de mauvaise qualité et d'erreurs de rapprochement quand les références ne correspondaient pas exactement.

Après avoir implémenté les 5 patterns décrits ci-dessus, le taux de succès est passé à 96%. Les 4% restants correspondent à des cas nécessitant une intervention humaine, détectés automatiquement et routés vers une file d'attente de traitement manuel. Le temps moyen de traitement a augmenté de 15% à cause des retries, mais le nombre de factures traitées avec succès a augmenté de 23%.

La clé du succès a été de traiter l'extraction OCR comme critique avec circuit breaker, et le rapprochement comme non-critique avec dégradation gracieuse. Quand le rapprochement automatique échoue, le système suggère des correspondances potentielles plutôt que de bloquer le workflow.

Architecture recommandée

Voici l'architecture que nous recommandons pour les systèmes multi-agents en production :

┌─────────────────────────────────────────────────────────┐
│                    Orchestrateur                         │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐ │
│  │  Circuit    │  │ Checkpoint  │  │   Dead Letter   │ │
│  │  Breakers   │  │   Store     │  │     Queue       │ │
│  └─────────────┘  └─────────────┘  └─────────────────┘ │
└─────────────────────────────────────────────────────────┘
           │                │                │
     ┌─────┴────────────────┴────────────────┴─────┐
     │              Message Bus (Redis/Kafka)       │
     └─────────────────────────────────────────────┘
           │              │              │
    ┌──────┴──────┐ ┌─────┴─────┐ ┌─────┴─────┐
    │   Agent A   │ │  Agent B  │ │  Agent C  │
    │  (Worker)   │ │  (Worker) │ │  (Worker) │
    └─────────────┘ └───────────┘ └───────────┘

Composants clés :

1. Orchestrateur : Gère le flux global et les décisions de récupération
2. Circuit Breakers : Un par service externe
3. Checkpoint Store : Redis ou PostgreSQL pour la persistance
4. Dead Letter Queue : Messages qui ont échoué après tous les retries
5. Message Bus : Communication asynchrone entre agents

Dimensionnement et choix technologiques

Le choix des technologies dépend de votre échelle et de vos contraintes :

Pour les petites équipes (moins de 1000 requêtes par heure) : Redis comme message bus et checkpoint store. Simple à opérer, suffisamment performant pour la plupart des cas d'usage.

Pour les équipes moyennes (1000 à 10000 requêtes par heure) : Redis pour les checkpoints, Kafka ou RabbitMQ pour le message bus. La séparation améliore la résilience.

Pour les grandes équipes (plus de 10000 requêtes par heure) : Infrastructure dédiée avec Kafka pour le messaging, PostgreSQL pour les checkpoints avec réplication, et monitoring distribué avec Prometheus et Grafana.

Quelle que soit l'échelle, commencez simple et évoluez en fonction des besoins réels. Un système sur-ingéniéré est aussi problématique qu'un système sous-dimensionné.

Métriques à surveiller

Pour détecter les problèmes avant qu'ils ne deviennent des pannes en cascade :

| Métrique | Seuil d'alerte | Action | |----------|----------------|--------| | Taux d'erreur par agent | Plus de 5% | Investiguer la cause | | Latence P99 | Plus de 2x la normale | Vérifier les dépendances | | Circuits ouverts | Plus de 1 | Escalade immédiate | | Taille de la DLQ | Plus de 100 messages | Traitement manuel requis | | Retries par minute | Plus de 50 | Possible problème systémique |

Implémentation avec les frameworks populaires

Avec LangGraph

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver

def create_resilient_graph():
    graph = StateGraph(State)

    # Checkpointing intégré
    memory = MemorySaver()

    graph.add_node("extract", extract_with_retry)
    graph.add_node("validate", validate_with_circuit_breaker)
    graph.add_node("enrich", enrich_with_fallback)

    # Edges avec conditions d'erreur
    graph.add_conditional_edges(
        "extract",
        should_continue_or_retry,
        {"continue": "validate", "retry": "extract", "fail": END}
    )

    return graph.compile(checkpointer=memory)

Avec CrewAI

from crewai import Crew, Task, Agent

class ResilientCrew(Crew):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.circuit_breakers = {}
        self.checkpoint_store = CheckpointStore()

    def kickoff(self, inputs: dict):
        checkpoint = self.checkpoint_store.load(inputs.get("workflow_id"))

        if checkpoint:
            # Reprendre depuis le checkpoint
            return self._resume_from_checkpoint(checkpoint, inputs)

        return super().kickoff(inputs)

Tests de résilience : l'injection de chaos

Un système n'est résilient que si sa résilience est testée régulièrement. L'injection de chaos consiste à provoquer délibérément des pannes pour vérifier que le système récupère correctement.

Voici les scénarios de test que nous recommandons :

Test de timeout d'agent : Introduisez un délai artificiel de 30 secondes dans un agent. Vérifiez que le circuit breaker s'ouvre et que les requêtes sont reroutées ou mises en attente.

Test de panne complète : Arrêtez un agent pendant 5 minutes. Vérifiez que les checkpoints permettent la reprise sans perte de données une fois l'agent redémarré.

Test de saturation : Envoyez 10 fois le volume normal de requêtes. Vérifiez que le système dégrade gracieusement sa capacité plutôt que de planter complètement.

Test de données corrompues : Envoyez des données invalides en entrée. Vérifiez que la validation rejette les données sans faire crasher le workflow.

Ces tests doivent être automatisés et exécutés régulièrement, idéalement à chaque déploiement en environnement de staging.

Checklist de mise en production

Avant de déployer votre système multi-agents :

• [ ] Circuit breakers configurés pour chaque API externe
• [ ] Checkpointing activé pour les workflows de plus de 2 minutes
• [ ] Dead Letter Queue configurée avec alertes
• [ ] Métriques de santé exposées (Prometheus/Datadog)
• [ ] Runbook documenté pour chaque type d'erreur
• [ ] Tests de chaos réguliers (injection de pannes)
• [ ] Budgets d'erreur définis par SLA
• [ ] Alertes configurées pour les seuils critiques
• [ ] Documentation des procédures de récupération manuelle

FAQ

Quelle est la différence entre retry et circuit breaker ?

Le retry relance l'opération immédiatement ou après un délai. Le circuit breaker bloque les appels pendant une période quand un seuil d'erreurs est atteint, protégeant le système d'une surcharge. Utilisez les deux ensemble : retries pour les erreurs transitoires, circuit breaker pour éviter d'aggraver une panne.

Comment choisir entre checkpointing et idempotence ?

L'idempotence permet de relancer une opération sans effet de bord. Le checkpointing sauvegarde l'état pour reprendre plus tard. L'idéal est de combiner les deux : opérations idempotentes + checkpoints pour les workflows longs. Si vous devez choisir, privilégiez l'idempotence pour les opérations courtes et le checkpointing pour les workflows de plus de 5 minutes.

Combien de retries configurer ?

La règle générale est 3 retries avec backoff exponentiel (1s, 2s, 4s). Pour les LLMs, augmentez à 5 retries car les timeouts sont fréquents. Pour les paiements ou opérations critiques, limitez à 1-2 retries et préférez l'intervention humaine.

Comment gérer les erreurs de LLM (hallucinations, refus) ?

Ces erreurs ne se résolvent pas avec des retries. Implémentez une validation de sortie et une logique de fallback : si le LLM refuse ou produit une sortie invalide, essayez un prompt alternatif ou un modèle de backup. Documentez les patterns de refus pour améliorer vos prompts.

Quel stockage utiliser pour les checkpoints ?

Redis pour les workflows de moins d'une heure (TTL automatique). PostgreSQL pour les workflows longs ou nécessitant un audit trail. Évitez le stockage en mémoire en production car vous perdez les checkpoints au redémarrage.