Flux d’agents avancés LangChain : plans en temps réel, appels d’outils et tokens.
Remplacer une invocation bloquante par un flux afin que les utilisateurs puissent voir le fonctionnement d’un agent avancé : les trois modes de flux, les noms d’espace pour les sous-agents, ainsi que le filtrage des tokens de synthèse.
Sources: LangChain Deep Agents: Your AI Agent Is Not Frozen. It Just Cannot Tell You It Is Working. II
Peu de choses érodent la confiance plus rapidement qu’un écran figé. Lorsqu’un utilisateur ne sait pas si un agent est en train de réfléchir, bloqué ou déjà en panne, il en vient au pire scénario. La diffusion en continu remplace ce silence par un flux en temps réel : le plan, les appels aux outils et les mots apparaissent au fur et à mesure, permettant aux gens de voir qu’il se passe quelque chose.
Pourquoi invoke fait croire qu’un agent avancé est en panne
Essayez de faire fonctionner un agent avancé sur une tâche réelle sous surveillance. L’appel passe par invoke, puis l’affichage reste immobile pendant quarante secondes. En réalité, l’agent lit des fichiers, élabore un plan, exécute des outils et raisonne. L’utilisateur ne voit qu’une boîte immobile et un indicateur de chargement, ce qui pourrait signifier un travail intense ou une panne silencieuse. Comme les deux situations ont l’air identiques, les gens rechargent la page et soumettent un ticket indiquant que l’application est bloquée.
Rien n’a été affiché. L’agent ne disposait tout simplement d’aucun canal pour se présenter. Les blocs invoke sont conçus de cette manière : ils exécutent toute la boucle en arrière-plan et renvoient une seule réponse finale, ce qui convient parfaitement à un script ou à une tâche cron que personne ne surveille. C’est cependant le mauvais choix lorsqu’un humain est à l’autre bout.
La solution proposée laisse la boucle intacte. Elle remplace l’appel bloquant par un flux, de sorte que les plans, les appels aux outils et les tokens apparaissent en temps réel.
De invoke à stream
Le streaming est étroitement lié à invoke. Au lieu d’attendre un état final, on itère sur agent.stream(...) et un morceau de données arrive chaque fois que la boucle effectue une action. Une version asynchrone, agent.astream(...), a la même structure et c’est celle que serait appelée un serveur web.
Le modèle mental est simplifié. Le LangGraph durable qui sous-tend un agent avancé émet un événement à chaque étape, et le flux transmet ces événements au fur et à mesure qu’ils sont générés. Il ne s’agit pas d’un agent différent ni d’une boucle différente, mais simplement d’une vue sur celui que vous connaissez déjà.
Un détail doit d’abord être clarifié, car tous les exemples s’y réfèrent : le format de flux v2, activé avec version="v2". Chaque morceau est un petit dictionnaire contenant trois clés :
typeindique le type d’événement.ns, le nom d’espace, indique qui l’a généré.datacontient la charge utile.
Ces trois clés restent identiques quel que soit le flux transmis, ce qui permet au code de rester lisible plutôt que d’être un enchevêtrement de déballages de tuples.
Trois modes, trois niveaux
Un flux n’est pas un simple tuyau d’incendie. La valeur stream_mode définit le type d’événements que vous recevez, et les trois choix courants correspondent à trois niveaux de détail.
updates correspond à une vue par étapes. Un événement est généré chaque fois qu’un nœud du graphe termine son traitement, ce qui, en pratique, signifie que l’agent a effectué une action ou que l’outil a fonctionné. C’est le niveau idéal pour un journal de progression, suffisant pour montrer comment l’agent avance dans sa tâche sans encombrer l’affichage de trop de données, et c’est également la valeur par défaut appropriée pour un flux clair indiquant « voici ce que fait l’agent ».
messages représente la vue par tokens. La sortie du modèle arrive par morceaux au fur et à mesure de sa génération, exactement comme on le voit lorsque l’assistant de chat tape. Utilisez-la lorsque quelqu’un lit le texte généré par l’agent au fur et à mesure de sa création. Les appels des outils en flux continu arrivent également ici, ce qui vous permet d’annoncer « appel à read_file » dès que le modèle décide d’effectuer l’appel, et non après son achèvement.
values représente la vue par snapshot. Après chaque étape, vous recevez l’état complet de l’agent, et non seulement ce qui a changé. Cette méthode est plus lourde et doit être utilisée moins souvent, mais elle est idéale lorsque l’ensemble de la situation est important à chaque instant, par exemple pour afficher en temps réel la liste des tâches ou les fichiers en cours d’utilisation par l’agent.
En passant une liste de modes, plusieurs sont retournés en même temps, c’est ainsi que fonctionnent généralement les interfaces utilisateur soignées : updates gère l’indicateur d’avancement tandis que messages alimente le texte visible. L’exemple le plus simple correspond à un journal d’avancement clair généré par updates:
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Summarize the repo’s failing tests”}]},
stream_mode=”updates”,
version=”v2”,
):
if chunk[”type”] == “updates”:
for node_name in chunk[”data”]:
print(f”step: {node_name}”)
Ce qui s’affiche est une liste des étapes du graphe au fur et à mesure qu’elles sont terminées, une ligne pour chaque tour du modèle et chaque exécution d’outil. Une interface qui semblait figée avance désormais visiblement. En revanche, en définissant stream_mode sur "messages", on obtient le texte lui-même, prêt à être imprimé pendant que l’agent écrit.
Distinguer l’agent principal de ses sous-agents
Un agent avancé travaille rarement seul pendant longtemps. Dès qu’il confie une sous-tâche à un sous-agent, deux ou plusieurs agents génèrent des résultats en même temps. Si l’on envoie tout dans un seul flux, le résultat est un enchevêtrement : une partie d’une phrase du coordinateur, quelques tokens d’un chercheur, une appel d’outil provenant on ne sait où. Le champ ns résout ce problème. Chaque morceau de texte contient ce champ lorsque l’on active le streaming de sous-graphe avec subgraphs=True.
La règle est facile à appliquer. Si l’espace de noms est vide, c’est l’agent principal qui a émis l’événement. S’il contient un segment tools:, c’est un sous-agent qui l’a fait, et ce segment indique lequel. Les résultats peuvent ensuite être triés par origine : le texte du coordinateur ici, et le travail de chaque sous-agent dans sa propre section.
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “Research and summarize quantum computing”}]},
stream_mode=”messages”,
subgraphs=True, # ①
version=”v2”,
):
if chunk[”type”] == “messages”: # ②
token, metadata = chunk[”data”] # ③
is_subagent = any(s.startswith(”tools:”) for s in chunk[”ns”]) # ④
source = “subagent” if is_subagent else “main” # ⑤
if token.content:
print(f”[{source}] {token.content}”, end=”“, flush=True) # ⑥
Six détails sont importants dans ce fragment. Premièrement, subgraphs=True est ce qui confère un sens à ns, car sans cela les événements de subagent ne disposent d’aucun espace de noms distinctif. Deuxièmement, la branche ne conserve que les événements messages, car cette boucle gère les tokens et non les étapes. Troisièmement, la charge utile de chaque morceau se décompose en un token ainsi que ses métadonnées. Quatrièmement, c’est le test d’espace de noms qui effectue le travail principal : un segment tools: n’importe où dans ns indique que le token provient d’un subagent. Cinquièmement, la valeur booléenne résultante devient une étiquette source lisible. Sixièmement, l’affichage avec cette étiquette en préfixe transforme un flux mélangé en deux flux attribués.
Chaque token est étiqueté avec son émetteur avant d’être affiché, de sorte qu’un coordinateur qui transmet les données à un chercheur apparaît sous forme de deux flux nommés. Une interface web complète représente une tâche plus complexe, mais l’espace de noms de chaque morceau de données est le même signal que l’interface React lirait.
Les tokens isolés qui ne proviennent pas de l’agent
Ce phénomène surprend les utilisateurs lors de leur première expérience de streaming prolongé. Afin de gérer son propre contexte, le système condense les messages anciens lorsque la conversation approche de la limite du modèle. L’étape de condensation constitue également une requête au modèle, ce qui génère des tokens qui se retrouvent dans le flux messages à côté de la véritable réponse de l’agent. Si ce flux est envoyé directement à l’utilisateur, celui-ci verra l’agent réciter soudainement un résumé de son propre historique, ce qui peut sembler être une erreur.
Heureusement, les tokens sont étiquetés. Chaque événement messages possède des métadonnées, et les tokens de résumé portent une valeur lc_source égale à "summarization". Ignorez-les et conservez le reste.
for chunk in agent.stream(
{”messages”: [{”role”: “user”, “content”: “...”}]},
stream_mode=”messages”,
version=”v2”,
):
token, metadata = chunk[”data”]
if metadata.get(”lc_source”) == “summarization”:
continue
# ... handle real agent tokens here
Un simple contrôle des métadonnées suffit pour distinguer l’agent qui s’adresse à l’utilisateur de l’agent qui communique avec lui-même. Seul le premier est affiché. Il s’agit d’une tâche mineure qui apporte cependant des améliorations notables en termes de qualité.
Observer une exécution par rapport à maintenir une conversation
Le streaming et les conversations sont souvent confondus, il convient donc d’établir une distinction claire. Tout ce qui a été mentionné concerne l’observation d’une exécution en temps réel. Chaque appel stream, tout comme chaque appel invoke, est par défaut unique. L’agent effectue la tâche, la termine et l’oublie ; ainsi, toute suite dans un nouvel appel débute sans aucun souvenir de l’échange précédent.
Pour maintenir une conversation ouverte au fil des tours, de manière à ce que l’instruction « corrigez aussi l’autre test » fonctionne car l’agent se souvient de la première correction, il existe un mécanisme distinct. Celui-ci repose sur des threads et un checkpointer, la couche de persistance où le graphe durable enregistre son état. Il convient de distinguer : le streaming assure la visibilité au sein d’une exécution, tandis que les threads assurent la continuité entre les exécutions. Ils sont indépendants, et vous aurez souvent besoin des deux.
Une petite interface de terminal affichant l’avancement du travail
Ensuite, assemblez tout cela en une console que vous pourrez réellement exécuter. Elle doit afficher une ligne par étape à mesure que l’agent progresse, ainsi que le texte généré par l’agent au fur et à mesure de sa rédaction ; elle doit donc demander ces deux modes en même temps et prendre une décision en fonction de la valeur de type du chunk.
from deepagents import create_deep_agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
system_prompt="You are a code-maintenance assistant for a small Python repo.",
) # ①
prompt = "List the files you'd look at first to find a failing test, and explain why."
for chunk in agent.stream(
{"messages": [{"role": "user", "content": prompt}]},
stream_mode=["updates", "messages"], # ②
version="v2",
):
if chunk["type"] == "updates": # ③
for node_name in chunk["data"]:
print(f"\n[step] {node_name}")
elif chunk["type"] == "messages": # ④
token, metadata = chunk["data"]
if metadata.get("lc_source") == "summarization": # ⑤
continue
if token.content:
print(token.content, end="", flush=True) # ⑥
Le code est court, mais chaque partie a une fonction précise. La structure reste inchangée, car create_deep_agent crée l’agent exactement comme avant. Deux modes demandés en une seule appelation signifient qu’une seule boucle peut alimenter à la fois un indicateur d’étape et un flux de texte. Lorsqu’un morceau représente un événement updates, la boucle affiche une ligne [step] pour le nœud qui vient de s’achever. Lorsqu’il s’agit d’un événement messages, la boucle affiche le texte généré. La vérification de synthèse de la section précédente se trouve à l’intérieur de cette branche, de sorte que les tâches d’entretien du système n’atteignent jamais l’affichage final, et les tokens vides sont ignorés avant d’être affichés.
Le résultat est une ligne [step] chaque fois que l’agent avance, entremêlée de son texte qui arrive un caractère à la fois, avec le filtre déjà en place pour éviter toute fuite d’informations. Ce n’est pas très esthétique, mais c’est honnête. Un utilisateur peut voir que l’agent est actif et le regarder réfléchir.
Les nouvelles versions de Deep Agents proposent une API de niveau supérieur à cet effet, agent.stream_events(...) sur version="v3", avec des projections typées comme stream.subagents. La documentation oriente les nouvelles applications vers cette approche, car chaque projection constitue un itérateur propre et il n’est pas nécessaire de gérer soi-même le mode de flux stream_mode ou de décoder les noms d’espace. Il reste néanmoins utile d’apprendre d’abord la méthode stream_mode, puisque les projections s’appuient sur ces mêmes événements. Vérifiez la documentation actuelle avant de vous engager avec une interface complexe.
Une liste de contrôle à appliquer dès aujourd’hui
- Commencez par identifier chaque
invokepour lequel une personne attend une réponse, puis remplacez-les paragent.stream(...)en modeupdates. Un journal des étapes approximatif vaut toujours mieux qu’une page vide. - Examinez un morceau de données réel. Lancez-le avec
version="v2", affichez-le, et consultez les champstype,nsetdataafin de connaître précisément sa structure. - Avant que les utilisateurs réels ne voient les résultats, supprimez tous les morceaux
messagesdont les métadonnées indiquentlc_source == "summarization". - Si l’agent délègue des tâches à des sous-agents, activez
subgraphs=Trueet classez les résultats par espace de noms, un groupe par agent. - Gardez une petite console similaire à celle ci-dessus dans le dépôt. Elle vous permet de suivre l’exécution de chaque nouvelle fonctionnalité sans devoir deviner à partir du dernier message.
Qu’est-ce qui a changé et qu’est-ce qui n’a pas changé
Résultat final : une appel de blocage s’est transformé en flux en direct. stream_mode détermine le niveau de détail obtenu, ns indique qui a généré chaque événement, le filtre de synthèse élimine les bavardages propres au système, et il devrait désormais être clair que observer une exécution et s’en souvenir sont deux problèmes distincts.
Rien n’a changé concernant l’agent ou son boucle de traitement. La seule ajoutation a été une fenêtre, et cette fenêtre marque la frontière entre un outil en qui les gens ont confiance et un outil qu’ils consultent dans la panique. Lorsqu’un agent montre le déroulement de son travail, les utilisateurs pensent qu’il fonctionne correctement.
Ainsi, la prochaine fois que quelqu’un signale que l’agent « bloque », évitez l’outil d’analyse et essayez d’abord un flux en direct. L’agent parlait constamment ; il avait besoin d’un canal pour le faire.
Lectures complémentaires
- Le système de fichiers de votre agent est une décision : explication des backends des Deep Agents — Comment les backends des LangChain Deep Agents décident de ce qui est conservé : cinq fonctionnalités intégrées, le routage CompositeBackend, le piège du nom d’espace pour les bases de données multi-utilisateurs, et les modèles sandbox.
- Managed Deep Agents Beta : une méthode basée sur les dossiers pour déployer des agents LangChain — Managed Deep Agents de LangChain transforme un dossier de projet en un agent hébergé. La fonction de chaque fichier, le comportement du déploiement mda, ainsi que les lacunes de la version bêta à connaître.