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 du côté « Dev » du développement d’agents, plus précisément, comment nous allons bâtir des agents fiables à partir de pièces peu fiables. Oui, vous m’avez bien entendu. Parce qu’soyons honnêtes, 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, de niveau entreprise. Plus souvent qu’autrement, nous assemblons des modèles open source, des API de tiers avec des limites de taux discutables, et peut-être même quelques microservices faits maison qui, disons simplement, ont une personnalité. Et pourtant, l’attente est toujours que nos agents devraient juste… fonctionner. De manière cohérente. Fiable. Même lorsque les composants sous-jacents sont en train de faire une crise de nerfs.

Je suis passé par là 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 plan gratuit, et un service de notification personnalisé que j’avais fabriqué en un week-end. Chacun de ces éléments avait ses bizarreries. GitHub me limitait parfois de manière inattendue, le modèle de sentiment se déconnectait de temps en temps, et mon service de notification… eh bien, disons simplement qu’il avait l’habitude d’oublier ses manières après une heure de fonctionnement. Si je n’avais pas mis en place de sérieuses mesures de protection, le tout se serait effondré comme un château de cartes.

Aujourd’hui, je veux partager quelques stratégies pratiques et modèles que j’ai adoptés pour rendre mes agents plus résilients, même lorsque les pièces dont ils sont construits ne le sont pas du tout.

La vérité inévitable : Les choses vont se casser

D’abord, acceptons cela comme un dogme. Aucune API n’est disponible à 100 %. Aucun modèle n’est 100 % précis. Aucun réseau n’est 100 % stable. Une fois que vous acceptez cela, vous pouvez commencer à concevoir pour l’échec, ce qui, de manière contre-intuitive, rend votre agent plus performant.

Le problème que je vois souvent, surtout avec les développeurs plus récents explorant le travail des agents, c’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 procéder
processed_data = process_response(response)

Et puis, quand external_api.call_method lance une erreur de connexion, ou renvoie un 500, ou renvoie simplement un JSON mal formé, tout l’agent s’arrête. Nous pouvons faire mieux.

Stratégie 1 : Reprises solides avec délai progressif

C’est probablement la technique la plus fondamentale, et pourtant elle est souvent mal mise en œuvre ou pas du tout. Simplement réessayer immédiatement après un échec est généralement une mauvaise idée. Si le service externe est en panne, vous le frappez simplement davantage, ce qui pourrait aggraver les choses ou vous rendre limités dans votre taux.

La clé est le délai exponentiel. Cela signifie attendre des périodes de plus en plus longues entre les tentatives. Cela donne au service externe une chance de se rétablir et réduit la charge que vous lui imposez.

Exemple : Python avec Tenacity

Pour Python, ma bibliothèque de prédilection pour cela est Tenacity. Cela rend l’ajout de logique de reprise 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

# Simule un appel API externe peu fiable
def call_unreliable_api(data):
 if random.random() < 0.6: # 60 % de chance d'échec
 logger.warning(f"L'appel API a échoué pour les données : {data}")
 raise ExternalServiceError("Échec ou délai d'attente de l'API simulé")
 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 de 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"Échec après plusieurs tentatives : {e}")
 except Exception as e:
 print(f"Une erreur inattendue est survenue : {e}")

Dans cet extrait :

• wait_exponential rend les attentes plus longues à chaque essai (4s, puis ~8s, puis ~16s, etc., jusqu’à un maximum de 10s).
• stop_after_attempt(5) signifie qu’il tentera un maximum de 5 fois.
• retry_if_exception_type(ExternalServiceError) garantit qu’il ne réessaie que pour des erreurs spécifiques, pas pour, disons, un KeyboardInterrupt.

Ce modèle est un véritable sauveur. Je l’utilise pour les connexions à la 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 éviter les échecs en cascade

Les reprises sont excellentes pour les erreurs transitoires. Mais que faire si le service est complètement en panne ? Réessayer continuellement ne fera qu’épuiser vos ressources et potentiellement aggraver le problème pour le service externe s’il a du mal à se rétablir. C’est là 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 « saute », empêchant plus de courant de passer et protégeant le système. Après un certain temps, il peut être réinitialisé, mais il ne continuera pas à envoyer du courant à travers un fil court-circuité.

Pour les agents, un disjoncteur surveille les appels à un service externe. Si le taux d’échec dépasse un certain seuil dans une fenêtre de temps donnée, le circuit « s’ouvre ». Lorsque c’est ouvert, tous les appels ultérieurs à ce service échouent immédiatement sans même tenter l’appel. Après une période de « délai » configurable, le circuit passe en état « semi-ouvert », permettant un nombre limité d’appels d’essai 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 compte pour les agents :

