Six primitives LangGraph et le mode de défaillance caché dans chacune d’elles

Ce guide construit ce modèle à partir des six primitives qui composent LangGraph : l’état, les nœuds, les arêtes directes, les arêtes conditionnelles, le système de points de contrôle et l’intervention humaine. Pour chacune d’elles, vous trouverez un exemple minimal, les erreurs que les équipes commettent le plus fréquemment avec elle, ainsi que la version à mettre en production. Si vous souhaitez une présentation plus complète des modèles d’agents basés sur ces éléments, LangGraph en pratique : état, nœuds, arêtes et cinq modèles d’agents aborde ce sujet ; ici, l’accent est mis sur les modes de défaillance.

Pourquoi un graphe plutôt qu’une chaîne

La syntaxe de pipeline de LangChain, prompt | llm | parser, est pratique pour une seule passe par le modèle. Elle cesse de fonctionner dès que l’agent doit prendre une décision : rechercher ou répondre directement, réessayer ou abandonner, demander de l’aide à une personne ou continuer. Une chaîne ne comprend pas le concept de « cela dépend », c’est pourquoi les développeurs encadrent les appels de la chaîne dans des instructions if ; rapidement, ils créent ainsi une machine à états non documentée et plus difficile à déboguer.

LangGraph rend cette machine à états explicite. On y trouve des nœuds, des arêtes et un objet d’état partagé que l’on peut examiner à tout moment. Il n’y a rien de magique là-dedans, et c’est justement l’avantage : chaque décision prise par l’agent correspond à quelque chose que l’on peut lire dans la définition du graphe.

1. État : un objet partagé et des réducteurs importants

L’état est l’unique objet dont chaque nœud lit et écrit. Sans lui, le contexte a tendance à être transmis sous forme d’arguments de fonction, ce qui rend difficile de déterminer ce que savait réellement une étape donnée. La définition ci-dessous est un TypedDict contenant une question, une réponse et une liste de messages dont les mises à jour sont fusionnées par le réducteur add_messages.

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
    question: str
    answer: str
    messages: Annotated[list, add_messages]

operator.add n’est pas un réducteur de messages

De nombreux tutoriels annotent le champ des messages par operator.add à la place. Cela semble correct : add ajoute un élément à la liste au lieu de l’écraser, ce qui est nécessaire pour une conversation en progression. Le problème, c’est qu’il effectue une concaténation de manière aveugle. Dès que l’on a besoin de mettre à jour ou de supprimer un message existant, par exemple pour raccourcir l’historique ou remplacer le résultat d’une appel d’outil, il ajoute plutôt un duplicata, et l’historique de la conversation se remplit d’entrées obsolètes sans aucun erreur.

add_messages a été conçu spécifiquement à cet effet. Il identifie les messages par leur ID et remplace un message existant lorsque cet ID est déjà présent, n’ajoutant que les messages véritablement nouveaux. La règle est simple : utilisez add_messages pour les champs contenant des objets HumanMessage et AIMessage, et conservez operator.add pour les listes d’accumulation classiques, comme un registre des outils qui ont été appelés.

Privilégiez une structure d’état minimale

La deuxième erreur courante consiste à concevoir l’état comme un schéma de base de données, avec un champ pour chaque besoin que quelqu’un pourrait avoir plus tard. N’ajoutez un champ que lorsque un nœud le lit ou l’écrit réellement. Le coût d’ignorer cela est concret : prenons un graphe de traitement de documents qui stocke dans l’état les réponses brutes complètes d’un LLM, y compris les métadonnées d’utilisation des tokens. Le traitement de 50 documents en boucle a fait monter chaque point de contrôle à environ 180 KB, et les écritures dans Postgres ont dépassé 400 ms, un temps suffisamment lent pour que les utilisateurs en attente d’une réponse s’en aperçoivent. La solution n’était pas spectaculaire : réduire l’état aux trois champs réellement utilisés par les nœuds suivants. N’oubliez pas qu’avec un point de contrôle associé, tout ce qui se trouve dans l’état est serialisé et enregistré à chaque étape.

2. Nœuds : ne renvoyer que ce qui a changé

Un nœud est une fonction Python ordinaire. Il reçoit l’état, effectue son travail et renvoie un dictionnaire contenant uniquement les champs qu’il a modifiés. C’est tout le contrat. Le premier exemple appelle un modèle de chat d’OpenAI avec une question et écrit la réponse dans answer.

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini")
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Lorsque vous itérez sur la structure du graphe, ce qui constitue la majeure partie des premiers travaux, vous ne voulez peut-être pas que chaque version provisoire fasse appel à une API payante. Un modèle local fourni par Ollama implémente la même interface, de sorte que le corps du nœud reste identique et que le débogage ne coûte rien :

