动态断点：interrupt()
让图在运行时自行决定是否暂停

1 静态断点的局限性：为什么还需要动态断点

在 Module 3 的前几节课中，我们学习了静态断点（Static Breakpoints）：在编译图时，通过 compile(interrupt_before=["node_name"]) 或 interrupt_after= 来指定哪些节点会触发暂停。

静态断点有一个根本性的限制：它是"编译期"决定的，与运行时的数据完全无关。

这就是动态断点（Dynamic Breakpoints）存在的原因：让节点函数内部的代码，根据运行时数据，自主决定是否触发中断，并且能向外界传达中断的具体原因。

2 动态断点的核心：interrupt() 函数

LangGraph 通过 langgraph.types.interrupt 函数实现动态断点。当节点函数在执行过程中调用 interrupt(...) 时，LangGraph 运行时会：

• 1

  触发受控暂停，而不是崩溃

  interrupt() 会在内部触发 LangGraph 可识别的受控中断，运行时会接管它，不会让程序以未捕获异常的方式崩溃。

• 2

  保存当前检查点状态

  图的完整状态（包括已执行节点的输出）被持久化到 MemorySaver 或其他 checkpointer 中。

• 3

  将中断信息写入任务记录

  传递给 interrupt() 的 payload（建议使用 JSON 可序列化的字符串、字典等值）会暴露给调用方，并记录在 state.tasks 的 interrupts 字段中。

• 4

  图暂停执行，等待外部操作

  graph.stream() 或 graph.invoke() 调用返回，流程停在触发中断的节点上；后续可以通过人工输入、状态修改或恢复命令继续。

导入方式

from langgraph.types import interrupt

# 调用 interrupt() 会暂停当前图执行
# payload 会作为中断原因暴露给外部调用方
interrupt("需要人工审核：输入内容超出长度限制")
interrupt({"reason": "high_risk", "score": 0.95})

3 完整示例：构建带动态断点的图

Notebook 中的示例构建了一个三步骤的简单图：step_1 → step_2 → step_3，其中 step_2 会在输入字符串超过 5 个字符时调用 interrupt()。

第一步：定义 State 和节点函数

from typing_extensions import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import START, END, StateGraph
from langgraph.types import interrupt   # ← 导入动态中断函数

# State 只有一个字段：input 字符串
class State(TypedDict):
    input: str

# ── step_1：普通节点，直接返回 state ──
def step_1(state: State) -> State:
    print("---Step 1---")
    return state

# ── step_2：包含动态断点逻辑的关键节点 ──
def step_2(state: State) -> State:
    # 运行时条件判断：如果输入超过 5 个字符，则中断
    if len(state['input']) > 5:
        # 调用 interrupt()，并附带说明原因的 payload
        interrupt(
            f"Received input that is longer than 5 characters: {state['input']}"
        )
    # 如果没有触发中断，则正常执行
    print("---Step 2---")
    return state

# ── step_3：普通节点 ──
def step_3(state: State) -> State:
    print("---Step 3---")
    return state

第二步：构建图并编译

# 构建图结构
builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)

builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)

# 准备 MemorySaver（动态断点依赖 checkpointer 保存状态）
memory = MemorySaver()

# 编译时 ★ 不需要 ★ 指定任何 interrupt_before / interrupt_after！
# 动态断点完全由节点内部的 interrupt() 调用控制
graph = builder.compile(checkpointer=memory)

图的可视化结构

从图结构本身看不出任何"中断"的痕迹——动态断点在运行时才会"显现"。这与静态断点（编译期即固定了中断节点）的可见性是不同的。

4 触发中断：检查状态与中断原因

运行图，触发 interrupt()

# 配置：使用 thread_id 追踪这次执行的状态
initial_input = {"input": "hello world"}  # 11 个字符，超过 5
thread_config = {"configurable": {"thread_id": "1"}}

# 使用 stream 运行图
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

观察输出：

• 第一行 {'input': 'hello world'}：图启动时的初始 State 事件
• ---Step 1---：step_1 正常执行并打印
• 第二行 {'input': 'hello world'}：step_1 执行完后的 State 事件
• 第三个事件包含 __interrupt__：step_2 调用了 interrupt()，payload 被返回给调用方
• 之后没有 step_3 的输出：图暂停，stream() 返回

