LangGraph en Python : États, nœuds parallèles, points de contrôle et révision humaine

Tout ceux qui ont développé plus qu’un simple jouet en s’appuyant sur un LLM connaissent trois problèmes récurrents. Les appels successifs augmentent la latence. Un flux de travail trop long peut planter et annuler tous les progrès réalisés. De plus, le modèle peut, au pire moment possible, fournir une réponse affirmative qui est complètement fausse.

LangGraph résout ces trois problèmes. Il s’agit d’un cadre d’orchestration basé sur des graphes, construit à partir de quelques éléments simples : l’état est représenté par un TypedDict, les nœuds sont des fonctions ordinaires, et les arêtes représentent la logique de routage. Ces éléments de base permettent de créer des flux de travail qui exécutent les nœuds en parallèle, enregistrent l’état à des points de contrôle afin de pouvoir se remettre d’un plantage, et permettent une pause pour que quelqu’un puisse examiner la situation lorsque les enjeux sont importants.

Les sections suivantes expliquent le processus étape par étape. D’abord, un seul nœud est présent, puis vient l’exécution en parallèle, le routage conditionnel et la mémoire persistante, suivis des interruptions avec intervention humaine. Un flux de travail complet de tri des e-mails conclut le tout en combinant tous ces éléments. Chaque exemple est conçu pour fonctionner sans modification. Le SDK TypeScript expose presque les mêmes API, ce qui permet d’appliquer le même modèle de pensée.

Prérequis

Deux paquets sont nécessaires en premier :

pip install langgraph langchain-openai

Ensuite, les variables d’environnement pour OpenAI et, éventuellement, le suivi de LangSmith :

export OPENAI_API_KEY="your-key-here"
# Optional: enable LangSmith tracing for debugging
export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="your-langsmith-key"

Les exemples ont été testés avec langgraph 1.1.3 (mars 2026) et langchain-openai 1.1.10, et nécessitent Python 3.10 ou une version ultérieure. Les guides plus anciens utilisent parfois LANGCHAIN_TRACING_V2 et LANGCHAIN_API_KEY ; ces derniers continuent de fonctionner, mais LANGSMITH_TRACING et LANGSMITH_API_KEY sont désormais les noms préférés.

Tout dans LangGraph repose sur deux idées fondamentales : l’état contient les données, et les nœuds sont des fonctions qui les transforment. Une fois ces concepts maîtrisés, le reste du framework s’organise naturellement.

Définir l’état avec TypedDict

L’état dans LangGraph est un simple TypedDict Python, sans classes de base particulières ni décorateurs spécifiques au framework, il s’agit simplement de dictionnaires typés :

from typing import Annotated
from typing_extensions import TypedDict
import operator

class State(TypedDict):
    messages: Annotated[list[str], operator.add]
    summary: str

La classe définit la structure des données qui circulent à travers le graphe, et chaque nœud lit dans et écrit dans cette structure partagée.

Pourquoi un TypedDict et non un dictionnaire ordinaire ou une dataclass ? Il permet un contrôle de type statique, puisque mypy et Pyright le comprennent tous deux, ainsi que des indications de type, et il évite la charge en temps de exécution propre aux dataclasses ou aux modèles Pydantic. LangGraph peut également examiner les annotations pour comprendre le schéma, ce qui est essentiel pour une sortie structurée et une validation efficace.

La ligne à comprendre est Annotated[list[str], operator.add], qui définit un réducteur. C’est le concept le plus important dans l’état de LangGraph, et il pose problème à presque tout le monde la première fois.

En l’absence de réducteur, un nœud renvoyant {"summary": "new value"} remplace simplement la valeur stockée. C’est exactement ce que l’on souhaite pour des champs scalaires comme summary. Les problèmes surviennent lorsque deux nœuds parallèles écrivent tous deux dans messages : seul celui qui termine en dernier est conservé. En déclarant plutôt operator.add, on indique à LangGraph d’ajouter les valeurs, de sorte que chaque contribution à messages s’accumule en toute sécurité.

Une logique de fusion plus complexe nécessite un réducteur personnalisé :

def keep_latest_n(existing: list[str], new: list[str]) -> list[str]:
    """Keep only the last 10 messages."""
    combined = existing + new
    return combined[-10:]

class State(TypedDict):
    messages: Annotated[list[str], keep_latest_n]

Ce réducteur combine la liste existante avec les nouveaux messages, puis la réduit aux dix messages les plus récents, ce qui est utile pour les conversations longues où le nombre de tokens augmenterait sinon de manière exponentielle. operator.add permet une simple ajoutition en fin de liste. Un réducteur personnalisé est nécessaire pour éliminer les doublons, limiter la taille de la liste ou effectuer des fusionnements allant au-delà d’une simple ajoutition.

La erreur classique consiste à oublier d’utiliser un réducteur pour un champ auquel des nœuds parallèles écrivent des données. Sans réducteur, les branches écrasent silencieusement les unes les autres, celle qui termine en dernier l’emporte, et aucun erreur n’apparaît jamais.

TypedDict versus Pydantic

Les deux apparaissent dans LangGraph, et la séparation entre eux est intentionnelle. TypedDict est utilisé pour l’état du graphe, le conteneur qui se déplace à travers celui-ci. LangGraph a été conçu autour de lui : ses champs prennent en charge la syntaxe du réducteur Annotated, il se serialise proprement pour la création de points de contrôle, et il ne génère aucun surcoût en temps de exécution.

Pydantic est utilisé pour les schémas destinés aux LLM. L’appel à .with_structured_output(SomeModel) fournit à l’LLM un schéma JSON à respecter et vérifie que la réponse correspond. Une réponse qui ne correspond pas provoque immédiatement une erreur et empêche les données mal formatées de circuler dans le graphe.

Le flux de travail final suit précisément cette division : EmailState est un TypedDict servant de conteneur pour le graphe, tandis que EmailClassification est un modèle Pydantic décrivant la sortie de l’LLM.

