为什么 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"
    }
}

这里有两个关键点：

1. checkpointer 用来保存 Agent 的执行状态。
2. 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 不为空时，说明当前执行过程中存在需要人工介入的动作。

七、查看待审批动作