第 5 章 · 人工介入（Human-in-the-Loop）

🎯 本章目标

理解为什么「能暂停、能等人类拍板」是智能体真正上线的必要条件
掌握 interrupt()：在节点里让图暂停，把一段 payload（问题、草稿、待审批的操作）交给人类
看懂「暂停那一刻 invoke 返回的对象长什么样」（__interrupt__ 到底是什么），以及 state.next 如何告诉你停在哪
用 new Command({ resume }) 从断点恢复，把人类的答复接力给下游节点
学会给「危险工具」装审批闸门（approve / reject），并用副作用日志证明工具到底执行没执行
结合第 4 章的存档能力，做「人工编辑 + 时间旅行重放」：人类改一句，后半段自动重跑
掌握审批的超时兜底策略：人类不响应时，用「安全默认值」（默认拒绝）让流程继续

本章代码

📌 代码约定：正文的代码块都保持可直接运行（含 import 与模型初始化）。若某段为聚焦概念的节选，会明确标注「节选」并指向同名的 code/ 完整文件。所有示例的输出都是真实运行结果，不是编造的示例输出。

文件	内容
`code/01-interrupt-basics.ts`	HITL 最小闭环：`interrupt` → `__interrupt__` → `Command` 恢复
`code/02-interrupt-edit-and-resume.ts`	人类修改内容后再继续，对比「同意原稿」与「改写」
`code/03-tool-approval.ts`	危险工具的人工审批，用副作用账本证明执行与否
`code/04-time-travel-replay.ts`	人工编辑 + 时间旅行重放（结合第 4 章存档）
`code/05-approval-with-timeout.ts`	审批超时兜底：超时默认拒绝

一、HITL：为什么「暂停」这么重要

前面几章，我们的图都是「一口气跑到底」：invoke() → 走完所有节点 → 返回。智能体越强，这个假设就越危险 —— 因为它开始自己动手做事了。

真实业务里，有几件事必须让人参与：

场景	人类要做的动作
审批：智能体要转账 / 删文件 / 发邮件	点「同意」或「拒绝」
修改：智能体写了一段文案 / 一封邮件	改成人类满意的版本，再让它继续
审阅工具调用：智能体打算调用某工具	先看看参数对不对，再放行
多轮澄清：需求含糊，缺关键信息	补充信息，让它接着往下做

这四件事有一个共同点：流程在中间必须停下来，等一个人。这就是 Human-in-the-Loop（HITL，人类在环）。

LangGraph 给出的答案是一个函数：interrupt()。它的工作方式很朴素：

       正常执行                          遇到 interrupt()
   ┌──────────────┐                  ┌──────────────┐
   │   节点 A      │                  │   节点 B      │
   └──────┬───────┘                  └──────┬───────┘
          │                                  │  interrupt(payload)
          ▼                                  ▼
       继续走                     图在此"暂停"，payload 交给人类
                                   （状态已存进 checkpointer）
                                            │
                                            │  人类处理完
                                            ▼
                              graph.invoke(new Command({ resume: 人类答复 }))
                                            │
                                            ▼
                                  从断点继续往下跑

三个必须记住的要点：

interrupt() 是靠「抛出一个特殊信号」实现暂停的。 所以不要用 try/catch 把它吞掉 —— 一旦被 catch，暂停就失效了。
暂停后，图的状态被存进 checkpointer。 恢复时靠 configurable.thread_id 找回现场。没有 checkpointer，interrupt 无法工作（这是第 4 章埋的伏笔）。
恢复用的是 new Command({ resume: 值 })。 这个 值 会成为被中断那个节点里 interrupt() 的返回值。

💡 一句话记忆：interrupt() 把「一个问题」交出去，Command({ resume }) 把「一个答案」送回来；中间隔着的可能是几秒，也可能是几天 —— 因为状态存在 saver 里，进程重启都不怕（第 4 章的 SqliteSaver）。

二、最小闭环：interrupt → interrupt → resume

先看一个最干净的 HITL 例子：写草稿 → 人工审核 → 定稿。审核节点里调用 interrupt()，图就在这里停下。

示例 1：把「暂停时拿到的对象」完整打印出来

typescript

/**
 * 示例 1：HITL 最小闭环 —— interrupt 暂停 → invoke 拿到 __interrupt__ → Command 恢复
 *
 * 这张图讲清楚人工介入（Human-in-the-Loop）最核心的三步：
 *   ① 节点里调用 interrupt(payload)  →  图在这里"暂停"，把 payload 交给人类
 *   ② 调用方 invoke 拿到结果里的 __interrupt__  →  这就是"暂停时拿到的对象"
 *   ③ 人类给出答复后，用 new Command({ resume: 答复 }) 再次 invoke  →  图从断点续跑
 *
 *    START → draft → humanReview(interrupt) → finalize → END
 *
 * ★ 关键前提：interrupt 必须配合 checkpointer 使用（否则无法暂停后续跑）。
 *   暂停后状态会被存进 checkpointer，恢复时靠 configurable.thread_id 找回。
 */
import {
  StateGraph,
  START,
  END,
  Annotation,
  MemorySaver,
  Command,
  interrupt,
  isInterrupted,
  type Interrupt,
} from "@langchain/langgraph";
import { banner } from "../../lib/model.js";

// ── 状态：一个话题、一份草稿、一个人类决定 ─────────────────────
const ReviewState = Annotation.Root({
  /** 要写的话题 */
  topic: Annotation<string>,
  /** 自动生成的草稿 */
  draft: Annotation<string>,
  /** 人类的决定（审核节点里由 interrupt 的 resume 值填回） */
  decision: Annotation<string>,
  /** 最终定稿 */
  finalText: Annotation<string>,
});

// ── 节点 1：生成草稿（这里不调模型，专注讲 interrupt 机制）──────
function draftNode(state: typeof ReviewState.State) {
  console.log("   [节点 writeDraft] 正在起草…");
  return { draft: `《${state.topic}》初稿：这是自动生成的一段内容，等待人工确认。` };
}

// ── 节点 2：人工审核 —— interrupt 就发生在这里 ─────────────────
function humanReviewNode(state: typeof ReviewState.State) {
  console.log("   [节点 review] 即将调用 interrupt，让图在这里暂停…");

  // ★ interrupt() 做了两件事：
  //   • 第一次跑到这里：抛出特殊信号，把 payload 交出去，图暂停
  //   • 恢复后重跑这个节点：不再暂停，而是把人类 resume 的值原样返回
  const humanAnswer = interrupt({
    kind: "approval_request",
    question: "这份草稿可以发布吗？请回复 approve 或 reject。",
    draft: state.draft,
  });

  // 这一行只有在"人类给了答复、图恢复执行"之后才会打印
  console.log(`   [节点 review] 收到人类答复（resume 原样返回）：${JSON.stringify(humanAnswer)}`);
  return { decision: String(humanAnswer) };
}

// ── 节点 3：根据人类决定定稿 ───────────────────────────────────
function finalizeNode(state: typeof ReviewState.State) {
  const approved = state.decision.toLowerCase().includes("approve");
  console.log(`   [节点 finalize] 人类决定 = ${state.decision} → ${approved ? "发布" : "退回"}`);
  return {
    finalText: approved
      ? `✅ 已发布：${state.draft}`
      : `❌ 已退回，未发布：${state.draft}`,
  };
}

