Fiabilité de LangChain : streaming, agrégation en lots, tentatives répétées, solutions de secours, middleware

Jusqu’à présent, l’agent du service d’assistance disposait d’une liberté délibérément restreinte. Il peut consulter les responsabilités liées aux services et les incidents récents, mais avant de rédiger une note de ticket, il doit s’arrêter, présenter les arguments qu’il compte utiliser et attendre une intervention humaine. Ce système fonctionne tant que chaque appel à un modèle ou outil est traité rapidement. En pratique, ce n’est pas le cas.

Une demande à un modèle peut prendre quinze secondes. Un fournisseur peut bloquer l’accès pendant une minute entière en raison d’une limitation de fréquence. Un seul ticket dans un lot peut contenir des données incorrectes. Un modèle de secours pourrait répondre différemment du modèle principal, et un utilisateur pourrait fermer son navigateur alors qu’un outil en cours d’exécution n’a pas encore terminé. Chacun de ces cas mérite un comportement spécifié accompagné d’un test correspondant.

Cette étape ajoute cinq contraintes supplémentaires pour l’agent. Pour une seule demande :

one request
  -> stream honest progress
  -> retry a small transient failure
  -> use a labeled fallback if attempts are exhausted
  -> stop at time and call limits

Pour une série de demandes non liées entre elles :

many independent requests
  -> run a bounded batch
  -> keep one result or error beside each input

Règle générale : ne cachez pas une erreur derrière une réponse enjouée.

Le ticket que nous continuons de tester

L’incident de test reste inchangé : les demandes de paiement en UE ne fonctionnent plus après le déploiement à 14:05, avec l’erreur « database connection refused » dans les journaux de checkout-api. L’utilisateur demande à l’agent de rechercher le responsable et les incidents récents, de suggérer une prochaine étape et d’ajouter une note à INC-2048 une fois son accord obtenu.

Dans le scénario idéal, l’agent consulte le responsable et les incidents récents, rédige une recommandation, propose une note, attend l’approbation, écrit une fois et renvoie le résultat. Pour rendre les choses plus difficiles : soumettez trois autres tickets en même temps, faites échouer deux fois une appel à un fournisseur, rendez un ticket invalide et interrompez la connexion au client au milieu de la première demande. Tout ce qui suit a été conçu pour faire face précisément à cela.

Donnez d’abord des noms aux erreurs

« L’appel a échoué » ne vous indique pas quoi faire. Classez les échecs en catégories qui conduisent à des actions différentes.

• Échec temporaire du fournisseur. Une connexion à réinitialiser, un délai d’attente dépassé, ou une réponse du fournisseur vous demandant explicitement de revenir plus tard. Essayez à plusieurs reprises en faisant une pause entre chaque tentative.
• Limite de fréquence. Le fournisseur protège une quota ou sa capacité maximale. Suivez les indications de tentative de réessai qu’il envoie, ralentissez les appels et diminuez la concurrence. Tenter immédiatement à nouveau ne fera qu’empirer les choses.
• Données invalides. Le ticket ne possède pas d’ID, le schéma ne correspond pas, ou le texte dépasse la limite imposée par l’application. Répondez en indiquant clairement une erreur côté client, et n’envoyez pas à nouveau la même demande mal formatée.
• Échec d’autorisation. L’utilisateur n’a pas le droit de lire un guide opérationnel ou d’écrire dans le ticket. Arrêtez-vous là ; ne changez pas de identifiants et ne demandez à aucun autre modèle de contourner ce refus.

Erreur liée à l’outil. Le ticket est déjà fermé, la transition n’est pas autorisée, ou une note identique existe. Transformez la réponse de l’outil en une erreur d’application significative, car tenter à nouveau pourrait s’avérer erroné.

Echec de la validation des résultats du modèle. Le fournisseur a renvoyé des données qui violent le contrat de sortie. Pour un résultat uniquement en lecture, une tentative de correction peut être acceptable, mais il doit toujours exister un chemin d’échec final.

Echec inconnu. L’application n’a pas pu classifier l’erreur. Clôture de l’échec, enregistrement du type d’erreur et enquête ; ne traitez jamais chaque exception comme temporaire.

Ces catégories déterminent ensemble quels échecs seront réessayés, lesquels feront l’objet d’un recours et lesquels seront transmis directement à l’appelant.