Rédaction du premier nœud

Chaque nœud prend l’état complet en entrée et renvoie un dictionnaire d’actualisation partielle. LangGraph intègre cette actualisation dans l’état et applique tous les réducteurs qui ont été déclarés :

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def chatbot(state: ChatState) -> dict:
    """Call the LLM with the conversation history."""
    response = llm.invoke(state["messages"])
    return {"messages": [response.content]}

# Build the graph
graph = StateGraph(ChatState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()
# Run it
result = app.invoke({"messages": ["What is LangGraph?"]})
print(result["messages"][-1])

Examen du code :

• StateGraph(ChatState) crée un graphe qui utilise ChatState comme schéma, et LangGraph lit les annotations pour en apprendre davantage sur les réducteurs et les types de champs.
• graph.add_node("chatbot", chatbot) enregistre la fonction sous le nom "chatbot", qui est celui auquel se réfèrent les définitions d’arêtes.
• graph.add_edge(START, "chatbot") fait de "chatbot" le premier nœud à être exécuté. START est un sentinelle intégré.
• graph.add_edge("chatbot", END) termine l’exécution une fois que chatbot a fini. END est une autre sentinelle intégrée.
• graph.compile() valide la structure, en vérifiant l’absence d’arêtes orphelines et de nœuds inaccessibles, et retourne un CompiledGraph prêt à être exécuté.

app.invoke({"messages": [...]}) exécute le graphe de manière synchrone et renvoie l’état final.

Le nœud enveloppe sa chaîne de caractères dans une liste, car messages utilise operator.add comme réducteur. En renvoyant {"messages": [response.content]}, on ajoute une liste contenant un seul élément à la liste existante, tandis que le renvoi d’une simple chaîne de caractères échouerait en temps de exécution, car operator.add ne peut pas ajouter une chaîne à une liste.

La règle consiste à faire correspondre ce qu’un nœud renvoie au type du champ, y compris le réducteur. Un champ de type list[str] utilisant operator.add attend une valeur de type list[str], tandis qu’un champ simple de type str sans réducteur attend une valeur de type str.

C’est là une application LangGraph complète et fonctionnelle en environ 20 lignes. La question suivante porte sur ce qui se passe à l’intérieur de invoke().

Étapes super : comment fonctionne l’exécution

LangGraph exécute les nœuds en étapes super. Chaque nœud programmé pour la même étape s’exécute en parallèle, et tous doivent être terminés avant que la étape suivante ne commence. L’exemple à un seul nœud comporte une étape super contenant un seul nœud. L’ajout d’arêtes parallèles place plusieurs nœuds dans la même étape super.

Cela est important pour deux raisons. L’exécution parallèle est intégrée au modèle dès le départ et n’est pas ajoutée ultérieurement. De plus, grâce à un checkpointer, LangGraph enregistre un instantané complet de l’état après chaque fin d’étape super. C’est ce qui permet la récupération en cas de panne : si le processus s’arrête en cours de route, l’exécution reprend à partir de la dernière étape super terminée.

Une fois que le flux d’état à travers un seul nœud est compris, la question suivante est de savoir comment contrôler l’ordre ou exécuter les nœuds en même temps. C’est là que interviennent les arêtes.

Arêtes : séquentielles, parallèles et conditionnelles

Les arêtes contrôlent la manière dont l’exécution se déplace entre les nœuds. LangGraph propose trois types, qui couvrent ensemble tous les schémas de routage possibles. La forme la plus simple est déjà apparue dans l’exemple du chatbot.

Arêtes sérielles

Le cas le plus simple consiste en un nœud suivi du suivant.

class DraftState(TypedDict):
    topic: str
    draft: str
    review: str

def generate_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Write a short paragraph about: {state['topic']}")
    return {"draft": response.content}

def review_draft(state: DraftState) -> dict:
    response = llm.invoke(f"Review this draft and suggest improvements:\\n{state['draft']}")
    return {"review": response.content}

graph = StateGraph(DraftState)
graph.add_node("generate", generate_draft)
graph.add_node("review", review_draft)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", END)

app = graph.compile()
result = app.invoke({"topic": "Why Python developers love type hints"})
print(result["review"])

generate_draft s’exécute en premier. Une fois qu’il a terminé et que LangGraph a fusionné le champ draft dans l’état, review_draft s’exécute et peut lire le brouillon depuis state["draft"]. L’ordre séquentiel est garanti, car le bord impose cette dépendance, et aucun code de séquencement n’est nécessaire. Les bords séquentiels conviennent chaque fois que le nœud B dépend de la sortie du nœud A : générer puis examiner, récupérer puis traiter, classifier puis diriger.

Les cycles accidentels constituent un piège courant. Par défaut, LangGraph attend un graphe acyclique dirigé, donc l’ajout d’un bord allant de review vers generate provoque une erreur au moment de la compilation, à moins que le graphe ne soit explicitement conçu pour être cyclique. Les boucles itératives sont généralement construites à l’aide de bords conditionnels qui peuvent rediriger vers un nœud antérieur.

Bords parallèles

Lorsque deux arêtes partent d’un même nœud, leurs cibles sont traitées en parallèle au cours de la même super-étape. Les réducteurs deviennent essentiels dans ce cas :

class ResearchState(TypedDict):
    topic: str
    findings: Annotated[list[str], operator.add]
    outline: str
    final_summary: str

def research(state: ResearchState) -> dict:
    response = llm.invoke(f"List 3 key facts about: {state['topic']}")
    return {"findings": [f"Research: {response.content}"]}

def create_outline(state: ResearchState) -> dict:
    response = llm.invoke(f"Create a brief outline for an article about: {state['topic']}")
    return {"outline": response.content}

def combine(state: ResearchState) -> dict:
    research_text = "\\n".join(state["findings"])
    response = llm.invoke(
        f"Combine this research and outline into a summary:\\n"
        f"Research:\\n{research_text}\\n\\n"
        f"Outline:\\n{state['outline']}"
    )
    return {"final_summary": response.content}

graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("outline", create_outline)
graph.add_node("combine", combine)
# Fan-out: both start after START
graph.add_edge(START, "research")
graph.add_edge(START, "outline")
# Fan-in: both feed into combine
graph.add_edge("research", "combine")
graph.add_edge("outline", "combine")
graph.add_edge("combine", END)
app = graph.compile()
result = app.invoke({"topic": "LangGraph parallel execution"})
print(result["final_summary"])

Deux arêtes partant de START planifient research et outline dans la même étape super, et LangGraph les exécute en parallèle. Ces deux tâches possèdent également une arête menant à combine ; LangGraph ne planifiera combine que lorsque tous les nœuds qui y font référence auront terminé, ce qui agit comme une barrière de synchronisation automatique.

findings dispose d’un réducteur même si seuls research y écrit ici, car un système réel pourrait comporter quatre agents de recherche parallèles ajoutant des éléments à la même liste, et sans réducteur un seul résultat survivrait.

Le calcul de la latence est simple : si research prend 2 secondes et outline 3, la version séquentielle nécessite 5 secondes tandis que la version parallèle n’en nécessite que 3, soit la durée de la tâche la plus longue. Pour des appels indépendants à un LLM, l’exécution parallèle permet des économies de latence presque nulles.

Les candidats à la parallélisation sont tous les paires de nœuds qui ne dépendent pas de la sortie l’un de l’autre. Même le fait de regrouper deux appels d’une seconde dans la même étape superposée a un impact significatif lorsqu’il s’agit de milliers de requêtes. Faites attention aux nœuds qui semblent indépendants mais partagent des états de manières moins évidentes. Si deux nœuds parallèles écrivent dans un même champ sans réducteur, il s’agit d’une condition de course : c’est le dernier à terminer qui détermine la valeur, et LangGraph ne donne aucun avertissement.

Ce schéma de diffusion et de réception constitue le fondement du projet final relatif aux e-mails, qui effectue en parallèle une recherche dans une base de connaissances et une vérification dans un suivi de bogues avant de rédiger une réponse.

Arêtes conditionnelles

Lorsque le nœud suivant dépend de ce que produit le nœud actuel, la fonction add_conditional_edges est appliquée :

class ClassifyState(TypedDict):
    text: str
    sentiment: str
    result: str

def classify(state: ClassifyState) -> dict:
    response = llm.invoke(
        f"Classify the sentiment of this text as 'positive' or 'negative'. "
        f"Reply with just the word.\\n\\n{state['text']}"
    )
    return {"sentiment": response.content.strip().lower()}

def handle_positive(state: ClassifyState) -> dict:
    return {"result": f"Great feedback! Text was: {state['text']}"}

def handle_negative(state: ClassifyState) -> dict:
    return {"result": f"Escalating concern. Text was: {state['text']}"}

def route_by_sentiment(state: ClassifyState) -> str:
    """Routing function: inspect state, return a string key."""
    if state["sentiment"] == "positive":
        return "positive"
    return "negative"

graph = StateGraph(ClassifyState)
graph.add_node("classify", classify)
graph.add_node("handle_positive", handle_positive)
graph.add_node("handle_negative", handle_negative)
graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    route_by_sentiment,
    {"positive": "handle_positive", "negative": "handle_negative"}
)
graph.add_edge("handle_positive", END)
graph.add_edge("handle_negative", END)
app = graph.compile()
result = app.invoke({"text": "This product is absolutely fantastic!"})
print(result["result"])
# Great feedback! Text was: This product is absolutely fantastic!

