中断 | LangChain 中文文档

中断允许你在特定点暂停图执行，并在继续之前等待外部输入。这使你能够在需要外部输入才能继续的场景中实现人在回路模式。当触发中断时，LangGraph 会使用其持久化层保存图状态，并无限期等待，直到你恢复执行。

中断的工作原理是在图节点的任意位置调用 interrupt() 函数。该函数接受任何可 JSON 序列化的值，并将其呈现给调用方。当你准备好继续时，可以通过使用 Command 重新调用图来恢复执行，该值随后成为节点内部 interrupt() 调用的返回值。

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

**检查点持久化保留你的进度：**检查点器会写入精确的图状态，这样即使处于错误状态，你之后也可以恢复。
**thread_id 是你的指针：**设置 config={"configurable": {"thread_id": ...}}，告知检查点器要加载哪个状态。
**中断载荷通过 stream.interrupts 呈现：**使用事件流（graph.stream_events(..., version="v3")）时，你传给 interrupt() 的值会出现在 stream.interrupts 上，当运行因等待输入而暂停时，stream.interrupted 为 True。
**检查点持久化保留你的进度：**检查点器会写入精确的图状态，这样即使处于错误状态，你之后也可以恢复。
**thread_id 是你的指针：**在 invoke 方法的选项中设置 { configurable: { thread_id: ... } }，告知检查点器要加载哪个状态。
**中断载荷以 __interrupt__ 形式呈现：**你传给 interrupt() 的值会在 __interrupt__ 字段中返回给调用方，这样你就知道图在等待什么。

你选择的 thread_id 实际上就是你的持久化游标。复用它会恢复同一检查点；使用新值则会以空状态启动一个全新的线程。

使用 `interrupt` 暂停

interrupt 函数会暂停图执行并向调用方返回一个值。当你在节点内调用 interrupt 时，LangGraph 会保存当前的图状态，并等待你用输入恢复执行。

要使用 interrupt，你需要：

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

python

from langgraph.types import interrupt

def approval_node(state: State):
    # 暂停并请求审批
    approved = interrupt("Do you approve this action?")

    # 当你恢复时，Command(resume=...) 会将该值返回到这里
    return {"approved": approved}

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 的确切位置被挂起
状态已保存，通过检查点器保存，以便之后恢复执行。在生产环境中，这应该是一个持久化检查点器（例如由数据库支持）
值返回给调用方，在使用事件流（graph.stream_events(..., version="v3")）时出现在 stream.interrupts 上，或在默认的 invoke() API 下出现在 __interrupt__ 中；它可以是任何可 JSON 序列化的值（字符串、对象、数组等）
值返回给调用方，出现在 __interrupt__ 中；它可以是任何可 JSON 序列化的值（字符串、对象、数组等）
图无限期等待，直到你用响应恢复执行
响应传回节点中，在你恢复时，成为 interrupt() 调用的返回值

恢复中断

中断暂停执行后，你可以使用包含恢复值的 Command 再次调用图来恢复它。恢复值会传回 interrupt 调用，使节点能够使用外部输入继续执行。

驱动可能中断的图的推荐方式是事件流——它通过 stream.interrupts 和 stream.interrupted 呈现中断，并通过 stream.output 暴露最终状态。

python

from langgraph.types import Command

# 首次运行——命中 interrupt 并暂停
# thread_id 是持久化指针（在生产环境中存储稳定的 ID）
config = {"configurable": {"thread_id": "thread-1"}}
stream = graph.stream_events({"input": "data"}, config=config, version="v3")

# 排空流以驱动运行；stream.output 等待最终状态。
final = stream.output

# 当运行因等待人工输入而暂停时，stream.interrupted 为 True，并且
# stream.interrupts 包含传给 interrupt() 的载荷。
if stream.interrupted:
    print(stream.interrupts)
    # > (Interrupt(value='Do you approve this action?'),)

# 用人工响应恢复运行
# resume 载荷会成为节点内部 interrupt() 的返回值
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")
final = resumed.output

INFO

默认的 graph.invoke(...) API 仍然有效，并在 result["__interrupt__"] 下呈现中断。当你不需要流式投影时可以使用它；否则请优先使用 graph.stream_events(..., version="v3")。

关于恢复的关键要点：

恢复时必须使用与中断发生时相同的线程 ID
传给 Command(resume=...) 的值会成为 interrupt 调用的返回值
恢复时，节点会从调用 interrupt 的节点的开头重新开始，因此 interrupt 之前的任何代码都会再次运行
你可以将任何可 JSON 序列化的值作为恢复值传入

WARNING

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

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 序列化的值作为恢复值传入

WARNING

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

常见模式

中断解锁的关键能力是在执行过程中暂停并等待外部输入。这在多种用例中非常有用，包括：

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

使用人在回路（HITL）中断进行流式输出

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

在循环中使用 graph.stream_events(..., version="v3") 返回的类型化投影，直到运行结束：

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

python

from langgraph.types import Command

stream_input: dict | Command = initial_input

while True:
    stream = graph.stream_events(stream_input, config=config, version="v3")

    # 流式输出 LLM 消息数据块（包括子图中的任何数据块），当它们逐块到达时。
    for message in stream.messages:
        for token in message.text:
            display_streaming_content(token)

    # 运行结束（或暂停）后，检查中断并恢复。
    if not stream.interrupted:
        final_state = stream.output
        break

    interrupt_info = stream.interrupts[0].value
    user_response = get_user_input(interrupt_info)
    stream_input = Command(resume=user_response)

stream.messages：对话模型输出作为内容块；遍历每个 message.text 以获取 token 增量。对于嵌套子图，从 stream.subgraphs[*].messages 读取消息数据块。
stream.values：每一步之后的完整状态快照
stream.interrupted / stream.interrupts：每次运行后，检查图是否暂停；从 stream.interrupts 读取载荷
Command(resume=...)：作为下一次 stream_events 的输入以恢复执行；循环直到运行在未中断的情况下完成

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

