LangGraph人工干预模式：interrupt()、状态编辑和时间回溯。
LangGraph人工干预机制：通过interrupt()批准工具调用，添加反馈机制，手动编辑状态并通过时间回溯功能恢复，附有实际应用示例。
来源: LangGraph Human-in-the-Loop: Pausing, Reviewing, and Rewinding Your Agent
以下内容均假设您已了解 LangGraph 文件标准的七部分结构（导入、状态、工具、节点、边、组装、入口点），并且使用过带有 thread_id 的检查点机制。这里的每种模式都是建立在检查点机制之上的，没有检查点机制，这一切都无法运作。
为何需要人工介入
一旦大型语言模型能够执行操作，比如发送邮件、删除文件、扣款或推送代码，它就可能在无人监督的情况下以机器速度犯下代价高昂的错误。人工介入机制并非附加在智能体上的额外功能，而是确保配备真实工具的智能体能够安全部署的设计基础，LangGraph 的设计也正是着眼于这一问题。其检查点系统主要就是为了实现暂停功能而存在的。
根据出现频率，有四种模式涵盖了大多数实际系统：
- 使用当前标准函数
interrupt()在执行高风险操作前暂停。 - 在工作流中设置一个必然会经过的检查点。
- 在暂停状态下修改状态，以便在智能体继续执行前进行修正。
- 时间回溯：从对话的任意早期节点重新播放或分支。
有些内容是故意省略的，因为它们在实际代码库中很少出现：令牌级流式处理机制、完整的LangGraph Studio SDK，以及已被interrupt()取代的旧版NodeInterrupt异常。
中断与恢复
每种模式最终都归结为相同的两个步骤：
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.
其他所有内容都属于变体：即调用 interrupt() 的位置、传递给它的数据以及如何处理返回结果。
术语表
interrupt(value) 来自 langgraph.types，在节点内部被调用。它会立即停止图的计算，通过检查指针保存整个状态，然后将 value 传递给启动运行的人。这种暂停是真实的——没有线程在后台空转，也不会占用任何资源，因此一个线程可以暂停数月之久。
Command(resume=value) 会被传递到 graph.invoke() 或 graph.stream() 中，用以唤醒处于暂停状态的图。value 会成为 interrupt() 调用的返回值，就好像 interrupt() 是一个已经收到回复的同步 input() 函数一样。
__interrupt__ 是在执行暂停时出现在输出中的关键信息。通过查看它可以判断代理是否已停止并需要输入。
检查指针（MemorySaver、SqliteSaver、PostgresSaver）是必需的。没有这些检查指针，就无法暂停或恢复已保存的状态。初学者编写的HITL代码失败的最常见原因就是忘记为图表添加相应的检查指针。
模式1：在工具调用执行前进行批准
这是目前最常见的实际应用场景。代理决定调用那些会产生实际效果的工具，比如发送邮件、查询数据库或下订单，而必须先由人工进行批准。
可以想象成经理需要共同签署支票的场景：员工（即大语言模型）填写好支票并交上去，但在经理（即人类操作者）签字之前一切都不会进展。如果经理拒绝签字，支票就会作废。
评审节点：
# ── 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"]
)
]
}
然后将其接入图表中：
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)
与基本代理流程相比，路由略有不同：现在工具请求会在工具节点之前先发送到review_tool_call。代理的选择与实际执行之间多了审核环节，即提现前的共同签名步骤。
运行并恢复：
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)
这就是完整的模式。其他部分的变化在于传递给 interrupt() 的内容，以及你对返回结果的处理方式。
模式2：强制反馈环节
有时你并不希望根据大语言模型的选择来决定是否暂停。你希望每次都在固定的阶段暂停。无论模型写了什么，那些在发布前总会停下来进行审核的内容生成系统都是典型的例子。
可以想象一下装配线上的质量检测站。每个产品都必须经过那里检查，没有例外。工人可以将其直接通过，要求返工，或直接进行修改。
反馈节点：
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']}")
]
}
还有回路接线：
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())
这几乎是所有内容审核、代码审查或研究智能体中都存在的“审核与修改”循环：生成内容，暂停等待人工审核，必要时进行修改，再次暂停，重复此过程直至内容获得批准。
模式3：手动修复状态
有时，人工审核者并不想直接批准或拒绝，而是希望在智能体继续运行之前修复其状态，因为智能体的状态存在轻微偏差，修正它比重新尝试要快。
这里有三个关键的调用：
graph.get_state(config)会返回当前的StateSnapshot，其中包含.values（状态字典）和.next（恢复运行时将处理的节点）。graph.update_state(config, values)会在当前检查点写入新值。它会创建一个新的检查点，同时保留历史记录，这对于后续的时间回溯非常重要。
as_node="node_name"是update_state的一个可选参数，用于告知LangGraph该更新来自指定节点。LangGraph会根据上一次更新的源节点来选择后续的边，因此设置错误可能会导致图在非预期的位置继续更新。实际应用中：
# ── 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)
有两个容易让人混淆的点。首先，interrupt_before=[“node_name”]是在编译时设置的，每次运行时都会在该节点之前暂停；而interrupt()则存在于节点内部，仅在你设定的逻辑条件满足时才会暂停。对于简单的调试或需要确保暂停的场景，可使用interrupt_before；而当暂停需要根据运行时条件决定时——这涵盖了大多数实际应用场景——则应使用interrupt()。
其次，graph.invoke(None, config)的含义是“不添加新输入，从上一个检查点继续执行”。这与Command(resume=...)不同，后者是用于响应尚未处理的interrupt()请求。在使用了interrupt_before并调用update_state之后，应使用None来继续执行；而在调用interrupt()之后，则应使用Command(resume=...)来继续。
模式4：时间旅行
检查点历史的功能类似于 Git 中的提交记录。你可以查看任何过去的状态，并从中创建分支以形成不同的时间线，而不会影响原始状态。另一个类比是电子游戏中的保存点。get_state_history 会列出所有的保存点，你可以加载其中的任意一个，而不仅仅是最新的那个。
graph.get_state_history(config)是一个生成器，会按最新优先的顺序输出为该线程保存的每一个StateSnapshot。每个快照都包含带有唯一checkpoint_id的config。checkpoint_id的位置在snapshot.config[“configurable”][“checkpoint_id”]中。将包含特定checkpoint_id的配置传递给invoke或update_state函数时，操作将针对该指定的检查点，而非最新的那个。
graph.invoke(None, config_with_old_checkpoint_id)，从而从该点开始重新执行流程。虽然该点之后的节点逻辑会再次运行，但该分支已存储的检查点会被直接复用，而无需重新计算。update_state，这样会在原有检查点及时间线之外生成一个新的检查点与新的时间线。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.
大多数聊天机器人运行时并不需要此功能。它的价值体现在三个方面：调试——重现失败运行中的完整过程，找出故障点；评估——使用不同的提示词或模型重新进行相同对话并加以比较，这相当于无需完整重放即可进行的A/B测试；错误恢复——当某个节点在运行过程中出错时，可回退到出错前的状态，修复问题后继续运行，而无需重新开始整个对话。
具备工具使用审批功能的完整智能体
这是一个结合了基本结构、内存管理以及工具调用审批功能的完整模板，代表了带有人工安全机制的实用化生产级智能体的真实形态。
# ============================================================
# 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}")
快速参考
暂停与继续。 interrupt(value) 会在节点内部暂停执行，保存当前状态并将 value 传递给调用方，之后再返回继续执行的值。Command(resume=value) 用于处理待处理的中断。“__interrupt__” 是表示暂停的输出键，而 result[“__interrupt__”][0].value 可显示被暂停时所处理的值。interrupt_before=[“node_name”] 总会在指定节点之前暂停执行。
编辑状态。 graph.get_state(config) 会返回一个 StateSnapshot（包含 .values、.next 和 .config）。graph.update_state(config, values) 可在不删除历史记录的情况下添加检查点。as_node 会影响恢复时遵循的边。graph.invoke(None, config) 会在没有新输入的情况下继续运行。
时间旅行。 graph.get_state_history(config) 会按最新顺序列出所有快照。checkpoint_id 用于标识某个保存点。回放功能可从旧检查点继续时间线，而分支功能则会创建新的时间线。
适用的模式
- 那些需要花费资金、发送消息或修改外部系统的工具：应使用
interrupt()进行中断处理。对于具有实际副作用的生产级代理，必须如此处理。
interrupt_before与update_state即可实现。get_state_history实现时间回溯。安全性即图的结构
在LangGraph中，记忆、总结、持久化以及人工监督都体现为结构形式：此处为一个节点，此处为一条边，状态中则包含相应字段。无需开启特殊的安全模式，安全性本身就是图可以呈现的另一种结构形态。
在实际应用中，代理很少会因为大语言模型做出了错误决策而失败，这种情况本就是预期之中且可以挽回的。它们真正失败的情况是，在错误决策与其后果之间没有设置检查点。interrupt()功能正是用于设置这样的检查点。它需要多一个节点和一条边，但或许会成为图中最关键的节点。
通过标准结构、让对话保持可行且持久的方法，以及对所有重要内容的安全保障，几乎所有成熟的LangGraph应用都拥有了基础框架。接下来的自然发展方向是带有监督节点和子图的多代理协调机制，以及跨会话用户资料等长期记忆功能，这些都需要依托状态、节点和检查点机制来实现。
相关阅读
- LangGraph Agent 的 MCP：从基础到实战——服务器、传输机制与适配器桥梁 — 从主机、客户端和服务器的角色出发，学习模型上下文协议，了解 FastMCP 服务器、stdio 与 HTTP 的区别、langchain-mcp-adapters 的用法，以及 LangGraph 在实际应用中可能遇到的问题。
- LangGraph 深入解析：图结构、状态管理、路由机制及生产环境应用模式 — 说明 LangGraph 如何将智能体视为具有类型化状态、归约器、检查点、路由功能、工具以及 ReAct 循环的循环图结构，同时介绍适用于生产环境的各种模式。