中断（Interrupts）

中断允许你在特定点暂停图执行，并在继续之前等待外部输入。这支持了在需要外部输入才能继续的人机协同（human-in-the-loop）模式。触发中断时，LangGraph 使用其持久化层保存图状态，并无限期等待，直到你恢复执行。

中断通过在图节点的任何位置调用 interrupt() 函数来工作。该函数接受向调用者公开的任何可 JSON 序列化的值。当你准备好继续时，可以通过使用 Command 重新调用图来恢复执行，该值随后成为节点内部 interrupt() 调用的返回值。

与静态断点（在特定节点之前或之后暂停）不同，中断是动态的：它们可以放置在代码中的任何位置，并且可以根据你的应用程序逻辑附加条件。

检查点记录你的位置：检查点保存器写入确切的图状态，以便你稍后恢复，即使在错误状态下也是如此。
thread_id 是你的指针：使用 { configurable: { thread_id: ... } } 作为 invoke 方法的选项，以告诉检查点保存器加载哪个状态。
中断载荷作为 __interrupt__ 显露：传递给 interrupt() 的值在 __interrupt__ 字段中返回给调用者，以便你知道图在等待什么。

你选择的 thread_id 实际上是你的持久游标。重用它可以恢复相同的检查点；使用新值将启动具有空状态的全新线程。

使用 `interrupt` 暂停

interrupt 函数暂停图执行并将值返回给调用者。当你在节点内调用 interrupt 时，LangGraph 会保存当前图状态并等待你携带输入恢复执行。

要使用 interrupt，你需要：

一个检查点保存器来持久化图状态（在生产中使用持久化检查点保存器）
配置中的线程 ID，以便运行时知道从哪个状态恢复
在你想要暂停的位置调用 interrupt()（载荷必须可 JSON 序列化）

typescript

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

async function approvalNode(state: State) {
    // 暂停并请求批准
    const approved = interrupt("Do you approve this action?");

    // Command({ resume: ... }) 提供返回到该变量中的值
    return { approved };
}

当你调用 interrupt 时，会发生以下情况：

图执行暂停在调用 interrupt 的确切点
保存状态使用检查点保存器，以便以后可以恢复执行。在生产中，这应该是一个持久化检查点保存器（例如由数据库支持）
将值返回给调用者下的 __interrupt__；它可以是任何可 JSON 序列化的值（字符串、对象、数组等）
图无限期等待，直到你携带响应恢复执行
当你恢复时，响应被传递回节点中，成为 interrupt() 调用的返回值

恢复中断（Resuming interrupts）

在中断暂停执行后，你可以通过再次使用包含恢复值的 Command 调用图来恢复该图。恢复值将传递回 interrupt 调用，允许节点使用外部输入继续执行。

typescript

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

// 初始运行 - 命中断点并暂停
// thread_id 是指向已保存检查点的持久指针
const config = { configurable: { thread_id: "thread-1" } };
const result = await graph.invoke({ input: "data" }, config);

// 检查被中断的内容
// __interrupt__ 镜像你传递给 interrupt() 的每个载荷
console.log(result.__interrupt__);
// [{ value: 'Do you approve this action?', ... }]

// 使用人类的响应恢复
// Command({ resume }) 从节点中的 interrupt() 返回该值
await graph.invoke(new Command({ resume: true }), config);

关于恢复的关键要点：

恢复时必须使用发生中断时使用的相同线程 ID
传递给 new Command({ resume: ... }) 的值成为 interrupt 调用的返回值
恢复执行时，节点从调用 interrupt 的节点开头重新启动，因此 interrupt 之前的任何代码都会再次运行
你可以将任何可 JSON 序列化的值作为恢复值传递，在指定 Zod responseSchema 时会受到校验

WARNING

new Command({ resume: ... }) 是唯一旨在作为 invoke()/stream()/stream_events() 输入的 Command 模式。其他 Command 参数（update、goto、graph）设计用于从节点函数返回。不要传递 new Command({ update: ... }) 作为输入来继续多轮对话——请改用普通输入对象。

定义中断响应架构（responseSchema）

interrupt 的 responseSchema 选项描述恢复图时预期的输入。客户端可以使用该模式渲染类型化的输入表单。

INFO

中断响应架构需要 @langchain/langgraph 版本 1.4.16 或更高版本。

传递 Zod 架构以验证恢复值。LangGraph 将其转换为 JSON Schema 并将其包含在 Interrupt.response_schema 中，与中断的 value 载荷分开。该架构在返回给调用者的中断中以及检查点任务中断中可用。

此示例使用来自 LangGraph 中断测试的审批模式：

typescript

import {
  Annotation,
  Command,
  MemorySaver,
  START,
  StateGraph,
  interrupt,
} from "@langchain/langgraph";
import { z } from "zod/v4";

const Decision = z.object({
  approved: z.boolean(),
  note: z.string().default(""),
});

const State = Annotation.Root({
  answer: Annotation<unknown>(),
});

