人机协作
为敏感的工具操作配置人工审批,让 Agent 可控又安全
从"无人值守"到"关键节点人工把关"的关键一步为什么需要它
你有没有试过让 Agent 自作主张删除文件、发送邮件?
无人值守的风险
- Agent 删了不该删的文件,没有确认机制
- 发了错误的邮件给全公司,无法拦截
- 执行了危险的操作,事后才发现
- 参数传错了(比如发给了全组而非个人),无法修正
- 有些场景需要人在中间补充信息,没有交互渠道
Human-in-the-loop 的解法
- 敏感工具执行前自动暂停,等待人工确认
- 支持批准、拒绝、编辑三种决策
- 可以修改参数后再执行,避免错误操作
- 基于 LangGraph 的
interrupt能力,可靠持久化 - 每个工具可配置不同的审批策略
核心概念
interrupt_on 参数——告诉 Agent 哪些工具需要人工审批
Deep Agents 通过 interruptOn 参数为每个工具配置审批策略。当 Agent 调用配置了中断的工具时,执行会自动暂停,等待人工做出决策后继续。
const agent = createDeepAgent({
model: "google-genai:gemini-3.1-pro-preview",
tools: [removeFile, fetchFile, notifyEmail],
interruptOn: {
// true = 允许全部三种决策(approve/edit/reject)
remove_file: true,
// false = 无需中断,Agent 直接执行
fetch_file: false,
// 对象形式 = 自定义允许的决策类型
notify_email: {
allowedDecisions: ["approve", "reject"] // 只允许批准或拒绝
},
},
checkpointer, // 必需!interrupt 依赖 checkpointer 持久化 Agent 状态
});
MemorySaver 或其他 checkpointer 来在中断和恢复之间持久化 Agent 状态。如果不配置 checkpointer,中断将无法正常工作。
基本配置
定义工具、配置中断策略、创建 Agent——三步完成
下面是一个完整的示例:定义三个工具,每个工具配置不同的审批策略。
import { tool } from "langchain";
import { createDeepAgent } from "deepagents";
import { MemorySaver } from "@langchain/langgraph";
import { z } from "zod";
// 工具 1:删除文件(高风险,允许所有决策)
const removeFile = tool(
async ({ path }: { path: string }) => {
return `Deleted ${path}`;
},
{
name: "remove_file",
description: "Delete a file from the filesystem.",
schema: z.object({ path: z.string() }),
},
);
// 工具 2:读取文件(低风险,无需中断)
const fetchFile = tool(
async ({ path }: { path: string }) => {
return `Contents of ${path}`;
},
{
name: "fetch_file",
description: "Read a file from the filesystem.",
schema: z.object({ path: z.string() }),
},
);
// 工具 3:发送邮件(中等风险,不允许编辑参数)
const notifyEmail = tool(
async ({ to, subject, body }: { to: string; subject: string; body: string }) => {
return `Sent email to ${to}`;
},
{
name: "notify_email",
description: "Send an email.",
schema: z.object({ to: z.string(), subject: z.string(), body: z.string() }),
},
);
// 人机协作必须使用 Checkpointer 来持久化中断时的 Agent 状态
const checkpointer = new MemorySaver();
const agent = createDeepAgent({
model: "google-genai:gemini-3.1-pro-preview",
tools: [removeFile, fetchFile, notifyEmail],
interruptOn: {
remove_file: true, // 高风险:允许全部决策
fetch_file: false, // 低风险:无需审批
notify_email: { allowedDecisions: ["approve", "reject"] }, // 中风险:禁止编辑参数
},
checkpointer, // 必需!用于在中断和恢复之间保存状态
});
remove_file 是高风险操作,允许批准、编辑、拒绝全部三种决策。fetch_file 只是读取,无需中断。notify_email 只允许批准或拒绝,不允许编辑收件人等敏感参数。
三种决策
approve / edit / reject——控制人工能做什么
approve(批准)
使用 Agent 提出的原始参数直接执行工具。
场景:参数没问题,放行
{ type: "approve" }
edit(编辑)
在执行前修改工具参数,比如改收件人、调路径。
场景:参数需要微调
{
type: "edit",
editedAction: {
name: "notify_email",
args: { to: "team@co.com", subject: "...", body: "..." }
}
}
reject(拒绝)
完全跳过此工具调用的执行,返回拒绝消息。
场景:不该执行这个操作
{ type: "reject" }
allowedDecisions 数组精确控制。
按风险配置
根据工具的风险等级定制允许的决策类型
不同工具的风险等级不同,配置策略也应该不同。下面展示了三种典型配置:
const interruptOn = {
// 高风险操作:允许所有选项(批准、编辑、拒绝)
delete_file: { allowedDecisions: ["approve", "edit", "reject"] },
// 中等风险:仅允许批准或拒绝(不允许修改参数)
write_file: { allowedDecisions: ["approve", "reject"] },
// 必须执行的操作:只允许批准(不允许拒绝)
critical_operation: { allowedDecisions: ["approve"] },
};
| 风险等级 | 工具示例 | 允许的决策 | 设计理由 |
|---|---|---|---|
| 高风险 | delete_file |
approve / edit / reject | 需要最大灵活性——可能要改参数或直接取消 |
| 中等风险 | write_file |
approve / reject | 二选一就够了——参数一般不需要人工修改 |
| 强制执行 | critical_operation |
approve | 操作必须执行,但需要人工确认后才能继续 |
true(允许全部决策)。对于"发邮件"这类工具,如果允许 edit,人工可能把收件人改成不该收到的人。按风险等级精细配置才是正确的做法。
处理中断
检测中断、展示信息、收集决策、恢复执行——完整流程
当中断触发时,Agent 暂停执行并返回包含 __interrupt__ 的结果。你的代码需要:检测中断、提取信息、获取用户决策、用 Command 恢复。
import { v7 as uuid7 } from "uuid";
import { Command } from "@langchain/langgraph";
// thread_id 用于标识对话线程,checkpointer 按此 ID 存储状态
const config = { configurable: { thread_id: uuid7() } };
// 第一步:发送用户消息,触发 Agent 执行
let result = await agent.invoke({
messages: [{ role: "user", content: "Delete the file temp.txt" }],
}, config);
// 第二步:检查 Agent 是否在工具调用前被中断
if (result.__interrupt__) {
// __interrupt__ 是 LangGraph 的内置字段,包含所有待审批的工具调用
const interrupts = result.__interrupt__[0].value;
const actionRequests = interrupts.actionRequests; // 待审批的工具调用列表
const reviewConfigs = interrupts.reviewConfigs; // 每个工具允许的决策类型
// 创建 工具名称 → 审查配置 的查找映射,方便后续查询
const configMap = Object.fromEntries(
reviewConfigs.map((cfg) => [cfg.actionName, cfg])
);
// 第三步:向用户展示每个待审批操作及其允许的决策
for (const action of actionRequests) {
const reviewConfig = configMap[action.name];
console.log(`工具: ${action.name}`);
console.log(`参数: ${JSON.stringify(action.args)}`);
console.log(`允许的决策: ${reviewConfig.allowedDecisions}`);
}
// 第四步:收集用户的审批决策(按 actionRequests 顺序)
const decisions = [
{ type: "approve" } // 用户批准了删除操作
];
// 第五步:用 Command({ resume }) 恢复 Agent 执行
result = await agent.invoke(
new Command({ resume: { decisions } }),
config // 必须使用相同的 config,否则找不到之前的状态
);
}
result.__interrupt__ → 提取 actionRequests → 展示给用户 → 收集 decisions → 用 Command({ resume }) 恢复。整个过程的核心就是"暂停-审批-恢复"。
批量审批
Agent 一次调用多个工具时,中断会批量合并
当 Agent 在一次循环中调用多个需要审批的工具时,所有中断会批量合并为一个。你需要按顺序为每个工具调用提供决策。
const config = { configurable: { thread_id: uuid7() } };
let result = await agent.invoke({
messages: [{
role: "user",
content: "Delete temp.txt and send an email to admin@example.com"
}]
}, config);
if (result.__interrupt__) {
const interrupts = result.__interrupt__[0].value;
const actionRequests = interrupts.actionRequests;
// Agent 一次调用了两个需要审批的工具,中断被批量合并
console.assert(actionRequests.length === 2);
// decisions 数组的顺序必须与 actionRequests 一一对应
const decisions = [
{ type: "approve" }, // 第 1 个工具 delete_file → 批准
{ type: "reject" } // 第 2 个工具 send_email → 拒绝
];
result = await agent.invoke(
new Command({ resume: { decisions } }),
config
);
}
决策顺序很重要
decisions数组的顺序必须和actionRequests一致- 第一个决策对应第一个工具调用
- 第二个决策对应第二个工具调用
- 顺序错乱会导致错误的决策应用到错误的工具
设计理念
- 批量合并而非逐个中断,减少交互次数
- 用户可以一次性审批所有操作
- 也可以选择性拒绝部分操作
- 保证了原子性——要么全部决策,要么不恢复
actionRequests 长度完全一致。少一个或多一个都会导致恢复失败。
编辑参数
使用 edit 决策在执行前修改工具参数
当 "edit" 在允许的决策列表中时,人工可以在执行前修改工具参数。比如 Agent 要给全公司发邮件,你觉得范围太广,可以改成只发给特定团队。
if (result.__interrupt__) {
const interrupts = result.__interrupt__[0].value;
const actionRequest = interrupts.actionRequests[0];
// 查看 Agent 原本想传给工具的参数
console.log(actionRequest.args);
// { to: "everyone@company.com", subject: "重要通知", body: "..." }
// 使用 edit 决策修改收件人范围,而不是直接拒绝
const decisions = [{
type: "edit",
editedAction: {
name: actionRequest.name, // 必须指定工具名称,与原始请求一致
args: {
to: "team@company.com", // 将收件人从全公司改为只发团队
subject: "重要通知", // 以下字段必须全量提供(非增量合并)
body: "..."
}
}
}];
// 用修改后的参数恢复执行,工具会使用 editedAction 中的参数
result = await agent.invoke(
new Command({ resume: { decisions } }),
config
);
}
| editedAction 字段 | 是否必需 | 说明 |
|---|---|---|
name | 是 | 工具名称,必须与原始请求一致 |
args | 是 | 修改后的完整参数对象(不是增量,是全量替换) |
editedAction.args 是全量替换,不是增量合并。你需要提供完整的参数对象,包括不需要修改的字段。修改前务必检查原始参数的完整内容。
子 Agent 中断
子 Agent 可以有独立的中断配置,覆盖主 Agent 的设置
使用子 Agent 时,中断可以发生在两个层面:工具调用上(通过 interruptOn 配置)和工具调用内部(通过 interrupt() 函数)。先看第一种。
const agent = createDeepAgent({
tools: [deleteFile, readFile],
interruptOn: {
delete_file: true, // 主 Agent:删除需要审批
read_file: false, // 主 Agent:读取无需审批
},
subagents: [{
name: "file-manager",
description: "Manages file operations",
systemPrompt: "You are a file management assistant.",
tools: [deleteFile, readFile],
interruptOn: {
// 子 Agent 覆盖:读取也需要审批!
delete_file: true,
read_file: true, // 与主 Agent 不同
}
}],
checkpointer
});
主 Agent 的规则
delete_file:需要审批read_file:无需审批- 适用于主 Agent 直接调用工具时
子 Agent 的覆盖规则
delete_file:需要审批(不变)read_file:也需要审批(更严格)- 只影响该子 Agent 内部
interruptOn 是完全覆盖而非合并。如果子 Agent 配置了 interruptOn,主 Agent 的规则不会继承到子 Agent 中。子 Agent 处理中断的方式与主 Agent 完全相同。
工具内部中断
用 interrupt() 函数在工具执行过程中暂停,等待人工响应
子 Agent 的工具可以直接调用 interrupt() 来暂停执行。这比 interruptOn 更灵活——你可以在工具内部的任意位置暂停,并获取人工的响应数据。
import { interrupt } from "@langchain/langgraph";
const requestApproval = tool(
async ({ actionDescription }: { actionDescription: string }) => {
// 调用 interrupt() 会在当前位置暂停工具执行,等待外部传入数据
// interrupt() 的参数会出现在 result.__interrupt__ 中供人类查看
const approval = interrupt({
type: "approval_request", // 自定义中断类型标识
action: actionDescription, // 需要审批的操作描述
message: `请审批: ${actionDescription}`, // 给人类看的提示信息
}) as { approved?: boolean; reason?: string }; // resume 时传入的数据会成为返回值
// interrupt() 恢复后,approval 就是 Command({ resume }) 中传入的对象
if (approval.approved) {
return `操作"${actionDescription}"已批准,继续执行...`;
} else {
return `操作"${actionDescription}"已拒绝。原因: ${
approval.reason || "未提供原因"
}`;
}
},
{
name: "request_approval",
description: "Request human approval before proceeding.",
schema: z.object({
actionDescription: z.string()
.describe("The action that requires approval"),
}),
}
);
interruptOn 在工具调用之前中断(工具还没执行),而 interrupt() 在工具内部中断(工具已经开始执行,执行到一半暂停)。两者适用于不同场景。
完整示例
从创建子 Agent 到处理中断、恢复执行的完整代码
import { createAgent, tool } from "langchain";
import { ChatOpenAI } from "@langchain/openai";
import { HumanMessage } from "@langchain/core/messages";
import { MemorySaver, Command, interrupt } from "@langchain/langgraph";
import { createDeepAgent } from "deepagents";
import { z } from "zod";
async function main() {
const checkpointer = new MemorySaver();
const model = new ChatOpenAI({ model: "gpt-4o-mini", maxTokens: 4096 });
// 编译子 Agent 图,包含带 interrupt() 的工具
const compiledSubagent = createAgent({
model,
tools: [requestApproval], // 使用上面定义的带 interrupt 的工具
name: "approval-agent",
});
// 创建父 Agent,将子 Agent 注册为可委派的目标
const parentAgent = await createDeepAgent({
checkpointer,
subagents: [{
name: "approval-agent",
description: "An agent that can request approvals", // 父 Agent 据此判断何时委派
runnable: compiledSubagent, // 传入编译后的子 Agent 图
}],
});
const config = { configurable: { thread_id: "test_interrupt" } };
// 调用父 Agent → 父 Agent 委派子 Agent → 子 Agent 调用工具 → interrupt 暂停
let result = await parentAgent.invoke({
messages: [new HumanMessage({
content: "Use the approval-agent to request approval for 'deploying to production'."
})],
}, config);
// 检查子 Agent 是否触发了 interrupt
if (result.__interrupt__) {
// interruptValue 就是工具中调用 interrupt() 时传入的对象
const interruptValue = result.__interrupt__[0].value;
console.log(`类型: ${interruptValue.type}`);
console.log(`操作: ${interruptValue.action}`);
console.log(`消息: ${interruptValue.message}`);
// 用 Command 恢复执行,resume 的数据会作为 interrupt() 的返回值
const result2 = await parentAgent.invoke(
new Command({ resume: { approved: true } }), // 传入审批结果
config
);
// 从结果中提取工具返回的消息
const toolMsgs = result2.messages?.filter(m => m.type === "tool");
if (toolMsgs?.length) {
console.log(`工具结果: ${toolMsgs[toolMsgs.length - 1].content}`);
}
}
}
main().catch(console.error);
运行时会输出以下内容:
Invoking agent - sub-agent will use request_approval tool...
Interrupt received!
Type: approval_request
Action: deploying to production
Message: Please approve or reject: deploying to production
Resuming with Command(resume={'approved': true})...
Execution completed!
Tool result: Approval for "deploying to production" has been granted. You can proceed with the deployment.
requestApproval 工具 → 工具内部 interrupt() 暂停 → 返回中断信息 → 人工审批 → Command({ resume }) 恢复 → 工具继续执行并返回结果。
最佳实践
四个关键原则,确保人机协作可靠运行
始终使用 Checkpointer
人机协作需要 Checkpointer 在中断和恢复之间持久化 Agent 状态。
// 必需!没有 checkpointer 就无法恢复
const checkpointer = new MemorySaver();
const agent = createDeepAgent({
tools: [...],
interruptOn: { ... },
checkpointer, // 不要忘记
});
使用相同的 thread ID
恢复执行时,必须使用与调用时相同的 thread_id。
// 调用时和恢复时用同一个 config
const config = {
configurable: { thread_id: uuid7() }
};
// 调用
let result = await agent.invoke(msg, config);
// 恢复 —— 必须用同一个 config!
result = await agent.invoke(
new Command({ resume: { decisions } }),
config // 同一个 config
);
决策顺序与操作匹配
决策列表必须与 actionRequests 的顺序完全一致。
// actionRequests: [delete_file, send_email]
const decisions = [
{ type: "approve" }, // 对应 delete_file
{ type: "reject" }, // 对应 send_email
];
// 长度必须相等,顺序不能错
按风险等级配置
不同工具的风险等级不同,不要一刀切。
const interruptOn = {
// 高风险:灵活审批
delete_file: {
allowedDecisions: ["approve", "edit", "reject"]
},
// 低风险:简单放行或拒绝
read_file: false,
};
thread_id,导致 Agent 找不到之前的状态,报错"no checkpoint found"。确保调用和恢复使用完全相同的 config 对象。
本章回顾
到这里,你已经掌握了 Human-in-the-loop 的完整用法
你学到了什么
interruptOn参数为每个工具配置审批策略- 三种决策:approve / edit / reject
- 通过
__interrupt__检测中断,Command({ resume })恢复执行 - 多工具调用时中断批量合并,决策按顺序匹配
edit决策可以修改参数后再执行- 子 Agent 可以有独立的中断配置
- 工具内部可用
interrupt()实现自定义中断
关键 API 速查
| API | 作用 |
|---|---|
interruptOn | 配置需要审批的工具 |
allowedDecisions | 限制允许的决策类型 |
result.__interrupt__ | 检测是否被中断 |
actionRequests | 获取待审批的工具调用 |
Command({ resume }) | 用决策恢复执行 |
interrupt() | 工具内部自定义中断 |
interruptOn 和 interrupt() 两种机制,覆盖了工具调用前后两种场景的审批需求。
延伸阅读
官方文档
- deepagents npm 包 — 安装和 API 参考
- GitHub 仓库 — 源码和示例
- LangGraph.js 文档 — 中断机制的底层实现
相关概念
- LangGraph Interrupt — 理解中断的底层原理
- LangGraph Command — 恢复执行的完整 API
- MemorySaver — 人机协作必需的 Checkpointer