检查 state.next：下一个等待执行的节点

state = graph.get_state(thread_config)
print(state.next)

state.next 显示 ('step_2',)，表示图停在了 step_2 "即将执行"的位置——step_2 触发了中断，但本次执行被中止，所以它仍然是"下一个待执行节点"。

检查 state.tasks：读取中断原因

print(state.tasks)

解读 state.tasks 的关键字段：

5 陷阱：简单恢复会重新触发中断

初学者容易犯的错误：以为不修改状态、也不提供 Command(resume=...)，直接 graph.stream(None, thread_config) 就能继续。让我们看看会发生什么：

# 尝试直接恢复（不修改 state）
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print(state.next)  # 仍然是 step_2！

这与静态断点的行为有所不同：

• 静态断点：interrupt_before=["step_2"] 是在 step_2 执行之前中断，恢复后 step_2 才开始执行，通过与否取决于 step_2 的代码逻辑。
• 动态断点：中断条件在 step_2 执行期间触发，直接用 None 恢复会重新跑 step_2，如果状态没变，中断条件仍成立，就会再次中断。

6 正确流程：修改状态后再恢复

正确的处理方式是：先用 graph.update_state() 修改状态，让中断条件不再满足，然后再恢复执行。

第一步：修改 state，消除中断条件

# 将 input 改为短字符串（5 个字符以内），让 step_2 的条件不再触发
graph.update_state(
    thread_config,
    {"input": "hi"},  # "hi" 只有 2 个字符，len("hi") = 2 ≤ 5
)

update_state 返回新的 checkpoint 配置，说明状态已成功写入。

第二步：恢复执行

# 传入 None 作为 input，表示"从上次中断处继续，不添加新输入"
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

这次执行成功了！执行流程分析：

• {'input': 'hi'}：恢复时的当前 State（已被修改）
• ---Step 2---：step_2 正常执行（len("hi") = 2 ≤ 5，不触发中断）
• {'input': 'hi'}：step_2 执行后的 State 事件
• ---Step 3---：step_3 正常执行
• 最后的 State 事件：图执行完毕

完整操作流程总结

7 通过 LangGraph API 使用动态断点

在生产环境中，图通常部署在 LangGraph 服务端（通过 langgraph dev 启动本地开发服务器），客户端通过 LangGraph SDK 与其交互。动态断点在 API 场景下同样适用，操作逻辑与本地完全一致。

启动本地开发服务器

在模块的 /studio 目录下运行：

langgraph dev

服务启动后会显示：

通过 SDK 连接并运行图（触发动态断点）

from langgraph_sdk import get_client

# 连接到本地开发服务器
URL = "http://127.0.0.1:2024"
client = get_client(url=URL)

# 创建新线程（相当于一次独立的图执行会话）
thread = await client.threads.create()
input_dict = {"input": "hello world"}  # 超过 5 字符，会触发 interrupt()

async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="dynamic_breakpoints",  # 图的名称（在 langgraph.json 中定义）
    input=input_dict,
    stream_mode="values",
):
    print(f"Receiving new event of type: {chunk.event}...")
    print(chunk.data)

流事件结束——图在 step_2 的 interrupt() 处暂停。

检查远程线程状态

current_state = await client.threads.get_state(thread['thread_id'])
print(current_state['next'])

远程修改状态，然后恢复

# 更新 state：将 input 改为短字符串
await client.threads.update_state(thread['thread_id'], {"input": "hi!"})

# 恢复执行：传入 input=None 表示从中断处继续
async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="dynamic_breakpoints",
    input=None,  # ← 关键：None 表示恢复，而非新的输入
    stream_mode="values",
):
    print(f"Receiving new event of type: {chunk.event}...")
    print(chunk.data)

图成功恢复并执行完毕（step_2 和 step_3 均正常执行）。

8 静态断点 vs 动态断点：完整对比

两种断点机制各有适用场景，理解它们的差异有助于在实际项目中做出正确选择。

如何选择？

真实工程应用场景举例

interrupt({
    "reason": "high_risk_content",
    "risk_score": 0.94,
    "flagged_text": state["generated_content"][:200],
    "suggested_action": "请人工审核后选择批准或拒绝"
})