const graph = new StateGraph(State)
  .addNode("node", () => ({
    answer: interrupt({ question: "approve?" }, { responseSchema: Decision }),
  }))
  .addEdge(START, "node")
  .compile({ checkpointer: new MemorySaver() });

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

const result = await graph.invoke({ answer: null }, config);
console.log(result.__interrupt__?.[0].response_schema);

const resumed = await graph.invoke(
  new Command({ resume: { approved: true } }),
  config
);
console.log(resumed.answer);

使用 Zod 架构，interrupt() 返回解析后的值。在此示例中，结果是 { approved: true, note: "" }，包括默认的 note。无效输入会引发 ZodError；你可以使用相同的线程 ID 使用更正后的 new Command({ resume: ... }) 重试。

你还可以将 JSON Schema 对象作为 responseSchema 传递。LangGraph 原封不动地公开该对象，但不会针对它验证恢复值。当你省略 responseSchema 时，中断没有 response_schema 字段，恢复值无需架构验证即可通过。

常见模式

中断解锁的关键是暂停执行并等待外部输入的能力。这对于各种使用案例非常有用，包括：

审批工作流：在执行关键操作（API 调用、数据库更改、金融交易）之前暂停
处理多个中断：在单次调用中恢复多个中断时，将中断 ID 与恢复值配对
审查和编辑：让人类在继续之前审查和修改 LLM 输出或工具调用
在工具中中断：在执行工具调用之前暂停以审查和编辑工具调用
验证人类输入：在继续下一步之前暂停以验证人类输入

流式传输配合人机协同（HITL）中断

构建具有人机协同工作流的交互式智能体时，你可以使用事件流式传输在处理中断的同时并发消费消息块和状态快照。

在循环中使用 graph.streamEvents(..., { version: "v3" }) 返回的类型化投影，直到运行完成：

通过 stream.messages 逐 Token 流式传输 AI 响应
通过 stream.values 观察每个步骤的状态快照
通过 stream.interrupted 检测中断并从 stream.interrupts 读取其载荷
通过使用 Command(resume=...) 再次调用 streamEvents 来恢复执行，并重复直到 stream.interrupted 为 false

typescript

let stream = await graph.streamEvents(input, { version: "v3" });

while (true) {
  for await (const message of stream.messages) {
    for await (const token of message.text) {
      process.stdout.write(token);
    }
  }

  if (!stream.interrupted) {
    break;
  }

  // 读取中断载荷
  const payload = stream.interrupts[0].value;
  const humanResponse = await getHumanInput(payload);

  stream = await graph.streamEvents(
    new Command({ resume: humanResponse }),
    { version: "v3" }
  );
}

const finalState = await stream.output;

处理多个中断

当并行分支同时中断时（例如扇出到各自调用 interrupt() 的多个节点），你可能需要在单次调用中恢复多个中断。将每个中断 ID 映射到其恢复值，以确保每个响应在运行时与正确的中断配对：

typescript

import {
  Annotation,
  Command,
  END,
  INTERRUPT,
  MemorySaver,
  START,
  StateGraph,
  interrupt,
  isInterrupted,
} from "@langchain/langgraph";

const State = Annotation.Root({
  vals: Annotation<string[]>({
    reducer: (left, right) =>
      left.concat(Array.isArray(right) ? right : [right]),
    default: () => [],
  }),
});

function nodeA(_state: typeof State.State) {
  const answer = interrupt("question_a") as string;
  return { vals: [`a:${answer}`] };
}

function nodeB(_state: typeof State.State) {
  const answer = interrupt("question_b") as string;
  return { vals: [`b:${answer}`] };
}

const graph = new StateGraph(State)
  .addNode("a", nodeA)
  .addNode("b", nodeB)
  .addEdge(START, "a")
  .addEdge(START, "b")
  .addEdge("a", END)
  .addEdge("b", END)
  .compile({ checkpointer: new MemorySaver() });

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

// 步骤 1：调用 - 两个并行节点均命中 interrupt() 并暂停
const interruptedResult = await graph.invoke({ vals: [] }, config);

// 步骤 2：一次性恢复所有未决中断
const resumeMap: Record<string, string> = {};
if (isInterrupted(interruptedResult)) {
  for (const i of interruptedResult[INTERRUPT]) {
    if (i.id != null) {
      resumeMap[i.id] = `answer for ${i.value}`;
    }
  }
}

const resumedResult = await graph.invoke(new Command({ resume: resumeMap }), config);

审批或拒绝

在节点执行关键外部副作用之前，暂停并请求审批：

typescript

const stepWithApproval = async (state: typeof State.State) => {
  const isApproved = interrupt({
    action: "delete_database",
    reason: "Maintenance",
  });

  if (!isApproved) {
    return { status: "rejected" };
  }

  await performAction();
  return { status: "approved" };
};

审查并编辑状态

让人类在模型生成之后审查内容并可选地进行编辑，然后继续后续流程：

typescript

const reviewNode = async (state: typeof State.State) => {
  const editedText = interrupt({
    originalText: state.generatedText,
    instruction: "Please edit or approve the draft",
  });

  return { finalText: editedText };
};