La fonction route_by_sentiment examine l’état et renvoie une clé de type chaîne de caractères, et le dictionnaire de mappage fourni à add_conditional_edges transforme cette clé en un nom de nœud. Une fonction de routage qui reste pure (entrée : état, sortie : chaîne de caractères, sans effets secondaires) est facile à tester, car elle peut être testée unitairement sans faire appel à un LLM.

add_conditional_edges prend trois arguments : le nœud source, la fonction de routage, et un dictionnaire qui associe des clés de type chaîne de caractères aux noms des nœuds de destination. Chaque clé que la fonction de routage pourrait renvoyer doit figurer dans ce dictionnaire, sinon LangGraph génère une erreur en temps de exécution.

L’alternative Command

Lorsqu’un nœud doit mettre à jour son état et effectuer le routage dans la même étape, il peut renvoyer directement une Command et omettre add_conditional_edges :

from langgraph.types import Command

def classify_and_route(state: ClassifyState) -> Command:
    response = llm.invoke(
        f"Classify sentiment as 'positive' or 'negative'.\\n\\n{state['text']}"
    )
    sentiment = response.content.strip().lower()
    if sentiment == "positive":
        return Command(update={"sentiment": sentiment}, goto="handle_positive")
    else:
        return Command(update={"sentiment": sentiment}, goto="handle_negative")

Une valeur de retour Command contient à la fois une mise à jour d’état (update=) et un choix de routage (goto=), de sorte que le nœud écrit des données et choisit la destination suivante en une seule opération.

add_conditional_edges est le choix approprié lorsque la logique de routage ne lit que l’état sans écrire rien de nouveau, lorsque le routage doit rester séparé de la logique du nœud, ou lorsque plusieurs nœuds sources partagent la même fonction de routage.

Command est le choix approprié lorsque le nœud génère de nouvelles données qui influencent son propre routage, lorsque un cas isolé ne justifie pas une fonction de routage distincte, ou lorsque la destination dépend de quelque chose calculé à l’intérieur du nœud et qui n’est pas encore présent dans l’état.

Pour le contrôle humain, l’email capstone s’appuie sur add_conditional_edges, car le routage y est simple : il ne vérifie que la priorité et la confiance déjà stockées dans l’état.

