用 LangGraph.js 实现持久化人工审批

系列文章：TypeScript 中的 AI——共 5 本书，从第一次调用 LLM，到让 Agent 上线生产环境——五本都在这里。

我的项目：Hermes IDE | GitHub——一款面向开发者的 IDE，帮助你使用 Claude Code 和其他 AI 编程工具交付软件。

关于我：xgabriel.com | GitHub

一个能够发起退款的 Agent，在做出决定和实际退款之间，必须有人类参与审核。最简单的实现方式，是在等待某人查看 Slack 消息时阻塞一个 Promise。这种方法在等待 90 秒时还算可行，但只要时间再长一点，就会彻底失效——一次部署、一次重启，或者用户出门吃午饭，都会让这次运行终止。

我们真正需要的，是一种能够跨进程存续的暂停机制。LangGraph.js 已经内置了这一能力。

interrupt 会暂停运行

import { interrupt, Command } from "@langchain/langgraph";

async function approval(state: typeof State.State) {
  const decision = interrupt({
    kind: "refund_approval",
    orderId: state.orderId,
    amount: state.refundAmount,
    reason: state.refundReason,
  });

  return {
    approved: decision === "approve",
    messages: [
      { role: "user", content: `Human decision: ${String(decision)}` },
    ],
  };
}

interrupt 会抛出一个特殊信号，由 graph 捕获。执行会在当前节点停止，state 被写入 checkpoint，随后 invoke 携带一个 __interrupt__ payload 返回给调用方。整个过程中，没有任何东西被阻塞，也没有任何内容需要一直保存在内存中。

传给 interrupt 的参数，是你希望审核人员看到的数据。应该传入 UI 渲染决策界面所需的字段——订单 ID、金额、原因——而不是一段 prompt 字符串，因为另一端的某个系统需要真正把这些内容渲染出来。

const config = { configurable: { thread_id: `refund-${orderId}` } };
const result = await graph.invoke({ orderId, refundAmount }, config);

if (result.__interrupt__) {
  const [pending] = result.__interrupt__;
  await queue.enqueueReview({
    threadId: config.configurable.thread_id,
    payload: pending.value,
    interruptId: pending.id,
  });
  return { status: "awaiting_approval" };
}

result.__interrupt__ 是一个数组——一个 graph 可能同时在多个分支中暂停。这个库还导出了 isInterrupted 和 INTERRUPT key，可以用它们进行类型安全的检查，而不是直接访问这个属性。

thread_id 是唯一必须持久化保存的东西。它是指向 checkpoint 的指针，也是审批 UI 返回决策时需要提交的值。与其使用随机 UUID，不如根据领域 ID 派生它，例如 refund-${orderId}。这样一来，即使没有查询映射表，你也能找到某次暂停的运行。

恢复运行，即使已经过去数天

export async function applyDecision(
  threadId: string,
  decision: "approve" | "reject",
) {
  const config = { configurable: { thread_id: threadId } };
  const result = await graph.invoke(
    new Command({ resume: decision }),
    config,
  );
  return result;
}

Command({ resume }) 会让原来的 interrupt() 调用返回这个值。之后，该节点会从这一行继续执行——即使换了进程、换了机器，或者中间已经过去了任意长的时间。

完整的对话、tool 执行结果以及累积的 state，都会从 checkpointer 中恢复。这样一来，处理“批准”按钮的 HTTP handler 只需要四行代码。

一个足以浪费你一下午的陷阱

在已有 thread 上调用 invoke 有两种方式，它们的含义完全不同。如果把两者用反了，你会得到一个看起来像是卡死的 graph。

// resuming a pending interrupt — use Command
await graph.invoke(new Command({ resume: "approve" }), config);

// continuing a multi-turn conversation — use a plain object
await graph.invoke({ messages: [{ role: "user", content: "..." }] }, config);

在后续轮次中使用 Command({ update }) 是错误的选择：它会从最新的 checkpoint——也就是上一个已经运行过的步骤——恢复，而不是从 graph 的入口点重新开始执行。表现出来的症状是：运行立刻返回，却什么都没做。