let streamInput: Record<string, unknown> | Command = initialInput;

while (true) {
  const stream = await graph.streamEvents(streamInput, {
    ...config,
    version: "v3",
  });

  // 流式输出 LLM 消息数据块（包括子图中的任何数据块），当它们逐块到达时。
  for await (const message of stream.messages) {
    for await (const token of message.text) {
      displayStreamingContent(token);
    }
  }

  // 运行结束（或暂停）后，检查中断并恢复。
  if (!stream.interrupted) {
    const finalState = await stream.output;
    break;
  }

  const interruptInfo = stream.interrupts[0].payload;
  const userResponse = await getUserInput(interruptInfo);
  streamInput = new Command({ resume: userResponse });
}

stream.messages：对话模型输出作为内容块；遍历 message.text 以获取 token 增量。对于嵌套子图，从 stream.subgraphs[*].messages 读取消息数据块。
stream.values：每一步之后的完整状态快照
stream.interrupted / stream.interrupts：每次运行后，检查图是否暂停；从 stream.interrupts 读取载荷
Command(resume=...)：作为下一次 streamEvents 的输入以恢复执行；循环直到运行在未中断的情况下完成

处理多个中断

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

python

from typing import Annotated, TypedDict
import operator

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class State(TypedDict):
    vals: Annotated[list[str], operator.add]

def node_a(state):
    answer = interrupt("question_a")
    return {"vals": [f"a:{answer}"]}

def node_b(state):
    answer = interrupt("question_b")
    return {"vals": [f"b:{answer}"]}

graph = (
    StateGraph(State)
    .add_node("a", node_a)
    .add_node("b", node_b)
    .add_edge(START, "a")
    .add_edge(START, "b")
    .add_edge("a", END)
    .add_edge("b", END)
    .compile(checkpointer=InMemorySaver())
)

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

# 第 1 步：流式处理事件以驱动运行；两个并行节点都会命中 interrupt() 并暂停
stream = graph.stream_events({"vals": []}, config, version="v3")
_ = stream.output  # 驱动流直到完成
# stream.interrupts 包含挂起的 Interrupt 载荷
print(stream.interrupts)
# > (Interrupt(value='question_a', id='...'), Interrupt(value='question_b', id='...'))

# 第 2 步：一次性恢复所有挂起的中断
resume_map = {
    i.id: f"answer for {i.value}" for i in stream.interrupts
}
resumed = graph.stream_events(Command(resume=resume_map), config, version="v3")

print("Final state:", resumed.output)
# Final state: {'vals': ['a:answer for question_a', 'b:answer for question_b']}

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" } };

async function main() {
  // 第 1 步：invoke——两个并行节点都会触发 interrupt() 并暂停
  const interruptedResult = await graph.invoke({ vals: [] }, config);
  console.log(interruptedResult);
  /*
  {
    vals: [],
    __interrupt__: [
      { id: '...', value: 'question_a' },
      { id: '...', value: 'question_b' }
    ]
  }
  */

  // 第 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 result = await graph.invoke(new Command({ resume: resumeMap }), config);

  console.log("Final state:", result);
  //> Final state: { vals: ['a:answer for question_a', 'b:answer for question_b'] }
}

main().catch(console.error);

审批或拒绝

中断最常见的用途之一是在关键操作之前暂停并请求审批。例如，你可能希望请求人类审批 API 调用、数据库更改或任何其他重要决策。

python

from typing import Literal
from langgraph.types import interrupt, Command

def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]:
    # 暂停执行；载荷会出现在 stream.interrupts（使用 stream_events 时）或 result["__interrupt__"]（使用 invoke 时）上
    is_approved = interrupt({
        "question": "Do you want to proceed with this action?",
        "details": state["action_details"]
    })

    # 根据响应进行路由
    if is_approved:
        return Command(goto="proceed")  # 在提供恢复载荷后运行
    else:
        return Command(goto="cancel")

typescript

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

const approvalNode: typeof State.Node = (state) => {
  // 暂停执行；载荷会出现在 result.__interrupt__ 中
  const isApproved = interrupt({
    question: "Do you want to proceed?",
    details: state.actionDetails
  });

  // 根据响应进行路由
  if (isApproved) {
    return new Command({ goto: "proceed" }); // 在提供恢复载荷后运行
  } else {
    return new Command({ goto: "cancel" });
  }
}

恢复图时，传入 True 表示审批，或传入 False 表示拒绝：

python

# 表示审批
graph.stream_events(Command(resume=True), config=config, version="v3").output

# 表示拒绝
graph.stream_events(Command(resume=False), config=config, version="v3").output

恢复图时，传入 true 表示审批，或传入 false 表示拒绝：

typescript

// 表示审批
await graph.invoke(new Command({ resume: true }), config);

// 表示拒绝
await graph.invoke(new Command({ resume: false }), config);

完整示例

python

from typing import Literal, Optional, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class ApprovalState(TypedDict):
action_details: str
status: Optional[Literal["pending", "approved", "rejected"]]

def approval_node(state: ApprovalState) -> Command[Literal["proceed", "cancel"]]:
# 暴露细节，以便调用方在 UI 中渲染它们
decision = interrupt(
    {
        "question": "Approve this action?",
        "details": state["action_details"],
    }
)

# 恢复后路由到适当的节点
return Command(goto="proceed" if decision else "cancel")

def proceed_node(state: ApprovalState):
return {"status": "approved"}

def cancel_node(state: ApprovalState):
return {"status": "rejected"}

builder = StateGraph(ApprovalState)
builder.add_node("approval", approval_node)
builder.add_node("proceed", proceed_node)
builder.add_node("cancel", cancel_node)
builder.add_edge(START, "approval")
builder.add_edge("proceed", END)
builder.add_edge("cancel", END)

# 在生产环境中使用更持久的检查点器
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "approval-123"}}
initial = graph.stream_events(
{"action_details": "Transfer $500", "status": "pending"},
config=config,
version="v3",
)
_ = initial.output  # 驱动流直到完成
print(initial.interrupts)  # -> (Interrupt(value={'question': ..., 'details': ...}),)

