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

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

本次示例基于：

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

一、为什么 Agent 需要人工介入

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 可能会操作对象存储或服务器文件系统。这类工具就非常适合加人工审批。

四、为什么必须配置 Checkpointer

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

这就需要保存执行状态。

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

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

config = {
   
    "configurable": {
   
        "thread_id": "123"
    }
}

这里有两个关键点：

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

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

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

五、创建支持人工介入的 DeepAgent

接下来创建 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..."
                },
                {
   
                    "name": "delete_file",
                    "args": {
   "file_name": "zhaoweifeng.txt"},
                    "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 申请了两个动作：

删除 user 表
删除 zhaoweifeng.txt 文件

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

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

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

九、第二次执行：恢复 Agent

审批完成后，不需要重新传入用户消息，而是通过 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 并恢复执行

十三、和 Spring AI Alibaba 版本的对比

从思想上看，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 状态会丢失。

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

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

比如：

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

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

关于DeepAgents的练习代码已经上传到了我的GitHub，欢迎fork & star
https://github.com/Jucunqi/deepagents-practice.git

DeepAgents 人工介入实战｜LangGraph 实现 Agent 高危工具人工审批