之前一篇文章里，我使用 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`：删除文件，高风险，需要人工审批

用户输入的任务是：

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

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

# 二、初始化大模型

首先初始化模型：

```python
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 的兼容接口。

需要提前配置环境变量：

```bash
export AliQwen_API="你的 API Key"
```

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

# 三、定义工具

示例中定义了三个工具：

```python
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 提供的内存检查点：

```python
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

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

这里有两个关键点：

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

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

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

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

接下来创建 Agent：

```python
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
)
```

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

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

以及：

```python
checkpointer=checkpointer
```

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

在这个例子中：

- 调用 `query_table_data` 不会中断
- 调用 `delete_table` 会中断
- 调用 `delete_file` 会中断

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

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

# 六、第一次执行：触发中断

第一次执行 Agent：

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

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

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

示例代码中这样获取：

```python
interrupt = result_1["__interrupt__"]
```

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

# 七、查看待审批动作

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

```python
[
    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`：拒绝执行

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

# 八、人工审批：拒绝高危操作

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

```python
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。

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

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

# 九、第二次执行：恢复 Agent

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

```python
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` 的工具会使用人工修改后的参数执行

最后输出结果：

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

# 十、完整代码

完整示例代码如下：

```python
"""
演示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}")
```

![python-hitl.png](https://developer.qcloudimg.com/http-save/yehe-11778967/345f9f70f05d0caa859ca1e5cf974704.png)
# 十一、如果想修改参数后再执行

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

比如 Agent 原本想删除：

```text
zhaoweifeng.txt
```

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

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

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

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

例如：

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

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

# 十二、执行流程总结

整体流程可以概括为：

```text
用户输入任务
    ↓
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 为什么要执行这个工具
- 当前用户是谁
- 当前会话上下文

第三，审批结果要落库。

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

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

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

本文为了演示使用的是：

```python
InMemorySaver()
```

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

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

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

比如：

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

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

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


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

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

本文详解基于 LangChain+LangGraph+DeepAgents 实现 Python 智能体人工介入实战，配置高风险工具中断审批、状态检查点保存与恢复，支持同意 / 拒绝 / 参数编辑，对比 Spring AI Alibaba 方案，附完整可运行源码与生产落地建议。

Java开发

人工智能

后端

编程语言

本文演示了如何用Python实现AI智能体执行高风险动作时的人工审批机制。通过LangChain、LangGraph和DeepAgents技术栈，当Agent执行删除表、删除文件等危险操作时，系统会自动暂停并等待人工确认。文章详细介绍了检查点保存、中断恢复、参数修改等核心功能实现，为AI系统安全执行提供了可靠方案。

对象存储

人机交互

文件系统

数据库

服务器

智能体

Python

GitHub

Redis

Agent

Java

xiaowei

tencentdb-catalog

对象存储COS新用户低至1元！


文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

DeepAgents 人工介入实战｜LangGraph 实现 Agent 高危工具人工审批-腾讯云开发者社区-腾讯云

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

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

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