# 使用决定恢复执行；True 路由到 proceed，False 路由到 cancel
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")
print(resumed.output["status"])

typescript

import {
  Command,
  MemorySaver,
  START,
  END,
  StateGraph,
  StateSchema,
  interrupt,
} from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  actionDetails: z.string(),
  status: z.enum(["pending", "approved", "rejected"]).nullable(),
});

const graphBuilder = new StateGraph(State)
  .addNode("approval", async (state) => {
    // 暴露细节，以便调用方在 UI 中渲染它们
    const decision = interrupt({
      question: "Approve this action?",
      details: state.actionDetails,
    });
    return new Command({ goto: decision ? "proceed" : "cancel" });
  }, { ends: ['proceed', 'cancel'] })
  .addNode("proceed", () => ({ status: "approved" }))
  .addNode("cancel", () => ({ status: "rejected" }))
  .addEdge(START, "approval")
  .addEdge("proceed", END)
  .addEdge("cancel", END);

// 在生产环境中使用更持久的检查点器
const checkpointer = new MemorySaver();
const graph = graphBuilder.compile({ checkpointer });

const config = { configurable: { thread_id: "approval-123" } };
const initial = await graph.invoke(
  { actionDetails: "Transfer $500", status: "pending" },
  config,
);
console.log(initial.__interrupt__);
// [{ value: { question: ..., details: ... } }]

// 使用决定恢复执行；true 路由到 proceed，false 路由到 cancel
const resumed = await graph.invoke(new Command({ resume: true }), config);
console.log(resumed.status); // -> "approved"

审查与编辑状态

有时你可能希望在继续之前，让人类审查和编辑部分图状态。这对于纠正 LLM、添加缺失信息或进行调整非常有用。

python

from langgraph.types import interrupt

def review_node(state: State):
    # 暂停并显示当前内容以供审查（载荷会出现在 stream.interrupts 上）
    edited_content = interrupt({
        "instruction": "Review and edit this content",
        "content": state["generated_text"]
    })

    # 使用编辑后的版本更新状态
    return {"generated_text": edited_content}

typescript

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

const reviewNode: typeof State.Node = (state) => {
  // 暂停并显示当前内容以供审查（会出现在 result.__interrupt__ 中）
  const editedContent = interrupt({
    instruction: "Review and edit this content",
    content: state.generatedText
  });

  // 使用编辑后的版本更新状态
  return { generatedText: editedContent };
}

恢复时，请提供编辑后的内容：

python

graph.stream_events(
    Command(resume="The edited and improved text"),  # 该值将成为 interrupt() 的返回值
    config=config,
    version="v3",
).output

typescript

await graph.invoke(
  new Command({ resume: "The edited and improved text" }), // 该值将成为 interrupt() 的返回值
  config
);

完整示例

python

from typing import TypedDict

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class ReviewState(TypedDict):
generated_text: str

def review_node(state: ReviewState):
# 让审查者编辑生成的内容
updated = interrupt(
    {
        "instruction": "Review and edit this content",
        "content": state["generated_text"],
    }
)
return {"generated_text": updated}

builder = StateGraph(ReviewState)
builder.add_node("review", review_node)
builder.add_edge(START, "review")
builder.add_edge("review", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "review-42"}}
initial = graph.stream_events(
{"generated_text": "Initial draft"}, config=config, version="v3"
)
_ = initial.output  # 驱动流直到完成
print(initial.interrupts)  # -> (Interrupt(value={'instruction': ..., 'content': ...}),)

# 使用审查者编辑后的文本恢复执行
final_state = graph.stream_events(
Command(resume="Improved draft after review"),
config=config,
version="v3",
)
print(final_state.output["generated_text"])  # -> "Improved draft after review"

typescript

import {
  Command,
  MemorySaver,
  START,
  END,
  StateGraph,
  StateSchema,
  interrupt,
} from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  generatedText: z.string(),
});

const builder = new StateGraph(State)
  .addNode("review", async (state) => {
    // 让审查者编辑生成的内容
    const updated = interrupt({
      instruction: "Review and edit this content",
      content: state.generatedText,
    });
    return { generatedText: updated };
  })
  .addEdge(START, "review")
  .addEdge("review", END);

const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });

const config = { configurable: { thread_id: "review-42" } };
const initial = await graph.invoke({ generatedText: "Initial draft" }, config);
console.log(initial.__interrupt__);
// [{ value: { instruction: ..., content: ... } }]

// 使用审查者编辑后的文本恢复执行
const finalState = await graph.invoke(
  new Command({ resume: "Improved draft after review" }),
  config,
);
console.log(finalState.generatedText); // -> "Improved draft after review"

工具中的中断

你还可以直接在工具函数内部放置中断。这使得工具在每次被调用时都会暂停等待审批，并允许人类在执行前审查和编辑工具调用。

首先，定义一个使用 interrupt 的工具：

python

from langchain.tools import tool
from langgraph.types import interrupt

@tool
def send_email(to: str, subject: str, body: str):
    """Send an email to a recipient."""

        # 发送前暂停；使用事件流时，载荷会出现在 stream.interrupts 上
    response = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Approve sending this email?"
    })

    if response.get("action") == "approve":
        # 恢复值可以在执行前覆盖输入
        final_to = response.get("to", to)
        final_subject = response.get("subject", subject)
        final_body = response.get("body", body)
        return f"Email sent to {final_to} with subject '{final_subject}'"
    return "Email cancelled by user"

typescript

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

const sendEmailTool = tool(
  async ({ to, subject, body }) => {
        // 发送前暂停；载荷会出现在 result.__interrupt__ 中
    const response = interrupt({
      action: "send_email",
      to,
      subject,
      body,
      message: "Approve sending this email?",
    });

    if (response?.action === "approve") {
      // 恢复值可以在执行前覆盖输入
      const finalTo = response.to ?? to;
      const finalSubject = response.subject ?? subject;
      const finalBody = response.body ?? body;
      return `Email sent to ${finalTo} with subject '${finalSubject}'`;
    }
    return "Email cancelled by user";
  },
  {
    name: "send_email",
    description: "Send an email to a recipient",
    schema: z.object({
      to: z.string(),
      subject: z.string(),
      body: z.string(),
    }),
  },
);