async function main() {
  banner("示例 1：HITL 最小闭环（interrupt → __interrupt__ → Command 恢复）");

  const graph = new StateGraph(ReviewState)
    .addNode("writeDraft", draftNode)
    .addNode("review", humanReviewNode)
    .addNode("finalize", finalizeNode)
    .addEdge(START, "writeDraft")
    .addEdge("writeDraft", "review")
    .addEdge("review", "finalize")
    .addEdge("finalize", END)
    // ★ 必须挂 checkpointer，interrupt 才能"暂停->续跑"
    .compile({ checkpointer: new MemorySaver() });

  const config = { configurable: { thread_id: "demo-1" } };

  // ── 第 1 步：正常 invoke，图会在 humanReview 处暂停 ───────────
  console.log("▶ 第 1 次 invoke（不传 resume）—— 图应该停在中途\n");
  const paused = await graph.invoke(
    { topic: "LangGraph 入门" },
    config
  );

  console.log("\n📦 invoke 返回的对象（完整打印，这就是「暂停时拿到的样子」）：");
  console.log(JSON.stringify(paused, null, 2));

  // ── 第 2 步：用 isInterrupted 判断"确实被暂停了" ──────────────
  if (isInterrupted(paused)) {
    const itr: Interrupt = paused.__interrupt__[0];
    console.log("\n🔍 isInterrupted(paused) === true，图确实暂停了");
    console.log(`   暂停在节点：${(await graph.getState(config)).next.join(", ")}`);
    console.log(`   interrupt 的 value：${JSON.stringify(itr.value)}`);
  }

  // ── 第 3 步：人类答复，用 Command({ resume }) 恢复 ─────────────
  console.log("\n▶ 第 2 次调用：new Command({ resume: \"approve\" }) 恢复执行\n");
  const resumed = await graph.invoke(
    new Command({ resume: "approve" }),
    config
  );

  console.log("\n📦 恢复并跑完后的最终状态：");
  console.log(JSON.stringify(resumed, null, 2));
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash

npx tsx 05-human-in-the-loop/code/01-interrupt-basics.ts

真实输出：

========================================================================
  示例 1：HITL 最小闭环（interrupt → __interrupt__ → Command 恢复）
========================================================================

▶ 第 1 次 invoke（不传 resume）—— 图应该停在中途

   [节点 writeDraft] 正在起草…
   [节点 review] 即将调用 interrupt，让图在这里暂停…

📦 invoke 返回的对象（完整打印，这就是「暂停时拿到的样子」）：
{
  "topic": "LangGraph 入门",
  "draft": "《LangGraph 入门》初稿：这是自动生成的一段内容，等待人工确认。",
  "__interrupt__": [
    {
      "id": "09bb928a1df363fcf7f7596ec6334cf0",
      "value": {
        "kind": "approval_request",
        "question": "这份草稿可以发布吗？请回复 approve 或 reject。",
        "draft": "《LangGraph 入门》初稿：这是自动生成的一段内容，等待人工确认。"
      }
    }
  ]
}

🔍 isInterrupted(paused) === true，图确实暂停了
   暂停在节点：review
   interrupt 的 value：{"kind":"approval_request","question":"这份草稿可以发布吗？请回复 approve 或 reject。","draft":"《LangGraph 入门》初稿：这是自动生成的一段内容，等待人工确认。"}

▶ 第 2 次调用：new Command({ resume: "approve" }) 恢复执行

   [节点 review] 即将调用 interrupt，让图在这里暂停…
   [节点 review] 收到人类答复（resume 原样返回）："approve"
   [节点 finalize] 人类决定 = approve → 发布

📦 恢复并跑完后的最终状态：
{
  "topic": "LangGraph 入门",
  "draft": "《LangGraph 入门》初稿：这是自动生成的一段内容，等待人工确认。",
  "decision": "approve",
  "finalText": "✅ 已发布：《LangGraph 入门》初稿：这是自动生成的一段内容，等待人工确认。"
}

⚠️ 上面 __interrupt__[0].id 是一串随机哈希，每次运行都不同，这很正常；value 里装的就是你传进 interrupt() 的 payload。

这段输出里有三个读者最容易迷糊的地方，逐个拆开：

① 「暂停时拿到的对象」就是带 __interrupt__ 的那个普通状态对象。

{
  "topic": "...",              ← 正常状态字段，照常返回
  "draft": "...",              ← 正常状态字段
  "__interrupt__": [ {...} ]   ← ★ 额外的：说明"图停住了"，里面是 interrupt 的 payload
}

调用方不需要抛异常、不需要 try/catch —— invoke 正常返回，只是返回值里多了一个 __interrupt__ 字段。这就是「暂停」的全部形态。

② 用 isInterrupted() 判断是否暂停，比手写 "__interrupt__" in result 更规范。

typescript

import { isInterrupted } from "@langchain/langgraph";

if (isInterrupted(paused)) {
  const itr = paused.__interrupt__[0];  // 一个 Interrupt 对象
}

__interrupt__ 是一个数组 —— 因为一个 super-step 里可能有多个节点同时中断。数组里每个 Interrupt 形如：

typescript

type Interrupt<Value = any> = {
  id?: string;                 // 本次中断的唯一 id
  value?: Value;               // 你传给 interrupt() 的那个 payload
  response_schema?: JSONSchema; // 若用了 responseSchema 选项，这里是给前端的表单描述
};

③ state.next 告诉你「停在哪一步」。

暂停在节点：review

这行来自 (await graph.getState(config)).next —— 它是一个待执行节点名数组。图暂停在 review 节点内部，所以 next 就是 ["review"]。恢复时，LangGraph 会从 review 节点重新执行。

⚠️ 极其重要的一点：恢复不是「从 interrupt() 那一行之后接着跑」，而是把整个 review 节点从头再跑一遍。只不过这一次，interrupt() 不再暂停，而是直接返回人类的值。所以：
节点里 interrupt() 之前的代码会再执行一次（上面的日志里 [节点 review] 即将调用 interrupt… 确实打印了两次）。
因此 interrupt() 之前的代码要尽量无副作用；如果一定要有副作用，必须保证幂等（重跑不出错、不重复扣款）。
这条规律会贯穿本章所有示例。

三、人类修改后再继续

示例 1 里人类只回答了一个「决定」（approve / reject）。但很多时候，人是要动手改内容的。好消息是：做法一模一样，只是 resume 的值从「一个决定」变成「一段改好的正文」。

示例 2：resume 传回「人类改过的内容」，下游照单全收

typescript

/**
 * 示例 2：人类修改后再继续 —— resume 时把「人类改过的内容」传回下游
 *
 * 示例 1 里，人类只回答了 approve / reject（一个决定）。
 * 但很多时候，人类不只是"点同意/不同意"，而是要**动手改内容**。
 *
 * 做法完全一样，只是 resume 的值变成一个"改好的正文"：
 *   节点里 interrupt({...原始草稿...})  →  人类拿到草稿，改完返回
 *   graph.invoke(new Command({ resume: "人类改好的文字" }), config)
 *   节点里 interrupt() 的返回值 = 人类改后的文字  →  交给下游节点使用
 *
 * 本示例跑两遍，对比两种人类反应：
 *   A) 人类"同意原稿"（resume 传回原始草稿）      →  下游用原稿
 *   B) 人类"改写"（resume 传回一段自己写的文字）   →  下游用改写稿
 */
import {
  StateGraph,
  START,
  END,
  Annotation,
  MemorySaver,
  Command,
  interrupt,
  isInterrupted,
  type Interrupt,
} from "@langchain/langgraph";
import { HumanMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

// ── 状态：话题、模型初稿、人类定稿、发布结果 ───────────────────
const EditState = Annotation.Root({
  topic: Annotation<string>,
  /** 模型生成的初稿 */
  draft: Annotation<string>,
  /** 人类最终定下的正文（可能改过，也可能原样通过） */
  edited: Annotation<string>,
  /** 发布结果 */
  published: Annotation<string>,
});

async function main() {
  banner("示例 2：人类修改后再继续（interrupt 传草稿 → resume 传改稿）");
  assertChatKey();

  const model = createModel({ temperature: 0 });

  // ── 节点 1：模型写初稿 ─────────────────────────────────────
  async function writeDraft(state: typeof EditState.State) {
    console.log(`   [节点 writeDraft] 让模型为「${state.topic}」写一段 30 字左右的简介…`);
    const res = await model.invoke([
      new HumanMessage(`用中文写一段 30 字左右、面向初学者的简介，主题：${state.topic}。只输出正文，不要标题。`),
    ]);
    const draft = String(res.content).trim();
    console.log(`   [节点 writeDraft] 初稿：${draft}`);
    return { draft };
  }

  // ── 节点 2：人工编辑 —— interrupt 把草稿交给人类 ─────────────
  function humanEdit(state: typeof EditState.State) {
    // first run：把草稿交出去并暂停；resume 后：返回值 = 人类改好的正文
    const finalText = interrupt({
      kind: "edit_request",
      message: "请审阅下面的草稿：可以直接批准，也可以改写后把新正文返回。",
      draft: state.draft,
    });
    console.log("   [节点 humanEdit] 拿到的最终正文（来自人类 resume 值）：");
    console.log(`      ${String(finalText)}`);
    return { edited: String(finalText) };
  }

  // ── 节点 3：发布（下游用的就是人类定下的正文）────────────────
  function publish(state: typeof EditState.State) {
    console.log("   [节点 publish] 下游节点开箱即用的就是被改过的内容");
    return { published: `📢 已发布：${state.edited}` };
  }

  const graph = new StateGraph(EditState)
    .addNode("writeDraft", writeDraft)
    .addNode("humanEdit", humanEdit)
    .addNode("publish", publish)
    .addEdge(START, "writeDraft")
    .addEdge("writeDraft", "humanEdit")
    .addEdge("humanEdit", "publish")
    .addEdge("publish", END)
    .compile({ checkpointer: new MemorySaver() });

  /** 跑一轮：先暂停拿到草稿，再按给定策略 resume，返回最终状态 */
  async function runOnce(threadId: string, decision: (draft: string) => string) {
    const config = { configurable: { thread_id: threadId } };

    // ① 跑到暂停
    const paused = await graph.invoke({ topic: "LangGraph 是什么" }, config);
    if (!isInterrupted(paused)) throw new Error("预期会暂停，但没有暂停");
    const draft = (paused.__interrupt__[0] as Interrupt<{ draft: string }>).value?.draft ?? "";

    // ② 人类决定 resume 什么内容
    const resumeWith = decision(draft);
    console.log(`   👤 人类的回复：${resumeWith}\n`);

    // ③ 用人类内容恢复，跑完
    return await graph.invoke(new Command({ resume: resumeWith }), config);
  }

  // ── A) 人类同意原稿：resume 原样传回草稿 ────────────────────
  console.log("━━━ A) 人类同意原稿 ━━━");
  console.log("▶ 第 1 次 invoke（暂停）\n");
  const a = await runOnce("edit-thread-A", (draft) => draft);
  console.log("\n📦 A 的最终状态：");
  console.log(`   初稿 draft   = ${a.draft}`);
  console.log(`   定稿 edited  = ${a.edited}`);
  console.log(`   发布 published = ${a.published}`);
  console.log(`   → draft === edited ? ${a.draft === a.edited ? "是（原样通过）" : "否"}`);

  // ── B) 人类改写：resume 一段自己写的正文 ────────────────────
  const HUMAN_REWRITE =
    "LangGraph 是 LangChain 生态里的「智能体编排框架」，用图的方式组织多步骤流程。" +
    "（这段是人工改写的，和模型原稿不一样。）";
  console.log("\n━━━ B) 人类改写 ━━━");
  console.log("▶ 第 1 次 invoke（暂停）\n");
  const b = await runOnce("edit-thread-B", () => HUMAN_REWRITE);
  console.log("\n📦 B 的最终状态：");
  console.log(`   初稿 draft   = ${b.draft}`);
  console.log(`   定稿 edited  = ${b.edited}`);
  console.log(`   发布 published = ${b.published}`);
  console.log(`   → draft === edited ? ${b.draft === b.edited ? "是（原样通过）" : "否（被人类改过）"}`);

  // ── 对照结论 ───────────────────────────────────────────────
  console.log("\n📌 对照结论：");
  console.log("   A 组 resume 传回原稿 → 下游 publish 用的就是模型初稿；");
  console.log("   B 组 resume 传回改写稿 → 下游 publish 用的是「人类改后的内容」。");
  console.log("   interrupt 的返回值，就是「人类交给下游的接力棒」。");
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash

npx tsx 05-human-in-the-loop/code/02-interrupt-edit-and-resume.ts

真实输出：

========================================================================
  示例 2：人类修改后再继续（interrupt 传草稿 → resume 传改稿）
========================================================================

━━━ A) 人类同意原稿 ━━━
▶ 第 1 次 invoke（暂停）

   [节点 writeDraft] 让模型为「LangGraph 是什么」写一段 30 字左右的简介…
   [节点 writeDraft] 初稿：LangGraph 是搭建有状态 AI 智能体工作流的图框架，像流程图一样组织多步骤任务。
   👤 人类的回复：LangGraph 是搭建有状态 AI 智能体工作流的图框架，像流程图一样组织多步骤任务。

   [节点 humanEdit] 拿到的最终正文（来自人类 resume 值）：
      LangGraph 是搭建有状态 AI 智能体工作流的图框架，像流程图一样组织多步骤任务。
   [节点 publish] 下游节点开箱即用的就是被改过的内容

📦 A 的最终状态：
   初稿 draft   = LangGraph 是搭建有状态 AI 智能体工作流的图框架，像流程图一样组织多步骤任务。
   定稿 edited  = LangGraph 是搭建有状态 AI 智能体工作流的图框架，像流程图一样组织多步骤任务。
   发布 published = 📢 已发布：LangGraph 是搭建有状态 AI 智能体工作流的图框架，像流程图一样组织多步骤任务。
   → draft === edited ? 是（原样通过）

━━━ B) 人类改写 ━━━
▶ 第 1 次 invoke（暂停）

   [节点 writeDraft] 让模型为「LangGraph 是什么」写一段 30 字左右的简介…
   [节点 writeDraft] 初稿：LangGraph 是用图结构编排大模型流程的框架，帮你构建多步骤、有状态的智能体。
   👤 人类的回复：LangGraph 是 LangChain 生态里的「智能体编排框架」，用图的方式组织多步骤流程。（这段是人工改写的，和模型原稿不一样。）

   [节点 humanEdit] 拿到的最终正文（来自人类 resume 值）：
      LangGraph 是 LangChain 生态里的「智能体编排框架」，用图的方式组织多步骤流程。（这段是人工改写的，和模型原稿不一样。）
   [节点 publish] 下游节点开箱即用的就是被改过的内容

📦 B 的最终状态：
   初稿 draft   = LangGraph 是用图结构编排大模型流程的框架，帮你构建多步骤、有状态的智能体。
   定稿 edited  = LangGraph 是 LangChain 生态里的「智能体编排框架」，用图的方式组织多步骤流程。（这段是人工改写的，和模型原稿不一样。）
   发布 published = 📢 已发布：LangGraph 是 LangChain 生态里的「智能体编排框架」，用图的方式组织多步骤流程。（这段是人工改写的，和模型原稿不一样。）
   → draft === edited ? 否（被人类改过）

📌 对照结论：
   A 组 resume 传回原稿 → 下游 publish 用的就是模型初稿；
   B 组 resume 传回改写稿 → 下游 publish 用的是「人类改后的内容」。
   interrupt 的返回值，就是「人类交给下游的接力棒」。

💡 模型的初稿每次措辞会不一样（推理型模型的正常随机性），但 A / B 两组的关键结论永远成立：A 组 draft === edited，B 组不相等。

三个要点：

① interrupt() 的返回值，就是下游要用的「人类内容」。 它不拘泥于字符串 —— 你可以 resume 一个对象、数组、甚至富文本，只要下游节点认得就行。

typescript

const finalText = interrupt({ draft: state.draft }); // 交出去
// ... resume 后 ...
return { edited: String(finalText) };                // 拿回来，往下游传

② interrupt() 的返回值类型是 any（或由 responseSchema 推断）。 本示例用 String(finalText) 显式收口，避免把 unknown 一路带进状态。想让它带类型、还能校验，可以用第二个参数 responseSchema（后面小结会提）。

③ resume 只对「当前被中断的那个节点」生效。 人类改了内容，不需要自己去改状态 —— 节点的返回值（{ edited }）会照常写进 state，下游自然读到新值。人类编辑 = 把新内容从 interrupt 的返回值里流进来，仅此而已。

四、危险工具的人工审批（真实场景）

现在到本章最实用的场景：智能体要调用一个危险工具（转账、删文件、发邮件），绝不能让它自己说执行就执行。

做法：把 ReAct 图里的 tools 节点换成一个自定义节点 —— 在真正调用危险工具之前先 interrupt()，等人类回复 approve / reject 再决定。为了证明工具到底执行没执行，危险工具的实现里会写一笔副作用账本。

示例 3：在工具执行前拦一道

typescript

/**
 * 示例 3：给「危险工具」加人工审批 —— 在工具真正执行前 interrupt
 *
 * 这是 HITL 最实用的一个场景：智能体可以调用各种工具，但其中有些是
 * "危险动作"（转账、删文件、发邮件……），绝不能让它自己说执行就执行。
 *
 * 做法：把 ReAct 里的 tools 节点换成一个**自定义节点** —— 里面在真正
 * 调用危险工具之前先 interrupt，等人类回复 approve / reject 再决定。
 *
 *    START → agent ──有工具调用?──► tools(interrupt 审批)
 *              ▲                      │
 *              └──────────────────────┘   （工具结果回传给模型）
 *
 * ★ 关键：为了证明工具"到底有没有被执行"，危险工具内部会写一笔"账本"
 *   （side-effect 日志）。最后把账本打印出来，谁执行过一目了然。
 */
import {
  StateGraph,
  START,
  END,
  MessagesAnnotation,
  MemorySaver,
  Command,
  interrupt,
  isInterrupted,
  type Interrupt,
} from "@langchain/langgraph";
import { HumanMessage, SystemMessage, AIMessage, ToolMessage, type BaseMessage } from "@langchain/core/messages";
import { tool } from "@langchain/core/tools";
import * as z from "zod";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

// ── 副作用账本：真正执行过的危险操作会登记在这里 ───────────────
const LEDGER: string[] = [];
function record(op: string) {
  LEDGER.push(op);
  console.log(`   💸 [副作用] 危险操作真正执行了：${op}`);
}

// ── 工具的真实实现 ─────────────────────────────────────────────
type ToolImpl = (args: Record<string, unknown>) => string;
const impls: Record<string, ToolImpl> = {
  // 危险工具：转账。只有人类批准后才会被调用到。
  transferMoney: ({ to, amount }) => {
    record(`transferMoney(to=${to}, amount=${amount})`);
    return `已向 ${to} 转账 ${amount} 元。`;
  },
  // 安全工具：查余额，不需要审批。
  getBalance: ({ account }) => `账户 ${account} 当前余额：1000 元。`,
};

// ── 工具声明（给模型看的 schema）──────────────────────────────
const transferMoney = tool((args) => impls.transferMoney(args as Record<string, unknown>), {
  name: "transferMoney",
  description: "向指定账户转账。属于危险操作，调用前需要人工审批。",
  schema: z.object({
    to: z.string().describe("收款账户名"),
    amount: z.number().describe("转账金额（元）"),
  }),
});

const getBalance = tool((args) => impls.getBalance(args as Record<string, unknown>), {
  name: "getBalance",
  description: "查询账户余额。安全操作，无需审批。",
  schema: z.object({
    account: z.string().describe("账户名"),
  }),
});

const tools = [transferMoney, getBalance];

/** 危险工具清单：出现在这里的工具必须人工审批 */
const DANGEROUS = new Set(["transferMoney"]);

async function main() {
  banner("示例 3：危险工具的人工审批（approve / reject）");
  assertChatKey();

  const model = createModel({ temperature: 0 }).bindTools(tools);

  // ── 节点 agent：调模型；模型可能返回 tool_calls ─────────────
  async function agentNode(state: typeof MessagesAnnotation.State) {
    console.log(`   [节点 agent] 把 ${state.messages.length} 条消息交给模型…`);
    const res = await model.invoke(state.messages);
    const calls = (res as AIMessage).tool_calls ?? [];
    console.log(
      `   [节点 agent] 模型决定：${
        calls.length ? `调用 ${calls.map((c) => c.name).join(", ")}` : "给出最终回答"
      }`
    );
    return { messages: [res] };
  }

  // ── 节点 tools：自定义！在真正执行前，对危险工具先 interrupt ─
  async function toolsNode(state: typeof MessagesAnnotation.State) {
    const last = state.messages[state.messages.length - 1] as AIMessage;
    const out: ToolMessage[] = [];

    for (const call of last.tool_calls ?? []) {
      const args = (call.args ?? {}) as Record<string, unknown>;
      const callId = call.id ?? call.name;

      if (DANGEROUS.has(call.name)) {
        console.log(`   [节点 tools] 「${call.name}」是危险工具 → 调用 interrupt 等人工审批`);
        // ★ 在真正执行工具之前先暂停，把"想做什么"交给人类
        const decision = interrupt({
          kind: "tool_approval",
          tool: call.name,
          args,
          question: `智能体想调用危险工具 ${call.name}，参数 ${JSON.stringify(args)}。是否批准？（approve / reject）`,
        });

        const approved = String(decision).toLowerCase().includes("approve");
        if (approved) {
          console.log(`   [节点 tools] ✅ 人类批准 → 执行 ${call.name}`);
          const result = impls[call.name](args);
          out.push(new ToolMessage({ content: result, tool_call_id: callId }));
        } else {
          console.log(`   [节点 tools] ❌ 人类拒绝 → 跳过 ${call.name}（不产生任何副作用）`);
          out.push(
            new ToolMessage({
              content: `该操作已被人工拒绝，${call.name} 未执行。`,
              tool_call_id: callId,
            })
          );
        }
      } else {
        console.log(`   [节点 tools] 「${call.name}」是安全工具 → 直接执行`);
        const result = impls[call.name](args);
        out.push(new ToolMessage({ content: result, tool_call_id: callId }));
      }
    }
    return { messages: out };
  }

  // ── 条件边：有工具调用就去 tools，否则结束 ──────────────────
  function shouldContinue(state: typeof MessagesAnnotation.State): "tools" | typeof END {
    const last = state.messages[state.messages.length - 1] as AIMessage;
    return last.tool_calls?.length ? "tools" : END;
  }

  const graph = new StateGraph(MessagesAnnotation)
    .addNode("agent", agentNode)
    .addNode("tools", toolsNode)
    .addEdge(START, "agent")
    .addEdge("tools", "agent")
    .addConditionalEdges("agent", shouldContinue, ["tools", END])
    .compile({ checkpointer: new MemorySaver() });

  // ★ 强约束系统提示：让模型稳定地"决定调用工具"（推理型模型对工具调用有随机性，
  //   加一条明确的系统指令后，本示例的模型调用就稳定复现了）。
  const SYSTEM =
    "你是一个转账执行助手。只要用户提出转账请求，你就必须调用 transferMoney 工具来真正执行，" +
    "绝不能只用文字描述或声称自己已经转账。";
  const PROMPT = "请给账户 alice 转账 100 元。";

  /** 跑一个场景：人类对危险工具给出某决定，返回（最终回答 + 本场景的账本） */
  async function scenario(threadId: string, humanDecision: string) {
    const ledBefore = LEDGER.length;
    const config = { configurable: { thread_id: threadId } };

    const paused = await graph.invoke(
      { messages: [new SystemMessage(SYSTEM), new HumanMessage(PROMPT)] },
      config
    );
    if (isInterrupted(paused)) {
      const itr = paused.__interrupt__[0] as Interrupt<{ tool: string }>;
      console.log(`\n   ⏸  图已暂停，等待人类审批工具：${itr.value?.tool}`);
      console.log(`   👤 人类决定：${humanDecision}\n`);
    } else {
      throw new Error("预期会暂停等待审批，但没有暂停（模型可能没调用工具）");
    }

    const done = await graph.invoke(new Command({ resume: humanDecision }), config);
    const last = done.messages[done.messages.length - 1] as BaseMessage;
    return { answer: String(last.content), newOps: LEDGER.slice(ledBefore) };
  }

  // ── 场景 A：人类 APPROVE ───────────────────────────────────
  console.log("━━━ 场景 A：人类批准（approve）━━━\n");
  const a = await scenario("approval-A", "approve");

  // ── 场景 B：人类 REJECT ────────────────────────────────────
  console.log("\n━━━ 场景 B：人类拒绝（reject）━━━\n");
  const b = await scenario("approval-B", "reject");

  // ── 用"账本"证明到底执行没执行 ─────────────────────────────
  console.log("\n📌 副作用账本对照（谁真正执行了危险工具）：");
  console.log("   场景 A（approve）新登记：", a.newOps.length ? a.newOps : "（无）");
  console.log("   场景 B（reject）新登记：", b.newOps.length ? b.newOps : "（无）");
  console.log("\n   ✅ approve：transferMoney 真的改了账 / 转了钱。");
  console.log("   ✅ reject ：transferMoney 完全没被调用，账本没有新增 —— 这就是审批的意义。");

  console.log("\n🤖 场景 A 的最终回答：", a.answer.replace(/\n/g, " ").slice(0, 120));
  console.log("🤖 场景 B 的最终回答：", b.answer.replace(/\n/g, " ").slice(0, 120));
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202

运行：

bash

npx tsx 05-human-in-the-loop/code/03-tool-approval.ts

真实输出：

========================================================================
  示例 3：危险工具的人工审批（approve / reject）
========================================================================

━━━ 场景 A：人类批准（approve）━━━

   [节点 agent] 把 2 条消息交给模型…
   [节点 agent] 模型决定：调用 transferMoney
   [节点 tools] 「transferMoney」是危险工具 → 调用 interrupt 等人工审批

   ⏸  图已暂停，等待人类审批工具：transferMoney
   👤 人类决定：approve

   [节点 tools] 「transferMoney」是危险工具 → 调用 interrupt 等人工审批
   [节点 tools] ✅ 人类批准 → 执行 transferMoney
   💸 [副作用] 危险操作真正执行了：transferMoney(to=alice, amount=100)
   [节点 agent] 把 4 条消息交给模型…
   [节点 agent] 模型决定：给出最终回答

━━━ 场景 B：人类拒绝（reject）━━━

   [节点 agent] 把 2 条消息交给模型…
   [节点 agent] 模型决定：调用 transferMoney
   [节点 tools] 「transferMoney」是危险工具 → 调用 interrupt 等人工审批

   ⏸  图已暂停，等待人类审批工具：transferMoney
   👤 人类决定：reject

   [节点 tools] 「transferMoney」是危险工具 → 调用 interrupt 等人工审批
   [节点 tools] ❌ 人类拒绝 → 跳过 transferMoney（不产生任何副作用）
   [节点 agent] 把 4 条消息交给模型…
   [节点 agent] 模型决定：给出最终回答

📌 副作用账本对照（谁真正执行了危险工具）：
   场景 A（approve）新登记： [ 'transferMoney(to=alice, amount=100)' ]
   场景 B（reject）新登记： （无）

   ✅ approve：transferMoney 真的改了账 / 转了钱。
   ✅ reject ：transferMoney 完全没被调用，账本没有新增 —— 这就是审批的意义。

🤖 场景 A 的最终回答： 已完成：向账户 **alice** 转账 **100 元**，操作成功。
🤖 场景 B 的最终回答： 转账未能执行。❌  系统返回信息：**该操作已被人工拒绝，transferMoney 未执行。**  由于转账属于危险操作，需要人工审批。本次审批未通过，因此并未向 alice 转账 100 元，您的账户资金没有发生任何变动。  如果您仍需

💡 两个场景的唯一差别，只是 resume 传的值不同（approve vs reject）。但副作用账本清清楚楚：A 组多了一笔 transferMoney(to=alice, amount=100)，B 组什么都没有 —— 工具真的没被执行，而不是"执行了又撤销"。

四个关键点：

① 为什么不用官方 ToolNode？因为它会「立刻执行」。 ToolNode 拿到 tool_calls 就无条件执行所有工具。要「执行前先问人」，就得自己写这个节点 —— 循环从 tool_calls 里读调用、逐个处理、把结果包成 ToolMessage 返回。结构上它和 ToolNode 完全等价（都接在 agent → tools → agent 的循环里），只是中间多了一道闸门。

② 审批要发生在「调用实现之前」。 注意代码顺序：先 interrupt(...) 拿到 decision，再决定要不要调 impls[call.name](args)。把闸门放在执行之后，就等于没审批。

③ 拒绝时也要回一个 ToolMessage。 被拒绝的工具不能"不作声" —— 否则模型在 tools → agent 回到 agent 时，会看到 tool_calls 没有对应的 ToolMessage 而报错。正确做法是回一条说明「已被人工拒绝」的 ToolMessage，让模型知道发生了什么（上面 B 组的最终回答就是这么来的）。

④ 一个真实的坑：推理型模型有时不调用工具。 第一次写这个例子时，模型偶尔会「用文字描述转账」而不是真的调用 transferMoney —— 那样就永远不会触发 interrupt。解决办法是加一条强约束系统提示（本示例的 SYSTEM），把「必须调用工具」写死。加完后连跑多次都稳定。记住：HITL 的前提是流程真的走到了 interrupt 那一行。

⚠️ 从安全角度补一句：审批的「判断」不要只看字符串里有没有 approve。示例为了简短用了 includes("approve")；生产里应该用结构化的 resume 值（如 { decision: "approve" }）并配合 interrupt 的 responseSchema 做校验，避免人工输入五花八门时误判。

五、人工编辑 + 时间旅行重放

💡 本章开头呼应一下第 4 章：第 4 章已经讲过 getStateHistory / updateState 的基础（存档、从旧存档分叉、时间旅行）。本节把同一套机制和 HITL 结合起来，讲一个非常实用的模式 —— 「人类对中间产物不满意，但又不想从头再来」。

场景：图先生成大纲、再写正文、最后排版。人类看完觉得大纲不对，但正文还没写、后面还有一大段流程。与其从头重跑，不如回到「大纲刚定、正文未写」那一刻，把大纲改成人类手写的版本，再让它往后重放。

示例 4：改掉中间那一步，再重跑到结束

typescript

/**
 * 示例 4：人工编辑后「重放」—— 回到中间某一步、改掉当时的产物、再重跑
 *
 * 第 4 章已经讲过 getStateHistory / updateState 的基础（存档、时间旅行）。
 * 本章把它们和 HITL 结合起来，讲一个非常实用的模式：
 *
 *   「人类对中间产物不满意，但又不想从头再来。」
 *
 * 做法（编辑 + 重放）：
 *   ① 正常跑一遍，图会留下若干 checkpoint；
 *   ② 用 getStateHistory 找到「大纲已生成、正文还没写」那一刻的存档；
 *   ③ 用 updateState 把当时的大纲改成「人类手写的新大纲」；
 *   ④ 从分叉出来的存档 invoke(null, ...) 重放后面的步骤；
 *   ⑤ 对比"修改前"与"重放后"的最终结果。
 *
 *    START → plan → write → format → END
 *
 * ★ 与第 4 章的区别：这里 updateState 写入的，是"人类审阅后给出的修改"，
 *   本质就是 HITL；而"从存档重跑"正是把人类编辑接回流水线的方式。
 */
import {
  StateGraph,
  START,
  END,
  Annotation,
  MemorySaver,
  type StateSnapshot,
} from "@langchain/langgraph";
import { HumanMessage } from "@langchain/core/messages";
import { createModel, banner } from "../../lib/model.js";
import { assertChatKey } from "../../lib/env.js";

const THREAD_ID = "hitl-time-travel";

// ── 状态：话题、大纲、正文、最终文本 ───────────────────────────
const OutlineState = Annotation.Root({
  topic: Annotation<string>,
  /** 大纲：写作前的要点清单（人类可以改它） */
  plan: Annotation<string>,
  /** 模型依据大纲写出的正文 */
  draft: Annotation<string>,
  /** 加壳后的最终文本 */
  finalText: Annotation<string>,
});

async function main() {
  banner("示例 4：人工编辑大纲后的时间旅行重放");
  assertChatKey();

  const model = createModel({ temperature: 0 });

  // ── 节点 makePlan：生成大纲（这里刻意用固定文案，方便对比改动）──
  function planNode(state: typeof OutlineState.State) {
    const plan = `1) 定义：${state.topic} 是什么；2) 价值：解决了什么问题；3) 场景：一个典型用法。`;
    console.log("   [节点 makePlan] 自动生成的大纲：");
    console.log(`      ${plan}`);
    return { plan };
  }

  // ── 节点 write：依据大纲写正文（会调用模型）──────────────────
  async function writeNode(state: typeof OutlineState.State) {
    console.log("   [节点 write] 依据当前大纲写作…");
    const res = await model.invoke([
      new HumanMessage(
        `请根据下面的提纲，写一段 60 字左右的简体中文介绍，主题是「${state.topic}」。只输出正文，不要标题。\n提纲：${state.plan}`
      ),
    ]);
    const draft = String(res.content).trim();
    console.log(`   [节点 write] 正文：${draft}`);
    return { draft };
  }

  // ── 节点 format：把正文包一层标题，产出最终文本 ───────────────
  function formatNode(state: typeof OutlineState.State) {
    return { finalText: `【标题】${state.topic}\n【正文】${state.draft}` };
  }

  const graph = new StateGraph(OutlineState)
    .addNode("makePlan", planNode)
    .addNode("write", writeNode)
    .addNode("format", formatNode)
    .addEdge(START, "makePlan")
    .addEdge("makePlan", "write")
    .addEdge("write", "format")
    .addEdge("format", END)
    .compile({ checkpointer: new MemorySaver() });

  const baseConfig = { configurable: { thread_id: THREAD_ID } };

  // ── ① 正常跑一遍 ───────────────────────────────────────────
  console.log("━━━ ① 修改前：正常跑一遍 ━━━\n");
  const original = await graph.invoke({ topic: "LangGraph" }, baseConfig);
  console.log("\n📦 修改前最终结果：");
  console.log(original.finalText);

  // ── ② 用 getStateHistory 找到「大纲已生成、正文未写」的存档 ──
  console.log("\n━━━ ② 定位中间存档（next=[write] 那一刻）━━━\n");
  const history: StateSnapshot[] = [];
  for await (const s of graph.getStateHistory(baseConfig)) history.push(s);
  const chrono = [...history].reverse(); // 从早到晚

  chrono.forEach((s, i) => {
    const cpId = String(s.config.configurable?.checkpoint_id ?? "");
    const next = s.next.length ? s.next.join(",") : "(结束)";
    console.log(`   #${i}  cp=…${cpId.slice(-8)}  next=[${next}]`);
  });

  // 目标存档：下一步是 write（说明 plan 刚做完，正文还没写）
  const beforeWrite = chrono.find((s) => s.next.includes("write"));
  if (!beforeWrite) throw new Error("没找到 next=[write] 的存档");
  const cpId = String(beforeWrite.config.configurable?.checkpoint_id ?? "");
  console.log(`\n   🎯 选中存档：cp=…${cpId.slice(-8)}，此时 plan =`);
  console.log(`      ${beforeWrite.values.plan as string}`);

  // ── ③ 人类出手：手写一份不满意原稿的新大纲 ─────────────────
  const HUMAN_PLAN =
    "1) 一句话定义；2) 用「状态 + 节点 + 边」三件套打比方；3) 强调它能把流程画成图。";
  console.log("\n━━━ ③ 人类人工编辑大纲 ━━━\n");
  console.log(`   👤 人类改成：${HUMAN_PLAN}`);

  // ★ updateState 在旧存档上「分叉」出一个新存档，返回它的 config
  const forkedConfig = await graph.updateState(beforeWrite.config, {
    plan: HUMAN_PLAN,
  });

  // ── ④ 从分叉存档重放（input 传 null：状态已在 checkpointer 里）──
  console.log("\n━━━ ④ 从存档重放（重新执行 write、format）━━━\n");
  const replayed = await graph.invoke(null, forkedConfig);
  console.log("\n📦 重放后的最终结果：");
  console.log(replayed.finalText);

  // ── ⑤ 对比 ─────────────────────────────────────────────────
  console.log("\n📌 对比：");
  console.log(`   修改前大纲：${original.plan}`);
  console.log(`   重放后大纲：${replayed.plan}`);
  console.log(`   两者最终文本是否相同？${original.finalText === replayed.finalText ? "相同" : "不同（人类编辑生效了）"}`);
  console.log("   → 人类只改了一句话，后面的 write / format 自动按新大纲重跑了一遍，无需从头再来。");
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash

npx tsx 05-human-in-the-loop/code/04-time-travel-replay.ts

真实输出：

========================================================================
  示例 4：人工编辑大纲后的时间旅行重放
========================================================================

━━━ ① 修改前：正常跑一遍 ━━━

   [节点 makePlan] 自动生成的大纲：
      1) 定义：LangGraph 是什么；2) 价值：解决了什么问题；3) 场景：一个典型用法。
   [节点 write] 依据当前大纲写作…
   [节点 write] 正文：LangGraph 是构建有状态、多步骤 LLM 应用的图框架。它解决复杂 Agent 的状态管理、循环与分支难题，常用于搭建带人工审核的 ReAct 工作流。