Mémoire et points de contrôle

Avec le contrôle de l’ordre et des branches assuré, le prochain problème est la persistance : que se passe-t-il lorsque le processus redémarre au milieu d’une conversation ?

Jusqu’à présent, chaque invocation a commencé à zéro. Les conversations et les workflows à plusieurs étapes nécessitent un état qui survive entre les appels. LangGraph gère cela grâce aux points de contrôle, et l’API est suffisamment simple pour que le passage de la phase de développement à celle de production ne nécessite qu’un seul changement.

MemorySaver pour le développement

MemorySaver conserve l’état en mémoire, ce qui convient au développement et aux tests :

from langgraph.checkpoint.memory import MemorySaver

class ConversationState(TypedDict):
    messages: Annotated[list[str], operator.add]

def chatbot(state: ConversationState) -> dict:
    history = "\\n".join(state["messages"])
    response = llm.invoke(f"Conversation so far:\\n{history}\\n\\nRespond to the latest message.")
    return {"messages": [response.content]}

graph = StateGraph(ConversationState)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)

# Compile WITH a checkpointer
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# Thread 1: a conversation about Python
config_1 = {"configurable": {"thread_id": "user-alice"}}
app.invoke({"messages": ["Hi, I'm learning Python"]}, config=config_1)
result = app.invoke({"messages": ["What should I learn first?"]}, config=config_1)
print(result["messages"][-1])

# The response knows about the previous message because state persisted
# Thread 2: a completely independent conversation
config_2 = {"configurable": {"thread_id": "user-bob"}}
result = app.invoke({"messages": ["Tell me about Rust"]}, config=config_2)
print(result["messages"][-1])
# This conversation has no knowledge of Alice's conversation

Avec checkpointer=memory, le graphe compilé enregistre une capture d’écran complète après chaque étape super. Une nouvelle appelation de invoke() utilisant le même thread_id amène LangGraph à charger en premier le dernier point de contrôle correspondant à ce thread. Le thread_id fonctionne comme un identifiant de session : réutilisez-le pour continuer, changez-le pour commencer à zéro.

La deuxième appelation peut omettre les messages précédents, car le point de contrôle les contient déjà. Si Alice lance l’appel avec {"messages": ["What should I learn first?"]}, LangGraph charge le point de contrôle "user-alice", trouve la liste enregistrée et ajoute la nouvelle entrée au moyen du réducteur operator.add. Une fois que le chatbot est lancé, state["messages"] contient toute l’histoire des messages.

Reutiliser un thread_id pour des conversations non liées est une erreur. Si "user-alice" est réutilisé pour un sujet différent, le chatbot prend en compte l’intégralité des conversations précédentes d’Alice, ce qui signifie que chaque nouvelle session doit disposer d’un thread_id frais. Pendant le développement, app.get_state_history(config) affiche l’historique complet des points de contrôle d’un thread, et c’est le même mécanisme qui permet le débogage par voyage dans le temps dans LangSmith.

Points de contrôle durables pour la production

MemorySaver existe dans la mémoire du processus, de sorte que lorsque ce dernier s’arrête, l’état est également perdu. La production nécessite un backend durable :

# Install: pip install langgraph-checkpoint-postgres
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = "postgresql://user:pass@localhost:5432/mydb"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # Creates tables on first run
    app = graph.compile(checkpointer=checkpointer)
    # Same API, durable state
    result = app.invoke({"messages": ["Hello"]}, config=config_1)

checkpointer.setup() crée les tables de base de données nécessaires lors du premier exécution et ne fait rien par la suite, il suffit donc de l’appeler une seule fois lors de l’initialisation. L’API reste identique quel que soit le backend, car les outils de vérification se trouvent derrière une abstraction commune. MemorySaver, PostgresSaver et RedisSaver suivent le même protocole, de sorte que le code du graphe ne sait pas avec lequel d’entre eux il communique.

Choix parmi eux :

• MemorySaver convient au développement, aux tests et aux démonstrations, où la survie après un redémarrage n’a pas d’importance.
• PostgresSaver est adapté aux workflows de production qui nécessitent de la durabilité ainsi que la possibilité de consulter l’historique des points de contrôle.
• RedisSaver, provenant de langgraph-redis, convient aux cas à fort débit qui exigent des lectures rapides des points de contrôle et peuvent accepter les compromis liés à la persistance de Redis.

Ce qui est enregistré

Une capture d’écran complète est stockée à chaque étape majeure. Cela permet de voyager dans le temps : on peut examiner l’état à n’importe quel moment de l’historique, le rejouer à partir d’un point de contrôle choisi, ou le déboguer pour identifier précisément où se sont produites les erreurs. Le même mécanisme sous-tend les interruptions. Lorsqu’un nœud appelle interrupt(), LangGraph a besoin d’un endroit pour conserver l’état en pause afin de pouvoir le reprendre plus tard. Sans point de contrôle, rien ne peut le conserver, et le schéma d’interruption s’effondre.

Intervention humaine avec les interruptions

Certaines décisions sont trop importantes pour être automatisées. Avant qu’un agent n’envoie un e-mail à un client, n’approuve un remboursement ou ne déploie du code, une personne doit les examiner et les approuver. LangGraph gère cela grâce à la fonction interrupt().

La fonction interrupt()

Placer interrupt() à l’intérieur d’un nœud arrête l’exécution, écrit l’état dans le checkpointer et renvoie un chargement utile à l’appelant :

from langgraph.types import interrupt, Command

class ReviewState(TypedDict):
    messages: Annotated[list[str], operator.add]
    draft: str
    approved: bool

def generate(state: ReviewState) -> dict:
    response = llm.invoke("Write a short thank-you email to a customer.")
    return {"draft": response.content}

def review(state: ReviewState) -> dict:
    """Pause here for human review."""
    decision = interrupt(f"Please review this draft:\\n\\n{state['draft']}")
    if decision == "approve":
        return {"approved": True}
    return {"approved": False}