from langchain_ollama import ChatOllama
llm = ChatOllama(model="llama3.1", temperature=0)
def answer_node(state: AgentState) -> dict:
    response = llm.invoke([HumanMessage(content=state["question"])])
    return {"answer": response.content}

Cette version nécessite que Ollama soit en cours d’exécution localement avec le modèle téléchargé (ollama pull llama3.1) et que le paquet d’intégration soit installé (pip install langchain-ollama). Définir temperature=0 rend également les exécutions plus reproductibles, ce qui est utile lors du test de la logique de routage.

Rétablir l’ensemble de l’état perturbe les autres mises à jour

Une erreur fréquente consiste à renvoyer tout le dictionnaire de l’état depuis un nœud au lieu des seules clés modifiées. Dans un petit graphe linéaire, cela semble fonctionner, car rien d’autre n’affecte ces champs. Dès que deux nœuds mettent à jour des champs qui se chevauchent, le renvoi complet d’un nœud écrase les modifications de l’autre avec des valeurs anciennes. Le symptôme ressemble à un problème de routage, ce qui pousse les développeurs à chercher dans la logique des arêtes, alors que la cause réelle est le fait qu’un nœud renvoie trop d’informations. Un renvoi minimal des mises à jour permet également aux réducteurs de faire leur travail : un champ sans réducteur est simplement remplacé par ce que le nœud renvoie.

3. Arêtes directes : connectez toujours la sortie

Les arêtes déterminent ce qui s’exécute ensuite. Une arête directe est inconditionnelle : lorsque le nœud A termine, le nœud B s’exécute. Le graphique ci-dessous définit deux nœuds, relie answer à refine, relie refine à END, définit le point d’entrée et compile le tout.

from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
graph.add_node("answer", answer_node)
graph.add_node("refine", refine_node)
graph.add_edge("answer", "refine")
graph.add_edge("refine", END)
graph.set_entry_point("answer")
app = graph.compile()

Le bord END est celui que les gens oublient, et c’est la cause classique d’un graphe qui semble tourner indéfiniment. L’habitude vraiment fiable consiste à faire en sorte que chaque chemin dans le graphe se termine explicitement par END, afin de pouvoir déterminer la fin en lisant la définition. Cela devient particulièrement important dès l’apparition de cycles : une boucle de tentative sans chemin menant à END, ou avec une condition qui ne devient jamais vraie, continue de tourner jusqu’à ce que la limite de récursivité de LangGraph l’arrête en générant une GraphRecursionError. Cette limite n’est qu’un filet de sécurité, pas une caractéristique fondamentale du design ; chacune de ces itérations coûte toujours des tokens. Lorsqu’un graphe semble bloqué, vérifiez d’abord sa définition.

4. Bords conditionnels : où l’agent prend réellement une décision

Les arêtes conditionnelles sont ce qui fait d’un graphe un agent plutôt qu’un pipeline fixe. Une fonction de routage examine l’état et renvoie une étiquette ; une transformation convertit chaque étiquette en le nœud suivant. Dans cet exemple, une réponse courte (moins de 50 caractères) est envoyée vers refine, tandis que tout le reste est dirigé vers END.

def route_based_on_quality(state: AgentState) -> str:
    if len(state["answer"]) < 50:
        return "refine"
    return "done"
graph.add_conditional_edges(
    "answer",
    route_based_on_quality,
    {"refine": "refine", "done": END},
)

Notez que cette arête conditionnelle remplace l’arête directe answer vers refine du snippet précédent. Si vous enregistrez les deux, les deux chemins sont empruntés, ce qui est rarement souhaité.

Les étiquettes de route incompatibles provoquent des erreurs claires mais difficiles à diagnostiquer

L’erreur récurrente ici est une fonction de routage qui renvoie une chaîne non présente dans le dictionnaire de mappage. L’erreur qui en résulte est une erreur de clé assez générique, cachée à plusieurs niveaux profonds dans l’historique des appels, et un détail aussi insignifiant qu’un espace en fin de chaîne peut coûter étonnamment beaucoup de temps. Une habitude fiable : écrire d’abord le mappage, puis le routeur en copiant les clés exactes à partir de celui-ci. Mieux encore, définir les étiquettes une fois en tant que constantes ou annoter le type de retour du routeur avec Literal["refine", "done"] afin que les vérificateurs de types et les lecteurs voient immédiatement les valeurs autorisées.