经验法则很简单：Command({ resume }) 用来回答一个待处理的 interrupt()；普通对象用来启动 graph 的新一轮执行。除此之外，不要使用其他方式。

应该把 interrupt 放在哪里

interrupt 的位置决定了这次暂停是否真正有用。

const graph = new StateGraph(State)
  .addNode("assess", assess)
  .addNode("approval", approval)
  .addNode("issueRefund", issueRefund)
  .addEdge(START, "assess")
  .addConditionalEdges("assess", (s) =>
    s.refundAmount > 100 ? "approval" : "issueRefund",
  )
  .addConditionalEdges("approval", (s) =>
    s.approved ? "issueRefund" : END,
  )
  .addEdge("issueRefund", END)
  .compile({ checkpointer });

这里有两点做对了。

第一，interrupt 是一个独立节点，并且位于副作用之前。如果把 interrupt() 放进 issueRefund，而且是在调用退款 API 之后才执行，那么这次暂停什么也保护不了。

第二，它是有条件触发的。小额退款可以完全跳过审批。一个每次运行都会触发的 human-in-the-loop 步骤，只会变成一个不到一周就再也没人处理的队列。

对 checkpointer 的要求

这才是这种模式在生产环境中真正容易失败的地方。

// works in tests, loses every pending approval on deploy
.compile({ checkpointer: new MemorySaver() })

MemorySaver 只在当前进程内有效。如果一次暂停需要存续数天，就必须使用持久化的 checkpointer，例如来自 @langchain/langgraph-checkpoint-* packages 的 Postgres 或 SQLite。此时，这个存储就成了真正的生产数据，也必须满足相应的生产要求：

需要备份，因为 checkpoint 一旦丢失，退款就会永远卡住。需要谨慎迁移，因为代码变更可能导致旧 checkpoint 无法恢复。需要有意识地制定保留策略，因为暂停的运行会不断累积。

还有一个无法仅靠技术解决的运维问题：暂停的运行必须有超时策略。一笔等待审批长达三个月的退款，不叫暂停，而叫被遗弃。

const stale = await db.query(
  `SELECT thread_id FROM checkpoints
   WHERE status = 'interrupted' AND updated_at < now() - interval '7 days'`,
);
for (const { thread_id } of stale.rows) {
  await applyDecision(thread_id, "reject");
  await notify(thread_id, "auto-rejected after 7 days");
}

使用默认决策恢复运行，总比让 state 永远留在那里更好。至于默认使用哪种决策，这是一个产品问题——涉及资金时，reject 是更安全的选择。

如果两个分支在同一个 superstep 中暂停，你会得到两个条目，并且必须根据 interrupt ID 分别回答它们：

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

const resumeMap: Record<string, string> = {};
if (isInterrupted(result)) {
  for (const i of result[INTERRUPT]) {
    if (i.id) resumeMap[i.id] = await decisionFor(i.value);
  }
}
await graph.invoke(new Command({ resume: resumeMap }), config);

当有两个待处理的 interrupt 时，如果只传入一个标量 resume 值，就只会回答其中一个，另一个仍会一直挂起——看起来和 graph 卡死完全一样。

这种模式的价值

它的核心能力并不是“向人类提一个问题”，而是让一次运行可以暂停并写入持久化存储，然后在未来任意时刻，由另一个进程正确恢复。

要在普通循环中后期补上这种能力，确实非常困难。这也是采用 graph framework 最有说服力的一条理由。如果你的 Agent 会处理资金、向客户发送内容，或者修改基础设施，那么这条理由足以决定技术选型。

《AI That Plans》完整介绍了 human-in-the-loop 的端到端实现——包括 interrupt 的放置位置、持久化 checkpointer、审批 UI、超时策略，以及当暂停期间外部世界已经发生变化时，如何安全地恢复运行。

完整系列位于 xgabriel.com/ai-in-typescript。

如需采取进一步措施，你可以考虑屏蔽此人和／或举报滥用行为。