Structure et configuration du projet

Le snapshot de cette étape contient :

langchain-helpdesk/
├── app.py
├── reliability.py
├── batch_runner.py
├── stream_view.py
├── middleware_config.py
└── tests/
    ├── test_reliability.py
    └── test_stream_view.py

Installez les paquets :

python -m pip install -U \
  langchain \
  langchain-core \
  langgraph \
  pytest

Les tests n’appellent jamais de fournisseur.

Le streaming fournit du contenu ; il ne le vérifie pas

Un écran vide pendant que les outils s’exécutent et que le modèle travaille constitue une mauvaise expérience. Le streaming résout ce problème en émettant des événements au fur et à mesure que le travail progresse. Le streaming d’agent actuel de LangChain vous offre trois vues :

• updates : étapes de l’agent terminées
• messages : tokens du modèle ainsi que des métadonnées
• custom : données de progression définies par votre application

Ces éléments contiennent différents types d’informations, il ne faut donc pas les afficher tous comme s’il s’agissait d’une seule réponse.

Une boucle de flux de base

for chunk in agent.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": request_text,
            }
        ]
    },
    config,
    stream_mode=["updates", "custom"],
    version="v2",
):
    send_to_client(chunk)

Décortiquons l’appel : agent.stream(...) lance la même exécution que invoke(...), mais génère des fragments au fur et à mesure de son exécution. Le premier argument correspond à la mise à jour d’état, qui est ici un message d’utilisateur. config contient des paramètres spécifiques à la requête, tels que l’ID de thread, et les secrets ne doivent pas y figurer. stream_mode=["updates", "custom"] demande des données relatives aux étapes terminées ainsi que vos propres messages de progression, de sorte qu’aucun texte token par token n’est retourné dans cet exemple. version="v2" sélectionne le format actuel des fragments, comprenant un type, un espace de noms et un chargement de données dans chaque fragment. send_to_client(chunk) fait référence à votre propre code, qui peut émettre des Server-Sent Events, un message WebSocket ou une ligne de terminal.

Traduire les fragments bruts pour l’interface utilisateur

Le snapshot convertit les fragments avant qu’ils n’atteignent l’interface utilisateur :

@dataclass(frozen=True)
class ProgressEvent:
    kind: str
    label: str
    payload: Any

def readable_events(chunks):
    for chunk in chunks:
        chunk_type = chunk.get("type")
        data = chunk.get("data")        if chunk_type == "updates" and isinstance(data, dict):
            for step_name, step_data in data.items():
                yield ProgressEvent(
                    kind="step",
                    label=step_name,
                    payload=step_data,
                )

Cet adaptateur fournit à l’interface utilisateur un contrat petit et stable. Il peut afficher des lignes comme celles-ci :

Checking service ownership...
Reading recent incidents...
Waiting for ticket-note approval...
Ticket note approved and written.

Ce qu’il ne doit absolument pas afficher avant que l’exécution ne soit vraiment terminée, c’est ceci :

Done.

Le texte partiel n’est pas le texte final

Les jetons de streaming donnent l’impression que la réponse est plus rapide, mais ils n’améliorent en rien la sécurité ou la précision. Le modèle peut commencer une phrase avant de s’arrêter ailleurs, un outil peut tomber en panne après que les jetons ont déjà été envoyés, et une vérification de politique peut empêcher l’exécution de l’action proposée.

Utilisez des états d’événement explicites : running, waiting_for_approval, completed, failed et cancelled. Seul l’état completed indique un résultat final.

Décidez de ce que signifie une interruption de connexion

Le fait que le navigateur soit fermé ne prouve pas pour autant que le serveur ait cessé de fonctionner. Définissez la politique à l’avance :

• Annulez les tâches en lecture seule lorsqu’une annulation est possible et que personne n’attend la réponse.
• Laissez une tâche durable approuvée se terminer si l’utilisateur compte y revenir.
• N’assumez jamais qu’une écriture n’a pas eu lieu simplement parce que sa réponse a été perdue.

Vérifiez l’opération à l’aide de la clé d’idempotence avant de tenter à nouveau l’écriture.

Le streaming et l’annulation sont liés, mais ce ne sont pas une seule et même fonctionnalité.

Grouper des entrées indépendantes

