精通人机协同：使用 LangGraph 构建交互式智能体的综合指南

第一节：自主智能体中人类监督的必要性

在人工智能（AI）领域，我们正迅速从简单的自动化脚本转向能够自主规划、推理和执行复杂任务的智能体。然而，随着智能体能力的增强，其潜在风险也呈指数级增长。将 AI 智能体视为不会犯错的“黑箱”是一种危险的误解。实际上，它们是强大的工具，在关键应用场景中，人类的监督不仅是可取的，更是必不可少的。人机协同（Human-in-the-Loop, HITL）并非一个附加功能，而是构建负责任、安全和可信赖 AI 系统的核心架构模式。

超越完全自动化

人机协同的核心理念是，在自动化流程的关键节点引入人类判断。这并非要削弱 AI 的能力，而是要为其增加一道安全屏障，确保其行为符合预期，尤其是在那些后果严重的领域。以下几个典型场景凸显了人类监督的迫切性：

• 金融服务：设想一个旨在执行股票交易的 AI 智能体。如果该智能体错误地解读了市场信号或收到了有误导性的新闻，可能会自动执行一笔灾难性的交易，导致巨大的财务损失。在这里，一个人机协同步骤是至关重要的：对于超过特定金额阈值的交易，系统必须暂停并等待人类交易员的批准。这确保了策略的执行既利用了 AI 的速度，又受到了人类经验的约束。

• 医疗健康：一个 AI 助手可以极大地提高医生的工作效率，例如通过分析病历和影像学报告来起草诊断摘要。然而，最终的诊断责任必须由人类医生承担。一个强制性的人类审查步骤可以确保 AI 的输出没有遗漏关键信息或产生误导性结论，从而保障患者的生命安全和诊断的准确性。

• 企业自动化：在现代 IT 运维中，智能体可以被授权管理云基础设施，例如配置服务器或删除未使用的资源。一个没有监督的智能体可能会因一个错误的指令或对环境的误判，意外删除关键的生产数据库或网络配置，导致整个服务中断。因此，在执行任何破坏性操作（如删除资源）之前，必须暂停并请求人类工程师的明确授权。

LangGraph 的设计哲学

LangGraph 从设计之初就深刻理解了这一现实。它并非将人机协同视为一个后来添加的补丁或一种变通方法，而是将其作为框架的核心设计原则。LangGraph 提供的工具旨在支持构建安全、可交互和可审计的工作流。通过其状态化的持久性架构，LangGraph 使开发者能够以一种原生且优雅的方式将人类智慧融入复杂的计算图中，确保技术的力量始终在人类的掌控之下。

第二节：基础概念：状态、持久化与中断

要在 LangGraph 中实现有效的人机协同，必须首先理解其背后的三大架构支柱：状态（State）、持久化（Persistence）和中断（Interrupts）。人机协同并非仅仅是一个函数调用，而是 LangGraph 状态化和持久化设计所带来的一个强大特性。

将 StateGraph 作为系统记忆

每个 LangGraph 工作流都围绕一个中心化的状态对象构建，通常使用 TypedDict 来定义。这个状态对象远不止是一个简单的数据容器；它是智能体在执行任务过程中的持久、演化的“记忆”。图中的每个节点在执行时都会读取当前状态，并可以对其进行更新。这种以状态为中心的设计意味着智能体的完整上下文（包括对话历史、中间计算结果、待处理任务等）都集中在一个地方，为暂停和恢复操作提供了坚实的基础。

检查点（Checkpointer）：持久化的引擎

检查点是实现人机协同的核心技术，也是 LangGraph 持久化机制的关键。它的核心职责是在图的每个计算步骤之后，将当前的状态对象快照保存到一个持久化存储中（如 SQLite、Postgres 或内存）。

这个机制之所以至关重要，是因为它赋予了图“无限期”暂停的能力。当一个中断被触发时，图的执行会停止，但由于检查点已经将最新的状态安全地保存了下来，整个计算上下文得以保留。这意味着应用程序进程可以完全终止，甚至系统可以重启。几天后，当人类准备好提供输入时，可以通过相同的线程 ID 从检查点加载保存的状态，使图从中断处无缝恢复。