当你希望审批逻辑与工具本身共存，使其可以在图的不同部分复用时，这种方法非常有用。LLM 可以自然地调用该工具，而每当工具被调用时，中断都会暂停执行，允许你审批、编辑或取消该操作。

完整示例

python

import sqlite3
import operator
from typing import TypedDict, Annotated, Literal
from langchain.tools import tool
from langchain_anthropic import ChatAnthropic
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langchain.messages import AnyMessage, SystemMessage, ToolMessage

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

@tool
def send_email(to: str, subject: str, body: str):
    """Send an email to a recipient."""

# 发送前暂停；使用事件流时，载荷会出现在 stream.interrupts 上
    response = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Approve sending this email?",
    })

    if response.get("action") == "approve":
        final_to = response.get("to", to)
        final_subject = response.get("subject", subject)
        final_body = response.get("body", body)

        # 实际发送电子邮件（这里是你自己的实现）
        print(f"[send_email] to={final_to} subject={final_subject} body={final_body}")
        return f"Email sent to {final_to}"

    return "Email cancelled by user"

model = ChatAnthropic(model="claude-sonnet-4-6").bind_tools([send_email])
tools_by_name = {"send_email": send_email}

def agent_node(state: AgentState):
    # LLM 可能决定调用该工具；中断会在发送前暂停
    result = model.invoke(state["messages"])
    return {"messages": [result]}