📦 修改前最终结果：
【标题】LangGraph
【正文】LangGraph 是构建有状态、多步骤 LLM 应用的图框架。它解决复杂 Agent 的状态管理、循环与分支难题，常用于搭建带人工审核的 ReAct 工作流。

━━━ ② 定位中间存档（next=[write] 那一刻）━━━

   #0  cp=…0273559c  next=[__start__]
   #1  cp=…d8479b24  next=[makePlan]
   #2  cp=…8c96edbf  next=[write]
   #3  cp=…9854f119  next=[format]
   #4  cp=…be0c8248  next=[(结束)]

   🎯 选中存档：cp=…8c96edbf，此时 plan =
      1) 定义：LangGraph 是什么；2) 价值：解决了什么问题；3) 场景：一个典型用法。

━━━ ③ 人类人工编辑大纲 ━━━

   👤 人类改成：1) 一句话定义；2) 用「状态 + 节点 + 边」三件套打比方；3) 强调它能把流程画成图。

━━━ ④ 从存档重放（重新执行 write、format）━━━

   [节点 write] 依据当前大纲写作…
   [节点 write] 正文：LangGraph 是编排 LLM 智能体工作流的框架。它用「状态+节点+边」三件套打比方：状态是共享记忆，节点是动作，边是路径，从而把复杂流程画成一张可执行图。