可以断言，没有检查点，中断将毫无意义。因为如果没有保存状态的机制，任何暂停都将导致上下文的丢失，从而无法恢复。检查点的存在将 LangGraph 的人机协同从简单的内存中暂停提升到了一个能够支持异步、长周期人类任务的强大架构特性。例如，一个需要经理在第二天上班后才能批准的请求，可以安全地暂停，而不会占用任何计算资源。

中断的分类

LangGraph 提供了两种类型的中断机制，它们服务于不同的目的。理解它们的区别对于编写清晰、可维护的代码至关重要。

• 动态中断 (interrupt())：这是实现人机协同应用逻辑的主要工具。动态中断是在图的节点内部通过编程方式调用的，其触发是基于运行时的当前状态。例如，if tool_call.is_sensitive: interrupt()。这种中断是应用程序控制流的一部分，用于处理需要人类干预的业务逻辑，如审批、编辑或提供额外信息 。本指南的重点将围绕动态中断展开。

• 静态中断 (interrupt_before / interrupt_after)：这主要是一种用于调试和开发的工具，可以被看作是代码中的“断点”。静态中断在图编译时或运行时指定，用于在特定节点执行之前或之后无条件地暂停图的执行 。它们不属于应用程序的业务逻辑，而是为了让开发者能够检查特定步骤的状态，从而调试图的行为。

为了更清晰地区分这两种机制，下表进行了详细的比较：

特性	动态中断 (interrupt())	静态中断 (interrupt_before/_after)
主要用例	实现应用逻辑中的人机协同（HITL），如审批、编辑、数据收集。	调试和测试图的执行流程，检查特定节点的状态。
触发条件	条件性触发，基于图在运行时的当前状态。	无条件触发，在编译时或运行时指定的目标节点处暂停。
实现方式	在节点函数内部调用 interrupt() 函数。	在 app.compile() 或 app.invoke() 的 interrupt_before/_after 参数中指定节点名称。
恢复机制	使用 Command(resume=value) 恢复，可以将人类输入的值传递回图中。	通常使用空的 Command() 或不带参数的 invoke 调用来恢复，仅为单步执行。
对应用逻辑的影响	是应用逻辑的一部分，直接影响控制流。	对应用逻辑透明，主要用于开发阶段，不应出现在生产代码中。

导出到 Google 表格

通过这个表格，可以清楚地看到，动态中断是构建交互式应用的核心，而静态中断则是保证这些应用正确运行的辅助工具。

第三节：核心机制：深入解析 interrupt() 与 Command 原语

理解了 LangGraph 的状态化和持久化基础后，我们可以深入探讨实现人机协同的核心 API：用于暂停的 interrupt() 函数和用于恢复的 Command 原语。这两者之间的交互遵循一个关键原则——“重执行原则”，理解该原则对于正确使用人机协同并避免常见陷阱至关重要。

使用 interrupt() 暂停执行

当在图的一个节点内部调用 interrupt() 函数时，图的执行会立即在该点停止。此时，LangGraph 引擎会执行以下操作：

1. 确保检查点已将当前状态成功保存到持久化存储中。
2. 向调用方（例如，调用 app.invoke() 的代码）抛出一个 Interrupt 异常。这个异常信号表明图已暂停，正在等待外部输入。

这个过程是同步的，调用方的代码可以通过 try...except 块来捕获这个中断信号，从而知道何时需要与用户进行交互。

使用 Command 恢复执行

一个被中断的图是通过再次调用 invoke 或 stream 方法来恢复的。关键在于，这次调用时需要传入一个 Command 对象，并使用 resume 参数携带人类的输入值，例如 Command(resume="approve") 。这个传递给 resume 的值，将成为中断点的“返回值”。

重执行原则（The Re-execution Principle）

这是理解 interrupt() 和 Command 交互时最核心、也最容易被误解的概念。官方文档指出：“图的执行会从最初调用 interrupt(...) 的那个节点的开头处恢复” 。

