中断

中断允许你在特定点暂停图执行，并在继续之前等待外部输入。这实现了你需要外部输入才能继续的人在回路模式。当中断被触发时，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 实际上就是你的持久游标。复用它会在同一个检查点恢复；使用新值则会以空状态启动一个全新的线程。

使用 `interrupt` 暂停

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

要使用 interrupt，你需要：

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

python

from langgraph.types import interrupt

def approval_node(state: State):
    # Pause and ask for approval
    approved = interrupt("Do you approve this action?")

    # When you resume, Command(resume=...) returns that value here
    return {"approved": approved}

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

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

恢复中断

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

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

python

from langgraph.types import Command

# Initial run - hits the interrupt and pauses
# thread_id is the persistent pointer (stores a stable ID in production)
config = {"configurable": {"thread_id": "thread-1"}}
stream = graph.stream_events({"input": "data"}, config=config, version="v3")

# Drain the stream to drive the run; stream.output awaits the final state.
final = stream.output

# stream.interrupted is True when the run paused for human input, and
# stream.interrupts contains the payloads passed to interrupt().
if stream.interrupted:
    print(stream.interrupts)
    # > (Interrupt(value='Do you approve this action?'),)

# Resume with the human's response
# The resume payload becomes the return value of interrupt() inside the node
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")
final = resumed.output

View example trace

打开本示例的一个公开 LangSmith 运行记录。

NOTE

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

关于恢复的关键点：

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

WARNING

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

常见模式

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

审批工作流：在执行关键操作（API 调用、数据库变更、金融交易）之前暂停 *

处理多个中断：在单次调用中恢复多个中断时，将中断 ID 与 resume 值配对 *

审查和编辑：让人类在继续之前审查和修改 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")

    # Stream LLM message chunks (including any in subgraphs) as they arrive.
    for message in stream.messages:
        for token in message.text:
            display_streaming_content(token)

    # After the run finishes (or pauses), check for interrupts and resume.
    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)

View example trace

打开本示例的一个公开 LangSmith 运行记录。

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

处理多个中断

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

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

# Step 1: stream events to drive the run; both parallel nodes hit interrupt() and pause
stream = graph.stream_events({"vals": []}, config, version="v3")
_ = stream.output  # drive the stream to completion
# stream.interrupts contains the pending Interrupt payloads
print(stream.interrupts)
# > (Interrupt(value='question_a', id='...'), Interrupt(value='question_b', id='...'))

# Step 2: resume all pending interrupts at once
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']}

View example trace

打开本示例的一个公开 LangSmith 运行记录。

批准或拒绝

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

python

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

def approval_node(state: State) -> Command[Literal["proceed", "cancel"]]:
    # Pause execution; payload shows up on stream.interrupts (with stream_events) or result["__interrupt__"] (with invoke)
    is_approved = interrupt({
        "question": "Do you want to proceed with this action?",
        "details": state["action_details"]
    })

    # Route based on the response
    if is_approved:
        return Command(goto="proceed")  # Runs after the resume payload is provided
    else:
        return Command(goto="cancel")

当你恢复图时，传入 True 表示批准，传入 False 表示拒绝：

python

# To approve
graph.stream_events(Command(resume=True), config=config, version="v3").output

# To reject
graph.stream_events(Command(resume=False), config=config, version="v3").output

完整示例

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"]]:
    # Expose details so the caller can render them in a UI
    decision = interrupt(
        {
            "question": "Approve this action?",
            "details": state["action_details"],
        }
    )

    # Route to the appropriate node after resume
    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)

# Use a more durable checkpointer in production
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  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'question': ..., 'details': ...}),)

# Resume with the decision; True routes to proceed, False to cancel
resumed = graph.stream_events(Command(resume=True), config=config, version="v3")
print(resumed.output["status"])

查看示例追踪

打开此示例的公开 LangSmith 运行记录。

审查和编辑状态

有时你想让人类在继续之前审查和编辑部分图状态。这对于纠正 LLM、补充缺失信息或进行调整很有用。

python

from langgraph.types import interrupt

def review_node(state: State):
    # Pause and show the current content for review (payload surfaces on stream.interrupts)
    edited_content = interrupt({
        "instruction": "Review and edit this content",
        "content": state["generated_text"]
    })

    # Update the state with the edited version
    return {"generated_text": edited_content}

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

python

graph.stream_events(
    Command(resume="The edited and improved text"),  # Value becomes the return from interrupt()
    config=config,
    version="v3",
).output

完整示例

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):
    # Ask a reviewer to edit the generated content
    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  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'instruction': ..., 'content': ...}),)