5. Point d’étape : mémoire conservée entre les appels

Un checkpointer transforme une appel de fonction sans état en une conversation avec la mémoire. Sans lui, chaque app.invoke() commence à zéro. Avec lui, l’état est sauvegardé pour chaque thread, et tout appel qui transmet le même thread_id dans sa configuration reprend là où l’appel précédent s’est arrêté. Dans l’exemple, la deuxième invocation sur le thread user-session-42 se souvient de la première question.

from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-session-42"}}
app.invoke({"question": "What is LangGraph?"}, config)
app.invoke({"question": "Show me a code example"}, config)   # remembers the first turn

InMemorySaver convient uniquement au développement local. Il existe en mémoire de processus, donc un redémarrage du serveur efface toutes les conversations. Tout ce dont dépendent les utilisateurs réels nécessite un backend persistant : SQLite pour un seul serveur, ou Postgres lorsque plusieurs instances doivent partager l’état.

# single-server production — pip install langgraph-checkpoint-sqlite
from langgraph.checkpoint.sqlite import SqliteSaver
# multi-instance production, needs shared state across servers
# pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

Chaque backend est fourni en tant que package indépendant, comme l’indiquent les commentaires d’installation. Dans les versions actuelles, ces sauvegardeurs sont généralement créés à partir d’une chaîne de connexion (par exemple via from_conn_string) et Postgres nécessite une appel unique à la fonction setup() pour créer ses tables ; consultez donc la documentation de checkpointer pour connaître l’initialisation exacte dans votre version.

Le problème potentiel réside dans l’utilisation du sauvegardeur en mémoire en environnement de production, ce qui ne sera découvert que lorsque une réinitialisation d’environnement de test effacera une démonstration en ligne. La bonne nouvelle, c’est que le changement est peu coûteux si le graphique est bien conçu : checkpointer est un paramètre de compilation, pas une refonte complète, et le passage à SqliteSaver peut prendre moins d’une heure.

6. Intervention humaine : points d’arrêt statiques versus interruptions dynamiques

Le schéma présenté dans la plupart des tutoriels est interrupt_before, une liste de noms de nœuds où le graphe compilé s’arrête avant d’exécuter :

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"],
)

Cela fonctionne et est facile à expliquer, mais c’est une approche statique. Le point d’arrêt est déterminé par le nom du nœud ; on ne peut pas en faire une condition, ni y associer de charge utile décrivant ce que l’examinateur doit examiner. Les exigences réelles le rendent rapidement insuffisant, car « s’arrêter avant ce nœud » et « s’arrêter uniquement lorsque le remboursement dépasse 500 $ » sont des règles différentes, et seule la première peut être exprimée de cette manière.

Arrêt depuis l’intérieur du nœud avec interrupt()

Le modèle plus flexible consiste à appeler interrupt() depuis l’intérieur du nœud. Le nœud ci-dessous vérifie le montant du remboursement ; s’il dépasse 500 $, il pause et présente le projet ainsi que le montant à un humain. La première invocation se poursuit jusqu’à cette pause. La deuxième appel transmet Command(resume="approve") sur le même thread, et la valeur fournie à resume devient la valeur de retour de interrupt() ; ainsi, le nœud soit continue d’envoyer les données, soit renvoie un statut de cancellation. Un pointeur de vérification est nécessaire, car l’état en pause doit être stocké quelque part pendant cette attente.

from langgraph.types import interrupt, Command
def send_email_node(state: AgentState) -> dict:
    if state["refund_amount"] > 500:
        decision = interrupt({
            "draft": state["draft"],
            "amount": state["refund_amount"],
        })
        if decision != "approve":
            return {"status": "cancelled"}
    # send the email
    return {"status": "sent"}
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "task-99"}}
app.invoke({"task": "Draft and send a refund email"}, config)
# graph pauses inside send_email_node, surfaces the interrupt payload
app.invoke(Command(resume="approve"), config)

L’état de l’exemple utilise des champs tels que refund_amount, draft et task qui ne figurent pas dans l’ancien AgentState ; dans un vrai graphe, ils devraient y être déclarés.

La reprise exécute à nouveau l’ensemble du nœud