这并非一个偶然的副作用，而是一个深思熟虑的设计选择。其背后的逻辑如下：

1. 首次执行：当节点代码第一次执行到 approval = interrupt() 时，interrupt() 函数的作用是暂停执行并抛出异常。approval 变量在此时并未被赋值。
2. 恢复执行：当客户端使用 Command(resume="approve") 来恢复图时，引擎会重新从该节点的开头开始执行。
3. 二次执行：当代码再次执行到 approval = interrupt() 这一行时，interrupt() 函数的行为发生了变化。因为它检测到了一个待处理的 resume 值，它不再暂停，而是立即返回这个值（在此例中是 “approve”）。
4. 继续执行：approval 变量现在被成功赋值为 “approve”，节点中后续的代码（如 if approval == "approve":...）可以基于这个人类输入的值继续执行。

这种设计将人类交互优雅地抽象成了一个函数调用。interrupt() 在第一次调用时请求数据（通过暂停），在第二次调用时提供数据（通过返回值）。通过重新执行节点，代码可以自然地、函数式地接收和使用来自人类的输入，而无需复杂的异步回调或状态管理。

然而，这一原则也带来了重要的实践影响：节点开头到 interrupt() 调用处之间的所有代码都会被执行两次。这直接引出了关于副作用的关键最佳实践，我们将在第六节中详细探讨。

第四节：分步教程：构建一个带有人类审批步骤的智能体

理论知识需要通过实践来巩固。本节将提供一个完整、端到端的教程，演示如何构建一个简单的智能体，该智能体在使用工具前需要获得人类的批准。这个例子将填补文档中缺失的详细步骤，将所有概念串联起来 。

场景

我们将构建一个使用搜索工具的智能体。在执行可能产生费用或访问敏感信息的搜索查询之前，它必须暂停以征求人类用户的批准。

1. 环境设置

首先，确保已安装必要的库。

pip install langgraph langchain langchain-openai tavily-python

同时，请设置你的 OpenAI 和 Tavily API 密钥作为环境变量（OPENAI_API_KEY 和 TAVILY_API_KEY）。

2. 定义状态

我们将使用 TypedDict 来定义图的中心状态。它将包含一个 messages 列表，用于存储对话历史。

import os
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph.message import add_messages# 设置 API 密钥
# os.environ["OPENAI_API_KEY"] = "..."
# os.environ["TAVILY_API_KEY"] = "..."class AgentState(TypedDict):messages: Annotated[List[BaseMessage], add_messages]

add_messages 是一个方便的累加器，它会将新消息追加到 messages 列表中，而不是覆盖它们。

3. 定义工具

我们将创建一个简单的搜索工具，使用 Tavily Search API。

from langchain_community.tools.tavily_search import TavilySearchResultstool = TavilySearchResults(max_results=2)

4. 实现智能体节点

这是实现人机协同逻辑的核心部分。我们将创建一个节点，该节点负责调用大语言模型（LLM）。如果 LLM 的响应包含工具调用，我们就在执行该工具调用之前插入一个 interrupt()。

from langchain_openai import ChatOpenAI
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.interrupt import interrupt, Interrupt# 初始化模型和工具执行器
llm = ChatOpenAI(model="gpt-4o")
model_with_tools = llm.bind_tools([tool])
tool_node = ToolNode([tool])# 定义智能体节点
def agent_node(state: AgentState):response = model_with_tools.invoke(state["messages"])# 如果 LLM 决定调用工具if response.tool_calls:# *** 关键的人机协同逻辑 ***print(f"智能体想要运行工具: {response.tool_calls[0]['name']}，参数: {response.tool_calls[0]['args']}")# 暂停图的执行，等待人类输入# 这里的 interrupt() 会抛出 Interrupt 异常try:interrupt()except Interrupt:# 这个异常会被外部的循环捕获，这里只是为了演示raise# 当图恢复时，interrupt() 会返回 resume 的值# 但在这个简单的审批模式中，我们不需要返回值，因为拒绝逻辑在外部处理# 如果恢复执行，意味着已经批准，直接返回工具调用消息return {"messages": [response]}# 如果 LLM 没有调用工具，直接返回其响应return {"messages": [response]}