在工具中中断

不仅可以在图节点中调用 interrupt()，还可以在 Tool 定义的内部直接调用 interrupt()。工具内部的中断会自动向外挂起整个图：

typescript

import { tool } from "@langchain/core/tools";
import { interrupt } from "@langchain/langgraph";
import * as z from "zod";

const sensitiveOperation = tool(
  async ({ query }) => {
    const confirmation = interrupt({
      message: `Are you sure you want to run query: ${query}?`,
    });
    if (!confirmation) {
      return "Operation cancelled by user.";
    }
    return `Query ${query} executed successfully.`;
  },
  {
    name: "sensitive_operation",
    description: "Run a sensitive operation requiring approval.",
    schema: z.object({ query: z.string() }),
  }
);

验证人类输入

在循环中验证人类输入时，不要在节点内使用 while 循环调用 interrupt()。请使用带有条件边的反馈分支：

中断的规则

当你在节点内调用 interrupt 时，LangGraph 通过引发通知运行时暂停的异常来挂起执行。当执行恢复时，运行时从头重新启动整个节点——它不会从调用 interrupt 的确切行恢复。这意味着在 interrupt 之前运行的任何代码都将再次执行。因此在处理中断时必须遵循一些重要规则：

切勿在裸 try/catch 中包装 `interrupt` 调用

interrupt 暂停执行的方式是抛出特殊异常。如果在 try/catch 块中包装 interrupt 调用，你将捕获此异常并且中断将无法传递回图。

✅ 将 interrupt 调用与易出错的代码分开
✅ 必要时有条件地捕获错误并重新抛出

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ✅ 良好：先中断，然后分别处理错误条件
  const name = interrupt("What's your name?");
  try {
    await fetchData(); // 可能会失败
  } catch (err) {
    console.error(err);
  }
  return state;
}

🔴 切勿将 interrupt 调用包装在简单的 try/catch 块中

typescript

async function nodeA(state: State) {
  // ❌ 错误：在简单 try/catch 中包装 interrupt 会捕获中断异常
  try {
    const name = interrupt("What's your name?");
  } catch (err) {
    console.error(err);
  }
  return state;
}

切勿对节点内的 `interrupt` 调用重新排序

当节点包含多个中断调用时，LangGraph 会保留特定于执行该节点的任务的恢复值列表。无论何时恢复执行，它都会从节点的开头开始。对于遇到的每个中断，LangGraph 都会检查任务的恢复列表中是否存在匹配的值。匹配严格基于索引，因此节点内中断调用的顺序至关重要。

✅ 保持跨节点执行的 interrupt 调用一致
🔴 不要根据动态状态在节点内部条件跳过某个 interrupt
🔴 不要在节点内部使用非确定性的 while 循环执行 interrupt

切勿在 `interrupt` 调用中返回复杂不可序列化的值

interrupt 传递的值和恢复值必须可被 JSON 序列化，以便检查点保存器存储并供客户端读取。

在 `interrupt` 之前调用的副作用必须是幂等的

因为节点在恢复时会重新运行，所以在 interrupt 之前调用的任何代码都会再次执行。外部数据库写入或支付 API 等副作用必须保持幂等，或者放在 interrupt 调用之后。

与作为函数调用的子图配合使用

当作为函数调用的子图内部包含 interrupt() 时，中断信号将自动向上冒泡穿过父图并在父图级别挂起。恢复时，提供的值将向下传递到相应的子图中。

使用中断进行调试

在调试过程中，中断可以充当动态断点，允许你检查图的状态快照并验证中间步骤。配合 LangSmith Studio，中断会被渲染为交互式表单控件，极大简化了人工协同与调试体验。

中断（Interrupts） ​

使用 interrupt 暂停 ​

恢复中断（Resuming interrupts） ​

定义中断响应架构（responseSchema） ​

常见模式 ​

流式传输配合人机协同（HITL）中断 ​

处理多个中断 ​

审批或拒绝 ​

审查并编辑状态 ​

在工具中中断 ​

验证人类输入 ​

中断的规则 ​

切勿在裸 try/catch 中包装 interrupt 调用 ​

切勿对节点内的 interrupt 调用重新排序 ​

切勿在 interrupt 调用中返回复杂不可序列化的值 ​

在 interrupt 之前调用的副作用必须是幂等的 ​

与作为函数调用的子图配合使用 ​

使用中断进行调试 ​

中断（Interrupts）

使用 `interrupt` 暂停

恢复中断（Resuming interrupts）

定义中断响应架构（responseSchema）

常见模式

流式传输配合人机协同（HITL）中断

处理多个中断

审批或拒绝

审查并编辑状态

在工具中中断

验证人类输入

中断的规则

切勿在裸 try/catch 中包装 `interrupt` 调用

切勿对节点内的 `interrupt` 调用重新排序

切勿在 `interrupt` 调用中返回复杂不可序列化的值

在 `interrupt` 之前调用的副作用必须是幂等的

与作为函数调用的子图配合使用

使用中断进行调试