def send(state: ReviewState) -> dict:
    if state["approved"]:
        return {"messages": ["Email sent successfully."]}
    return {"messages": ["Email was rejected. Not sent."]}

En ordre, l’appel effectue quatre actions :

1. Il bloque l’exécution à la ligne exacte où il est appelé.
2. Il enregistre l’état complet dans le checkpointer, d’où la nécessité de celui-ci.
3. Il renvoie la charge utile qui a été transmise, ici le texte provisoire, à celui qui a appelé invoke().
4. Il attend, laissant le graphe en suspens.

La variable decision ne reçoit rien au début, car interrupt() ne renvoie que lorsque le graphe est repris avec Command(resume=value). À ce moment-là, decision reçoit ce qui a été transmis à resume=.

La charge utile indique au réviseur humain ce qu’il est en train d’examiner, ce qui lui permet de contenir tout ce dont il a besoin : le texte provisoire, un résumé, ou un dictionnaire structuré avec du contexte. Celui qui a appelé la fonction la reçoit dans le cadre de la notification d’interruption.

Reprise avec Command

Pour continuer après une interruption, il faut invoquer à nouveau le graphe avec Command(resume=value) :

memory = MemorySaver()

graph = StateGraph(ReviewState)
graph.add_node("generate", generate)
graph.add_node("review", review)
graph.add_node("send", send)
graph.add_edge(START, "generate")
graph.add_edge("generate", "review")
graph.add_edge("review", "send")
graph.add_edge("send", END)
app = graph.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "email-review-1"}}

# Step 1: Invoke the graph. It will run generate, then pause at review.
result = app.invoke({"messages": ["Start email workflow"]}, config=config)
# result contains the interrupt payload with the draft

# Step 2: The human reviews the draft and approves.
result = app.invoke(Command(resume="approve"), config=config)
print(result["messages"][-1])
# "Email sent successfully."

Dans la première étape, LangGraph exécute generate, enregistre le brouillon dans l’état actuel et atteint la fonction interrupt() du nœud review. Il crée un point de contrôle de l’état et revient, le resultat de ce premier invoke() contenant le chargement d’interruption, c’est-à-dire le brouillon destiné à l’examinateur. Dans la deuxième étape, Command(resume="approve") fait charger par LangGraph le point de contrôle correspondant à "email-review-1", reprend l’exécution à l’appel de interrupt() et renvoie "approve" comme valeur de retour. La valeur decision du nœud review devient "approve" ; il définit alors approved=True et l’exécution passe à l’étape send.

Le même config doit être utilisé pour les deux appels, car LangGraph trouve le bon point de vérification grâce au thread_id. Un thread_id différent pour le deuxième appel pointerait vers un point de vérification qui n’existe pas, ce qui entraînerait l’échec de l’appel.

Un point de vérification est obligatoire pour les interruptions. Sans lui, l’état en pause n’a nulle part où aller, c’est pourquoi app.compile(checkpointer=memory) n’est pas optionnel dans ce cas.

Depuis LangGraph 1.1.0, invoke() accepte version="v2" et renvoie un GraphOutput doté des attributs .value et .interrupts, ce qui permet un traitement des interruptions plus sûr du point de vue du type et mérite d’être essayé une fois les bases maîtrisées. Le chargement d’une interruption peut être n’importe quelle valeur serialisable, qu’il s’agisse d’une chaîne de caractères, d’un dictionnaire ou d’une liste. Les environnements de production ont tendance à transmettre un dictionnaire structuré contenant tout ce dont un réviseur a besoin, comme le brouillon, la classification, l’expéditeur et l’URL de révision, que la couche d’application peut afficher dans une interface utilisateur ou transmettre via Slack ou par e-mail.

Il existe encore un piège. Appeler à nouveau invoke() avec l’état initial complet à la place de Command(resume=value) est une erreur : app.invoke({"messages": [...]}, config=config) sur le même thread_id redémarre le graphe correspondant à ce thread et élimine l’état en pause. Un graphe interrompu ne doit jamais être repris que via Command(resume=value).

Projet final : un flux de travail de tri des e-mails

Lorsque toutes les composantes sont en place, elles peuvent être intégrées pour former un système de tri des e-mails de type production. Ce système lit un e-mail reçu, classe son intention grâce à des résultats structurés, recherche en parallèle dans une base de connaissances et un suivi de bogues, rédige une réponse, et peut s’arrêter pour une révision humaine avant d’envoyer.

Chaque concept présenté précédemment correspond à une partie de ce flux de travail :

• La structure de TypedDict avec des réducteurs apparaît sous le nom de EmailState, avec operator.add appliqué à search_results.
• Les arêtes sérielles relient read_email à classify_intent.
• Les arêtes parallèles permettent à classify_intent de se diviser en search_docs et bug_tracking.
• La synchronisation par arrivée de données signifie que les deux nœuds parallèles alimentent write_response.
• Le routage conditionnel est géré par route_by_priority, qui choisit entre une révision humaine et un envoi automatique.
• La sortie structurée générée par Pydantic est EmailClassification, qui encadre la réponse du LLM.
• L’interruption est représentée par human_review, qui pause le processus en attendant l’approbation.
• Le mécanisme de point de contrôle est MemorySaver, qui permet le cycle d’interruption et de reprise.

États et schémas

Le champ d’état et le schéma Pydantic pour la classification structurée viennent en premier :

from typing import Annotated
from typing_extensions import TypedDict
import operator
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import MemorySaver

# Pydantic model for structured classification
class EmailClassification(BaseModel):
    intent: str = Field(description="The intent: 'bug_report', 'question', 'feedback', or 'other'")
    priority: str = Field(description="Priority level: 'high', 'medium', or 'low'")
    confidence: float = Field(description="Confidence score between 0.0 and 1.0")