注意：在实际代码中，interrupt() 会抛出一个 Interrupt 异常，该异常会被 LangGraph 引擎捕获并传递给外部调用者。节点函数本身不会继续执行。当图恢复时，它会从节点开头重新执行，但这一次 interrupt() 不会再抛出异常，而是允许代码继续执行。

5. 构建图

现在，我们将节点连接起来，构建一个完整的图。我们需要一个条件边来决定在智能体节点之后是调用工具还是结束流程。

def should_continue(state: AgentState) -> str:if state["messages"][-1].tool_calls:return "call_tool"else:return "end"# 创建图
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", agent_node)
graph_builder.add_node("tools", tool_node)# 定义图的结构
graph_builder.set_entry_point("agent")
graph_builder.add_conditional_edges("agent",should_continue,{"call_tool": "tools", "end": "__end__"},
)
graph_builder.add_edge("tools", "agent")

6. 编译图并配置检查点

为了使 interrupt() 工作，我们必须为图配置一个检查点。这里我们使用一个简单的内存检查点 SqliteSaver（内存模式），在生产环境中，您应该使用 SqliteSaver（文件模式）或其他持久化后端。

import uuid# 使用内存中的检查点
memory_saver = SqliteSaver.from_conn_string(":memory:")# 编译图
app = graph_builder.compile(checkpointer=memory_saver)# 为每个对话创建一个唯一的线程ID
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

7. 完整的交互循环

这是驱动智能体的客户端代码。它负责调用图、捕获中断、获取用户输入，并使用 Command 恢复图的执行。

from langgraph.step import Command# 启动对话
user_input = "你好，我想知道 LangGraph 的最新更新是什么？"
events = app.stream([HumanMessage(content=user_input)], config, stream_mode="values")for event in events:if "messages" in event:event["messages"][-1].pretty_print()# 检查图的状态，看是否被中断
try:snapshot = app.get_state(config)last_message = snapshot.values["messages"][-1]if last_message.tool_calls:print("\n--- 图已暂停，等待您的批准 ---")# 获取用户输入user_approval = input("您是否批准执行该工具？(yes/no): ")if user_approval.lower() == "yes":print("--- 批准执行，正在恢复图 ---")# 使用 Command 恢复执行# 注意：这里的输入为空，因为我们的 agent_node 没有处理 resume 的值# 恢复操作本身就是一种批准信号events = app.stream(None, config, stream_mode="values")for event in events:if "messages" in event:event["messages"][-1].pretty_print()else:print("--- 您已拒绝执行，工作流结束 ---")# 如果拒绝，我们可以选择更新状态或直接结束# 这里我们简单地结束交互passexcept Interrupt:# 这个代码块在 stream 模式下通常不会被直接触发# 在 invoke 模式下会更有用print("捕获到中断！")

这个完整的例子展示了从图的定义到客户端交互的整个流程。它清晰地分离了智能体的内部逻辑（图的定义）和外部的控制逻辑（交互循环）。这种解耦是一个强大的架构优势，意味着同一个编译好的图可以被命令行界面、Web 后端或任何其他系统所控制，具有极高的通用性。

第五节：高级模式与实际应用

掌握了基础的审批流程后，我们可以探索 LangGraph 的人机协同功能所支持的更复杂、更强大的模式。这些模式将 interrupt() 视为一个低级的构建块，可以组合成高级、可复用的解决方案，从而构建出更具扩展性和可维护性的系统。

模式一：用于状态修正的人机协同

有时，人类的干预不仅仅是“是”或“否”的批准，而是需要直接修正智能体的状态。例如，一个智能体从非结构化文本中提取信息，但可能弄错某个字段。我们可以暂停图，让用户提供修正后的数据。

场景

智能体从一段文本中提取用户的姓名和邮箱，但将姓名识别错了。

实现

