05-LangGraph人工干预：实现人机协作的AI助手
## 前言
在构建 AI Agent 的过程中，我们经常会遇到一些复杂场景，比如：
- AI 需要在执行关键操作前获得人类确认
- 遇到不确定的问题时需要人工专业指导
- 需要人工审核 AI 生成的敏感内容
- 业务流程中需要人工决策的节点
这些场景都涉及到 **Human-in-the-loop（人在环中）** 的概念，即在 AI 自动执行的过程中引入人工干预机制。
LangGraph 作为 LangChain 团队推出的强大 Agent 编排框架，提供了完善的 **人工干预（Human-in-the-loop）** 功能。通过 `interrupt` 和 `Command` 两个核心 API，我们可以轻松实现人机协作的 AI 助手。
本文将详细介绍如何在 LangGraph 中实现人工干预功能，并提供完整的代码实现和详细解析。
---
## 核心概念
在深入代码之前，我们需要先理解三个核心概念：
### 1. interrupt - 中断执行
`interrupt` 是 LangGraph 提供的一个函数，用于在图执行过程中暂停当前流程，等待外部输入。
```python
from langgraph.types import interrupt
# 在工具中调用 interrupt，暂停执行并等待外部输入
human_response = interrupt({
"query": "请确认是否执行此操作",
"status": "waiting_for_human_input"
})
```
调用 `interrupt` 后，图的执行会暂停在当前节点，状态会被保存到检查点（checkpoint）中，可以随时恢复。
### 2. Command - 恢复执行
`Command` 是 LangGraph 的一个类型，用于恢复被中断的执行流程，并向图传递数据。
```python
from langgraph.types import Command
# 创建恢复命令，传入人工回复内容
command = Command(resume={"data": "人工确认执行"})
graph.stream(command, config)
```
`Command` 有两种模式：
- `resume`: 恢复被中断的执行，传递数据
- `goto`: 跳转到指定节点重新执行
### 3. Human-in-the-loop - 人在环中
Human-in-the-loop 是一种 AI 系统设计模式，核心思想是在 AI 自动处理大部分任务的同时，保留人类参与关键决策点的能力。
典型应用场景包括：
- **审批流程**：AI 预处理 + 人工最终审批
- **质量控制**：AI 生成内容 + 人工审核
- **复杂决策**：AI 建议 + 人工选择
- **错误处理**：AI 无法处理时转人工
---
## 代码实现
下面是基于 LangGraph 实现人工干预功能的完整代码：
```python
"""
LangGraph 教程 - 添加人工干预 (Human-in-the-loop)
本示例在 04添加记忆.py 的基础上，添加人工干预功能。
通过 interrupt 函数实现人机协作，允许在关键节点暂停执行并等待人工输入。
"""
# 过滤警告信息
import warnings
warnings.filterwarnings("ignore", category=DeprecationWarning)
warnings.filterwarnings("ignore", category=UserWarning)
# 设置标准输入输出编码
import sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8')
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import AIMessage
from langchain_core.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt
from dotenv import load_dotenv
import os
# 加载环境变量
load_dotenv()
# ==================== 1. 定义状态 ====================
class State(TypedDict):
"""
定义图的状态结构。
messages: 消息列表，使用 add_messages reducer 函数
确保新消息追加到列表，而不是覆盖
"""
messages: Annotated[list, add_messages]
# ==================== 2. 定义人工协助工具 ====================
@tool
def human_assistance(query: str) -> str:
"""
请求人工协助工具。
当 AI 遇到复杂问题或需要人工确认时，调用此工具暂停执行，
等待人工输入后再继续。
Args:
query: 向人工提出的问题或请求
Returns:
人工提供的回复内容
"""
# interrupt 函数会暂停执行，等待外部输入
human_response = interrupt({
"query": query,
"status": "waiting_for_human_input"
})
return human_response["data"]
# ==================== 3. 创建图 ====================
def create_graph():
"""
创建并编译 StateGraph，添加人工干预功能。
Returns:
编译后的图对象
"""
# 创建图构建器
graph_builder = StateGraph(State)
# 初始化模型
llm = ChatOpenAI(
model="Qwen/Qwen3-Next-80B-A3B-Instruct",
openai_api_key=os.getenv("SILICONFLOW_API_KEY"),
openai_api_base="https://api.siliconflow.cn/v1",
temperature=0.7
)
# 创建工具列表
search_tool = TavilySearchResults(max_results=2)
tools = [search_tool, human_assistance]
# 绑定工具到 LLM
llm_with_tools = llm.bind_tools(tools)
# 定义聊天机器人节点
def chatbot(state: State):
"""
聊天机器人节点。
注意：由于我们会在工具执行期间中断，
为避免恢复时重复调用工具，需要禁用并行工具调用。
"""
message = llm_with_tools.invoke(state["messages"])
# 确保最多只有一个工具调用
assert len(message.tool_calls) <= 1, "禁用并行工具调用以避免重复执行"
return {"messages": [message]}
# 添加节点
graph_builder.add_node("chatbot", chatbot)
# 使用 LangGraph 预定义的 ToolNode
tool_node = ToolNode(tools=tools)
graph_builder.add_node("tools", tool_node)
# 添加边
graph_builder.add_edge(START, "chatbot")
# 添加条件边：从 chatbot 到 tools 或 END
graph_builder.add_conditional_edges(
"chatbot",
tools_condition,
)
# 添加边：从 tools 回到 chatbot（形成循环）
graph_builder.add_edge("tools", "chatbot")
# ==================== 关键：添加 MemorySaver 检查点 ====================
# 创建内存中的检查点（支持执行中断和恢复）
memory = MemorySaver()
# 编译图时传入 checkpointer
return graph_builder.compile(checkpointer=memory)
# ==================== 4. 运行聊天机器人 ====================
def stream_graph_updates(graph, user_input: str, config: dict):
"""
流式处理图更新，支持中断和恢复。
Args:
graph: 编译后的图对象
user_input: 用户输入的消息（或 Command 对象用于恢复）
config: 包含 thread_id 的配置字典
Returns:
是否发生了中断 (True/False)
"""
is_interrupted = False
for event in graph.stream(
user_input,
config,
stream_mode="values"
):
if "messages" in event:
last_message = event["messages"][-1]
# 只打印 AI 消息
if isinstance(last_message, AIMessage):
if last_message.content:
print("助手:", last_message.content)
# 检查是否有工具调用
if last_message.tool_calls:
for tool_call in last_message.tool_calls:
print(f"\n🔧 工具调用: {tool_call['name']}")
print(f" 参数: {tool_call['args']}")
# 检查是否发生中断
try:
snapshot = graph.get_state(config)
if snapshot.next:
is_interrupted = True
print(f"\n⏸️ 执行已暂停，等待人工输入...")
print(f" 当前节点: {snapshot.next}")
except Exception:
pass
return is_interrupted
def main():
"""主函数 - 运行交互式聊天机器人（支持人工干预）。"""
print("🤖 LangGraph 人工干预演示已启动！")
print("=" * 60)
print("提示：")
print(" - 输入 'quit'、'exit' 或 'q' 退出对话")
print(" - 输入 'resume' 恢复中断的执行")
print(" - 输入 'state' 查看当前状态")
print(" - 当 AI 调用 human_assistance 工具时，执行会暂停")
print(" 此时你可以输入回复内容，然后输入 'resume' 继续\n")
# 创建图
graph = create_graph()
# 配置
current_thread_id = "1"
config = {"configurable": {"thread_id": current_thread_id}}
# 用于存储人工回复
human_response = None
is_waiting_for_human = False
print(f"当前会话 ID: {current_thread_id}")
print("-" * 60)
while True:
try:
# 获取用户输入
if is_waiting_for_human:
user_input = input("人工回复: ")
else:
user_input = input("用户: ")
# 检查退出命令
if user_input.lower() in ["quit", "exit", "q"]:
print("\n👋 再见！")
break
# 检查状态
if user_input.lower() == "state":
snapshot = graph.get_state(config)
print(f"\n📊 当前会话状态 (thread_id={current_thread_id}):")
print(f" 消息数量: {len(snapshot.values.get('messages', []))}")
print(f" 下一步: {snapshot.next}")
print(f" 是否等待人工: {is_waiting_for_human}")
print("-" * 60)
continue
# 恢复执行
if user_input.lower() == "resume" and is_waiting_for_human:
if human_response:
# 创建 Command 对象恢复执行
command = Command(resume={"data": human_response})
print(f"\n▶️ 恢复执行，传入人工回复: {human_response}")
is_interrupted = stream_graph_updates(graph, command, config)
is_waiting_for_human = is_interrupted
if not is_interrupted:
human_response = None
print()
else:
print("⚠️ 请先输入人工回复内容，再输入 'resume'")
continue
# 如果正在等待人工输入，保存回复
if is_waiting_for_human:
human_response = user_input
print(f"✅ 已记录人工回复: {human_response}")
print(" 输入 'resume' 继续执行")
continue
# 正常对话流程
is_interrupted = stream_graph_updates(
graph,
{"messages": [{"role": "user", "content": user_input}]},
config
)
is_waiting_for_human = is_interrupted
if is_waiting_for_human:
print(" 请输入你的回复，然后输入 'resume' 继续")
print()
except KeyboardInterrupt:
print("\n\n👋 再见！")
break
except Exception as e:
print(f"发生错误: {e}")
import traceback
traceback.print_exc()
break
if __name__ == "__main__":
main()
```
---
## 关键代码解析
### 1. interrupt 函数的使用
`interrupt` 是实现人工干预的核心，它有以下特点：
```python
# 在工具内部调用 interrupt
@tool
def human_assistance(query: str) -> str:
# interrupt 会暂停整个图的执行
# 传入的参数会存储在中断状态中，可供后续使用
human_response = interrupt({
"query": query, # 传递给外部的问题
"status": "waiting_for_human_input" # 当前状态标识
})
# 当执行恢复时，这里会继续执行
# human_response 包含外部传入的数据
return human_response["data"]
```
**关键点**：
- `interrupt` 只能在工具（Tool）内部调用
- 调用后，当前图的状态会被保存到检查点
- 执行不会自动恢复，需要外部显式调用 `Command` 来恢复
### 2. Command 对象的使用
`Command` 用于恢复被中断的执行：
```python
from langgraph.types import Command
# 恢复模式：resume
command = Command(resume={"data": "人工回复内容"})
graph.stream(command, config)
# 跳转模式：goto（可以重新导向到任意节点）
command = Command(goto="chatbot")
graph.stream(command, config)
```
### 3. 检查点的重要性
要支持中断和恢复，必须配置检查点（Checkpoint）：
```python
from langgraph.checkpoint.memory import MemorySaver
# 创建内存检查点
memory = MemorySaver()
# 编译图时传入 checkpointer
graph = graph_builder.compile(checkpointer=memory)
```
**为什么需要检查点？**
- 当 `interrupt` 被调用时，图的执行状态需要被保存
- 检查点保存了完整的图状态（节点、边、变量等）
- 恢复时可以从保存的状态继续执行
### 4. 禁用并行工具调用
在有人工干预的场景中，需要禁用并行工具调用：
```python
def chatbot(state: State):
message = llm_with_tools.invoke(state["messages"])
# 确保最多只有一个工具调用
assert len(message.tool_calls) <= 1, "禁用并行工具调用以避免重复执行"
return {"messages": [message]}
```
**原因**：
- 如果并行调用多个工具，恢复时可能导致重复执行
- 人工干预场景通常需要串行执行，确保每步都可控
---
## 运行效果
启动程序后，你可以体验以下交互流程：
从运行效果可以看到：
1. 用户发送请求后，AI 决定调用 `human_assistance` 工具
2. 执行暂停，显示等待人工输入
3. 人工输入回复后，输入 `resume` 恢复执行
4. AI 收到人工回复后，继续生成最终响应
---
## 踩坑记录
### 1. 忘记配置检查点
**错误**：运行时报错或中断后无法恢复
**原因**：没有在编译图时传入 `checkpointer`
**解决**：
```python
# 错误写法
graph = graph_builder.compile()
# 正确写法
memory = MemorySaver()
graph = graph_builder.compile(checkpointer=memory)
```
### 2. 在非工具函数中调用 interrupt
**错误**：`interrupt() can only be called inside a tool`
**原因**：`interrupt` 只能在 `@tool` 装饰的函数内部调用
**解决**：将需要中断的逻辑封装到工具函数中
### 3. 并行工具调用导致重复执行
**错误**：恢复执行后，工具被调用多次
**原因**：启用了并行工具调用，恢复时可能重复执行
**解决**：禁用并行工具调用
```python
assert len(message.tool_calls) <= 1
```
### 4. Command 格式错误
**错误**：恢复执行后没有正确传递数据
**原因**：`Command` 格式不正确
**解决**：
```python
# 正确格式
command = Command(resume={"data": "人工回复"})
```
### 5. 检查中断状态时机不对
**错误**：无法正确检测到中断
**原因**：在 `stream` 之前就检查中断状态
**解决**：在 `stream` 执行完成后检查
```python
# 执行 stream 后检查
for event in graph.stream(user_input, config):
# 处理事件...
# 然后检查中断状态
snapshot = graph.get_state(config)
if snapshot.next:
# 发生了中断
```
---
## 总结
本文详细介绍了 LangGraph 中人工干预功能的实现，主要包含以下内容：
### 核心要点
1. **`interrupt` 函数**：在工具内部调用，暂停图执行并等待外部输入
2. **`Command` 对象**：用于恢复被中断的执行，可传递数据
3. **检查点（Checkpoint）**：必须配置才能支持中断和恢复
4. **禁用并行工具调用**：避免恢复时重复执行
### 应用场景
人工干预机制适用于：
- 需要人工确认的关键操作
- 专业领域的问答系统
- 内容审核和过滤
- 复杂决策辅助
- 错误处理和人工接管
### 最佳实践
1. 合理设计人工干预的触发条件，不要过度使用
2. 提供清晰的中断提示信息
3. 保存完整的对话上下文，确保恢复后连贯性
4. 考虑超时机制，避免无限等待
5. 实现优雅的错误处理
通过 LangGraph 的人工干预机制，我们可以构建更加可靠、可控的 AI Agent 系统，在自动化和人工监督之间找到完美的平衡点。
---
**相关文章推荐**：
- [LangGraph 快速入门：构建你的第一个 AI Agent]()
- [LangGraph 状态管理：如何设计高效的状态流]()
- [LangGraph 工具使用：让 AI 具备行动能力]()
---
*本文作者：码上AI_123*
*原文链接：https://blog.csdn.net/...*
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
11
0- 0
已为社区贡献3条内容
所有评论(0)