Je construis des agents fiables à partir de pièces peu fiables : ma stratégie de développement

Salut tout le monde, Leo ici de agntdev.com ! Aujourd’hui, je veux parler de quelque chose qui me préoccupe beaucoup ces derniers temps, surtout alors que je travaille sur quelques nouveaux projets d’agents. Il s’agit de la partie “Dev” du développement d’agents, spécifiquement, comment nous allons construire des agents fiables à partir de composants non fiables. Oui, vous m’avez bien entendu. Parce qu’en toute honnêteté, c’est la réalité pour la plupart d’entre nous, non ?

Nous ne travaillons généralement pas avec des API et des services parfaitement élaborés et de niveau entreprise. Plus souvent qu’autrement, nous assemblons des modèles open-source, des API tierces avec des limites de taux douteuses, et peut-être même quelques microservices maison qui, disons-le, ont une personnalité. Et pourtant, l’attente est toujours que nos agents devraient simplement… fonctionner. De manière cohérente. Fiable. Même lorsque les composants sous-jacents font des siennes.

J’ai emprunté ce chemin tellement de fois. Je me souviens d’un projet l’année dernière où je construisais un agent pour aider à gérer mes contributions open-source. Il devait interagir avec l’API de GitHub, un modèle d’analyse de sentiments hébergé sur un niveau gratuit, et un service de notification personnalisé que j’avais concocté en un week-end. Chacun de ceux-ci avait ses particularités. GitHub me limitait parfois de manière inattendue, le modèle de sentiments pouvait parfois expirer, et mon service de notification… eh bien, disons juste qu’il avait l’habitude d’oublier ses manières après une heure de fonctionnement. Si je n’avais pas mis en place des protections sérieuses, tout cela se serait effondré comme un château de cartes.

Aujourd’hui, je veux partager quelques stratégies et modèles pratiques que j’ai adoptés pour rendre mes agents plus résilients, même lorsque les éléments dont ils sont constitués sont tout sauf fiables.

La vérité inévitable : les choses vont casser

Tout d’abord, acceptons cela comme un aveu. Aucune API n’est disponible à 100 %. Aucun modèle n’est précis à 100 %. Aucun réseau n’est stable à 100 %. Une fois que vous embrassez cela, vous pouvez commencer à concevoir pour l’échec, ce qui, contre-intuitivement, rend votre agent plus efficace.

Le problème que je vois souvent, surtout avec les développeurs plus récents explorant le travail d’agent, est qu’ils supposent le succès à chaque appel externe. Ils écrivent du code comme ceci :

response = external_api.call_method(data)
# Supposer que la réponse est toujours parfaite et continuer
processed_data = process_response(response)

Et puis, lorsque external_api.call_method lève une erreur de connexion, ou renvoie un 500, ou envoie simplement un JSON malformé, l’agent entier s’arrête. Nous pouvons faire mieux.

Stratégie 1 : tentatives solides avec retour progressif

C’est probablement la technique la plus fondamentale, et pourtant elle est souvent mal mise en œuvre ou pas du tout. Tenter immédiatement après un échec est généralement une mauvaise idée. Si le service externe est en panne, vous ne faites que le frapper davantage, ce qui peut aggraver les choses ou vous vaut d’être soumis à un taux limite.

La clé est le retour progressif. Cela signifie attendre des périodes de plus en plus longues entre les tentatives. Cela donne au service externe une chance de récupérer et réduit la charge que vous lui imposez.

Exemple : Python avec Tenacity

Pour Python, ma bibliothèque de choix pour cela est Tenacity. Elle facilite l’ajout de logique de tentative de manière incroyablement propre.

import random
import logging
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ExternalServiceError(Exception):
 """Exception personnalisée pour les échecs de service externe."""
 pass

# Simuler un appel API externe peu fiable
def call_unreliable_api(data):
 if random.random() < 0.6: # 60 % de chances d'échec
 logger.warning(f"L'appel API a échoué pour les données : {data}")
 raise ExternalServiceError("Échec ou délai d'attente simulé de l'API")
 logger.info(f"L'appel API a réussi pour les données : {data}")
 return {"status": "success", "result": f"processed_{data}"}

