8.人工介入控制
在构建复杂的 AI 工作流时，完全依赖自动化并不总是最佳选择。有些场景下，我们需要让人类参与进来，对关键决策进行审查、批准或修正。LangGraph 的人工介入控制机制，正是为了解决这一需求而设计的。它允许我们在图的执行过程中设置中断点，暂停工作流，等待人类输入，然后基于反馈继续执行。这种机制不仅提升了系统的可靠性，也为处理敏感操作提供了必要的安全保障。
中断点设置
中断点是人工介入控制的核心。LangGraph 提供了两种类型的中断：动态中断和静态中断。动态中断基于运行时的状态条件触发，而静态中断则在编译时或运行时指定在特定节点前后暂停。
动态中断
动态中断通过 interrupt 函数实现。这个函数可以在节点内部的任何位置调用，当执行到该位置时，图会暂停并等待人类输入。调用 interrupt 时，可以传递任意 JSON 可序列化的数据，这些数据会作为中断信息展示给人类审查者。
使用 interrupt 需要满足几个前提条件。首先，必须配置检查点（checkpointer）来持久化图状态。其次，在调用图时需要指定线程 ID，这样系统才能正确追踪和恢复执行上下文。
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START
class State(TypedDict):
text: str
def human_review_node(state: State):
# 在这里暂停，等待人类输入
value = interrupt({
"text_to_revise": state["text"],
"action": "请审查并修改这段文本"
})
return {"text": value}
# 构建图
graph_builder = StateGraph(State)
graph_builder.add_node("human_review", human_review_node)
graph_builder.add_edge(START, "human_review")
# 编译时必须指定检查点
checkpointer = InMemorySaver()
graph = graph_builder.compile(checkpointer=checkpointer)
# 使用线程 ID 运行图
config = {"configurable": {"thread_id": "review_session_1"}}
result = graph.invoke({"text": "原始文本"}, config=config)
# 输出中断信息
print(result['__interrupt__'])
代码执行到 interrupt 调用时会立即暂停。返回结果中的 __interrupt__ 字段包含了中断的详细信息，包括传递的值、中断 ID 和命名空间等元数据。此时，工作流的状态已经被保存到检查点中，可以安全地等待任意长的时间。
要恢复执行，需要使用 Command 对象，并通过 resume 参数提供人类输入：
# 恢复执行，提供修改后的文本
final_result = graph.invoke(
Command(resume="修改后的人类文本"),
config=config
)
print(final_result) # {'text': '修改后的人类文本'}
恢复执行时，图会从 interrupt 所在的节点重新开始执行。这次，interrupt 函数不再暂停，而是直接返回 Command 中提供的 resume 值。需要注意的是，整个节点会重新执行一遍，因此建议将 interrupt 放在节点的开头，或者将节点设计为幂等的，避免副作用重复执行。
静态中断
静态中断在编译时或运行时指定，用于在特定节点执行前或执行后暂停。这种方式更适合调试和测试场景，因为中断点是固定的，不依赖于运行时状态。
在编译时设置静态中断：
graph = graph_builder.compile(
interrupt_before=["node_a"], # 在 node_a 执行前暂停
interrupt_after=["node_b", "node_c"] # 在 node_b 和 node_c 执行后暂停
)
也可以在运行时动态指定：
# 运行时指定中断点
result = await client.runs.wait(
thread_id,
assistant_id,
input=inputs,
interrupt_before=["node_a"],
interrupt_after=["node_b"]
)
静态中断的使用场景相对有限，因为它缺乏动态中断的灵活性。在生产环境中，动态中断是更常见的选择，因为它可以根据实际状态决定是否需要人工介入。
人工审批工作流设计
人工审批是人工介入最常见的模式。在关键操作执行前，系统暂停并请求人类批准，根据批准结果决定后续执行路径。这种模式特别适用于敏感操作，如金融交易、数据删除或重要配置变更。
批准或拒绝模式
实现审批模式的核心是在中断后根据人类输入进行条件路由。人类可以批准操作，让工作流继续执行；也可以拒绝操作，引导工作流走向替代路径。
from typing import Literal
from langgraph.types import interrupt, Command
def approval_node(state: State) -> Command[Literal["execute_action", "alternative_path"]]:
# 暂停并请求批准
is_approved = interrupt({
"question": "是否批准执行以下操作？",
"operation": state["pending_operation"],
"details": state["operation_details"]
})
# 根据批准结果路由到不同节点
if is_approved:
return Command(goto="execute_action")
else:
return Command(goto="alternative_path")
# 在图中使用审批节点
builder = StateGraph(State)
builder.add_node("approval", approval_node)
builder.add_node("execute_action", execute_node)
builder.add_node("alternative_path", alternative_node)
builder.add_edge("approval", "execute_action")
builder.add_edge("approval", "alternative_path")
这个模式的关键在于 Command 对象的 goto 参数，它允许我们在恢复执行时直接跳转到指定的节点，而不是按照常规的边连接顺序执行。这为实现复杂的审批逻辑提供了极大的灵活性。
状态审查与编辑
有时我们不仅需要批准或拒绝，还需要人类直接修改工作流的状态。这在纠正错误、补充信息或优化结果时非常有用。
def review_and_edit_node(state: State):
# 展示当前状态供审查
reviewed_data = interrupt({
"task": "请审查并编辑以下生成的摘要",
"generated_summary": state["summary"],
"original_text": state["source_text"]
})
# 使用人类编辑后的内容更新状态
return {
"summary": reviewed_data["edited_summary"],
"reviewed_by": "human",
"review_timestamp": reviewed_data.get("timestamp")
}
这种模式允许人类直接修改工作流中的任何数据，而不仅仅是提供简单的批准或拒绝。返回的数据结构可以根据具体需求灵活设计，系统会将人类提供的值整合回状态中。
状态手动编辑恢复
状态手动编辑是人工介入的深层应用。它不仅允许人类审查输出，还能直接修改图的内部状态，然后让工作流基于修改后的状态继续执行。这在调试和纠错场景中特别有价值。
在工具中实现状态编辑
在工具调用中集成状态编辑功能，可以让模型生成的参数经过人类验证后再执行实际操作。这种方式结合了自动化生成和人类验证的优势。
from langchain_core.tools import InjectedToolCallId, tool
from langchain_core.messages import ToolMessage
from typing_extensions import Annotated
@tool
def sensitive_operation(
param1: str,
param2: int,
tool_call_id: Annotated[str, InjectedToolCallId]
):
"""执行需要人工审查的敏感操作"""
# 暂停并展示操作详情
human_response = interrupt({
"action_request": {
"action": "sensitive_operation",
"args": {"param1": param1, "param2": param2}
},
"config": {
"allow_accept": True,
"allow_edit": True,
"allow_respond": True
},
"description": "请审查此敏感操作"
})
# 处理人类响应
if human_response["type"] == "accept":
# 执行原始操作
result = execute_with_original_params(param1, param2)
response_msg = "操作已执行"
elif human_response["type"] == "edit":
# 使用编辑后的参数
edited_params = human_response["args"]["args"]
result = execute_with_original_params(
edited_params["param1"],
edited_params["param2"]
)
response_msg = f"已使用编辑后的参数执行: {edited_params}"
else:
# 拒绝执行
result = None
response_msg = "操作已被拒绝"
# 返回包含状态更新的 Command
return Command(
update={
"operation_result": result,
"messages": [ToolMessage(content=response_msg, tool_call_id=tool_call_id)]
}
)
这个实现展示了三种可能的响应类型：接受、编辑和拒绝。每种类型都会导致不同的后续行为，最终通过 Command 对象更新图的状态。这种方式将人工审查深度集成到工作流中，确保关键操作得到适当监督。
批量处理多个中断
当图并行执行多个节点时，可能会同时触发多个中断。LangGraph 支持一次性恢复所有中断，通过映射中断 ID 到对应的恢复值来实现。
# 获取当前状态的所有中断
state = graph.get_state(config)
interrupts = state.interrupts
# 构建恢复映射
resume_map = {
interrupt.id: f"编辑后的值 for {interrupt.value}"
for interrupt in interrupts
}
# 一次性恢复所有中断
result = graph.invoke(Command(resume=resume_map), config)
这种模式在处理并行任务时特别有用，可以避免多次往返通信，提高整体效率。
工具调用人工审查
工具调用是 AI 代理与外部系统交互的主要方式，也是人工介入最关键的控制点。审查工具调用可以确保模型生成的参数正确、安全且符合业务规则。
基础工具审查
最简单的工具审查模式是在工具执行前插入中断，让人类确认或修改调用参数。
from langgraph.prebuilt import create_react_agent
from langgraph.types import interrupt
def book_hotel(hotel_name: str, nights: int):
"""预订酒店"""
# 在真正执行前暂停
response = interrupt(
f"准备调用 book_hotel，参数: hotel_name={hotel_name}, nights={nights}。 "
"请批准或建议修改。"
)
if response["type"] == "accept":
# 执行原始调用
return f"成功预订 {hotel_name}，{nights} 晚。"
elif response["type"] == "edit":
# 使用修改后的参数
new_name = response["args"]["hotel_name"]
new_nights = response["args"]["nights"]
return f"成功预订 {new_name}，{new_nights} 晚。"
else:
raise ValueError("操作被拒绝")
# 创建带有审查功能的代理
agent = create_react_agent(
model="anthropic:claude-3-5-sonnet-latest",
tools=[book_hotel],
checkpointer=InMemorySaver()
)
运行代理时，当模型尝试调用 book_hotel 工具，执行会在 interrupt 处暂停。人类审查者可以看到完整的调用详情，并决定接受、修改或拒绝。
通用工具审查包装器
为了避免为每个工具重复实现审查逻辑，可以创建一个通用的包装器，为任意工具添加审查功能。
from typing import Callable
from langchain_core.tools import BaseTool, tool as create_tool
from langgraph.prebuilt.interrupt import HumanInterruptConfig
def add_human_review(
tool: Callable | BaseTool,
*,
interrupt_config: HumanInterruptConfig = None
) -> BaseTool:
"""为任意工具添加人工审查功能"""
if not isinstance(tool, BaseTool):
tool = create_tool(tool)
if interrupt_config is None:
interrupt_config = {
"allow_accept": True,
"allow_edit": True,
"allow_respond": True
}
@create_tool(
tool.name,
description=tool.description,
args_schema=tool.args_schema
)
def reviewed_tool(**tool_input):
# 构建审查请求
request = {
"action_request": {
"action": tool.name,
"args": tool_input
},
"config": interrupt_config,
"description": f"请审查工具调用: {tool.name}"
}
# 等待人类响应
response = interrupt([request])[0]
# 根据响应类型处理
if response["type"] == "accept":
return tool.invoke(tool_input)
elif response["type"] == "edit":
edited_args = response["args"]["args"]
return tool.invoke(edited_args)
elif response["type"] == "respond":
# 返回自定义反馈给模型
return response["args"]
else:
raise ValueError(f"不支持的响应类型: {response['type']}")
return reviewed_tool
使用这个包装器，可以轻松地为现有工具添加审查功能，而无需修改工具本身的实现：
# 原始工具
def delete_user_account(user_id: str):
"""删除用户账户"""
return f"已删除用户 {user_id}"
# 添加审查功能
reviewed_delete_tool = add_human_review(delete_user_account)
# 在代理中使用
agent = create_react_agent(
model="anthropic:claude-3-5-sonnet-latest",
tools=[reviewed_delete_tool],
checkpointer=InMemorySaver()
)
这种方式将审查逻辑与业务逻辑解耦，提高了代码的可维护性和复用性。
审查响应的格式规范
为了实现工具审查的标准化，LangGraph 提供了 HumanInterrupt 和 HumanResponse 模式，定义了审查请求和响应的结构。
审查请求包含：
action_request: 工具调用的详细信息，包括名称和参数config: 审查配置，指定允许的操作类型（接受、编辑、响应）description: 人类可读的描述信息
审查响应包含：
type: 响应类型（"accept"、"edit"、"respond"）args: 相关参数，根据类型不同而变化
这种标准化格式使得不同的 UI 工具（如 Agent Chat UI）能够与 LangGraph 的审查机制无缝集成，提供一致的用户体验。
与持久化系统的协同
人工介入控制与 LangGraph 的持久化系统紧密集成。每次中断时，当前状态都会被保存到检查点中。这意味着即使系统重启，也能从上次中断的位置恢复执行。
长时间等待的处理
由于状态被持久化，工作流可以暂停任意长的时间。这对于需要等待人类审查的场景特别重要，审查可能需要几小时甚至几天。系统不需要保持运行状态，只需在恢复时加载检查点即可。
# 第一次调用，触发中断
result = graph.invoke(initial_input, config=config)
# 系统可以在这里完全关闭
# 几天后，重新初始化并恢复
# 加载保存的状态
state = graph.get_state(config)
# 检查是否处于中断状态
if state.interrupts:
# 获取审查结果（可能来自数据库或用户界面）
human_input = get_review_result_from_database()
# 恢复执行
final_result = graph.invoke(
Command(resume=human_input),
config=config
)
这种机制确保了工作流的可靠性和容错性，即使在长时间等待人类输入的情况下也能保持一致性。
避免副作用重复执行
由于中断后节点会重新执行，需要特别注意副作用的处理。任何具有副作用的操作（如 API 调用、文件写入）都应该放在 interrupt 调用之后，或者放在单独的节点中。
# 不推荐：副作用在 interrupt 之前
def problematic_node(state: State):
api_result = call_external_api(state["data"]) # 可能重复执行
human_input = interrupt({"api_result": api_result})
return {"result": process(api_result, human_input)}
# 推荐：副作用在 interrupt 之后
def better_node(state: State):
human_input = interrupt({"data": state["data"]})
if human_input["proceed"]:
api_result = call_external_api(state["data"]) # 只执行一次
return {"result": api_result}
return {"result": "cancelled"}
# 最佳：副作用在单独节点
def review_node(state: State):
return interrupt({"data": state["data"]})
def execution_node(state: State):
api_result = call_external_api(state["data"])
return {"result": api_result}
遵循这些原则可以确保工作流在恢复时行为正确，不会重复执行已经完成的操作。
人工介入控制为 LangGraph 工作流提供了关键的安全保障和灵活性。通过精心设计中断点和审查流程，我们可以在自动化和人类监督之间找到最佳平衡，构建既高效又可靠的 AI 系统。下一章将探讨如何实现流式交互，让工作流的执行过程更加透明和实时。