大家好，我是讯享网，很高兴认识大家。这里提供最前沿的Ai技术和互联网信息。



之前一篇文章里，我使用 Spring AI Alibaba 演示了智能体执行过程中的人工介入能力。那篇文章的核心思路是：当 Agent 准备执行某些高风险动作时，不要让它直接执行，而是先暂停下来，把待执行动作交给人工审批，审批通过后再继续执行。感兴趣的小伙伴可以通过下面链接回顾一下：https://www.lucaju.cn/index.php/archives/165/

这篇文章换一个技术栈，使用 Python 版本来实现同样的能力。

本次示例基于：

• LangChain：负责模型和工具抽象
• LangGraph：负责执行状态、检查点和恢复执行
• DeepAgents：负责创建支持工具调用和中断审批的 Agent
• 通义千问兼容 OpenAI API：作为底层大模型

Agent 最大的价值是可以根据用户目标自主规划并调用工具。

但并不是所有工具都适合完全自动执行。比如：

• 删除数据库表
• 删除文件
• 发起转账
• 修改线上配置
• 调用外部系统执行不可逆操作

这些动作一旦执行错误，影响可能非常大。

所以比较合理的模式是：

普通查询类动作可以让 Agent 自动执行，高风险动作必须先进入人工审批。

在本文示例中，我们定义了三个工具：

• query_table_data：查询表数据，低风险，可以自动执行
• delete_table：删除数据表，高风险，需要人工审批
• delete_file：删除文件，高风险，需要人工审批

用户输入的任务是：

先查询product表的数据！再删除user表，最后，删除lucaju.txt文件 

Agent 会先分析任务，并尝试依次调用工具。但当执行到删除表、删除文件这类高风险动作时，会被框架中断，等待人工确认。

首先初始化模型：