@retry(wait=wait_exponential(multiplier=1, min=4, max=10),
 stop=stop_after_attempt(5),
 retry=retry_if_exception_type(ExternalServiceError))
def get_processed_data_with_retries(input_data):
 logger.info(f"Tentative d'appel à l'API pour : {input_data}")
 return call_unreliable_api(input_data)

if __name__ == "__main__":
 try:
 result = get_processed_data_with_retries("some_important_item")
 print(f"Résultat final : {result}")
 except ExternalServiceError as e:
 print(f"Échoué après plusieurs tentatives : {e}")
 except Exception as e:
 print(f"Une erreur inattendue s'est produite : {e}")

Dans cet extrait :

• wait_exponential allonge les délais avec chaque essai (4s, puis ~8s, puis ~16s, etc., jusqu’à 10s max).
• stop_after_attempt(5) signifie qu’il essaiera un maximum de 5 fois.
• retry_if_exception_type(ExternalServiceError) s’assure qu’il ne réessaye que pour des erreurs spécifiques, pas, disons, pour un KeyboardInterrupt.

Ce modèle est un véritable sauveur. Je l’utilise pour les connexions de base de données, les requêtes HTTP, et même pour la communication interne entre les modules d’agents lorsque je sais que l’un d’eux pourrait être temporairement surchargé.

Stratégie 2 : Disjoncteurs pour prévenir les pannes en cascade

Les tentatives sont excellentes pour les erreurs transitoires. Mais que se passe-t-il si le service est complètement en panne ? Essayer de manière répétée épuisera vos ressources et pourrait aggraver le problème pour le service externe s’il a du mal à se rétablir. C’est ici qu’intervient le modèle de disjoncteur.

Pensez à cela comme à un disjoncteur électrique dans votre maison. S’il y a un défaut (trop d’échecs), il « se déclenche », empêchant plus de courant de circuler et protégeant le système. Après un certain temps, il peut être réinitialisé, mais il ne continuera pas à essayer de faire passer du courant par un fil court-circuité.

Pour les agents, un disjoncteur surveille les appels à un service externe. Si le taux d’échecs dépasse un certain seuil dans une fenêtre de temps donnée, le circuit « s’ouvre ». Quand il est ouvert, tous les appels suivants à ce service échouent immédiatement sans même essayer l’appel. Après une période de « délai » configurable, le circuit passe à un état « semi-ouvert », permettant un nombre limité d’appels de test pour voir si le service s’est rétabli. Si ceux-ci réussissent, il se ferme ; s’ils échouent, il s’ouvre à nouveau.

Pourquoi cela importe pour les agents :

• Conservation des ressources : Votre agent ne perd pas de temps et de ressources à essayer d’appeler un service défaillant.
• Échec plus rapide : Au lieu d’attendre un délai, votre agent reçoit un signal d’échec immédiat, lui permettant de gérer la situation (par exemple, utiliser une solution de secours, enregistrer le problème, notifier un opérateur).
• Protège les services externes : Empêche votre agent de DDOS un service en difficulté.

J’implémente généralement cela en utilisant des bibliothèques. Pour Python, Pybreaker est excellent.

import time
import random
import logging
from pybreaker import CircuitBreaker, CircuitBreakerError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ExternalAPIClient:
 def __init__(self):
 # Configurer le disjoncteur :
 # 3 échecs consécutifs en 60 secondes ouvriront le circuit.
 # Il reste ouvert pendant 5 secondes.
 self.breaker = CircuitBreaker(fail_max=3, reset_timeout=5, exclude=[TypeError]) # Ne pas couper sur les TypeErrors

 def _unreliable_call(self, data):
 if random.random() < 0.7: # 70 % de chances d'échec
 logger.warning(f"Simulation d'une erreur interne de l'API pour les données : {data}")
 raise ConnectionError("Service inaccessible")
 logger.info(f"L'appel API a réussi pour les données : {data}")
 return {"result": f"processed_{data}"}

 def process_data(self, data):
 try:
 return self.breaker.call(self._unreliable_call, data)
 except CircuitBreakerError:
 logger.error(f"Le circuit est ouvert ! Pas d'appel API pour les données : {data}")
 # Logique de repli ici : retourner des données mises en cache, une valeur par défaut, ou lever une erreur plus spécifique
 return {"result": "fallback_data", "source": "circuit_breaker"}
 except Exception as e:
 logger.error(f"Erreur lors de l'appel API (non liée au disjoncteur) : {e}")
 raise

