LangGraph avec intervention humaine : interrupt(), modifications de l’état et voyage dans le temps
LangGraph avec un humain dans la boucle : validation des appels d’outils via interrupt(), ajout de mécanismes de retour d’information, modification manuelle de l’état et réinitialisation grâce au voyage dans le temps, accompagnés d’exemples concrets.
Sources: LangGraph Human-in-the-Loop: Pausing, Reviewing, and Rewinding Your Agent
Tout ce qui suit suppose que vous connaissez la structure standard en sept parties d’un fichier LangGraph (imports, état, outils, nœuds, arêtes, assemblage, point d’entrée) et que vous avez utilisé des outils de sauvegarde avec un thread_id. Chacun des modèles présentés ici s’appuie sur un tel outil de sauvegarde ; sans lui, rien ne fonctionne.
Pourquoi un être humain doit rester impliqué
Dès qu’un LLM peut agir, qu’il s’agisse d’envoyer un e-mail, de supprimer un fichier, de passer une carte ou de déployer du code, il peut également commettre des erreurs coûteuses à une vitesse mécanique, sans que personne ne le surveille. La présence d’un être humain dans le processus (Human-in-the-Loop ou HITL) n’est pas une fonction supplémentaire ajoutée à un agent ; c’est plutôt la conception qui permet de rendre les agents dotés d’outils du monde réel sûrs à déployer. LangGraph a été conçu en tenant compte de ce problème. Son système de sauvegarde existe principalement pour permettre des pauses.
Quatre modèles couvrent la plupart des systèmes réels, classés selon leur fréquence d’apparition :
- Pause avant une action risquée à l’aide de
interrupt(), la méthode standard actuelle. - Un point de contrôle de révision garanti au sein d’un flux de travail.
- Modification de l’état pendant la pause, afin de corriger l’agent avant qu’il ne continue.
- Voyage dans le temps : rejouer et bifurquer à partir de n’importe quel point antérieur d’une conversation.
Quelques éléments ont été intentionnellement omis car ils apparaissent rarement dans des bases de code réelles : les mécanismes de streaming au niveau des tokens, le SDK complet de LangGraph Studio ainsi que l’exception plus ancienne NodeInterrupt, que interrupt() a remplacée.
Interrompre et reprendre
Chaque schéma se résume à la même séquence en deux étapes :
1. interrupt() → Graph pauses. State is saved. Control returns to your code.
2. Command(resume=...) → Graph picks up exactly where it paused, with your input.
Tout le reste relève de variations : l’endroit où l’on appelle interrupt(), les données que l’on lui transmet et ce que l’on fait avec la réponse.
Vocabulaire
interrupt(value) est importé depuis langgraph.types et appelé à l’intérieur d’un nœud. Il arrête immédiatement le graphe, enregistre l’ensemble de l’état via le checkpointer et transmet value à celui qui a lancé l’exécution. La pause est réelle : aucune thread ne tourne en arrière-plan et aucun ressource n’est bloquée, ce qui permet à une thread de rester en pause pendant des mois.
Command(resume=value) est utilisé dans graph.invoke() ou graph.stream() pour réactiver un graphe en pause. La valeur value devient alors la valeur de retour de l’appel à interrupt(), comme si interrupt() était une fonction input() synchrone qui venait de recevoir une réponse.
__interrupt__ est l’élément qui apparaît dans la sortie lorsque l’exécution est suspendue. Vérifiez sa présence pour savoir si l’agent s’est arrêté et attend des entrées.
Le checkpointer (MemorySaver, SqliteSaver, PostgresSaver) est obligatoire. Sans lui, il n’y a pas d’état enregistré permettant de suspendre ou de reprendre l’exécution. Oublier de compiler le graphe avec un tel checkpointer est la raison la plus fréquente d’échec du code HITL chez les débutants.
Modèle 1 : approuver une appel d’outil avant son exécution
C’est de loin l’usage le plus courant en environnement de production. L’agent décide d’appeler un outil qui a des effets réels, tels que l’envoi d’un e-mail, la consultation d’une base de données ou la passation d’une commande, et une personne doit d’abord l’approuver.
Imaginez un manager qui co-signe un chèque : l’employé (le LLM) le remplit et le lui donne, et rien ne se passe tant que le manager (l’être humain) ne l’a pas signé. Si le manager refuse, le chèque devient nul.
Le nœud d’avis :
# ── IMPORTS needed ───────────────────────────────────────────
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import MemorySaver
# ── MODULE 4: NODES ───────────────────────────────────────────
def review_tool_call(state: AgentState) -> dict:
"""Pauses execution to let a human approve, edit, or reject a tool call
before it actually runs."""
last_message = state["messages"][-1]
tool_call = last_message.tool_calls[0] # Assume one tool call for simplicity
# interrupt() pauses HERE. The dict below is shown to whoever resumes the graph.
human_response = interrupt({
"question": "Approve this action?",
"tool_name": tool_call["name"],
"tool_args": tool_call["args"],
})
# Execution resumes here, with human_response being whatever
# was passed to Command(resume=...)
if human_response["type"] == "approve":
# Do nothing - let the tool call proceed as-is
return {}
elif human_response["type"] == "edit":
# Replace the tool call's arguments with the human's edits
last_message.tool_calls[0]["args"] = human_response["edited_args"]
return {"messages": [last_message]}
elif human_response["type"] == "reject":
# Short-circuit: skip the tool entirely, tell the LLM why
return {
"messages": [
ToolMessage(
content="User rejected this action.",
tool_call_id=tool_call["id"]
)
]
}
Ensuite, le connecter au graphe :
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("review_tool_call", review_tool_call)
graph_builder.add_node("tools", tool_node)
graph_builder.set_entry_point("agent")
# Agent decides: call a tool (go through review) or finish
graph_builder.add_conditional_edges(
"agent",
should_continue,
{"tools": "review_tool_call", "__end__": END}
)
graph_builder.add_edge("review_tool_call", "tools")
graph_builder.add_edge("tools", "agent")
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
Le routage diffère légèrement de celui de l’agent de base : une demande d’outil est désormais acheminée vers review_tool_call avant d’arriver au nœud des outils. Le choix de l’agent et l’exécution réelle sont intercalés par l’examinateur, qui représente l’étape de co-signature avant le paiement.
Le lancer et la reprise :
config = {"configurable": {"thread_id": "session-001"}}
# ── First call: runs until interrupt() pauses it ─────────────
result = graph.invoke(
{"messages": [HumanMessage(content="Send an email to the team about the delay")]},
config=config
)
# Check if the graph paused
if "__interrupt__" in result:
interrupt_data = result["__interrupt__"][0].value
print(f"Agent wants to: {interrupt_data['tool_name']}")
print(f"With arguments: {interrupt_data['tool_args']}")
# Show this to a human (in a real app: render a UI button, Slack message, etc.)
user_decision = input("Approve? (yes/no): ")
# ── Resume with the human's decision ──────────────────────
if user_decision.lower() == "yes":
final_result = graph.invoke(
Command(resume={"type": "approve"}),
config=config
)
else:
final_result = graph.invoke(
Command(resume={"type": "reject"}),
config=config
)
print(final_result["messages"][-1].content)
That's the whole pattern. What changes elsewhere is the content passed to interrupt() and what you do with the answer that comes back.
Pattern 2: a mandatory feedback gate
Sometimes you don't want a conditional pause depending on what the LLM chose. You want a pause at a fixed stage, every time. A content-generation agent that always stops for review before publishing is the usual example, regardless of what it wrote.
Think of a quality-control station on an assembly line. Each item stops there, no exceptions. The worker can pass it as is, send it back for rework or adjust it directly.
The feedback node:
def human_feedback(state: AgentState) -> dict:
"""A guaranteed pause point. Shows the human the latest draft
and lets them approve, request changes, or edit directly."""
draft = state["messages"][-1].content
feedback = interrupt({
"question": "Review this draft. Approve, or provide edit instructions.",
"draft": draft,
})
if feedback.get("approved"):
return {} # Pass through unchanged
# Feedback contains revision instructions — feed back to the agent
return {
"messages": [
HumanMessage(content=f"Please revise based on this feedback: {feedback['notes']}")
]
}
Et le câblage du circuit bouclé :
graph_builder = StateGraph(AgentState)
graph_builder.add_node("writer", writer_node)
graph_builder.add_node("human_feedback", human_feedback)
graph_builder.set_entry_point("writer")
graph_builder.add_edge("writer", "human_feedback")
# After feedback, loop back to writer for revisions, OR end if approved
graph_builder.add_conditional_edges(
"human_feedback",
lambda state: "writer" if state["messages"][-1].content.startswith("Please revise") else "__end__",
{"writer": "writer", "__end__": END}
)
graph = graph_builder.compile(checkpointer=MemorySaver())
C’est le cycle d’examen et de révision que l’on retrouve dans presque tous les agents de gestion de contenu, d’examen de code ou de recherche : générer, faire une pause pour une personne, réviser si nécessaire, faire à nouveau une pause, et répéter jusqu’à l’approbation.
Modèle 3 : corriger manuellement l’état
Parfois, une personne ne souhaite ni approuver ni rejeter. Elle veut corriger quelque chose dans l’état de l’agent avant qu’il ne continue, car l’agent était légèrement défectueux et relancer les opérations est plus lent que de le corriger.
Trois appels sont importants ici :
graph.get_state(config)renvoie leStateSnapshotactuel, avec.values(le dictionnaire d’état) et.next(le nœud qui sera exécuté à la reprise).graph.update_state(config, values)enregistre de nouvelles valeurs au point de contrôle actuel. Il crée un nouveau point de contrôle tout en conservant l’historique, ce qui est utile par la suite pour les voyages dans le temps.as_node="node_name"est un argument optionnel deupdate_statequi indique à LangGraph de considérer que la mise à jour provient de ce nœud. LangGraph choisit les arêtes suivantes en fonction du nœud source de la dernière mise à jour ; une erreur à cet égard peut faire reprendre le traitement du graphe à un endroit inattendu.
En pratique :
# ── Pause first (using interrupt_before at compile time) ────
graph = graph_builder.compile(
checkpointer=memory,
interrupt_before=["agent"] # Always pause before the agent node runs
)
config = {"configurable": {"thread_id": "session-002"}}
# Run until the breakpoint
result = graph.invoke({"messages": [HumanMessage(content="What's 5 + 5?")]}, config)
# ── Inspect what's about to happen ───────────────────────────
snapshot = graph.get_state(config)
print(f"About to run: {snapshot.next}") # e.g. ('agent',)
print(f"Current messages: {snapshot.values['messages']}")
# ── Directly correct the state ───────────────────────────────
# Suppose the human wants to inject a clarification before the agent responds
graph.update_state(
config,
{"messages": [HumanMessage(content="Actually, I meant 5 * 5, not 5 + 5")]},
)
# ── Resume - agent now sees the corrected message ────────────
final = graph.invoke(None, config) # None = "just resume, no new input"
print(final["messages"][-1].content)
Deux distinctions peuvent prêter à confusion. Premièrement, interrupt_before=["node_name"] est défini lors de la compilation et provoque toujours une pause avant ce nœud à chaque exécution, tandis que interrupt() se trouve à l’intérieur d’un nœud et ne provoque une pause que lorsque votre logique le demande. Utilisez interrupt_before pour un débogage simple ou pour des arrêts garantis, et interrupt() chaque fois que la pause dépend de conditions en temps de exécution, ce qui couvre la plupart des applications réelles.
Deuxièmement, graph.invoke(None, config) signifie « ne pas ajouter de nouvelle entrée, reprendre à partir du dernier point de contrôle ». Cela diffère de Command(resume=...), qui répond à un interrupt() en attente. Après avoir utilisé interrupt_before suivi de update_state, reprenez l’exécution avec None. Après un interrupt(), reprenez l’exécution avec Command(resume=...).
Pattern 4 : voyage dans le temps
L’historique des points de contrôle fonctionne comme les commits dans Git. Vous pouvez consulter n’importe quel état antérieur et en créer une branche pour élaborer une timeline alternative sans perturber l’original. Une autre comparaison : les points de sauvegarde dans un jeu vidéo. get_state_history liste tous ces points, et vous pouvez en charger n’importe lequel, pas seulement le dernier.
graph.get_state_history(config)est un générateur qui produit chaqueStateSnapshotenregistré pour le thread, du plus récent au plus ancien. Chacun contient son propreconfigavec uncheckpoint_idunique.checkpoint_idse trouve à l’emplacementsnapshot.config[“configurable”][“checkpoint_id”]. Transmettez unconfigcontenant uncheckpoint_idspécifique àinvokeouupdate_stateafin que l’opération cible ce point précis, et non le plus récent.
graph.invoke(None, config_with_old_checkpoint_id). Cela exécute à nouveau le processus à partir de ce point. Les points de contrôle déjà enregistrés pour cette branche sont réutilisés plutôt que recalculés, bien que la logique des nœuds suivants soit exécutée à nouveau.update_state sur un ancien point de contrôle, ce qui crée un nouveau point de contrôle ainsi qu’une nouvelle chronologie parallèlement à la première.config = {"configurable": {"thread_id": "session-003"}}
# Run a few turns normally...
graph.invoke({"messages": [HumanMessage(content="My name is Sam")]}, config)
graph.invoke({"messages": [HumanMessage(content="What's my name?")]}, config)
# ── Find a past checkpoint ────────────────────────────────────
history = list(graph.get_state_history(config))
for snapshot in history:
print(f"Step {snapshot.metadata['step']}: {len(snapshot.values['messages'])} messages")
# Pick an earlier checkpoint - say, right after "My name is Sam"
target_checkpoint = history[-2] # Adjust index based on what you printed above
target_config = target_checkpoint.config
# ── Branch: update state AT that old checkpoint ──────────────
# This creates a NEW checkpoint that forks from the old one.
# The original conversation history is untouched.
branch_config = graph.update_state(
target_config,
{"messages": [HumanMessage(content="Actually, my name is Alex")]},
)
# ── Continue from the new branch ──────────────────────────────
result = graph.invoke(None, branch_config)
print(result["messages"][-1].content)
# Agent now responds based on the BRANCHED history (name = Alex),
# while the original thread (name = Sam) still exists if you go back to it.
La plupart des chatbots n’ont pas besoin de cela en temps de exécution. Sa valeur réside en trois domaines : le débogage, qui permet de rejouer exactement ce qui s’est passé lors d’une exécution échouée afin de déterminer l’endroit où cela a échoué ; l’évaluation, qui consiste à relancer la même conversation avec un autre prompt ou modèle pour comparer les résultats, ce qui constitue un test A/B sans avoir à rejouer tout le processus ; et la récupération en cas d’erreur, qui permet de revenir juste avant qu’un nœud ne tombe en panne, de corriger la cause et de reprendre la conversation au lieu de tout recommencer.
Un agent complet avec validation des outils
Voici un modèle complet intégrant la structure de base, la mémoire ainsi que la validation des appels aux outils. C’est la forme réelle d’un agent de production disposant d’une sécurité humaine.
# ============================================================
# LANGGRAPH AGENT WITH HUMAN-IN-THE-LOOP TOOL APPROVAL
# Extends: Part 1 (core structure) + Part 2 (memory)
# ============================================================
# ── MODULE 1: IMPORTS & CONFIGURATION ───────────────────────
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage, BaseMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
llm = ChatOpenAI(model="gpt-4o", temperature=0)
# ── MODULE 2: STATE ──────────────────────────────────────────
class State(MessagesState):
pass # messages field is inherited; add custom fields if needed
# ── MODULE 3: TOOLS ──────────────────────────────────────────
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""Send an email. This is a SENSITIVE action requiring human approval."""
return f"Email sent to {to} with subject '{subject}'"
tools = [send_email]
llm_with_tools = llm.bind_tools(tools)
tool_node = ToolNode(tools)
# ── MODULE 4: NODES ──────────────────────────────────────────
def agent_node(state: State) -> dict:
messages = [SystemMessage(content="You are a helpful assistant.")] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response]}
def review_tool_call(state: State) -> dict:
"""Pauses for human approval before any tool executes."""
last_message = state["messages"][-1]
tool_call = last_message.tool_calls[0]
decision = interrupt({
"tool_name": tool_call["name"],
"tool_args": tool_call["args"],
})
if decision["type"] == "reject":
return {
"messages": [
ToolMessage(
content="Action rejected by user.",
tool_call_id=tool_call["id"]
)
]
}
if decision["type"] == "edit":
last_message.tool_calls[0]["args"] = decision["edited_args"]
return {"messages": [last_message]}
return {} # approved - pass through unchanged
# ── MODULE 5: ROUTING ─────────────────────────────────────────
def should_continue(state: State) -> Literal["review_tool_call", "__end__"]:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "review_tool_call"
return "__end__"
def after_review(state: State) -> Literal["tools", "agent"]:
"""If the tool call was rejected, the last message is a ToolMessage -
skip the tools node and go straight back to the agent."""
last_message = state["messages"][-1]
if isinstance(last_message, ToolMessage):
return "agent"
return "tools"
# ── MODULE 6: GRAPH ASSEMBLY ──────────────────────────────────
graph_builder = StateGraph(State)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("review_tool_call", review_tool_call)
graph_builder.add_node("tools", tool_node)
graph_builder.add_edge(START, "agent")
graph_builder.add_conditional_edges(
"agent", should_continue,
{"review_tool_call": "review_tool_call", "__end__": END}
)
graph_builder.add_conditional_edges(
"review_tool_call", after_review,
{"tools": "tools", "agent": "agent"}
)
graph_builder.add_edge("tools", "agent")
graph = graph_builder.compile(checkpointer=MemorySaver())
# ── MODULE 7: ENTRYPOINT ──────────────────────────────────────
if __name__ == "__main__":
config = {"configurable": {"thread_id": "session-001"}}
result = graph.invoke(
{"messages": [HumanMessage(content="Email bob@example.com that the meeting moved to 3pm")]},
config
)
while "__interrupt__" in result:
interrupt_info = result["__interrupt__"][0].value
print(f"\n Approval needed:")
print(f" Tool: {interrupt_info['tool_name']}")
print(f" Args: {interrupt_info['tool_args']}")
choice = input("Approve (a) / Edit (e) / Reject (r)? ").strip().lower()
if choice == "a":
result = graph.invoke(Command(resume={"type": "approve"}), config)
elif choice == "e":
new_to = input("New 'to' address: ")
result = graph.invoke(
Command(resume={
"type": "edit",
"edited_args": {**interrupt_info["tool_args"], "to": new_to}
}),
config
)
else:
result = graph.invoke(Command(resume={"type": "reject"}), config)
print(f"\nAgent: {result['messages'][-1].content}")
Référence rapide
Pause et reprise. interrupt(value) pause l’exécution à l’intérieur d’un nœud, enregistre l’état actuel et transmet value à l’appelant, avant de restituer la valeur de reprise ultérieurement. Command(resume=value) traite une interruption en attente. "__interrupt__" est la clé de sortie indiquant une pause, et result["__interrupt__"][0].value affiche ce qui a été restitué. interrupt_before=["node_name"] provoque toujours une pause avant le nœud spécifié.
État de modification. graph.get_state(config) renvoie un StateSnapshot (.values, .next, .config). graph.update_state(config, values) ajoute un point de contrôle sans effacer l’historique. as_node influence les arêtes suivies lors de la reprise. graph.invoke(None, config) continue sans nouvelle entrée.
Voyage dans le temps. graph.get_state_history(config) affiche les snapshots du plus récent au plus ancien. checkpoint_id identifie un point de sauvegarde. La réexécution reprend une chronologie à partir d’un ancien point de contrôle, tandis qu’une branche en crée un nouveau.
Quel modèle convient
- Outils qui dépensent de l’argent, envoient des messages ou modifient des systèmes externes : approbation avec
interrupt(). Pour les agents de production ayant des effets secondaires réels, considérez cela comme obligatoire. - Une étape de contenu qui doit toujours être revue avant que le flux de travail ne progresse : le nœud dédié aux retours avec une boucle de révision.
- Une personne qui corrige ou ajoute des informations au cours d’une conversation sans étape d’approbation formelle :
interrupt_beforeassocié àupdate_state. - Le débogage d’une exécution échouée, la mise à l’épreuve de modifications de prompts ou la récupération après une erreur sans redémarrage : un voyage dans le temps grâce à
get_state_history.
La sécurité est la forme du graphe
La mémoire, la synthèse, la persistance et la supervision humaine s’expriment toutes sous forme de structure dans LangGraph : un nœud ici, un bord là, un champ dans l’état. Il n’existe pas de mode de sécurité spécifique à activer. La sécurité est simplement une autre forme que peut prendre le graphe.
Les agents échouent rarement en production à cause d’une mauvaise décision de l’LLM, ce qui est prévisible et réparable. Ils échouent lorsque cette mauvaise décision n’a pas de point de contrôle entre la décision elle-même et ses conséquences. interrupt() constitue ce point de contrôle. Il nécessite un nœud supplémentaire et un bord supplémentaire, et il peut être le nœud le plus important du graphe.
Avec une structure standard, un moyen de maintenir les conversations abordables et persistantes, ainsi qu’un filet de sécurité autour de tout ce qui est important, on dispose du cadre de base pour presque toutes les applications sérieuses LangGraph. Les sujets suivants naturels sont l’orchestration multi-agents avec des superviseurs et des sous-graphes, ainsi que la mémoire à long terme telle que les profils d’utilisateurs entre sessions, les deux s’appuyant sur l’état, les nœuds et les points de contrôle.
Lectures complémentaires
- MCP pour les agents LangGraph, étape par étape : serveurs, transports et le pont d’adaptation — Découvrez le Model Context Protocol en abordant les rôles hôte, client et serveur, un serveur FastMCP, stdio versus HTTP, langchain-mcp-adapters ainsi que les pièges courants en environnement de production dans LangGraph.
- LangGraph expliqué : graphes, état, routage et modèles de production — Comment LangGraph modélise les agents sous forme de graphes cycliques avec un état typé, des réducteurs, des points de contrôle, un routage, des outils et des boucles ReAct, ainsi que les modèles adaptés à un usage en production.