1. 在智能体节点中，提取信息后，调用 interrupt() 暂停。
2. 在客户端，显示提取出的数据，并请求用户确认或修正。
3. 用户提供一个修正后的字典，例如 {"name": "Correct Name"}。
4. 客户端使用 app.update_state() 方法将修正后的数据合并到当前状态中，然后使用 app.stream(None,...) 恢复执行。

# 客户端代码片段
#... 捕获中断后...
print(f"提取的数据: {current_state.values['extracted_data']}")
correction_json = input("请输入修正的 JSON (或直接回车跳过): ")
if correction_json:import jsoncorrection = json.loads(correction_json)# 使用 update_state 更新状态app.update_state(config, correction)print("--- 状态已更新 ---")# 恢复执行
app.stream(None, config)

这种模式对于数据清理、事实核查和需要人类提供领域知识的场景非常有用。

模式二：通用的“需要批准”工具装饰器

为了避免在每个需要审批的工具调用处都重复编写 interrupt() 逻辑，我们可以利用 Python 的装饰器（Decorator）来创建一个可复用的模式。这是一个非常强大的技术，可以将审批逻辑与工具的核心功能完全解耦。

实现

我们创建一个名为 @requires_approval 的装饰器。这个装饰器会包装任何工具函数，在其实际执行前自动插入 interrupt() 逻辑。

from functools import wrapsdef requires_approval(func):"""一个装饰器，用于在执行函数前添加一个中断以获取人类批准。"""@wraps(func)def wrapper(*args, **kwargs):# 打印提示信息print(f"--- 审批请求：即将执行工具 '{func.__name__}' ---")print(f"    参数: {args}, {kwargs}")# 暂停执行try:interrupt()except Interrupt:# 重新抛出，让 LangGraph 引擎处理raise# 如果恢复，意味着已批准，执行原始函数print(f"--- 审批通过，正在执行 '{func.__name__}' ---")return func(*args, **kwargs)return wrapper# 应用装饰器到任何工具上
@requires_approval
def sensitive_database_query(query: str) -> str:"""一个需要审批才能执行的敏感数据库查询工具。"""print(f"正在对数据库执行查询: {query}")#... 实际的数据库查询逻辑...return "查询结果..."# 在 ToolNode 中使用这个被装饰的工具
# tool_node = ToolNode([sensitive_database_query])

通过这种方式，我们可以轻松地为任何工具添加审批步骤，而无需修改智能体的主要逻辑。这极大地提高了代码的模块化和可维护性，体现了“关注点分离”和“不要重复自己”（DRY）的软件工程原则。

模式三：使用多个中断进行输入验证

文档中提到了在一个节点内使用多个中断的模式，这对于需要反复验证用户输入的场景非常有用 。

场景

智能体需要用户提供一个格式正确的电子邮件地址。

实现

在一个节点内部使用一个 while True 循环。循环内部，首先使用 interrupt() 请求输入，然后验证输入。如果格式无效，则发出另一个 interrupt() 来显示错误消息并再次暂停，等待下一次循环。

import redef is_valid_email(email: str) -> bool:return re.match(r"[^@]+@[^@]+\.[^@]+", email) is not Nonedef get_user_email_node(state: AgentState):while True:# 第一个中断，请求输入。它将返回 resume 的值。email = interrupt(message="请输入您的电子邮件地址:")if is_valid_email(email):# 验证通过，跳出循环print(f"邮箱 '{email}' 格式正确。")return {"email": email} # 更新状态else:# 第二个中断，显示错误消息并暂停。# 这个中断不需要返回值，只是为了暂停和通知。interrupt(message=f"'{email}' 不是一个有效的邮箱格式，请重试。")# 客户端恢复逻辑
# client_app.invoke(..., Command(resume="[e-mail]"))
# 如果格式错误，图会再次中断，客户端需要再次调用 invoke 提供新的输入

需要注意的是，当一个节点中有多个 interrupt() 调用时，Command(resume=...) 的值是严格按照索引匹配的。例如，Command(resume=["value1", "value2"]) 会将 “value1” 传递给第一个 interrupt()，将 “value2” 传递给第二个。这种模式虽然强大，但也可能变得脆弱，应谨慎使用。

