LangGraph con intervención humana: interrupt(), ediciones de estado y viaje en el tiempo
LangGraph con intervención humana: aprobar llamadas a herramientas mediante interrupt(), agregar mecanismos de retroalimentación, editar el estado manualmente y rebobinar con viajes en el tiempo, con ejemplos funcionales.
Fuentes: LangGraph Human-in-the-Loop: Pausing, Reviewing, and Rewinding Your Agent
Todo lo que se explica a continuación supone que conoces la estructura estándar de siete partes de un archivo LangGraph (importaciones, estado, herramientas, nodos, aristas, ensamblaje, punto de entrada) y que has utilizado sistemas de puntos de control con un thread_id. Cada patrón aquí se basa en un punto de control; sin él, nada funciona.
Por qué es necesario que un humano esté involucrado
En el momento en que un LLM puede actuar, ya sea enviando un correo electrónico, eliminando un archivo, cargando una tarjeta o enviando código, también puede cometer errores costosos a velocidad de máquina mientras nadie lo observa. La intervención humana no es una función adicional añadida al agente; es el diseño que permite que los agentes con herramientas reales sean seguros para su uso, y LangGraph fue creado teniendo ese problema en mente. Su sistema de puntos de control existe principalmente para hacer posibles las pausas.
Hay cuatro patrones que abarcan la mayoría de los sistemas reales, enumerados según la frecuencia con la que aparecen:
- Detenerse antes de realizar una acción arriesgada con
interrupt(), que es el estándar actual. - Un punto de control de revisión garantizado dentro de un flujo de trabajo.
- Editar el estado mientras está en pausa, para corregir al agente antes de que continúe.
- Viaje en el tiempo: reproducir y ramificarse desde cualquier punto anterior en una conversación.
Algunos elementos se han omitido intencionadamente porque rara vez aparecen en los repositorios de código reales: los mecanismos internos de transmisión a nivel de token, el SDK completo de LangGraph Studio y la excepción antigua NodeInterrupt, que fue reemplazada por interrupt().
Interrumpir y reanudar
Cada patrón se reduce a los mismos dos pasos:
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.
Todo lo demás es variación: dónde se llama a interrupt(), qué datos se le pasan y qué se hace con la respuesta.
Vocabulario
interrupt(value) se importa de langgraph.types y se llama desde dentro de un nodo. Detiene el grafo en ese punto, guarda todo el estado a través del checkpointer y entrega value a quien inició la ejecución. La pausa es real; no hay hilos que funcionen en segundo plano ni recursos retenidos, por lo que un hilo puede permanecer en pausa durante meses.
Command(resume=value) se utiliza en graph.invoke() o graph.stream() para reanudar un grafo en pausa. El value se convierte en el valor de retorno de la llamada a interrupt(), como si interrupt() fuera una función input() síncrona que acabara de recibir una respuesta.
__interrupt__ es el elemento que aparece en la salida cuando la ejecución se ha detenido. Busque este valor para saber si el agente se ha detenido y necesita entrada.
El checkpointer (MemorySaver, SqliteSaver, PostgresSaver) es obligatorio. Sin él, no existe ningún estado guardado del que poder pausar o reanudar. Olvidarse de compilar el grafo con uno de estos elementos es la causa más frecuente de que falle el código HITL de los principiantes.
Patrón 1: aprobar una llamada a una herramienta antes de que se ejecute
Es, con diferencia, el uso más común en entornos de producción. El agente decide llamar a una herramienta que tiene efectos reales, como enviar un correo electrónico, consultar una base de datos o realizar un pedido, y una persona debe aprobarla primero.
Imagínese a un gerente que firma como co-firmante un cheque. El empleado (el LLM) lo completa y se lo entrega, y nada avanza hasta que el gerente (el humano) lo firma. Si el gerente se niega, el cheque se invalida.
El nodo de reseña:
# ── 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"]
)
]
}
Luego conéctelo al gráfico:
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)
El enrutamiento difiere ligeramente del agente básico: una solicitud de herramienta ahora se dirige a review_tool_call antes del nodo de herramientas. La elección del agente y la ejecución real cuentan con la intervención del revisor, que representa el paso de co-firmación antes del cobro.
Ejecutarlo y reanudarlo:
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']}")
]
}
Y el cableado del bucle:
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())
Este es el ciclo de revisión y corrección que se encuentra en casi todos los agentes de contenido, revisión de código o investigación: generar, hacer una pausa para que una persona intervenga, revisar si es necesario, pausar de nuevo y repetir hasta que se apruebe.
Patrón 3: corregir el estado manualmente
A veces una persona no quiere aprobar ni rechazar. Quiere reparar algo en el estado del agente antes de que continúe, porque el agente estaba ligeramente desviado y volver a intentarlo es más lento que corregirlo.
Aquí son importantes tres llamadas:
graph.get_state(config)devuelve elStateSnapshotactual, con.values(el diccionario de estado) y.next(el nodo que se ejecutaría al reanudar).graph.update_state(config, values)escribe nuevos valores en el punto de control actual. Crea un nuevo punto de control y mantiene la historia intacta, lo cual es importante más adelante para los viajes en el tiempo.as_node="node_name"es un argumento opcional deupdate_stateque indica a LangGraph que considere la actualización como proveniente de ese nodo. LangGraph elige los siguientes enlaces en función del nodo de origen de la última actualización, por lo que un error al especificarlo puede hacer que el grafo continúe en un lugar inesperado.
En la práctica:
# ── 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)
Dos aspectos causan confusión. Primero, interrupt_before=["node_name"] se establece al compilar y siempre hace una pausa antes de ese nodo en cada ejecución, mientras que interrupt() se encuentra dentro de un nodo y solo hace una pausa cuando así lo indica la lógica. Utilice interrupt_before para depuración sencilla o paradas garantizadas, y interrupt() siempre que la pausa dependa de condiciones en tiempo de ejecución, lo cual cubre la mayoría de las aplicaciones reales.
Segundo, graph.invoke(None, config) significa “no agregar nuevas entradas, continuar desde el último punto de control”. Esto es diferente de Command(resume=...), que responde a un interrupt() pendiente. Después de usar interrupt_before junto con update_state, continúe con None. Después de interrupt(), continúe con Command(resume=...).
Patrón 4: viaje en el tiempo
El historial de puntos de control funciona como los commits en Git. Puedes consultar cualquier estado anterior y crear una rama a partir de él para generar una línea de tiempo alternativa sin alterar la original. Otra comparación: los puntos de guardado en un videojuego. get_state_history enumera todos ellos, y puedes cargar cualquiera de ellos, no solo el más reciente.
graph.get_state_history(config)es un generador que devuelve cadaStateSnapshotguardado para el hilo, del más reciente al más antiguo. Cada uno lleva su propioconfigcon uncheckpoint_idúnico.checkpoint_idse encuentra ensnapshot.config["configurable"]["checkpoint_id"]. Pasa un configuración que contenga uno específico ainvokeoupdate_statey la operación se dirigirá a ese punto, no al más reciente.
graph.invoke(None, config_with_old_checkpoint_id). Esto vuelve a ejecutar el proceso a partir de ese punto. Se reutilizan los puntos de control ya almacenados para esa rama en lugar de volver a calcularlos, aunque la lógica de los nodos a partir de ese punto se ejecuta nuevamente.update_state sobre un punto de control antiguo, lo que crea un nuevo punto de control y una nueva línea de tiempo junto con la primera.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 mayoría de los chatbots no necesitan esto en tiempo de ejecución. Su valor se manifiesta en tres aspectos: depuración: reproducir exactamente lo que ocurrió en una ejecución fallida para identificar dónde se produjo el error; evaluación: volver a ejecutar la misma conversación con otro prompt o modelo y comparar resultados, lo que equivale a pruebas A/B sin tener que reproducir todo desde cero; recuperación de errores: cuando un nodo falla a mitad de proceso, retroceder hasta justo antes de ese punto, corregir la causa y continuar en lugar de reiniciar toda la conversación.
Un agente completo con aprobación de herramientas
Aquí hay un template completo que combina la estructura básica, la memoria y la aprobación para el uso de herramientas. Representa la forma realista de un agente en producción con una red de seguridad humana.
# ============================================================
# 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}")
Referencia rápida
Pausa y reanudación. interrupt(value) pausa la ejecución dentro de un nodo, guarda su estado y pasa value al llamante, para luego devolver el valor de reanudación más tarde. Command(resume=value) responde a una interrupción pendiente. „__interrupt__“ es la clave de salida que indica una pausa, y result[„__interrupt__“][0].value muestra qué fue devuelto. interrupt_before=[„node_name“] siempre realiza una pausa antes del nodo especificado.
Estado de edición. graph.get_state(config) devuelve un StateSnapshot (.values, .next, .config). graph.update_state(config, values) agrega un punto de control sin borrar el historial. as_node determina qué aristas se seguirán al reanudar. graph.invoke(None, config) continúa sin nueva entrada.
Viaje en el tiempo. graph.get_state_history(config) muestra los snapshots del más reciente al más antiguo. checkpoint_id identifica un punto de guardado. La reproducción continúa una línea de tiempo a partir de un punto de control antiguo, mientras que una rama crea uno nuevo.
Qué patrón es adecuado
- Herramientas que gastan dinero, envían mensajes o modifican sistemas externos: aprobación con
interrupt(). En el caso de agentes de producción con efectos secundarios reales, considérelo obligatorio. - Un paso de contenido que siempre debe revisarse antes de que avance el flujo de trabajo: el nodo dedicado a los comentarios con un bucle de revisión.
- Una persona que corrige o agrega información durante la conversación sin un paso formal de aprobación:
interrupt_beforejunto conupdate_state. - Depuración de una ejecución fallida, pruebas de cambios en los prompts o recuperación de un error sin reiniciar: viaje en el tiempo mediante
get_state_history.
La seguridad es la forma del grafo
La memoria, la resumen, la persistencia y la supervisión humana se expresan todas como estructura en LangGraph: un nodo aquí, un edge allá, un campo en el estado. No existe un modo de seguridad especial que activar. La seguridad es otra forma que puede adoptar el grafo.
Los agentes rara vez fallan en producción debido a una mala decisión del LLM, lo cual es esperable y recuperable. Fallan cuando una mala decisión no cuenta con un punto de control entre la decisión y sus consecuencias. interrupt() es ese punto de control. Requiere un nodo adicional y un edge adicional, y puede ser el nodo más valioso en el grafo.
Con una estructura estándar, una forma de mantener las conversaciones a un costo razonable y persistentes, además de una red de seguridad alrededor de todo lo importante, se cuenta con la base para casi todas las aplicaciones serias de LangGraph. Los temas siguientes naturales son la orquestación multiagente con supervisores y subgrafos, así como la memoria a largo plazo como los perfiles de usuario entre sesiones, ambos basados en el estado, los nodos y los puntos de control.
Leer más
- MCP para agentes de LangGraph, nivel por nivel: servidores, transportes y el puente adaptador — Aprenda el Protocolo de Contexto de Modelo desde las roles de host, cliente y servidor hasta un servidor FastMCP, stdio vs HTTP, langchain-mcp-adapters y los problemas comunes en la producción con LangGraph.
- LangGraph explicado: grafos, estado, enrutamiento y patrones de producción — Cómo LangGraph modela a los agentes como grafos cíclicos con estado tipado, reductores, puntos de control, enrutamiento, herramientas y bucles ReAct, además de patrones para su uso en producción.