• Conservation des ressources : Votre agent ne perd pas de temps et de ressources à essayer d’appeler un service mort.
• Échec plus rapide : Au lieu d’attendre un délai d’attente, votre agent reçoit un signal d’échec immédiat, lui permettant de gérer la situation (par exemple, utiliser un secours, enregistrer le problème, notifier un opérateur).
• Protège les services externes : Empêche votre agent de DDOSer 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):
 # Configure 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 ouvrir sur les TypeErrors

 def _unreliable_call(self, data):
 if random.random() < 0.7: # 70 % de chance d'échec
 logger.warning(f"Simulation d'une erreur d'API interne 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 à l'API pour les données : {data}")
 # Logique de secours 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 à l'API (pas 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 refermer à nouveau si le service simulé commence à se comporter.

Stratégie 3 : Idempotence pour les opérations changeant d’é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 de réaliser une action, et que le réseau fléchit, ou que le service externe n’est pas disponible, comment savez-vous si l’action a réellement eu lieu ?

Si vous réessayez simplement sans considérer l’idempotence, vous pourriez accidentellement réaliser l’action deux fois. Imaginez envoyer le même e-mail deux fois, ou pire, facturer un client deux fois. Pas super.

Une opération est idempotente si l’effectuer plusieurs fois a le même effet que de 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 ID de requête uniques : Lors de l’appel d’une API qui modifie l’état, incluez un ID 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 demandes en double et renvoyer la réponse originale sans re-traitement.
• Concevez des APIs 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 ID de commande unique.
• Vérifiez l’état avant de réessayer : Après une opération échouée modifiant l’état, si l’API le supporte, interrogez l’état de la ressource en utilisant l’ID unique avant d’essayer de réessayer.

Bien que je n’aie pas de morceau de code direct pour cela (il s’agit plus de conception d’API et de logique côté client), voici à quoi pourrait ressembler le processus de pensée 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 ! Stockez transaction_id et response.
except (ConnectionError, TimeoutError, APIError) as e:
 # Oh non, cela 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 réussi lors de la vérification de réessai.")
 # Traitez comme un succès, stockez les informations.
 else:
 logger.info(f"Le paiement {transaction_id} a vraiment échoué, tentative de réessai (avec la même clé d'idempotence).")
 # C'est ici que vous réessayez 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érez le succès/échec de la tentative
 except Exception as check_e:
 logger.error(f"Impossible même de vérifier l'état de la transaction pour {transaction_id} : {check_e}")
 # Besoin de consigner pour un examen manuel, ou de passer à une file de lettres mortes

Cela nécessite la coopération du service externe, mais c’est un motif essentiel pour construire des agents vraiment fiables qui gèrent des opérations financières ou d’autres opérations sensibles.

Stratégie 4 : Solutions de repli et dégradation gracieuse

Parfois, un service externe est simplement totalement indisponible, et il n’y a pas d’espoir de réessayer ou d’attendre. Dans ces cas, un bon agent ne se contente pas de planter ; 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 hors service, pouvez-vous utiliser une version périmée d’un cache ?
• Fournir des valeurs par défaut : Si un modèle d’IA pour l’analyse de sentiment est hors service, pouvez-vous simplement classer toutes les entrées comme “neutre” ou “inconnu” pendant un certain temps, plutôt que d’échouer toute la chaîne de l’agent ?
• Passer à un service de secours : Si votre API de traduction principale est hors service, pouvez-vous rediriger les demandes vers une secondaire, 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, peut-être en enregistrant un avertissement ?
• Notifier les utilisateurs/opérateurs : Au minimum, échouez gracieusement et communiquez 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 repli était simple : si mon service de notification personnalisé tombait en panne, l’agent enregistrerait simplement l’événement localement et m’enverrait un email me disant “Salut, votre service de notification est probablement à nouveau en panne, vérifiez les journaux.” Pas idéal pour les utilisateurs finaux, mais cela empêchait l’ensemble de l’agent de se bloquer et me garantissait que je savais qu’il y avait un problème.

Conseils pratiques pour votre prochain projet d’agent

1. Supposer un échec : Concevez votre agent dès le départ en vous attendant à ce que les dépendances externes échouent.
2. Mettre en œuvre des réessais avec un retour exponentiel : Utilisez des bibliothèques comme Tenacity (Python) ou des motifs similaires dans d’autres langages pour les erreurs transitoires.
3. Déployer des disjoncteurs : Prévenez les défaillances en cascade et conservez les ressources en “déclenchant” le circuit lorsqu’un service échoue de manière constante. Pybreaker est un bon début.
4. Prioriser 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. Prévoir une dégradation gracieuse : Identifiez les dépendances critiques et non critiques et construisez des solutions de repli. Quelle est la “moins mauvaise” chose que votre agent peut faire lorsqu’une dépendance tombe en panne ?
6. Surveiller agressivement : 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. Il s’agit fondamentalement d’incorporer de la solidité dans chaque couche, surtout lorsqu’il s’agit de la réalité chaotique des dépendances externes. En appliquant ces stratégies, vous passerez moins de temps à déboguer des plantages mystérieux d’agents et plus de temps à construire des systèmes autonomes réellement utiles et fiables.

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

🕒 Published: March 26, 2026