# Resume with the edited text from the reviewer
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"

查看示例追踪

打开此示例的公开 LangSmith 运行记录。

工具中的中断

你也可以将中断直接放在工具函数内部。这使工具在被调用时自行暂停以请求批准，并允许在执行工具调用之前进行人工审查和编辑。

首先，定义一个使用 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."""

    # Pause before sending; payload surfaces on stream.interrupts when using event streaming
    response = interrupt({
        "action": "send_email",
        "to": to,
        "subject": subject,
        "body": body,
        "message": "Approve sending this email?"
    })

    if response.get("action") == "approve":
        # Resume value can override inputs before executing
        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"

当你希望批准逻辑与工具本身放在一起，使其可在图的各个部分复用时，这种方法很有用。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."""

    # Pause before sending; payload surfaces on stream.interrupts when using event streaming
    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)

        # Actually send the email (your implementation here)
        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 may decide to call the tool; interrupt pauses before sending
    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])  # Routes to "tools" or END
builder.add_edge("tool_node", "agent")  # Loop back after tools

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  # drive the stream to completion
print(initial.interrupts)  # -> (Interrupt(value={'action': 'send_email', ...}),)

# Resume with approval and optionally edited arguments
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

验证人类输入

有时你需要验证来自人类的输入，并在值无效时重新提示。推荐的做法是每次节点调用只调用一次 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)  # called exactly once per invocation
    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() 调用，然后退出。当答案无效时，条件边会循环回来，下一次中断会使用更新后的问题重新提示。每次恢复中没有任何代码会运行超过一次。

完整示例

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)  # called exactly once per node invocation
    print(f"I got {answer}")  # runs exactly once per resume
    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):
    # Loop back to collect_age until we have a valid 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  # drive the stream to completion
print(first.interrupts)  # -> (Interrupt(value='What is your age?', ...),)

# Provide invalid data; the node re-prompts via the conditional edge
retry = graph.stream_events(Command(resume="thirty"), config=config, version="v3")
_ = retry.output
print(retry.interrupts)  # -> (Interrupt(value="'thirty' is not a valid age...", ...),)

# Provide valid data; route() returns END and the graph finishes
final = graph.stream_events(Command(resume=30), config=config, version="v3")
print(final.output["age"])  # -> 30

中断规则

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

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

不要将 `interrupt` 调用包裹在 try/except 中

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

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

分离逻辑显式异常处理

python

def node_a(state: State):
    # ✅ Good: interrupting first, then handling
    # error conditions separately
    interrupt("What's your name?")
    try:
        fetch_data()  # This can fail
    except Exception as e:
        print(e)
    return state

python

def node_a(state: State):
    # ✅ Good: catching specific exception types
    # will not catch the interrupt exception
    try:
        name = interrupt("What's your name?")
        fetch_data()  # This can fail
    except NetworkException as e:
        print(e)
    return state

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

python

def node_a(state: State):
    # ❌ Bad: wrapping interrupt in bare try/except
    # will catch the interrupt exception
    try:
        interrupt("What's your name?")
    except Exception as e:
        print(e)
    return state

不要在节点内重新排序 `interrupt` 调用

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

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

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

python

def node_a(state: State):
    # ✅ Good: interrupt calls happen in the same order every time
    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
    }

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

跳过中断循环中断

python

def node_a(state: State):
    # ❌ Bad: conditionally skipping interrupts changes the order
    name = interrupt("What's your name?")

    # On first run, this might skip the interrupt
    # On resume, it might not skip it - causing index mismatch
    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):
    # ❌ Bad: looping based on non-deterministic data
    # The number of interrupts changes between executions
    results = []
    for item in state.get("dynamic_list", []):  # List might change between runs
        result = interrupt(f"Approve {item}?")
        results.append(result)

    return {"results": results}

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

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

✅ 向 interrupt 传递简单的、JSON 可序列化的类型
✅ 传递具有简单值的字典/对象

简单值结构化数据

python

def node_a(state: State):
    # ✅ Good: passing simple types that are serializable
    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):
    # ✅ Good: passing dictionaries with simple values
    response = interrupt({
        "question": "Enter user details",
        "fields": ["name", "email", "age"],
        "current_values": state.get("user", {})
    })

    return {"user": response}

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

函数类实例

python

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

def node_a(state: State):
    # ❌ Bad: passing a function to interrupt
    # The function cannot be serialized
    response = interrupt({
        "question": "What's your name?",
        "validator": validate_input  # This will fail
    })
    return {"name": response}