📦 重放后的最终结果：
【标题】LangGraph
【正文】LangGraph 是编排 LLM 智能体工作流的框架。它用「状态+节点+边」三件套打比方：状态是共享记忆，节点是动作，边是路径，从而把复杂流程画成一张可执行图。

📌 对比：
   修改前大纲：1) 定义：LangGraph 是什么；2) 价值：解决了什么问题；3) 场景：一个典型用法。
   重放后大纲：1) 一句话定义；2) 用「状态 + 节点 + 边」三件套打比方；3) 强调它能把流程画成图。
   两者最终文本是否相同？不同（人类编辑生效了）
   → 人类只改了一句话，后面的 write / format 自动按新大纲重跑了一遍，无需从头再来。

💡 cp=… 后面那 8 位是 checkpoint_id 的尾段，每次运行都不同；关键是 next 列 —— next=[write] 的那个存档，正是「makePlan 刚做完、write 还没跑」的那一刻。

四个要点：

① next 是找存档的「路标」。 getStateHistory() 返回的每个存档都带 next（下一步该谁执行）。要找「大纲已定、正文未写」，就找 next.includes("write") 的那个存档 —— 比按顺序数 #0 #1 #2 稳得多。

② 人类编辑 = updateState(旧存档, { plan: 人类的新大纲 })。 updateState 不会「改写历史」，而是在旧存档上分叉一个新存档，并返回它的 config。