# Graph state
class EmailState(TypedDict):
    email_subject: str
    email_body: str
    email_sender: str
    classification: dict
    search_results: Annotated[list[str], operator.add]
    bug_info: str
    draft_response: str
    approved: bool
    status: str
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

EmailClassification est un modèle Pydantic et ne fait pas partie de EmailState, qui illustre la séparation entre TypedDict et Pydantic. Il s’agit de la structure transmise au LLM en tant que schéma JSON, et Pydantic valide la réponse en levant immédiatement une erreur si elle ne correspond pas. EmailState est le conteneur du graphe, un TypedDict car LangGraph a besoin de la syntaxe de réducteur ainsi que d’une serialization efficace.

classification est stocké sous forme de dict dans EmailState car, après que Pydantic ait validé la réponse, la méthode .model_dump() la convertit en un simple dict avant qu’elle ne soit enregistrée dans l’état. Cela permet à l’état de rester serialisable par n’importe quel checkpointer, sans nécessiter de dépendance à Pydantic au moment du restauration.

Le champ le plus important est search_results: Annotated[list[str], operator.add]. Comme search_docs et bug_tracking s’exécutent en parallèle et peuvent tous deux enregistrer des résultats, le réducteur empile leurs sorties ensemble, tandis que sans lui une branche écraserait silencieusement l’autre.

Création des nœuds

Les nœuds restent petits et à vocation unique. Chacun ne récupère que les valeurs dont il a besoin dans l’état, effectue sa tâche et renvoie une mise à jour partielle :

def read_email(state: EmailState) -> dict:
    """Parse the incoming email into structured fields."""
    return {
        "email_subject": state["email_subject"],
        "email_body": state["email_body"],
        "email_sender": state["email_sender"],
    }


def classify_intent(state: EmailState) -> dict:
    """Use structured output to classify the email."""
    structured_llm = llm.with_structured_output(EmailClassification)
    classification = structured_llm.invoke(
        f"Classify this email.\\n\\n"
        f"Subject: {state['email_subject']}\\n"
        f"Body: {state['email_body']}"
    )
    return {"classification": classification.model_dump()}

def search_docs(state: EmailState) -> dict:
    """Search knowledge base for relevant documentation."""
    intent = state["classification"]["intent"]
    # In production, this would hit a real vector store or search API
    response = llm.invoke(
        f"Suggest 2 relevant documentation links for a {intent} "
        f"about: {state['email_subject']}"
    )
    return {"search_results": [f"[Docs] {response.content}"]}

def bug_tracking(state: EmailState) -> dict:
    """Look up related bugs in the tracking system."""
    # In production, this would query Jira, Linear, GitHub Issues, etc.
    response = llm.invoke(
        f"Summarize any known bugs related to: {state['email_subject']}"
    )
    return {"bug_info": response.content}

