第 12 章

人机协作

为敏感的工具操作配置人工审批,让 Agent 可控又安全

从"无人值守"到"关键节点人工把关"的关键一步

为什么需要它

你有没有试过让 Agent 自作主张删除文件、发送邮件?

无人值守的风险

  • Agent 删了不该删的文件,没有确认机制
  • 发了错误的邮件给全公司,无法拦截
  • 执行了危险的操作,事后才发现
  • 参数传错了(比如发给了全组而非个人),无法修正
  • 有些场景需要人在中间补充信息,没有交互渠道

Human-in-the-loop 的解法

  • 敏感工具执行前自动暂停,等待人工确认
  • 支持批准、拒绝、编辑三种决策
  • 可以修改参数后再执行,避免错误操作
  • 基于 LangGraph 的 interrupt 能力,可靠持久化
  • 每个工具可配置不同的审批策略
问题核心:当你让 Agent 执行"删文件、发邮件、改数据库"这类不可逆操作时,你需要一个"暂停按钮"——在 Agent 真正动手之前,人先看一眼,确认没问题再放行。Deep Agents 通过 LangGraph 的中断机制实现了这个能力。

核心概念

interrupt_on 参数——告诉 Agent 哪些工具需要人工审批

Deep Agents 通过 interruptOn 参数为每个工具配置审批策略。当 Agent 调用配置了中断的工具时,执行会自动暂停,等待人工做出决策后继续。

图:Human-in-the-loop 工作流——中断与恢复
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 状态
});
必须使用 Checkpointer:人机协作依赖 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,否则找不到之前的状态
  );
}
图:中断处理完整调用链
关键步骤:调用 Agent → 检查 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 内部
图:主/子 Agent 中断配置覆盖关系
覆盖逻辑:每个子 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.
执行流程:用户调用父 Agent → 父 Agent 委派子 Agent → 子 Agent 调用 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()工具内部自定义中断
核心结论:Human-in-the-loop 不是让 Agent 变慢,而是让 Agent 变可信。在不可逆操作前加一道人工确认,是 AI 系统从"能用"走向"敢用"的关键一步。Deep Agents 通过 interruptOn 和 interrupt() 两种机制,覆盖了工具调用前后两种场景的审批需求。

延伸阅读

继续本课程

官方文档

相关概念

  • LangGraph Interrupt — 理解中断的底层原理
  • LangGraph Command — 恢复执行的完整 API
  • MemorySaver — 人机协作必需的 Checkpointer