Le comportement qui surprend les gens : dans le mode de reprise, LangGraph ne continue pas à partir de la ligne interrupt(). Il réexécute l’ensemble du nœud depuis le début, et cette fois interrupt() renvoie la valeur de reprise au lieu de suspendre l’exécution. Tout code exécuté avant cet appel est relancé. Un nœud qui incrémente un compteur avant d’interruption l’incrémentera deux fois pour chaque approbation. Gardez tout ce qui précède interrupt() idempotent, ou déplacez les effets secondaires dans un nœud antérieur. La même logique s’applique aux appels API ou aux écritures dans la base de données placés avant la pause.

Un modèle qui s’approbe lui-même n’est pas un système avec une intervention humaine

Quel que soit le mécanisme choisi, demander au modèle « Devrais-je continuer ? » et faire confiance à sa réponse n’équivaut pas à une supervision humaine, quel que soit le nom donné à cette pratique. Il s’agit plutôt de l’agent qui valide sa propre décision. Une véritable étape d’approbation transfère le contrôle à une personne extérieure au réseau et attend sa réponse.

Un aperçu des six primitives

Le résumé ci-dessous associe chaque concept à sa fonction ainsi qu’à l’erreur typique qui lui est associée.

+----------------------+----------------------------------------+---------------------------+
| Concept              | What it does                            | The mistake I made        |
+----------------------+----------------------------------------+---------------------------+
| State                | Shared, typed dict every node touches   | operator.add instead of   |
|                      |                                          | add_messages for chat     |
+----------------------+----------------------------------------+---------------------------+
| Nodes                | Plain functions: state in, updates out  | Returning full state,     |
|                      |                                          | not just changed fields   |
+----------------------+----------------------------------------+---------------------------+
| Direct edges         | Always go to the same next node         | Forgetting to wire END    |
+----------------------+----------------------------------------+---------------------------+
| Conditional edges    | Function inspects state, picks next node| Return value doesn't      |
|                      |                                          | match a mapping key       |
+----------------------+----------------------------------------+---------------------------+
| Checkpointing        | Persists state per thread_id            | InMemorySaver in prod     |
+----------------------+----------------------------------------+---------------------------+
| Human-in-the-loop    | Pauses for a real person, then resumes  | Non-idempotent code       |
|                      |                                          | before interrupt()        |
+----------------------+----------------------------------------+---------------------------+

Un ordre de construction sensé

Pour votre premier graphe réel, assurez que le boucle complète fonctionne du début à la fin avec InMemorySaver sans interruptions. Gardez l’état minimal et limité aux seules informations nécessaires aux nœuds, et vérifiez que chaque arête conditionnelle renvoie exactement les étiquettes attendues par sa correspondance. Ce n’est qu’une fois que tout fonctionne correctement que vous devriez introduire un point de contrôle persistant et ajouter une interruption à l’étape qui nécessite réellement une intervention humaine, généralement toute opération impliquant le transfert d’argent, l’envoi d’un e-mail externe ou la suppression de données.

Des fonctionnalités plus avancées, telles que des réducteurs personnalisés au-delà de add_messages, des sous-graphes permettant de diviser un grand graphe en parties testables, ainsi que le streaming au niveau des tokens, reposent toutes sur la même structure de base. Elles deviennent beaucoup plus faciles à adopter une fois que vous avez construit, détruit et corrigé un graphe en utilisant uniquement ces six principes.

Points clés

• Utilisez add_messages pour l’historique de chat et operator.add uniquement pour des listes simples, et maintenez un état minimal car il est persisté à chaque étape.
• Renvoyez uniquement les champs modifiés des nœuds ; un retour de l’état complet écrasera silencieusement les mises à jour parallèles ou antérieures.
• Attribuez à chaque chemin une route explicite vers END, et utilisez des boucles de tentative limitées plutôt que de compter sur la limite de récursivité.
• Déduisez les étiquettes de routage à partir du mappage afin qu’elles ne s’éloignent pas l’une de l’autre.
• Tenez InMemorySaver réservé au développement uniquement ; le remplacement du checkpointer est peu coûteux, donc effectuez-le avant que les utilisateurs ne dépendent du graphe.
• Préférez une utilisation dynamique de interrupt() pour les approbations conditionnelles, et assurez-vous que le code soit idempotent avant de reprendre son exécution, car cela relance le nœud.

Documentation de référence : la documentation de l’API Graph pour LangGraph, le guide sur les interruptions ainsi que la API interrupt() de référence.