Les exécutables LangChain exposent batch et abatch. Par défaut, batch exécute des appels indépendants dans un pool de threads, ce qui convient à de nombreuses opérations liées au réseau ; un exécutable spécifique au fournisseur peut remplacer cela par une API de batch native. Le travailleur du service client accepte plusieurs tickets non liés entre eux :

outputs = runnable.batch(
    tickets,
    config={"max_concurrency": 4},
    return_exceptions=True,
)

tickets représente la liste ordonnée des entrées. max_concurrency limite le nombre d’appels qui peuvent s’exécuter en parallèle, et la valeur quatre n’est qu’un exemple ; déterminez ce chiffre en fonction des quotas du fournisseur, de la capacité de l’outil, des coûts et de la latence mesurée. Avec return_exceptions=True, l’échec d’un élément n’annule plus les résultats des autres. Par la suite, l’application associe chaque sortie à son entrée correspondante :

for ticket, output in zip(tickets, outputs, strict=True):
    if isinstance(output, BaseException):
        save_failure(ticket.ticket_id, type(output).__name__)
    else:
        save_recommendation(ticket.ticket_id, output)

Cela garantit un seul résultat par ticket.

Omettez le traitement en lot parallèle dans les cas suivants : l’élément deux dépend de l’élément un ; chaque élément met à jour le même enregistrement sans contrôle de concurrence ; l’ordre d’exécution constitue lui-même une règle métier ; le fournisseur utilise des transactions en masse avec des garanties différentes ; ou un utilisateur peut soumettre suffisamment d’éléments pour épuiser la capacité partagée. Le traitement en lot redistribue simplement la charge sans la faire disparaître.

Réessayer avec la limite la plus sûre

Réessayer signifie effectuer à nouveau le travail, il faut donc être prudent quant à ce que l’on répète. Relancer un flux de travail complet peut entraîner une nouvelle récupération des données, de nouvelles appels au modèle et une nouvelle écriture dans le ticket. Une approche plus sûre consiste à ne réessayer que l’étape de lecture connue pour échouer temporairement. Le snapshot définit une erreur spécifique à cet effet :

class TemporaryProviderError(RuntimeError):
    """A short-lived error that may succeed when tried again."""

et enrobe un seul élément exécutable :

primary_call = RunnableLambda(primary).with_retry(
    retry_if_exception_type=(TemporaryProviderError,),
    wait_exponential_jitter=False,
    stop_after_attempt=3,
)

RunnableLambda(primary) transforme une fonction Python ordinaire en élément exécutable. retry_if_exception_type fonctionne comme une liste d’autorisation : un ticket invalide ne correspond pas et est renvoyé directement à l’appelant. En définissant wait_exponential_jitter=False, on assure que le test hors ligne reste rapide et déterministe. stop_after_attempt=3 compte chaque tentative, ce qui signifie trois essais au total et non trois reprises supplémentaires par rapport à la première. Face à un service réseau réel, le recul exponentiel avec jitter est généralement le meilleur choix, car il empêche tous les travailleurs d’accéder au service en même temps.

Les écritures ne peuvent pas être reessayées aveuglément

Imaginez que le service de tickets ajoute la note, puis que la connexion se coupe avant l’arrivée de la réponse de succès. L’appelant voit un délai d’attente, mais l’écriture a déjà pu avoir lieu, et une tentative de réessai naïve ajouterait la note deux fois. Utilisez une clé d’idempotence :

operation_id = "ticket-note:INC-2048:approval-7331"

Le service de tickets enregistre cette clé avec la première écriture réussie. Si la même clé réapparaît, il renvoie le résultat stocké et n’applique pas le changement une seconde fois. Lorsque le système externe ne peut pas assurer l’idempotence, vérifiez son état actuel avant de réessayer, et acceptez que certaines fenêtres d’échec ne puissent pas être complètement résolues.

Une solution de secours consiste en un chemin différent

Lorsque les tentatives autorisées sont épuisées, un programme de secours peut prendre le relais :

reliable = primary_call.with_fallbacks(
    [RunnableLambda(conservative_fallback)],
    exceptions_to_handle=(TemporaryProviderError,),
)

Il ne s’exécute que pour l’erreur temporaire spécifiée, et ce qu’il renvoie est ceci :