python

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

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

    # ❌ Bad: passing a class instance to interrupt
    # The instance cannot be serialized
    response = interrupt({
        "question": "Enter data to process",
        "processor": processor  # This will fail
    })
    return {"result": response}

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

因为中断的工作方式是重新运行调用它们的节点，所以在 interrupt 之前调用的副作用（理想情况下）应该是幂等的。作为背景，幂等性意味着同一操作可以应用多次，而不会改变初始执行之后的结果。

举例来说，你可能在某个节点内有一个更新记录的 API 调用。如果在该调用完成之后才调用 interrupt，那么当节点恢复时，该 API 调用会被多次重新运行，可能会覆盖初始更新或创建重复记录。

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

幂等操作中断后的副作用拆分到不同的节点

python

def node_a(state: State):
    # ✅ Good: using upsert operation which is idempotent
    # Running this multiple times will have the same result
    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):
    # ✅ Good: placing side effect after the interrupt
    # This ensures it only runs once after approval is received
    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):
    # ✅ Good: only handling the interrupt in this node
    approved = interrupt("Approve this change?")

    return {"approved": approved}

def notification_node(state: State):
    # ✅ Good: side effect happens in a separate node
    # This runs after approval, so it only executes once
    if (state.approved):
        send_notification(
            user_id=state["user_id"],
            status="approved"
        )

    return state

🔴 在 interrupt 之前，不要执行非幂等操作
🔴 在创建新记录之前，先检查它们是否已存在

创建记录追加到列表

python

def node_a(state: State):
    # ❌ Bad: creating a new record before interrupt
    # This will create duplicate records on each resume
    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):
    # ❌ Bad: appending to a list before interrupt
    # This will add duplicate entries on each resume
    db.append_to_history(state["user_id"], "approval_requested")

    approved = interrupt("Approve this change?")

    return {"approved": approved}

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

在节点内调用子图时，父图将从调用子图并触发 interrupt 的节点开头恢复执行。同样，子图也将从调用 interrupt 的节点开头恢复执行。

python

def node_in_parent_graph(state: State):
    some_code()  # <-- This will re-execute when resumed
    # Invoke a subgraph as a function.
    # The subgraph contains an `interrupt` call.
    subgraph_result = subgraph.invoke(some_input)
    # ...

def node_in_subgraph(state: State):
    some_other_code()  # <-- This will also re-execute when resumed
    result = interrupt("What's your name?")
    # ...

使用中断进行调试

要调试和测试图，你可以使用静态中断作为断点，一次一个节点地单步执行图的执行过程。静态中断在节点执行之前或之后定义好的位置触发。你可以在编译图时通过指定 interrupt_before 和 interrupt_after 来设置这些断点。

NOTE

静态中断不推荐用于人在回路的工作流。请改用 interrupt 函数。

python

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

# Pass a thread ID to the graph
config = {
    "configurable": {
        "thread_id": "some_thread"
    }
}

# Run the graph until the breakpoint
graph.invoke(inputs, config=config)  

# Resume the graph
graph.invoke(None, config=config)

断点在 compile 时设置。
interrupt_before 指定在执行节点之前应暂停执行的那些节点。
interrupt_after 指定在执行节点之后应暂停执行的那些节点。
要启用断点，需要提供一个检查点器。
图会一直运行，直到命中第一个断点。
通过为输入传入 None 来恢复图。这会让图一直运行，直到命中下一个断点。

TIP

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

使用 LangSmith Studio

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

数据集

创建数据集

运行评估

评估技术

定义评估目标

打分方法

实验配置

多模态评估

分析实验结果

微调

教程

常见数据类型

评估器类型

UI

SDK

框架与集成

改进评估器

反馈

追踪设置

集成

手动埋点

配置与故障排除

项目与环境设置

高级追踪技术

数据与隐私

故障排除指南

查看与管理追踪

批量导出追踪数据

轨迹视图

数据类型参考

监控与告警

在线评估器

自动化

开发你的应用

设置依赖

持久化

能力

助手

运行

并发消息处理

如何构建

鉴权与访问控制

自定义鉴权教程

服务器定制

静态加密

替换内置后端

扩展 HTTP 服务器

请求头与日志

部署指南

参考

配置

参考

提示词

创建与管理提示词

连接模型

上下文中枢

教程

全栈 Web 应用

Agent Server API

控制平面 API

计费与用量

参考

按云服务商起步

用 Terraform 部署

AWS

GCP

Azure

安装实例

配置实例

连接外部服务

平台鉴权与访问控制

管理实例

自托管可观测性

混合部署

脚本

参考

组织