def tool_node(state: AgentState):
    """Performs the tool call"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

def should_continue(state: AgentState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""
    messages = state["messages"]
    last_message = messages[-1]

    if last_message.tool_calls:
        return "tool_node"
    return END

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tool_node", tool_node)

builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tool_node", END])  # 路由到 "tools" 或 END
builder.add_edge("tool_node", "agent")  # 在工具之后循环回去

checkpointer = SqliteSaver(
    sqlite3.connect("tool-approval.db", check_same_thread=False)
)
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "email-workflow"}}
initial = graph.stream_events(
    {
        "messages": [
            {"role": "user", "content": "Send an email to alice@example.com about the meeting"}
        ]
    },
    config=config,
    version="v3",
)
initial.output  # 驱动流直到完成
print(initial.interrupts)  # -> (Interrupt(value={'action': 'send_email', ...}),)

# 使用审批以及可选的编辑后参数恢复执行
resumed = graph.stream_events(
    Command(resume={"action": "approve", "subject": "Updated subject"}),
    config=config,
    version="v3",
)
print(resumed.output["messages"][-1])  # -> Tool result returned by send_email

typescript

import { tool } from "@langchain/core/tools";
import { ChatAnthropic } from "@langchain/anthropic";
import {
  Command,
  MemorySaver,
  START,
  END,
  StateGraph,
  StateSchema,
  MessagesValue,
  GraphNode,
  interrupt,
} from "@langchain/langgraph";
import * as z from "zod";

const sendEmailTool = tool(
  async ({ to, subject, body }) => {
// 发送前暂停；载荷会出现在 result.__interrupt__ 中
    const response = interrupt({
      action: "send_email",
      to,
      subject,
      body,
      message: "Approve sending this email?",
    });

    if (response?.action === "approve") {
      const finalTo = response.to ?? to;
      const finalSubject = response.subject ?? subject;
      const finalBody = response.body ?? body;
      console.log("[sendEmailTool]", finalTo, finalSubject, finalBody);
      return `Email sent to ${finalTo}`;
    }
    return "Email cancelled by user";
  },
  {
    name: "send_email",
    description: "Send an email to a recipient",
    schema: z.object({
      to: z.string(),
      subject: z.string(),
      body: z.string(),
    }),
  },
);

const model = new ChatAnthropic({ model: "claude-sonnet-4-6" }).bindTools([sendEmailTool]);

const State = new StateSchema({
  messages: MessagesValue,
});

const agent: typeof State.Node = async (state) => {
  // LLM 可能决定调用该工具；中断会在发送前暂停
  const response = await model.invoke(state.messages);
  return { messages: [response] };
};

const graphBuilder = new StateGraph(State)
  .addNode("agent", agent)
  .addEdge(START, "agent")
  .addEdge("agent", END);

const checkpointer = new MemorySaver();
const graph = graphBuilder.compile({ checkpointer });

const config = { configurable: { thread_id: "email-workflow" } };
const initial = await graph.invoke(
  {
    messages: [
      { role: "user", content: "Send an email to alice@example.com about the meeting" },
    ],
  },
  config,
);
console.log(initial.__interrupt__); // -> [{ value: { action: 'send_email', ... } }]

// 使用审批以及可选的编辑后参数恢复执行
const resumed = await graph.invoke(
  new Command({
    resume: { action: "approve", subject: "Updated subject" },
  }),
  config,
);
console.log(resumed.messages.at(-1)); // -> Tool result returned by send_email

验证人工输入

有时你需要验证来自人类的输入，并在值无效时重新提示。推荐的做法是在每次节点调用中调用一次 interrupt()，从节点返回时将错误消息存储在状态中，并使用条件边循环回到该节点，直到提供有效值。

WARNING

**避免在单个节点内使用 while True + interrupt() 循环。**因为每次恢复时节点都会从头重新运行（请参阅中断规则），一个多次调用 interrupt() 的循环会导致每次恢复都重放之前的所有迭代：第一次恢复重放 1 次迭代，第二次恢复重放 2 次，依此类推。结果是循环体内的任何代码都会被指数级地重复执行。

正确的模式：

将重新提示的问题存储在状态中（例如 pending_question）。
在节点中，恰好调用一次 interrupt()，传入状态中的当前问题。
如果答案无效，返回更新后的 pending_question，以便下一次调用重新提示。
使用 add_conditional_edges 路由回到该节点，直到收集到有效值。

python

from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt

class FormState(TypedDict):
    age: int | None
    pending_question: str | None

def get_age_node(state: FormState):
    question = state.get("pending_question") or "What is your age?"
    answer = interrupt(question)  # 每次调用恰好调用一次
    if isinstance(answer, int) and answer > 0:
        return {"age": answer, "pending_question": None}
    return {"pending_question": f"'{answer}' is not a valid age. Please enter a positive number."}

def route(state: FormState):
    return END if state.get("age") is not None else "collect_age"

builder = StateGraph(FormState)
builder.add_node("collect_age", get_age_node)
builder.add_edge(START, "collect_age")
builder.add_conditional_edges("collect_age", route)

每次恢复都会恰好调用一次 get_age_node、运行一次 interrupt() 调用，然后退出。当答案无效时，条件边会循环回去，下一次中断使用更新后的问题重新提示。每次恢复时没有任何代码运行超过一次。

正确的模式：

将重新提示的问题存储在状态中（例如 pendingQuestion）。
在节点中，恰好调用一次 interrupt()，传入状态中的当前问题。
如果答案无效，返回更新后的 pendingQuestion，以便下一次调用重新提示。
使用 addConditionalEdges 路由回到该节点，直到收集到有效值。

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

const getAgeNode: typeof State.Node = (state) => {
  const question = state.pendingQuestion ?? "What is your age?";
  const answer = interrupt(question); // 每次调用恰好调用一次

  if (typeof answer === "number" && answer > 0) {
    return { age: answer, pendingQuestion: null };
  }
  return {
    pendingQuestion: `'${answer}' is not a valid age. Please enter a positive number.`,
  };
};

// builder.addConditionalEdges("collectAge", (state) =>
//   state.age !== null ? END : "collectAge"
// );

每次恢复都会恰好调用一次 getAgeNode、运行一次 interrupt() 调用，然后退出。当答案无效时，条件边会循环回去，下一次中断使用更新后的问题重新提示。

完整示例

python

from typing import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class FormState(TypedDict):
age: int | None
pending_question: str | None

def get_age_node(state: FormState):
question = state.get("pending_question") or "What is your age?"
answer = interrupt(question)  # 每次节点调用恰好调用一次
print(f"I got {answer}")  # 每次恢复恰好运行一次
if isinstance(answer, int) and answer > 0:
    return {"age": answer, "pending_question": None}
return {"pending_question": f"'{answer}' is not a valid age. Please enter a positive number."}

def route(state: FormState):
# 循环回到 collect_age，直到获得有效的年龄
return END if state.get("age") is not None else "collect_age"

builder = StateGraph(FormState)
builder.add_node("collect_age", get_age_node)
builder.add_edge(START, "collect_age")
builder.add_conditional_edges("collect_age", route)

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "form-1"}}
first = graph.stream_events({"age": None, "pending_question": None}, config=config, version="v3")
_ = first.output  # 驱动流直到完成
print(first.interrupts)  # -> (Interrupt(value='What is your age?', ...),)

# 提供无效数据；节点通过条件边重新提示
retry = graph.stream_events(Command(resume="thirty"), config=config, version="v3")
_ = retry.output
print(retry.interrupts)  # -> (Interrupt(value="'thirty' is not a valid age...", ...),)

# 提供有效数据；route() 返回 END，图完成
final = graph.stream_events(Command(resume=30), config=config, version="v3")
print(final.output["age"])  # -> 30

typescript

import {
  Command,
  MemorySaver,
  START,
  END,
  StateGraph,
  StateSchema,
  interrupt,
} from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  age: z.number().nullable(),
  pendingQuestion: z.string().nullable(),
});

const builder = new StateGraph(State)
  .addNode("collectAge", (state) => {
    const question = state.pendingQuestion ?? "What is your age?";
    const answer = interrupt(question); // 每次调用只调用一次

    if (typeof answer === "number" && answer > 0) {
      return { age: answer, pendingQuestion: null };
    }
    return { pendingQuestion: `'${answer}' is not a valid age. Please enter a positive number.` };
  })
  .addEdge(START, "collectAge")
  .addConditionalEdges("collectAge", (state) =>
    state.age !== null ? END : "collectAge"
  );

const checkpointer = new MemorySaver();
const graph = builder.compile({ checkpointer });

const config = { configurable: { thread_id: "form-1" } };
const first = await graph.invoke({ age: null, pendingQuestion: null }, config);
console.log(first.__interrupt__); // -> [{ value: "What is your age?", ... }]

// 提供无效数据；节点通过条件边重新提示
const retry = await graph.invoke(new Command({ resume: "thirty" }), config);
console.log(retry.__interrupt__); // -> [{ value: "'thirty' is not a valid age...", ... }]

// 提供有效数据；路由返回 END，图完成
const final = await graph.invoke(new Command({ resume: 30 }), config);
console.log(final.age); // -> 30

中断规则

当你在节点内调用 interrupt 时，LangGraph 通过抛出一个向运行时发出暂停信号的异常来挂起执行。该异常沿调用栈向上传播，被运行时捕获，运行时随即通知图保存当前状态并等待外部输入。

当执行恢复时（在你提供所请求的输入之后），运行时从开头重新启动整个节点——它不会从调用 interrupt 的确切行继续。这意味着在 interrupt 之前运行的任何代码都会再次执行。因此，在使用中断时需要遵循一些重要规则，以确保它们按预期工作。

不要用 try/except 包裹 `interrupt` 调用

interrupt 在调用点暂停执行的方式是抛出一个特殊的异常。如果你用 try/except 块包裹 interrupt 调用，你会捕获到这个异常，中断就不会传回给图。

✅ 将 interrupt 调用与容易出错的代码分开
✅ 在 try/except 块中使用特定的异常类型

Python · Separating logicPython · Explicit exception handling

python

def node_a(state: State):
    # ✅ 正确：先中断，然后分别处理
    # 错误条件
    interrupt("What's your name?")
    try:
        fetch_data()  # 这可能会失败
    except Exception as e:
        print(e)
    return state

python

def node_a(state: State):
    # ✅ 正确：捕获特定的异常类型
    # 不会捕获中断异常
    try:
        name = interrupt("What's your name?")
        fetch_data()  # 这可能会失败
    except NetworkException as e:
        print(e)
    return state

🔴 不要用裸 try/except 块包裹 interrupt 调用

python

def node_a(state: State):
    # ❌ 错误：用裸 try/except 包裹 interrupt
    # 会捕获中断异常
    try:
        interrupt("What's your name?")
    except Exception as e:
        print(e)
    return state

不要用 try/catch 包裹 `interrupt` 调用

interrupt 在调用点暂停执行的方式是抛出一个特殊的异常。如果你用 try/catch 块包裹 interrupt 调用，你会捕获到这个异常，中断就不会传回给图。

✅ 将 interrupt 调用与容易出错的代码分开
✅ 如有需要，有条件地捕获错误

TypeScript · Separating logicTypeScript · Conditionally handling errors

typescript

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

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ✅ 正确：重新抛出异常会
  // 允许中断被传回给
  // 图
  try {
    const name = interrupt("What's your name?");
    await fetchData(); // 这可能会失败
  } catch (err) {
    if (error instanceof NetworkError) {
      console.error(error);
    }
    throw error;
  }
  return state;
}

🔴 不要用裸 try/catch 块包裹 interrupt 调用

typescript

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

不要重新排列节点内的 `interrupt` 调用

在单个节点中使用多个中断是很常见的，但如果处理不当，这可能会导致意外行为。

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

✅ 在节点的各次执行中保持 interrupt 调用一致

python

def node_a(state: State):
    # ✅ 正确：interrupt 调用每次都按相同顺序发生
    name = interrupt("What's your name?")
    age = interrupt("What's your age?")
    city = interrupt("What's your city?")

    return {
        "name": name,
        "age": age,
        "city": city
    }

typescript

async function nodeA(state: State) {
    // ✅ 正确：interrupt 调用每次都按相同顺序发生
    const name = interrupt("What's your name?");
    const age = interrupt("What's your age?");
    const city = interrupt("What's your city?");

    return {
        name,
        age,
        city
    };
}

🔴 不要在节点内条件性地跳过 interrupt 调用
🔴 不要使用在各次执行中不确定的逻辑循环 interrupt 调用，包括 while True 验证循环。请改用条件边（参见验证人工输入）

Python · Skipping interruptsPython · Looping interruptsTypeScript · Skipping interruptsTypeScript · Looping interrupts

python

def node_a(state: State):
    # ❌ 错误：条件性跳过中断会改变顺序
    name = interrupt("What's your name?")

    # 首次运行时，这可能会跳过中断
    # 恢复时它可能不会跳过，导致索引不匹配
    if state.get("needs_age"):
        age = interrupt("What's your age?")

    city = interrupt("What's your city?")

    return {"name": name, "city": city}

python

def node_a(state: State):
    # ❌ 错误：基于不确定的数据进行循环
    # 各次执行之间中断的数量会变化
    results = []
    for item in state.get("dynamic_list", []):  # 列表在两次运行之间可能会变化
        result = interrupt(f"Approve {item}?")
        results.append(result)

    return {"results": results}

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ❌ 错误：条件性跳过中断会改变顺序
  const name = interrupt("What's your name?");

  // 首次运行时，这可能会跳过中断
  // 恢复时它可能不会跳过，导致索引不匹配
  if (state.needsAge) {
    const age = interrupt("What's your age?");
  }

  const city = interrupt("What's your city?");

  return { name, city };
}

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ❌ 错误：基于不确定的数据进行循环
  // 各次执行之间中断的数量会变化
  const results = [];
  for (const item of state.dynamicList || []) {  // 列表在两次运行之间可能会变化
    const result = interrupt(`Approve ${item}?`);
    results.push(result);
  }

  return { results };
}

不要在 `interrupt` 调用中返回复杂值

取决于所使用的检查点器，复杂值可能无法序列化（例如你无法序列化函数）。为了让你的图能够适应任何部署，最佳实践是只使用可以合理序列化的值。

✅ 向 interrupt 传入简单、可 JSON 序列化的类型
✅ 传入包含简单值的字典/对象

Python · Simple valuesPython · Structured dataTypeScript · Simple valuesTypeScript · Structured data

python

def node_a(state: State):
    # ✅ 正确：传入可序列化的简单类型
    name = interrupt("What's your name?")
    count = interrupt(42)
    approved = interrupt(True)

    return {"name": name, "count": count, "approved": approved}

python

def node_a(state: State):
    # ✅ 正确：传入包含简单值的字典
    response = interrupt({
        "question": "Enter user details",
        "fields": ["name", "email", "age"],
        "current_values": state.get("user", {})
    })

    return {"user": response}

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ✅ 正确：传入可序列化的简单类型
  const name = interrupt("What's your name?");
  const count = interrupt(42);
  const approved = interrupt(true);

  return { name, count, approved };
}

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ✅ 正确：传入包含简单值的对象
  const response = interrupt({
    question: "Enter user details",
    fields: ["name", "email", "age"],
    currentValues: state.user || {}
  });

  return { user: response };
}

🔴 不要向 interrupt 传入函数、类实例或其他复杂对象

Python · FunctionsPython · Class instancesTypeScript · FunctionsTypeScript · Class instances

python

def validate_input(value):
    return len(value) > 0

def node_a(state: State):
    # ❌ 错误：将函数传给 interrupt
    # 函数无法被序列化
    response = interrupt({
        "question": "What's your name?",
        "validator": validate_input  # 这将会失败
    })
    return {"name": response}

python

class DataProcessor:
    def __init__(self, config):
        self.config = config

def node_a(state: State):
    processor = DataProcessor({"mode": "strict"})

    # ❌ 错误：将类实例传给 interrupt
    # 实例无法被序列化
    response = interrupt({
        "question": "Enter data to process",
        "processor": processor  # 这将会失败
    })
    return {"result": response}

typescript

function validateInput(value: string): boolean {
    return value.length > 0;
}

const nodeA: GraphNode<typeof State> = async (state) => {
  // ❌ 错误：将函数传给 interrupt
  // 函数无法被序列化
  const response = interrupt({
    question: "What's your name?",
    validator: validateInput  // 这将会失败
  });
  return { name: response };
}

typescript

class DataProcessor {
    constructor(private config: any) {}
}

const nodeA: GraphNode<typeof State> = async (state) => {
  const processor = new DataProcessor({ mode: "strict" });

  // ❌ 错误：将类实例传给 interrupt
  // 实例无法被序列化
  const response = interrupt({
    question: "Enter data to process",
    processor: processor  // 这将会失败
  });
  return { result: response };
}

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

由于中断的工作原理是重新运行调用它们的节点，因此在 interrupt 之前调用的副作用（理想情况下）应该是幂等的。顺便说明一下，幂等意味着同一操作可以被多次应用，而结果不会超出首次执行的效果。

例如，你可能在节点内部有一个用于更新记录的 API 调用。如果在该调用之后调用 interrupt，当节点恢复时，它会多次重新运行，可能覆盖最初的更新或创建重复的记录。

✅ 在 interrupt 之前使用幂等操作
✅ 将副作用放在 interrupt 调用之后
✅ 尽可能将副作用分离到单独的节点中

Python · Idempotent operationsPython · Side effects after interruptPython · Separating into different nodesTypeScript · Idempotent operationsTypeScript · Side effects after interruptTypeScript · Separating into different nodes

python

def node_a(state: State):
    # ✅ 正确：使用幂等的 upsert 操作
    # 多次运行都会得到相同的结果
    db.upsert_user(
        user_id=state["user_id"],
        status="pending_approval"
    )

    approved = interrupt("Approve this change?")

    return {"approved": approved}

python

def node_a(state: State):
    # ✅ 正确：将副作用放在 interrupt 之后
    # 这可以确保在收到审批后只运行一次
    approved = interrupt("Approve this change?")

    if approved:
        db.create_audit_log(
            user_id=state["user_id"],
            action="approved"
        )

    return {"approved": approved}

python

def approval_node(state: State):
    # ✅ 正确：只在此节点中处理中断
    approved = interrupt("Approve this change?")

    return {"approved": approved}

def notification_node(state: State):
    # ✅ 正确：副作用发生在单独的节点中
    # 这会在审批后运行，因此只执行一次
    if (state.approved):
        send_notification(
            user_id=state["user_id"],
            status="approved"
        )

    return state

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ✅ 正确：使用幂等的 upsert 操作
  // 多次运行都会得到相同的结果
  await db.upsertUser({
    userId: state.userId,
    status: "pending_approval"
  });

  const approved = interrupt("Approve this change?");

  return { approved };
}

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ✅ 正确：将副作用放在 interrupt 之后
  // 这可以确保在收到审批后只运行一次
  const approved = interrupt("Approve this change?");

  if (approved) {
    await db.createAuditLog({
      userId: state.userId,
      action: "approved"
    });
  }

  return { approved };
}

typescript

const approvalNode: GraphNode<typeof State> = async (state) => {
  // ✅ 正确：只在此节点中处理中断
  const approved = interrupt("Approve this change?");

  return { approved };
}

const notificationNode: GraphNode<typeof State> = async (state) => {
  // ✅ 正确：副作用发生在单独的节点中
  // 这会在审批后运行，因此只执行一次
  if (state.approved) {
    await sendNotification({
      userId: state.userId,
      status: "approved",
    });
  }

  return state;
}

🔴 不要在 interrupt 之前执行非幂等操作
🔴 不要在不检查记录是否已存在的情况下创建新记录

Python · Creating recordsPython · Appending to listsTypeScript · Creating recordsTypeScript · Appending to arrays

python

def node_a(state: State):
    # ❌ 错误：在 interrupt 之前创建新记录
    # 每次恢复时都会创建重复记录
    audit_id = db.create_audit_log({
        "user_id": state["user_id"],
        "action": "pending_approval",
        "timestamp": datetime.now()
    })

    approved = interrupt("Approve this change?")

    return {"approved": approved, "audit_id": audit_id}

python

def node_a(state: State):
    # ❌ 错误：在 interrupt 之前向列表追加
    # 每次恢复时都会添加重复条目
    db.append_to_history(state["user_id"], "approval_requested")

    approved = interrupt("Approve this change?")

    return {"approved": approved}

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ❌ 错误：在 interrupt 之前创建新记录
  // 每次恢复时都会创建重复记录
  const auditId = await db.createAuditLog({
    userId: state.userId,
    action: "pending_approval",
    timestamp: new Date()
  });

  const approved = interrupt("Approve this change?");

  return { approved, auditId };
}

typescript

const nodeA: GraphNode<typeof State> = async (state) => {
  // ❌ 错误：在 interrupt 之前向数组追加
  // 每次恢复时都会添加重复条目
  await db.appendToHistory(state.userId, "approval_requested");

  const approved = interrupt("Approve this change?");

  return { approved };
}

与作为函数调用的子图一起使用

当在节点内调用子图时，父图将从调用子图并触发 interrupt 的节点的开头恢复执行。类似地，子图也会从调用 interrupt 的节点的开头恢复执行。

python

def node_in_parent_graph(state: State):
    some_code()  # <-- 恢复时会重新执行
    # 将子图作为函数调用。
    # 该子图包含一个 `interrupt` 调用。
    subgraph_result = subgraph.invoke(some_input)
    # ...

def node_in_subgraph(state: State):
    some_other_code()  # <-- 恢复时也会重新执行
    result = interrupt("What's your name?")
    # ...

typescript

async function nodeInParentGraph(state: State) {
    someCode(); // <-- 恢复时会重新执行
    // 将子图作为函数调用。
    // 该子图包含一个 `interrupt` 调用。
    const subgraphResult = await subgraph.invoke(someInput);
    // ...
}

async function nodeInSubgraph(state: State) {
    someOtherCode(); // <-- 恢复时也会重新执行
    const result = interrupt("What's your name?");
    // ...
}

使用中断进行调试

要调试和测试图，你可以使用静态中断作为断点，一次一个节点地逐步执行图。静态中断在节点执行之前或之后的规定点触发。你可以在编译图时通过指定 interrupt_before 和 interrupt_after 来设置它们。要调试和测试图，你可以使用静态中断作为断点，一次一个节点地逐步执行图。静态中断在节点执行之前或之后的规定点触发。你可以在编译图时通过指定 interruptBefore 和 interruptAfter 来设置它们。

INFO

不建议在人在回路工作流中使用静态中断。请改用 interrupt 函数。

在编译时

python

graph = builder.compile(
    interrupt_before=["node_a"],  
    interrupt_after=["node_b", "node_c"],  
    checkpointer=checkpointer,
)

# 向图传递线程 ID
config = {
    "configurable": {
        "thread_id": "some_thread"
    }
}

# 运行图直到断点
graph.invoke(inputs, config=config)  

# 恢复图
graph.invoke(None, config=config)

断点是在 compile 期间设置的。
interrupt_before 指定执行应在哪些节点之前暂停。
interrupt_after 指定执行应在哪些节点之后暂停。
启用断点需要一个检查点器。
图一直运行到第一个断点。
通过传入 None 作为输入来恢复图。这将一直运行图直到下一个断点。

typescript

const graph = builder.compile({
    interruptBefore: ["node_a"],  
    interruptAfter: ["node_b", "node_c"],  
    checkpointer,
});

// 向图传递线程 ID
const config = {
    configurable: {
        thread_id: "some_thread"
    }
};

// 运行图直到断点
await graph.invoke(inputs, config);# [!code highlight]

await graph.invoke(null, config);  # [!code highlight]

断点是在 compile 期间设置的。
interruptBefore 指定执行应在哪些节点之前暂停。
interruptAfter 指定执行应在哪些节点之后暂停。
启用断点需要一个检查点器。
图一直运行到第一个断点。
通过传入 null 作为输入来恢复图。这将一直运行图直到下一个断点。

在运行时

python

config = {
    "configurable": {
        "thread_id": "some_thread"
    }
}

# 运行图直到断点
graph.invoke(
    inputs,
    interrupt_before=["node_a"],  
    interrupt_after=["node_b", "node_c"],  
    config=config,
)

# 恢复图
graph.invoke(None, config=config)

使用 interrupt_before 和 interrupt_after 参数调用 graph.invoke。这是运行时配置，可以针对每次调用进行更改。
interrupt_before 指定执行应在哪些节点之前暂停。
interrupt_after 指定执行应在哪些节点之后暂停。
图一直运行到第一个断点。
通过传入 None 作为输入来恢复图。这将一直运行图直到下一个断点。

typescript

// 运行图直到断点
graph.invoke(inputs, {
    interruptBefore: ["node_a"],  
    interruptAfter: ["node_b", "node_c"],  
    configurable: {
        thread_id: "some_thread"
    }
});

// 恢复图
await graph.invoke(null, config);

使用 interruptBefore 和 interruptAfter 参数调用 graph.invoke。这是运行时配置，可以针对每次调用进行更改。
interruptBefore 指定执行应在哪些节点之前暂停。
interruptAfter 指定执行应在哪些节点之后暂停。
图一直运行到第一个断点。
通过传入 null 作为输入来恢复图。这将一直运行图直到下一个断点。

TIP

要调试你的中断，请使用 LangSmith。

使用 LangSmith Studio

你可以使用 LangSmith Studio 在运行图之前通过界面在图中设置静态中断。你也可以使用该界面在执行过程中的任意时刻检查图状态。

Graph API

Functional API

使用 `interrupt` 暂停

恢复中断

常见模式

使用人在回路（HITL）中断进行流式输出

处理多个中断

审批或拒绝

完整示例

审查与编辑状态

完整示例

工具中的中断

完整示例

验证人工输入

完整示例

中断规则

不要用 try/except 包裹 `interrupt` 调用

不要用 try/catch 包裹 `interrupt` 调用

不要重新排列节点内的 `interrupt` 调用

不要在 `interrupt` 调用中返回复杂值

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

与作为函数调用的子图一起使用

使用中断进行调试

在编译时

在运行时

使用 LangSmith Studio

使用 interrupt 暂停 ​

恢复中断 ​

常见模式 ​

使用人在回路（HITL）中断进行流式输出 ​

处理多个中断 ​

审批或拒绝 ​

完整示例 ​

审查与编辑状态 ​

完整示例 ​

工具中的中断 ​

完整示例 ​

验证人工输入 ​

完整示例 ​

中断规则 ​

不要用 try/except 包裹 interrupt 调用 ​

不要用 try/catch 包裹 interrupt 调用 ​

不要重新排列节点内的 interrupt 调用 ​

不要在 interrupt 调用中返回复杂值 ​

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

与作为函数调用的子图一起使用 ​

使用中断进行调试 ​

在编译时 ​

在运行时 ​

使用 LangSmith Studio ​

使用 `interrupt` 暂停

恢复中断

常见模式

使用人在回路（HITL）中断进行流式输出

处理多个中断

审批或拒绝

完整示例

审查与编辑状态

完整示例

工具中的中断

完整示例

验证人工输入

完整示例

中断规则

不要用 try/except 包裹 `interrupt` 调用

不要用 try/catch 包裹 `interrupt` 调用

不要重新排列节点内的 `interrupt` 调用

不要在 `interrupt` 调用中返回复杂值

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

与作为函数调用的子图一起使用

使用中断进行调试

在编译时

在运行时

使用 LangSmith Studio