Recommendation(
    ticket_id=ticket.ticket_id,
    text=(
        "The recommendation service is temporarily unavailable. "
        "Keep the ticket open and compare the failed deployment "
        "with the last known-good release."
    ),
    source="fallback",
)

La réponse admet ce qui s’est passé. Elle ne prétend pas que des manuels de procédure ont été consultés ni que le ticket a été mis à jour.

Passer à un autre modèle n’est pas identique à redémarrer le premier. Le deuxième modèle peut se comporter différemment lors des appels aux outils, avoir une limite de contexte plus faible, gérer les sorties structurées différemment, appliquer d’autres comportements de sécurité, présenter des différences en termes de latence et de coût, ou être soumis à une politique de données différente du fournisseur. Traitez la route de secours comme une configuration à part entière et testez-la séparément. Le meilleur scénario de secours est parfois d’absolument pas utiliser de modèle, mais plutôt d’afficher un message fixe et prudent ou de placer le contenu en file d’attente pour une révision humaine.

Les délais de timeout existent à plusieurs niveaux

Une seule valeur de délai d’expiration n’est pas suffisante. Allouez des budgets distincts pour l’établissement de la connexion, une demande de modèle unique, une demande d’outil unique, l’exécution complète de l’agent, l’attente d’approbation et la connexion au client. Le budget pour l’exécution complète doit dépasser celui de chaque appel individuel, mais il doit néanmoins être limité.

Placez le comportement de délai d’expiration à côté de l’opération qu’il limite : les délais d’expiration du fournisseur sur le client du fournisseur, les délais d’expiration de l’outil autour du client de l’outil, ainsi que la date limite pour la demande web au niveau de la frontière du service. Lorsqu’un délai d’expiration est atteint, notez quel niveau a mis fin au traitement. Un simple message « délai d’expiration atteint » sans indication du niveau concerné est difficile à résoudre.

Le middleware enveloppe les appels de l’agent

Les étapes précédentes utilisaient principalement des chaînes fixes, tandis que l’étape d’approbation a introduit un middleware d’agent. Le middleware constitue également un excellent endroit pour les règles qui doivent couvrir de nombreuses appels à des modèles ou outils : réessayer les échecs liés aux modèles sélectionnés, limiter le nombre d’appels aux modèles, limiter le nombre d’appels aux outils, filtrer les outils en fonction de l’utilisateur actuel, ajouter des métadonnées aux requêtes pour les journaux et les traces, masquer certains données personnelles et interrompre le fonctionnement lorsque le budget de l’application est épuisé. Le snapshot fait usage du middleware intégré :

def reliability_middleware():
    return [
        ModelRetryMiddleware(
            max_retries=2,
            retry_on=(
                TemporaryProviderError,
                TimeoutError,
                ConnectionError,
            ),
            on_failure="error",
            initial_delay=0.5,
            backoff_factor=2.0,
            max_delay=4.0,
            jitter=True,
        ),
        ModelCallLimitMiddleware(
            run_limit=8,
            exit_behavior="error",
        ),
        ToolCallLimitMiddleware(
            run_limit=6,
            exit_behavior="end",
        ),
    ]

ModelRetryMiddleware entoure les appels aux modèles, et un appel initial ainsi que deux tentatives de réessai donnent au maximum trois essais. ModelCallLimitMiddleware empêche une boucle de modèles de fonctionner indéfiniment. ToolCallLimitMiddleware arrête une utilisation excessive d’outils au cours d’une exécution. Ces mécanismes de contrôle ont des fonctions distinctes : une limite d’appels ne remplace pas un délai de timeout, un délai de timeout ne remplace pas l’approbation, et l’approbation ne remplace pas l’autorisation.

L’ordre est important

Les hooks qui s’exécutent avant un appel sont ordonnés dans l’ordre de la liste, ceux qui s’exécutent après le sont dans l’ordre inverse, et les hooks d’enveloppement s’empilent de la même manière que les appels de fonction. Lorsque le middleware A enveloppe B, une tentative de réessai dans A peut refaire le travail effectué par B ; il convient donc d’établir l’ordre chaque fois que deux composants pourraient tous deux tenter un réessai, modifier des modèles ou changer l’état. L’approbation d’une écriture doit se faire en dehors de tout réessai susceptible de reproduire l’effet secondaire déjà approuvé.