def write_response(state: EmailState) -> dict:
    """Draft an email response using all gathered context."""
    context_parts = [
        f"Original email from {state['email_sender']}:",
        f"Subject: {state['email_subject']}",
        f"Body: {state['email_body']}",
        f"\\nClassification: {state['classification']}",
        f"\\nRelevant docs: {chr(10).join(state['search_results'])}",
        f"\\nBug info: {state['bug_info']}",
    ]
    context = "\\n".join(context_parts)
    response = llm.invoke(
        f"Draft a helpful, professional reply to this customer email.\\n\\n

À première vue, read_email semble inutile, car il ne renvoie que des champs déjà présents dans l’état, mais il sert de étape de parsing explicite. Une version en production prendrait un payload brut de webhook et en extrairait les champs, de sorte que cette fonction remplace cette opération d’extraction.

classify_intent est là que Pydantic se révèle utile. Dans structured_llm = llm.with_structured_output(EmailClassification), le LLM est enveloppé de manière à ce que les réponses respectent le schéma. Il reçoit un schéma JSON dérivé de EmailClassification et doit renvoyer un JSON valide correspondant ; LangChain tente automatiquement à nouveau avant de lever une erreur s’il ne le fait pas. .model_dump() convertit ensuite l’objet validé en un dictionnaire simple pour l’état.

Les deux nœuds parallèles écrivent dans des champs distincts, search_results et bug_info, de sorte que leurs appels à l’LLM s’exécutent en même temps. Pour search_results, search_docs renvoie une liste ([f"[Docs] {response.content}"]), car operator.add nécessite des entrées sous forme de liste. bug_tracking renvoie une chaîne de caractères simple pour bug_info, qui ne dispose d’aucun opérateur de réduction, donc une simple affectation suffit.

Le nœud de type fan-in est write_response. Lorsqu’il démarre, les deux nœuds parallèles ont déjà terminé leur travail et leurs résultats sont stockés dans l’état, ce qui lui permet de construire le prompt à partir de tout cela et d’élaborer la réponse.

human_review transmet un dictionnaire structuré à interrupt() et non une simple chaîne de caractères, ce qui constitue une bonne pratique en environnement de production. Ce dictionnaire contient le brouillon, la classification ainsi qu’un message destiné au réviseur, de sorte que le système qui se trouve de l’autre côté, qu’il s’agisse d’une interface web, d’un bot Slack ou d’une notification par e-mail, reçoit tout le contexte nécessaire.

send_reply utilise state.get("approved", True) comme valeur par défaut de sécurité ; ainsi, si la valeur approved fait défaut dans l’état du système, la réponse est envoyée malgré tout. Cette valeur par défaut peut être modifiée en False lorsque le profil de risque exige une approche plus prudente.

Dans un environnement de production, search_docs et bug_tracking appelleraient des systèmes réels tels qu’une base de données vectorielle, Jira ou GitHub Issues, et le décorateur @tool de LangChain facilite l’encapsulation de ces appels. La structure des nœuds reste identique, avec un appel à un outil remplaçant l’appel au LLM. Un autre piège : write_response lit state["search_results"] ; si cette liste est toujours vide, l’accès peut échouer. Cela peut être évité en initialisant search_results: [] dans le payload d’appel initial, afin que le réducteur commence à y ajouter des éléments.

Connexion des arêtes

Les arêtes séquentielles, parallèles et conditionnelles relient alors les nœuds :

def route_by_priority(state: EmailState) -> str:
    """Route based on priority and confidence."""
    classification = state["classification"]
    if classification["priority"] == "high" or classification["confidence"] < 0.8:
        return "needs_review"
    return "auto_send"


graph = StateGraph(EmailState)

# Add all nodes
graph.add_node("read_email", read_email)
graph.add_node("classify_intent", classify_intent)
graph.add_node("search_docs", search_docs)
graph.add_node("bug_tracking", bug_tracking)
graph.add_node("write_response", write_response)
graph.add_node("human_review", human_review)
graph.add_node("send_reply", send_reply)

# Serial edges
graph.add_edge(START, "read_email")
graph.add_edge("read_email", "classify_intent")

# Parallel fan-out from classify_intent
graph.add_edge("classify_intent", "search_docs")
graph.add_edge("classify_intent", "bug_tracking")

# Parallel fan-in to write_response
graph.add_edge("search_docs", "write_response")
graph.add_edge("bug_tracking", "write_response")

# Conditional routing after write_response
graph.add_conditional_edges(
    "write_response",
    route_by_priority,
    {"needs_review": "human_review", "auto_send": "send_reply"}
)

# After human review, send the reply
graph.add_edge("human_review", "send_reply")
graph.add_edge("send_reply", END)

# Compile with memory for interrupt support
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

route_by_priority envoie les e-mails à haute priorité à une révision humaine, quel que soit le niveau de confiance. Les classifications ayant un niveau de confiance inférieur à 0,8 sont également soumises à révision, car si le modèle n’est pas sûr du sujet d’un e-mail, c’est une personne qui doit trancher. Tout le reste est envoyé automatiquement, ce qui constitue une valeur par défaut raisonnable en environnement de production, à ajuster en fonction des taux d’erreur réels.

Cette routage passe par add_conditional_edges et non par Command, car route_by_priority est une fonction pure : elle ne fait que consulter l’état et renvoie une clé, sans créer de nouveaux données, ce qui correspond exactement à l’usage prévu pour add_conditional_edges.

Voici toutes les connexions dans le graphe :

• Une exécution séquentielle part de START, passe par read_email pour arriver à classify_intent.
• La sortie de classify_intent déclenche simultanément search_docs et bug_tracking.
• La entrée envoie les résultats de ces deux nœuds à write_response.
• Après write_response, un bord conditionnel choisit entre human_review et send_reply.
• Lorsque l’examen est terminé, human_review alimente send_reply, qui se termine à END.

Exécution du flux de travail de bout en bout

Un e-mail de exemple est d’abord envoyé, et la première invocation se poursuit jusqu’à l’interruption :

config = {"configurable": {"thread_id": "email-001"}}

# Invoke with a sample email
result = app.invoke({
    "email_subject": "Login page broken after update",
    "email_body": (
        "Hi, since yesterday's update I can't log into my account. "
        "I get a 500 error every time. This is blocking my team. "
        "Please fix ASAP."
    ),
    "email_sender": "frustrated-customer@example.com",
    "search_results": [],
    "bug_info": "",
    "draft_response": "",
    "approved": False,
    "status": "",
}, config=config)

# The graph will classify this as high-priority, so it pauses at human_review.
# The interrupt payload contains the draft for review.
print("Interrupt payload received. Draft response ready for review.")

À ce stade, le graphe a déjà effectué beaucoup de travail :

1. read_email a analysé le message.
2. classify_intent l’a classé comme un rapport de bug à haute priorité en utilisant une sortie structurée de Pydantic.
3. La recherche dans la documentation et la consultation du suivi des bugs se sont déroulées en parallèle au sein d’une seule étape.
4. write_response a rédigé une réponse en utilisant tout ce contexte.
5. Le nœud human_review a appelé interrupt() et a mis en pause le processus.

Tout le pipeline de traitement, y compris les deux appels en parallèle aux modèles de langage, s’est terminé avant la pause. Lorsque le graphe s’arrête, l’état est entièrement sauvegardé, permettant ainsi à l’examinateur de répondre en quelques minutes ou heures sans rien perdre.

L’examinateur approuve ensuite :

# Resume after human approval
final = app.invoke(Command(resume="approve"), config=config)
print(final["status"])
# "Reply sent to frustrated-customer@example.com"

LangGraph a chargé le point de contrôle pour "email-001", a repris le processus human_review à partir de son appel interrupt(), a renvoyé "approve" en tant que valeur de retour, a défini approved=True dans l’état et a poursuivi avec send_reply, qui a renvoyé l’état de confirmation.

Toutes les idées précédentes apparaissent dans cette solution finale : un état TypedDict soutenu par un réducteur, des arêtes sérielles et parallèles, un routage conditionnel, une sortie structurée via Pydantic, une révision humaine assurée par une interruption, ainsi que des données mémorisées grâce à un point de contrôle.

Pour un e-mail à faible priorité mais présentant un degré élevé de fiabilité, route_by_priority renvoie "auto_send" et la vérification par un humain est omise. Le processus s’exécute en parallèle : lecture, classification, recherche, rédaction et envoi, et l’interruption n’intervient que sur le chemin qui en a besoin. C’est là l’avantage du routage conditionnel.

Une dernière mise en garde concerne l’état initial. Contrairement à Pydantic, TypedDict ne définit pas de valeurs par défaut pour les champs manquants. Si write_response tente d’accéder à state["search_results"] et que cette clé n’était pas présente dans l’état initial, une erreur KeyError est générée. Chaque champ doit figurer dans le dictionnaire d’état initial, même s’il commence par une chaîne vide, une liste vide ou False.

Où aller ensuite

Le parcours décrit ici commence par un seul nœud et un TypedDict, et se termine par un flux de travail de type production capable d’exécuter des appels à des modèles de langage en parallèle, de valider les résultats structurés, de prendre des décisions conditionnelles, de surmonter les pannes grâce à des points de contrôle et d’attendre un réviseur. Plusieurs orientations sont possibles à partir de là :

• LangGraph Platform gère le déploiement, les points de contrôle automatisés et l’échelle horizontale, ce qui élimine la charge liée à l’exploitation de l’infrastructure lorsque l’objectif est de gérer un trafic réel.
• LangSmith traque et débogue chaque exécution de graphe et affiche l’état exact à chaque étape lorsque un nœud ne fonctionne pas correctement. Cela devient crucial lorsque les graphes comptent cinq nœuds ou plus avec des chemins conditionnels.
• Les sous-graphes composent des graphes plus petits en systèmes multi-agents plus importants. Chacun gère son propre état et communique via un graphe parent, ce qui constitue une étape naturelle après les flux de travail à un seul graphe.
• Les outils appelés via LangChain @tool et ToolNode permettent aux agents d’utiliser des API externes, des bases de données et des services directement depuis le graphe, et les nœuds phares search_docs et bug_tracking constituent des endroits évidents pour les tester.
• La persistance en production exige l’installation de langgraph-checkpoint-postgres afin de conserver un état fiable malgré les redémarrages, et le remplacement via l’API ne nécessite qu’une seule ligne de code.

Résumé rapide

• L’état est représenté par un TypedDict. Tout champ défini comme Annotated[list, operator.add] collecte les valeurs au fil des écritures, tandis qu’un champ sans fonction de réduction est remplacé à chaque écriture.
• Un nœud n’est rien d’autre qu’une fonction : elle reçoit l’état complet et renvoie un dictionnaire d’actualisation partielle, LangGraph s’occupant alors de la fusion des données.
• Les arêtes déterminent le flux. Utilisez des arêtes séquentielles lorsque l’ordre est important, des arêtes parallèles pour des tâches indépendantes exécutées simultanément, et des arêtes conditionnelles lorsque les données doivent choisir la branche à suivre.
• Utilisez un TypedDict pour représenter l’état du graphe, et employez Pydantic avec .with_structured_output() afin de définir précisément ce que le LLM renvoie.
• Les outils de sauvegarde assurent la persistance et permettent les interruptions : MemorySaver pendant le développement, PostgresSaver en production. Sans outil de sauvegarde, pas d’interruptions possibles.
• Une interruption n’est rien d’autre qu’un état sauvegardé et suspendu. interrupt(payload) arrête l’exécution et effectue une sauvegarde, tandis que Command(resume=value) redémarre exactement à cet endroit.
• Le parallélisme est intégré par défaut : deux arêtes partant d’un même nœud exécutent leurs cibles en même temps, et chaque champ écrit par ces nœuds nécessite un réducteur.

Questions à examiner

1. MemorySaver convient au développement et PostgresSaver convient à la production. Quels autres systèmes de stockage méritent d’être examinés pour un système à fort débit, et quels compromis permettraient de trancher le choix ?
2. route_by_priority envoie les e-mails à haute priorité à des fins de révision et envoie automatiquement les autres. Comment ce seuil pourrait-il être ajusté dans un système réel, et quels indicateurs permettraient de vérifier s’il est correctement défini ?
3. Le traitement final a lieu après write_response grâce à add_conditional_edges. Existe-t-il un scénario dans le flux de travail des e-mails où le pattern Command serait plus adapté qu’une fonction de routage distincte ?

Lectures complémentaires

• Notes pratiques : Construire un agent IA auto-correction avec Self-RAG et LangGraph — Guide pas à pas des notes pratiques : Construire un agent IA auto-correction avec Self-RAG et LangGraph : contrats, vérifications et emplacements de code prêts à l’emploi pour les équipes qui mettent en œuvre ce modèle.
• Construire un agent de recherche boursière avec ReAct et LangGraph — Un agent LangGraph ReAct prend un symbole boursier, consulte les données fondamentales, les actualités, les indicateurs techniques et les outils d’analystes dans son propre ordre, puis rédige un résumé, avec des pauses permettant l’intervention humaine.

Un flux de travail d’agent LLM local construit à partir de dossiers et de fichiers Markdown — Créez un pipeline d’agent local à étapes multiples avec des contrôles d’accès, où chaque étape correspond à un dossier, les prompts sont des fichiers Markdown modifiables, et un humain examine la sortie avant l’étape coûteuse.

Notes pratiques : Agent Text-to-SQL en Python : Tutoriel sur l’appel d’outils LLM — Guide pas à pas des Notes pratiques : Agent Text-to-SQL en Python : Tutoriel sur l’appel d’outils LLM : contrats, vérifications et emplacements de code prêts à l’emploi pour les équipes qui implémentent ce modèle.

Notes pratiques : Pourquoi personne ne parle de LangChain, LangGraph ou AutoAgent — Guide pas à pas des Notes pratiques : Pourquoi personne ne parle de LangChain, LangGraph ou AutoAgent : contrats, vérifications et emplacements de code prêts à l’emploi pour les équipes qui utilisent ce modèle.

Équipe multi-agent auto-réparatrice dans LangGraph : Architecte, Testeur, Débogueur — Créez une machine à états LangGraph en Python où un Architecte, un Développeur, un environnement de test pytest, un Débogueur et un Réviseur collaborent et itèrent jusqu’à ce que les tests réussissent.

Notes pratiques : Création d’un cycle de revue de code à deux agents avec A2A, LangGraph, et — Guide pas à pas des Notes pratiques : Création d’un cycle de revue de code à deux agents avec A2A, LangGraph, et : contrats, vérifications, ainsi que des emplacements de code prêts à l’emploi pour les équipes utilisant ce modèle.