typescript

const forkedConfig = await graph.updateState(beforeWrite.config, { plan: HUMAN_PLAN });

③ 重放必须用「分叉返回的 config」，且 invoke 的第一个参数传 null。 状态已经在 checkpointer 里了，不需要再喂输入：

typescript

const replayed = await graph.invoke(null, forkedConfig); // ★ 用 forkedConfig，不是旧 config

④ 和第 4 章示例 3 的差别，只在「改什么」。 第 4 章往 messages 里插了一条 HumanMessage；这里把 plan 字段换成了人类手写的大纲。机制一模一样：回到某一步 → 改状态 → 从那里重跑。区别在于，这里改动的内容来自 HITL 的人工审阅 —— 这正是把「人工编辑」接回自动流水线的方式。

⚠️ 和第 4 章一样：别自己拼 { thread_id, checkpoint_id }，直接用快照自带的 config（它含 thread_id + checkpoint_ns + checkpoint_id），否则容易漏 checkpoint_ns 而报错。

六、审批的超时兜底（进阶）

interrupt() 会把图挂起等人类。但人类可能几分钟后才回、干脆不回、或者审批系统宕了。如果无限期等待，流程就永远卡死。生产做法是给审批加一个截止时间：超时就用安全默认值（通常是「拒绝」）恢复，让流程继续跑。