第六节：关键考量与最佳实践

虽然 LangGraph 的人机协同机制非常强大，但如果不了解其背后的设计原则，很容易陷入一些常见的陷阱。本节将综合文档中的警告 ，并将其提炼为一套可操作的最佳实践，帮助您构建健壮、可预测的交互式智能体。

管理副作用（黄金法则）

这是使用 interrupt() 时最重要的规则，它直接源于我们之前讨论的“重执行原则”。由于节点在恢复时会从头开始重新执行，因此在 interrupt() 调用之前的任何具有副作用（Side Effect）的操作（如 API 调用、数据库写入、发送电子邮件）都会被执行两次。

为了避免这种情况，必须遵循**“决策与行动分离”**的设计模式。

错误模式：在一个节点内完成决策、审批和行动

# 错误示例
def agent_node(state):# 1. 决策：决定发送邮件email_content = "..."# 2. 副作用：在审批前发送了 API 请求（将被执行两次！）some_logging_api(message="Preparing to send email")# 3. 审批interrupt()# 4. 行动：发送邮件send_email(email_content)return...

正确模式：将决策和行动分离到不同的节点

• 节点 A（决策节点）：此节点负责生成行动计划（例如，确定要调用的工具及其参数），然后调用 interrupt() 进行审批。如果获得批准，它会将待执行的工具调用信息更新到状态中，然后将控制权传递给下一个节点。这个节点不执行任何有副作用的操作。
• 节点 B（行动节点）：此节点从状态中读取已经过批准的工具调用信息，并执行它。这个节点不包含任何中断，因此它保证只会被执行一次。

# 正确示例
def decision_node(state):# 1. 决策tool_call = {"name": "send_email", "args": {"to": "...", "body": "..."}}print("请求批准发送邮件...")# 2. 审批interrupt()# 3. 传递行动计划return {"approved_tool_call": tool_call}def action_node(state):# 1. 行动tool_call = state["approved_tool_call"]execute_tool(tool_call) # 执行实际的副作用操作return...# 在图构建中: graph.add_edge("decision_node", "action_node")

通过将“思考”与“行动”分离，我们确保了即使决策逻辑被执行两次，具有实际影响的行动也只会在得到明确批准后执行一次。

子图中的中断

当一个图（子图）被另一个图（父图）作为函数调用时，子图中的中断行为需要特别注意。如果子图中触发了中断，整个执行堆栈都会暂停。当父图恢复执行时，它会从调用子图的那个节点的开头重新执行。这会再次触发对子图的调用，进而使子图从其被中断的节点的开头恢复执行 。这个行为是符合逻辑且一致的，但开发者在设计复杂的嵌套图时需要意识到这一点，以避免意外的重执行。

多个中断的风险

如第五节所述，虽然可以在单个节点中使用多个 interrupt()，但这引入了复杂性。恢复值的匹配是严格基于索引的 。如果将来修改了节点逻辑，在现有中断之间添加或删除了一个新的 interrupt() 调用，那么所有客户端的恢复逻辑都可能被破坏。因此，一般建议将复杂的、多步骤的人类交互分解到多个独立的节点中，每个节点只包含一个 interrupt()。这样可以使状态转换更明确，代码也更易于维护。

结论

LangGraph 提供的人机协同（HITL）功能远不止是一个简单的暂停按钮。它是其状态化、持久化架构的自然延伸，为构建复杂、可靠且负责任的 AI 应用提供了坚实的基础。通过深入理解其核心机制——状态、检查点、interrupt() 与 Command 的交互，以及至关重要的“重执行原则”——开发者可以超越简单的审批流程，实现状态修正、可复用组件（如装饰器）和复杂的输入验证等高级模式。

成功的关键在于遵循最佳实践，尤其是“决策与行动分离”的黄金法则，以有效管理副作用。通过将 LangGraph 的人机协同功能视为一种强大的架构工具，而不仅仅是一个 API 调用，开发者可以构建出真正将人类智慧与机器智能相结合的下一代应用程序，确保技术在提供强大能力的同时，始终处于安全、可控和可信的范围之内。