Les hooks personnalisés ne sont que du code ordinaire

LangChain propose des hooks de type nœud (before_model, after_model) et des hooks de type enveloppement (wrap_model_call, wrap_tool_call). Les hooks de type nœud conviennent aux vérifications et mises à jour d’état en points fixes. Les hooks de type enveloppement servent aux codes qui doivent entourer une seule appel, ou qui déterminent si le gestionnaire s’exécute zéro, une ou plusieurs fois. Cependant, toutes les règles d’une application ne devraient pas devenir du middleware. Les transitions d’état des tickets, les décisions d’autorisation et les règles de propriété des données sont plus faciles à gérer en tant que services d’application nommés.

Test des chemins de défaillance hors ligne

Le snapshot comprend cinq tests, aucun n’exigeant de fournisseur.

Une erreur temporaire réussit à la deuxième tentative :

result = runnable.invoke(ticket())

assert calls == 2
assert result.source == "primary"

Lorsque les tentatives échouent, on utilise la solution de repli indiquée :

assert calls == 3
assert result.source == "fallback"
assert "temporarily unavailable" in result.text

Les entrées invalides ne sont ni réessayées ni masquées :

with pytest.raises(InvalidTicketError):
    runnable.invoke(ticket())

assert calls == 1

Une échec de lot ne remet pas en cause un autre succès :

assert results[0].status == "completed"
assert results[1].status == "failed"
assert results[1].error_type == "RuntimeError"

La progression reste distincte des tokens du modèle :

assert [(event.kind, event.label) for event in events] == [
    ("step", "model"),
    ("progress", "custom"),
    ("token", "model"),
]

Exécutez-les :

pytest -q

Résultat attendu :

5 passed

Ces tests valident vos règles de contrôle locales. Ils ne prouvent pas que les erreurs d’un fournisseur sont correctement mappées, ce qui nécessite un petit test d’intégration avec le SDK réel du fournisseur ainsi qu’un compte de test contrôlé.

Rédigez la matrice des échecs

Insérez un tableau comme celui-ci dans le code ou la documentation avant la publication :

failure                         retry?   fallback?   client result
connection reset               yes      maybe       503 if exhausted
provider rate limit            yes      maybe       503 or queued
invalid ticket                 no       no          422
authorization denied           no       no          403
structured output invalid      limited  maybe       502 if exhausted
ticket already closed          no       no          409
unknown exception              no       no          500 + incident ID

Les codes d’état exacts dépendent de votre API. Ce qui importe, c’est que chaque catégorie d’échec soit soumise à une politique définie.

Toujours manquant

Le comportement en temps de exécution est désormais plus clair, mais il s’agit encore d’un aperçu pédagogique en ligne de commande. Il ne dispose pas d’API HTTP authentifiée, d’actions de vérification au démarrage ou de disponibilité, de chargement de secrets, de journaux de requêtes structurés, d’un stockage durable des tâches, de configurations de déploiement, d’essais d’intégration avec des fournisseurs réels, d’aucun ensemble d’évaluation vérifié avant la publication, ni d’analyse des traces en production. Une version du service incluant ces fonctionnalités suivra.

État actuel du flux de support

Le flux présente désormais des états visibles et une capacité de récupération limitée :

request
  -> progress events
  -> small retry boundary
  -> labeled fallback when allowed
  -> model and tool call limits
  -> approval before write
  -> final result or named failure

batch
  -> bounded concurrency
  -> one outcome per ticket

Aucun contrôle ne permet d’éliminer l’incertitude. L’objectif est de la maintenir visible, maîtrisée et testable. Viennent ensuite l’ensemble du flux de travail du service d’assistance derrière une API, incluant la configuration, les secrets, les identifiants de demande, les délais d’attente, les journaux et les mécanismes de suivi, suivis de tests déterministes, de cas d’évaluation et de vérifications de déploiement, ainsi que d’une liste de ce qu’une entreprise réelle doit fournir avant que le service ne traite des tickets en production. Tout ce qui précède a été vérifié par rapport à la documentation Python de LangChain concernant le streaming, les tentatives répétées exécutables, les solutions de secours, le traitement par lots, les middleware et les tests, le 31 août 2026 ; les API de paquets et les types d’erreurs des fournisseurs peuvent changer.