示例 5：到点还没人回？默认拒绝

typescript

/**
 * 示例 5（进阶）：审批的「超时兜底」—— 人类迟迟不响应怎么办？
 *
 * interrupt 会把图挂起，等待人类 resume。但现实中人类可能：
 *   • 几分钟后才回；• 干脆不回；• 审批系统宕了。
 * 如果无限期等待，流程就永远卡死。生产做法是给审批加一个**截止时间**：
 *   在截止时间内收到人类答复 → 用人类答复；
 *   超过截止时间仍没答复       → 用**安全默认值**（通常是"拒绝"）恢复，让流程继续。
 *
 * ★ 关键认知：超时逻辑写在**调用方**（你的应用/网关），不写在图里。
 *   图只负责 interrupt 等一个 resume 值；"等多久、等不到用什么"由调用方决定。
 *   本示例用一个"会睡一会儿的模拟人类"来制造"及时响应"与"超时"两种情况。
 *
 *    START → propose → approve(interrupt) → execute → END
 */
import {
  StateGraph,
  START,
  END,
  Annotation,
  MemorySaver,
  Command,
  interrupt,
  isInterrupted,
} from "@langchain/langgraph";
import { banner } from "../../lib/model.js";

const DEADLINE_MS = 120; // 审批截止时间（真实项目里可能是几分钟/几小时）