if __name__ == "__main__":
 client = ExternalAPIClient()
 for i in range(15):
 print(f"\n--- Tentative {i+1} ---")
 try:
 result = client.process_data(f"item_{i}")
 print(f"Résultat : {result}")
 except Exception as e:
 print(f"Erreur gérée : {e}")
 time.sleep(1) # Simuler un délai entre les appels

Exécutez cela, et vous verrez le circuit s’ouvrir après quelques échecs, puis finalement essayer de se semi-ouvrir, et peut-être même se fermer à nouveau si le service simulé commence à se comporter.

Stratégie 3 : Idempotence pour les opérations modifiant l’état

C’est crucial pour tout agent qui modifie l’état externe (par exemple, créer un enregistrement, envoyer un e-mail, initier un paiement). Si votre agent essaie d’effectuer une action, et que le réseau est instable, ou que le service externe expire, comment savez-vous si l’action a effectivement eu lieu ?

Si vous réessayez simplement sans prendre en compte l’idempotence, vous pourriez accidentellement effectuer l’action deux fois. Imaginez envoyer le même e-mail deux fois, ou pire, facturer un client deux fois. Ce n’est pas bon.

Une opération est idempotente si l’effectuer plusieurs fois a le même effet que l’effectuer une seule fois. Par exemple, définir une valeur (SET x = 5) est idempotent. Incrémenter une valeur (x = x + 1) ne l’est pas.

Comment atteindre l’idempotence :

• Utilisez des identifiants de requête uniques : Lors d’un appel API modifiant l’état, incluez un identifiant unique généré par le client dans l’en-tête de la requête (par exemple, X-Idempotency-Key). Le service externe peut alors utiliser cette clé pour détecter les requêtes en double et renvoyer la réponse originale sans re-traitement.
• Concevez des API idempotentes : Si vous contrôlez l’API, concevez des points de terminaison qui sont naturellement idempotents. Par exemple, au lieu d’un point de terminaison “créer une commande”, ayez un point de terminaison “upsert commande” qui peut créer ou mettre à jour en fonction d’un identifiant de commande unique.
• Vérifiez l’état avant de réessayer : Après une opération de modification de l’état échouée, si l’API le supporte, interrogez l’état de la ressource en utilisant l’identifiant unique avant de tenter un nouvel essai.

Bien que je n’aie pas de morceau de code direct pour cela (il s’agit davantage de conception d’API et de logique côté client), voici à quoi pourrait ressembler le raisonnement de votre agent :

# Pseudo-code de l'agent pour une opération idempotente
transaction_id = generate_unique_id()
payload = {"data": "some_value", "idempotency_key": transaction_id}

try:
 response = external_payment_api.process_charge(payload)
 # Succès ! Stocker transaction_id et response.
except (ConnectionError, TimeoutError, APIError) as e:
 # Oh non, ça a échoué. Le paiement a-t-il tout de même été effectué ?
 logger.warning(f"Le paiement a échoué, vérification de l'état avec l'ID : {transaction_id}")
 try:
 status_response = external_payment_api.get_transaction_status(transaction_id)
 if status_response.get("status") == "completed":
 logger.info(f"Le paiement {transaction_id} a en fait été réussi lors de la vérification de réessai.")
 # Considérer comme réussi, stocker les informations.
 else:
 logger.info(f"Le paiement {transaction_id} a réellement échoué, tentative de réessai (avec la même clé d'idempotence).")
 # C'est ici que vous réessayeriez avec le *même* transaction_id
 # L'API de paiement devrait le reconnaître et ne pas facturer deux fois.
 response = external_payment_api.process_charge(payload)
 # ... gérer le succès/l'échec du réessai
 except Exception as check_e:
 logger.error(f"Impossible de vérifier l'état de la transaction pour {transaction_id} : {check_e}")
 # Besoin de journaliser pour une révision manuelle, ou de passer à la file des lettres mortes