import os from langchain.chat_models import init_chat_model llm = init_chat_model( model="kimi-k2.5", model_provider="openai", api_key=os.getenv("AliQwen_API"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", model_kwargs={"reasoning_effort": "none"}, ) 

这里使用的是 OpenAI 兼容模式，所以 model_provider 设置为 openai，同时把 base_url 指向阿里云 DashScope 的兼容接口。

需要提前配置环境变量：

export AliQwen_API="你的 API Key" 

model_kwargs={"reasoning_effort": "none"} 是模型调用参数，可以根据实际模型支持情况调整。

示例中定义了三个工具：

from langchain.tools import tool @tool def delete_table(table_name: str) -> str: """ 删除指定的表 """ return f"删除表{table_name}" @tool def delete_file(file_name: str) -> str: """ 删除指定的文件 """ return f"删除文件{file_name}" @tool def query_table_data(table_name: str) -> str: """ 查询指定表的数据 """ return f"查询表{table_name}的数据" 

这里为了演示效果，工具内部只是返回字符串，并没有真的连接数据库或删除文件。

在真实项目中，delete_table 可能会执行 SQL，delete_file 可能会操作对象存储或服务器文件系统。这类工具就非常适合加人工审批。

Human-in-the-loop 的核心不是简单地打印一句“是否确认”，而是让 Agent 的执行流程真正暂停下来，并且后续可以从暂停点继续执行。

这就需要保存执行状态。

示例代码中使用了 LangGraph 提供的内存检查点：

from langgraph.checkpoint.memory import InMemorySaver checkpointer = InMemorySaver() config = { "configurable": { "thread_id": "123" } } 

这里有两个关键点：

1. checkpointer 用来保存 Agent 的执行状态。
2. thread_id 用来标识当前会话。

当 Agent 执行到需要人工介入的节点时，LangGraph 会把当前状态保存下来。等人工审批完成后，再通过同一个 thread_id 找回之前的状态，并继续执行。

如果没有检查点，框架就不知道应该从哪里恢复执行。

接下来创建 Agent：

from deepagents import create_deep_agent main_agent = create_deep_agent( model=llm, name="主智能体", system_prompt="回答使用中文，调用对应的工具实现对应的功能！", tools=[delete_table, delete_file, query_table_data], interrupt_on={"delete_table": True, "delete_file": True}, checkpointer=checkpointer ) 

这里最关键的是两个参数：

interrupt_on={"delete_table": True, "delete_file": True} 

以及：

checkpointer=checkpointer 

interrupt_on 用来声明哪些工具调用需要中断。

在这个例子中：

• 调用 query_table_data 不会中断
• 调用 delete_table 会中断
• 调用 delete_file 会中断

也就是说，Agent 可以自动查询数据，但不能自动删除表或文件。

这就是人工介入的核心配置。

第一次执行 Agent：

result_1 = main_agent.invoke( { "messages": [ { "role": "user", "content": "先查询product表的数据！再删除user表，最后，删除lucaju.txt文件", } ] }, config=config, ) 

这次调用并不一定会完整执行完所有工具。

如果执行链路中包含需要人工审批的工具，Agent 会暂停，并把中断信息放到返回结果的 __interrupt__ 字段中。

示例代码中这样获取：

interrupt = result_1["__interrupt__"] 

当 interrupt 不为空时，说明当前执行过程中存在需要人工介入的动作。

__interrupt__ 中会包含本次待审批的工具调用信息，例如：

[ Interrupt( value={ "action_requests": [ { "name": "delete_table", "args": {"table_name": "user"}, "description": "Tool execution requires approval..." }, , "description": "Tool execution requires approval..." } ], "review_configs": [ { "action_name": "delete_table", "allowed_decisions": ["approve", "edit", "reject"] }, { "action_name": "delete_file", "allowed_decisions": ["approve", "edit", "reject"] } ] } ) ] 

这里有两个比较重要的字段。

action_requests 表示 Agent 想要执行哪些工具：

• 工具名称
• 工具参数
• 工具描述

review_configs 表示这些动作允许哪些审批结果：

• approve：同意执行
• edit：修改参数后执行
• reject：拒绝执行

这三个动作基本覆盖了常见的人审场景。

示例代码中把删除表和删除文件都拒绝掉：

decisions = [] action_requests = interrupt[0].value["action_requests"] print(f"当前人机交互工具：{action_requests}") for action_request in action_requests: if action_request["name"] == "delete_table": decisions.append({"type": "reject"}) elif action_request["name"] == "delete_file": decisions.append({"type": "reject"}) 

这里的 decisions 顺序需要和 action_requests 对应。

也就是说，如果 Agent 申请了两个动作：

1. 删除 user 表
2. 删除 zhaoweifeng.txt 文件

那么人工审批结果也应该按顺序给出两个 decision。

在这个示例中，两个高风险动作都被拒绝：

[ {"type": "reject"}, {"type": "reject"} ] 

审批完成后，不需要重新传入用户消息，而是通过 Command(resume=...) 恢复执行：

from langgraph.types import Command result_2 = main_agent.invoke( Command( resume={ "decisions": decisions } ), config=config, ) 

这里一定要继续传入相同的 config，尤其是相同的 thread_id。

因为 Agent 要根据 thread_id 找到之前暂停的执行状态。

恢复执行后，框架会根据人工审批结果继续处理：

• 被 approve 的工具会继续执行
• 被 reject 的工具不会执行
• 被 edit 的工具会使用人工修改后的参数执行

最后输出结果：

print(f"最终结果{result_2['messages'][-1].content}") 

完整示例代码如下：

""" 演示human-in-the-loop模式, 必须使用记忆功能 """ import os from deepagents import create_deep_agent from langchain.chat_models import init_chat_model from langchain.tools import tool from langgraph.checkpoint.memory import InMemorySaver from langgraph.types import Command # 初始化大模型 llm = init_chat_model( model="kimi-k2.5", model_provider="openai", api_key=os.getenv("AliQwen_API"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", model_kwargs={"reasoning_effort": "none"}, ) # 删除表工具 @tool def delete_table(table_name: str) -> str: """ 删除指定的表 """ return f"删除表{table_name}" # 删除文件 @tool def delete_file(file_name: str) -> str: """ 删除指定的文件 """ return f"删除文件{file_name}" # 查询表数据 @tool def query_table_data(table_name: str) -> str: """ 查询指定表的数据 """ return f"查询表{table_name}的数据" # 设置检查点 checkpointer = InMemorySaver() # 配置检查点 config = { "configurable": { "thread_id": "123" } } # 创建deepagent，同时给高危工具设置人机交互 main_agent = create_deep_agent( model=llm, name="主智能体", system_prompt="回答使用中文，调用对应的工具实现对应的功能！", tools=[delete_table, delete_file, query_table_data], interrupt_on={"delete_table": True, "delete_file": True}, checkpointer=checkpointer ) # 预执行，本次不会真正的执行所有工具。 # 如果执行链路中存在人机交互节点，框架会暂停，并返回 __interrupt__。 result_1 = main_agent.invoke( { "messages": [ { "role": "user", "content": "先查询product表的数据！再删除user表，最后，删除lucaju.txt文件", } ] }, config=config, ) # 检查本次执行是否存在人机交互动作 interrupt = result_1["__interrupt__"] if interrupt: print("存在人机交互动作") # 定义一个列表，存储所有审批结果 decisions = [] # 获取所有待审批动作 action_requests = interrupt[0].value["action_requests"] print(f"当前人机交互工具：{action_requests}") for action_request in action_requests: if action_request["name"] == "delete_table": decisions.append({"type": "reject"}) elif action_request["name"] == "delete_file": decisions.append({"type": "reject"}) # 再次执行，不需要传会话内容，只需要传审批意见和 config result_2 = main_agent.invoke( Command( resume={ "decisions": decisions } ), config=config, ) print(f"最终结果{result_2['messages'][-1].content}") 

除了直接拒绝，我们也可以修改工具参数后再执行。

比如 Agent 原本想删除：

zhaoweifeng.txt 

人工审批时可以把文件名改成另一个值：

decisions.append({ "type": "edit", "edited_action": { "name": action_request["name"], "args": { "file_name": "new-file.txt" } } }) 

这时框架不会使用 Agent 原始生成的参数，而是使用人工编辑后的参数继续执行工具。

这在实际业务中非常有用。

例如：

• Agent 选择的表名不准确，人工改成正确表名
• Agent 生成的文件路径不安全，人工改成允许路径
• Agent 生成的金额过大，人工改成合理金额
• Agent 生成的收件人错误，人工改成正确收件人

相比简单的同意或拒绝，edit 让人工介入更灵活。

整体流程可以概括为：

用户输入任务 ↓ Agent 分析任务并规划工具调用 ↓ 普通工具自动执行 ↓ 遇到 interrupt_on 配置的高风险工具 ↓ LangGraph 保存状态并中断执行 ↓ 人工读取 __interrupt__ 中的 action_requests ↓ 人工给出 approve / edit / reject ↓ 使用 Command(resume=...) 恢复执行 ↓ Agent 根据审批结果继续完成任务 

这套机制的关键点是：

• interrupt_on：定义哪些工具需要人工审批
• checkpointer：保存执行状态
• thread_id：标识同一次会话
• __interrupt__：获取待审批动作
• Command(resume=...)：把审批结果送回 Agent 并恢复执行

从思想上看，Python 版本和 Spring AI Alibaba 版本是一致的：

Agent 不应该无限制地自动执行所有动作，高风险动作需要进入人工审批流程。

但实现方式上有所不同。

Spring AI Alibaba 更偏向 Java 生态，适合和 Spring Boot、企业系统、审批流、权限体系结合。

而 LangChain + LangGraph + DeepAgents 的 Python 方案更偏向实验、原型验证和 Agent 工作流编排。尤其是 LangGraph 的 checkpoint 和 resume 机制，让“暂停后恢复”这件事变得非常自然。

如果项目本身是 Java 技术栈，可以优先考虑 Spring AI Alibaba。

如果项目本身是 Python 技术栈，或者正在做 Agent 编排、工具调用、多步骤任务规划，那么 DeepAgents 这套方式会比较顺手。

在真实项目中使用人工介入时，建议注意以下几点。

第一，高风险工具要显式配置。

不要只依赖 prompt 告诉模型“删除前要确认”。更可靠的方式是像本文一样，在框架层面对工具进行拦截。

第二，审批信息要完整展示。

人工审批时至少要看到：

• 工具名称
• 工具参数
• Agent 为什么要执行这个工具
• 当前用户是谁
• 当前会话上下文

第三，审批结果要落库。

生产环境中，审批记录应该持久化，包括：

• 谁审批的
• 什么时间审批的
• 原始参数是什么
• 修改后的参数是什么
• 最终是通过、拒绝还是编辑后通过

第四，检查点不要只用内存。

本文为了演示使用的是：

InMemorySaver() 

生产环境更建议使用数据库、Redis 或其他持久化存储。否则服务重启后，暂停中的 Agent 状态会丢失。

第五，审批权限要和业务系统打通。

不是所有人都应该可以批准所有工具调用。

比如：

• 普通用户只能审批自己的任务
• 管理员可以审批团队任务
• 涉及资金、删除、发布的动作需要更高权限

人工介入不是简单的弹窗确认，而应该是完整的安全控制链路。