// ── 状态 ───────────────────────────────────────────────────────
const ActionState = Annotation.Root({
  /** 待审批的动作描述 */
  action: Annotation<string>,
  /** 人类决定：approve / reject / 超时兜底的 reject */
  decision: Annotation<string>,
  /** 执行结果 */
  result: Annotation<string>,
});

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

// 副作用账本：证明动作到底执行没有
const EXECUTED: string[] = [];

/** 模拟一个人类：delayMs 毫秒后才给出答复 */
async function mockHuman(delayMs: number, value: string): Promise<string> {
  await sleep(delayMs);
  return value;
}

async function main() {
  banner("示例 5：审批超时兜底（超时 → 安全默认拒绝）");

  // ── 节点 propose：提出一个待审批的动作 ─────────────────────
  function proposeNode(state: typeof ActionState.State) {
    console.log(`   [节点 propose] 提出待审批动作：${state.action}`);
    return {};
  }

  // ── 节点 approve：interrupt 请求人类审批 ───────────────────
  function approveNode(state: typeof ActionState.State) {
    console.log("   [节点 approve] 调用 interrupt，等待人类审批…");
    const answer = interrupt({
      kind: "approval_request",
      action: state.action,
      deadlineMs: DEADLINE_MS,
      question: `是否批准执行「${state.action}」？`,
    });
    console.log(`   [节点 approve] 收到的 resume 值：${JSON.stringify(answer)}`);
    return { decision: String(answer) };
  }

  // ── 节点 execute：只有批准才执行 ───────────────────────────
  function executeNode(state: typeof ActionState.State) {
    const approved = state.decision.toLowerCase().includes("approve");
    if (approved) {
      EXECUTED.push(state.action);
      console.log(`   [节点 execute] ✅ 已批准 → 真正执行：${state.action}`);
      return { result: `已执行：${state.action}` };
    }
    console.log(`   [节点 execute] ❌ 未批准 → 拒绝执行：${state.action}`);
    return { result: `未执行（被拒绝）：${state.action}` };
  }

  const graph = new StateGraph(ActionState)
    .addNode("propose", proposeNode)
    .addNode("approve", approveNode)
    .addNode("execute", executeNode)
    .addEdge(START, "propose")
    .addEdge("propose", "approve")
    .addEdge("approve", "execute")
    .addEdge("execute", END)
    .compile({ checkpointer: new MemorySaver() });

  /**
   * 跑一个审批场景：
   *   humanDelayMs = 模拟人类响应耗时；humanValue = 人类会回什么
   *   调用方用 Promise.race 给审批加截止时间，超时就用安全默认值
   */
  async function approveWithTimeout(
    threadId: string,
    humanDelayMs: number,
    humanValue: string
  ) {
    const config = { configurable: { thread_id: threadId } };
    const paused = await graph.invoke({ action: "删除生产数据库" }, config);
    if (!isInterrupted(paused)) throw new Error("预期会暂停等待审批");

    console.log(`   ⏳ 已暂停，开始等待人类响应（截止 ${DEADLINE_MS}ms）…`);

    // ★ 截止时间逻辑在调用方：人类答复 vs 超时，谁先到用谁
    const humanReply = mockHuman(humanDelayMs, humanValue).then((v) => ({ kind: "human" as const, v }));
    const timeout = sleep(DEADLINE_MS).then(() => ({ kind: "timeout" as const, v: "reject" }));
    const winner = await Promise.race([humanReply, timeout]);

    let decision: string;
    let source: string;
    if (winner.kind === "human") {
      decision = winner.v;
      source = "人类及时响应";
      console.log(`   👤 ${humanDelayMs}ms 时收到人类答复：${decision}`);
    } else {
      decision = winner.v; // 安全默认值
      source = `⏰ 超过 ${DEADLINE_MS}ms 未响应，兜底拒绝`;
      console.log(`   ⏰ 人类超过 ${DEADLINE_MS}ms 未响应 → 使用安全默认值：${decision}`);
    }

    // 把（人类答复 或 兜底值）作为 resume 传回去
    const done = await graph.invoke(
      new Command({ resume: decision }),
      config
    );
    // 记录来源，方便对照展示（写回状态）
    return { source, decision, result: done.result as string };
  }

  // ── 场景 A：人类及时响应 → approve ─────────────────────────
  console.log("━━━ 场景 A：人类在 30ms 内响应（< 截止时间）━━━\n");
  const a = await approveWithTimeout("timeout-A", 30, "approve");
  console.log(`   结果：${a.result}（来源：${a.source}）`);

  // ── 场景 B：人类迟迟不回 → 超时兜底 reject ──────────────────
  console.log("\n━━━ 场景 B：人类 500ms 才回（> 截止时间）━━━\n");
  const b = await approveWithTimeout("timeout-B", 500, "approve");
  console.log(`   结果：${b.result}（来源：${b.source}）`);

  // ── 对照 ───────────────────────────────────────────────────
  console.log("\n📌 对照：");
  console.log(`   场景 A：人类的 "approve" 在 ${DEADLINE_MS}ms 内到达 → 动作被执行。`);
  console.log(`   场景 B：人类虽然也会回 "approve"，但超过 ${DEADLINE_MS}ms，`);
  console.log("           调用方用兜底值 reject 恢复 → 动作没有执行。");
  console.log(`   副作用账本（真正执行过的动作）：${EXECUTED.length ? EXECUTED.join(", ") : "（无）"}`);
  console.log("\n   ⚠️ 兜底值一定要选「安全」的方向：删库/转账这类动作，超时应默认拒绝，");
  console.log("      绝不能默认放行 —— 这是 HITL 在生产里最重要的一条纪律。");
}