Cela nécessite la coopération du service externe, mais c’est un modèle critique pour construire des agents véritablement fiables qui gèrent des opérations financières ou autres sensibles.

Stratégie 4 : Solutions de secours et dégradations gracieuses

Parfois, un service externe est tout simplement complètement indisponible, et il n’y a aucun espoir de réessayer ou d’attendre. Dans ces cas, un bon agent ne plante pas simplement ; il trouve un moyen de fournir une expérience dégradée mais toujours utile.

Cela pourrait signifier :

• Utiliser des données mises en cache : Si votre agent a besoin de données spécifiques d’un service, mais que le service est en panne, pouvez-vous utiliser une version obsolète d’un cache ?
• Fournir des valeurs par défaut : Si un modèle d’IA pour l’analyse de sentiment est en panne, pouvez-vous simplement classer toutes les entrées comme “neutres” ou “inconnues” pendant une période, plutôt que de faire échouer l’ensemble du flux de l’agent ?
• Passer à un service de secours : Si votre API de traduction principale est en panne, pouvez-vous rediriger les demandes vers une seconde, peut-être moins performante ou plus coûteuse ?
• Sauter des étapes optionnelles : Si une étape d’enrichissement non critique échoue, l’agent peut-il simplement continuer sans cet enrichissement, en journalisant peut-être un avertissement ?
• Notifier les utilisateurs/opérateurs : Au minimum, échouer gracieusement et communiquer clairement le problème à l’utilisateur ou à l’opérateur du système.

Mon anecdote sur l’échec du service de notification ? Ma solution de secours était simple : si mon service de notification personnalisé était en panne, l’agent journalisait simplement l’événement localement et envoyait un e-mail à *moi* disant “Hé, votre service de notification est probablement en panne à nouveau, vérifiez les journaux.” Pas idéal pour les utilisateurs finaux, mais cela a empêché l’ensemble de l’agent de se bloquer et m’a assuré que je savais qu’il y avait un problème.

Mesures concrètes pour votre prochain projet d’agent

1. Supposez l’échec : Concevez votre agent dès le départ en vous attendant à ce que les dépendances externes échouent.
2. Mettez en œuvre des réessais avec un retour d’expérience exponentiel : Utilisez des bibliothèques comme Tenacity (Python) ou des modèles similaires dans d’autres langages pour les erreurs transitoires.
3. Déployez des disjoncteurs : Prévenez les échecs en cascade et préservez les ressources en “déclenchant” le circuit lorsqu’un service échoue de manière constante. Pybreaker est un bon début.
4. Priorisez l’idempotence pour les changements d’état : Assurez-vous que des opérations comme les paiements ou la création d’enregistrements ne se dupliquent pas si un réessai se produit. Utilisez des ID uniques.
5. Planifiez pour une dégradation gracieuse : Identifiez les dépendances critiques par rapport aux dépendances non critiques et construisez des solutions de secours. Quelle est la “moins mauvaise” chose que votre agent peut faire lorsqu’une dépendance tombe en panne ?
6. Surveillez de manière agressive : Toutes ces stratégies génèrent des journaux. Assurez-vous de collecter et d’analyser ces journaux pour comprendre *pourquoi* les choses échouent et à quelle fréquence.

Construire des agents fiables ne concerne pas seulement des algorithmes astucieux ou des modèles puissants. C’est fondamentalement une question d’ingénierie de solidité dans chaque couche, surtout lorsqu’il s’agit de la réalité troublée des dépendances externes. En appliquant ces stratégies, vous passerez moins de temps à déboguer des plantages d’agents mystérieux et plus de temps à créer des systèmes autonomes véritablement utiles et fiables.

Quelles sont vos stratégies préférées pour faire face à des services externes peu fiables ? Laissez un commentaire ci-dessous, j’aimerais entendre vos histoires de guerre et vos solutions !

🕒 Published: March 26, 2026