编译时的断点存在 Pregel 实例上（main.py:721-815）：

# main.py:721
    interrupt_after_nodes: All | Sequence[str]
    interrupt_before_nodes: All | Sequence[str]
# main.py:768（__init__ 参数，默认空元组）
        interrupt_after_nodes: All | Sequence[str] = (),
        interrupt_before_nodes: All | Sequence[str] = (),
# main.py:814
        self.interrupt_after_nodes = interrupt_after_nodes
        self.interrupt_before_nodes = interrupt_before_nodes

运行时（stream/invoke）可以传参覆盖，逻辑在 main.py:2549-2570：

# main.py:2549 (stream 内部的 defaults 解析)
        interrupt_before: All | Sequence[str] | None,
        interrupt_after: All | Sequence[str] | None,
        ...
# main.py:2569
        interrupt_before = interrupt_before or self.interrupt_before_nodes
        interrupt_after = interrupt_after or self.interrupt_after_nodes

断点在主循环 PregelLoop 里生效。before 在每个超步开始、任务已备好但还没执行时检查，见 _loop.py:666-671：

# _loop.py:666  （在 tick() 里，任务准备完之后）
        # before execution, check if we should interrupt
        if self.interrupt_before and should_interrupt(
            self.checkpoint, self.interrupt_before, self.tasks.values()
        ):
            self.status = "interrupt_before"
            raise GraphInterrupt()

after 在超步结束、写已 apply、checkpoint 已存之后检查，见 _loop.py:719-724（在 after_tick() 里）：

# _loop.py:717  （after_tick 尾部）
        # save checkpoint
        self._put_checkpoint({"source": "loop"})
        # after execution, check if we should interrupt
        if self.interrupt_after and should_interrupt(
            self.checkpoint, self.interrupt_after, self.tasks.values()
        ):
            self.status = "interrupt_after"
            raise GraphInterrupt()

判定核心 _algo.py:155-185：

# _algo.py:155
def should_interrupt(checkpoint, interrupt_nodes, tasks) -> list[PregelExecutableTask]:
    """Check if the graph should be interrupted based on current state."""
    version_type = type(next(iter(checkpoint["channel_versions"].values()), None))
    null_version = version_type()
    seen = checkpoint["versions_seen"].get(INTERRUPT, {})    # ① 上次中断时"看到"的版本
    # interrupt if any channel has been updated since last interrupt
    any_updates_since_prev_interrupt = any(
        version > seen.get(chan, null_version)                # ② 有通道版本比上次新?
        for chan, version in checkpoint["channel_versions"].items()
    )
    # and any triggered node is in interrupt_nodes list
    return (
        [
            task for task in tasks
            if (
                (not task.config or TAG_HIDDEN not in task.config.get("tags", EMPTY_SEQ))
                if interrupt_nodes == "*"                     # ③ "*"=所有(除隐藏)
                else task.name in interrupt_nodes             # ④ 或名字在名单里
            )
        ]
        if any_updates_since_prev_interrupt                   # ⑤ 且"确实有更新"才停
        else []
    )

L05 那个"自上次中断后有更新才停"的闸门，靠的是 versions_seen[INTERRUPT]。它是一本"观察者账本"——记录"INTERRUPT 这个观察者上次看到各通道到哪个版本了"。

静态断点不带值，所以恢复更简单——通常直接再跑一次（输入传 None），或用 Command(resume=None)：

app = builder.compile(checkpointer=saver, interrupt_before=["risky"])
cfg = {"configurable": {"thread_id": "t1"}}
app.invoke({"x": 0}, cfg)          # 跑到 risky 前停下
print(app.get_state(cfg).next)     # ('risky',) —— 下一步是 risky
# ...人工检查，甚至改状态...
app.update_state(cfg, {"x": 99})   # 可选：介入修改（D44 讲）
app.invoke(None, cfg)              # 传 None 恢复 → 从 risky 继续跑到底

# 1. 读两个触发点与判定
sed -n '666,671p' libs/langgraph/langgraph/pregel/_loop.py
sed -n '719,724p' libs/langgraph/langgraph/pregel/_loop.py
sed -n '155,185p' libs/langgraph/langgraph/pregel/_algo.py
# 2. 亲手设静态断点
python - <<'PY'
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from typing_extensions import TypedDict
class S(TypedDict):
    x: int
g=StateGraph(S)
g.add_node("a", lambda s:{"x":s["x"]+1})
g.add_node("b", lambda s:{"x":s["x"]*10})
g.add_edge(START,"a"); g.add_edge("a","b"); g.add_edge("b",END)
app=g.compile(checkpointer=InMemorySaver(), interrupt_before=["b"])
cfg={"configurable":{"thread_id":"t"}}
app.invoke({"x":0}, cfg)
print("停在 b 前, next =", app.get_state(cfg).next)   # ('b',)
print("此刻 x =", app.get_state(cfg).values["x"])       # 1（a 跑完，b 未跑）
app.invoke(None, cfg)                                    # 恢复
print("跑完 x =", app.get_state(cfg).values["x"])       # 10
PY