main().catch((e) => {
  console.error("❌ 调用失败：", e);
  process.exit(1);
});

运行：

bash

npx tsx 05-human-in-the-loop/code/05-approval-with-timeout.ts

真实输出：

========================================================================
  示例 5：审批超时兜底（超时 → 安全默认拒绝）
========================================================================

━━━ 场景 A：人类在 30ms 内响应（< 截止时间）━━━

   [节点 propose] 提出待审批动作：删除生产数据库
   [节点 approve] 调用 interrupt，等待人类审批…
   ⏳ 已暂停，开始等待人类响应（截止 120ms）…
   👤 30ms 时收到人类答复：approve
   [节点 approve] 调用 interrupt，等待人类审批…
   [节点 approve] 收到的 resume 值："approve"
   [节点 execute] ✅ 已批准 → 真正执行：删除生产数据库
   结果：已执行：删除生产数据库（来源：人类及时响应）

━━━ 场景 B：人类 500ms 才回（> 截止时间）━━━

   [节点 propose] 提出待审批动作：删除生产数据库
   [节点 approve] 调用 interrupt，等待人类审批…
   ⏳ 已暂停，开始等待人类响应（截止 120ms）…
   ⏰ 人类超过 120ms 未响应 → 使用安全默认值：reject
   [节点 approve] 调用 interrupt，等待人类审批…
   [节点 approve] 收到的 resume 值："reject"
   [节点 execute] ❌ 未批准 → 拒绝执行：删除生产数据库
   结果：未执行（被拒绝）：删除生产数据库（来源：⏰ 超过 120ms 未响应，兜底拒绝）

📌 对照：
   场景 A：人类的 "approve" 在 120ms 内到达 → 动作被执行。
   场景 B：人类虽然也会回 "approve"，但超过 120ms，
           调用方用兜底值 reject 恢复 → 动作没有执行。
   副作用账本（真正执行过的动作）：删除生产数据库

   ⚠️ 兜底值一定要选「安全」的方向：删库/转账这类动作，超时应默认拒绝，
      绝不能默认放行 —— 这是 HITL 在生产里最重要的一条纪律。

三个要点：

① 超时逻辑在「调用方」，不在图里。 图只负责 interrupt 等一个 resume 值；「等多久、等不到用什么」是你的应用/网关该管的事。示例用 Promise.race([人类答复, 超时]) 模拟这件事 —— 真实项目里，人类答复可能来自 webhook、审批台的轮询结果或消息队列。

② 超时时，照样用 Command({ resume }) 恢复，只是值换成兜底值。 图完全不知道「这是超时」—— 它只看到 resume 进来的是 "reject"，于是走了拒绝分支。同一个 interrupt 点，无论人类答复还是系统兜底，都走同一条恢复通道。

③ 兜底值必须选「安全」的方向。 删库、转账、发邮件这类动作，超时要默认拒绝，绝不能默认放行。这是 HITL 在生产里最重要的一条纪律：宁可让流程停一下、让人补一次审批，也不能因为「没人管」而放行一个破坏性操作。

💡 把示例 5 和示例 3 组合起来，就是一套完整的生产级危险操作闸门：先审批（示例 3），审批超时则安全拒绝（示例 5）。

七、本章小结

本章速查表

你要做的事	代码
在节点里暂停等人类	`const v = interrupt(payload)`（节点函数里调用）
判断图是否被暂停	`isInterrupted(result)` 或 `"__interrupt__" in result`
取出暂停信息	`result.__interrupt__[0].value`（数组，含 `id` / `value`）
看停在哪一步	`(await graph.getState(config)).next`
恢复执行	`graph.invoke(new Command({ resume: 值 }), config)`
人类「改内容」再继续	`resume` 传改好的内容，节点里当返回值用
危险工具审批	自定义 `tools` 节点，执行前 `interrupt`
拒绝工具调用	回一条说明「已拒绝」的 `ToolMessage`（不能沉默）
带校验的审批输入	`interrupt(payload, { responseSchema: zodSchema })`
人类编辑 + 重放	`updateState(旧存档config, {…})` → `invoke(null, 分叉config)`
审批超时兜底	调用方 `Promise.race`，超时 `resume` 一个安全默认值
前提	`compile({ checkpointer })` + `configurable.thread_id`

三句话总结

interrupt() 暂停，Command({ resume }) 恢复 —— 二者之间隔着的可以是几秒也可以是几天，因为状态存在 checkpointer 里，而连接它们的是 thread_id。
暂停不是「从中间接着跑」，而是「把中断的那个节点整体重跑一遍」（这次 interrupt 直接返回人类的值）—— 所以 interrupt 之前的代码必须无副作用、可重复执行。
HITL 的三种典型用法：审批决定（示例 1）、修改内容（示例 2）、拦截危险工具（示例 3）；再配合第 4 章的存档做「人工编辑 + 重放」（示例 4），以及生产必备的「超时默认拒绝」（示例 5）。

动手练习

把示例 1 的 resume 从 "approve" 改成 "reject"，跑一遍，观察 finalize 节点走了哪条分支、finalText 变成了什么。
改造示例 2：让 humanEdit 节点在 interrupt 时把整份草稿交出去，人类 resume 一个数组 ["第一段", "第二段"]，让下游节点把它们拼成带序号的最终正文。（提示：resume 可以传对象/数组，String(...) 收口前先 Array.isArray 判断。）
改造示例 3：再增加一个危险工具 deleteFile({ path })，让模型的 prompt 变成「先删掉 /tmp/x 再转账」；跑一次，会连续触发两次审批 —— 观察两次 interrupt 的 payload 分别是什么，以及人类的两次 resume 值是如何按顺序对应上的。（提示：一个节点里多次调用 interrupt，恢复时按调用顺序依次取 resume。）

下一章预告

到这里，我们已经能让图「停、等、续」。但你也发现了：跑完一次总要等到最后才看到全部输出，中间过程是黑盒。下一章我们讲流式输出（streaming）：用 graph.stream(...) 把「哪个节点在跑、token 一个个蹦出来、状态怎么变」实时推给前端 —— 让你的智能体从「等半天」变成「边想边说」。

← 上一章：持久化与记忆｜返回总目录｜下一章：流式输出 →

第 5 章 · 人工介入（Human-in-the-Loop） ​

🎯 本章目标 ​

一、HITL：为什么「暂停」这么重要 ​

二、最小闭环：interrupt → interrupt → resume ​

示例 1：把「暂停时拿到的对象」完整打印出来 ​

三、人类修改后再继续 ​

示例 2：resume 传回「人类改过的内容」，下游照单全收 ​

四、危险工具的人工审批（真实场景） ​

示例 3：在工具执行前拦一道 ​

五、人工编辑 + 时间旅行重放 ​

示例 4：改掉中间那一步，再重跑到结束 ​

六、审批的超时兜底（进阶） ​

示例 5：到点还没人回？默认拒绝 ​

七、本章小结 ​

第 5 章 · 人工介入（Human-in-the-Loop）

🎯 本章目标

一、HITL：为什么「暂停」这么重要

二、最小闭环：interrupt → interrupt → resume

示例 1：把「暂停时拿到的对象」完整打印出来

三、人类修改后再继续

示例 2：resume 传回「人类改过的内容」，下游照单全收

四、危险工具的人工审批（真实场景）

示例 3：在工具执行前拦一道

五、人工编辑 + 时间旅行重放

示例 4：改掉中间那一步，再重跑到结束

六、审批的超时兜底（进阶）

示例 5：到点还没人回？默认拒绝

七、本章小结