第9章 中断与人机协作

9.1 引言

在 AI Agent 的实际应用中，纯自动化的执行流程往往不够。当 Agent 需要进行高风险操作（如转账、删除数据）、当决策需要人类专业判断（如医疗诊断确认）、或者当信息不足需要用户补充时，系统必须能够优雅地暂停执行，等待人类介入，然后从暂停点恢复。

这就是 LangGraph 中断（Interrupt）机制要解决的问题。与传统的回调或轮询方案不同，LangGraph 的中断是一种持久化的暂停——执行状态被完整保存到 Checkpoint 中，使用持久化的 checkpointer 时，进程可以终止、重启，甚至迁移到另一台机器上，只要提供正确的 thread_id，就能从中断点恢复执行。

本章将从源码层面深入分析中断机制的完整实现：从 interrupt() 函数如何抛出异常、到 GraphInterrupt 如何被 Pregel 循环捕获、再到恢复时如何通过 Command(resume=...) 将值传回节点。我们还将探讨 interrupt_before/interrupt_after 的配置式中断，以及多中断场景下的匹配策略。

为了充分理解 LangGraph 中断机制的精妙设计，有必要先反思传统的暂停/恢复方案。最简单的做法是在每个可能中断的位置设置回调函数，让用户代码手动管理暂停逻辑。这种方法的问题在于：它将中断的责任分散到了业务代码中，增加了认知负担；而且回调模式难以持久化——一旦进程终止，回调的上下文就丢失了。另一种方案是使用协程或 async/await，但这要求所有节点函数都是异步的，而且 Python 的协程状态无法可靠地序列化。LangGraph 选择了第三种路径：基于异常的暂停加重新执行的恢复。这种方式让节点函数保持简单（可以是普通同步函数），同时通过 Checkpoint 实现真正的持久化暂停。代价是恢复时需要从头重新执行整个节点——通过 scratchpad 的索引追踪，已解决的中断会立即返回缓存值、不再暂停，但中断之前的代码（包括其副作用）会再跑一遍（见 §9.10.2）。

9.2 Interrupt 数据类型

9.2.1 Interrupt 类定义

Interrupt 是一个数据类（未设 frozen，字段可以重新赋值），用于封装中断信息：

# langgraph/types.py（节选，省略 docstring 与已弃用的 interrupt_id 属性）

@final
@dataclass(init=False, slots=True)
class Interrupt:
    value: Any
    """The value associated with the interrupt."""
    id: str
    """The ID of the interrupt. Can be used to resume the interrupt directly."""

    def __init__(self, value: Any, id: str = _DEFAULT_INTERRUPT_ID, **deprecated_kwargs):
        self.value = value
        if (
            (ns := deprecated_kwargs.get("ns", MISSING)) is not MISSING
            and (id == _DEFAULT_INTERRUPT_ID)
            and (isinstance(ns, Sequence))
        ):
            self.id = xxh3_128_hexdigest("|".join(ns).encode())
        else:
            self.id = id

    @classmethod
    def from_ns(cls, value: Any, ns: str) -> Interrupt:
        return cls(value=value, id=xxh3_128_hexdigest(ns.encode()))

两个核心属性：

value：中断携带的值，可以是任意类型。它被传递给客户端，用于展示中断原因或请求用户输入。例如 "请确认是否执行转账操作？" 或 {"question": "请选择以下选项", "options": ["A", "B", "C"]}。

id：中断的唯一标识符。注意它不是随机生成的 UUID，而是通过 xxh3_128_hexdigest 对检查点命名空间进行哈希计算得出的确定性 ID。命名空间形如 节点名:任务ID，而任务 ID 由检查点 ID、步数、节点名等确定性地算出，所以同一个任务被中断、恢复、重新执行时，ID 保持不变。反过来说，这个 ID 是任务级的：同一节点里的多次 interrupt() 共用一个 ID（pin 版实跑：§9.7.3 的三次中断 id 完全相同），不同 thread、不同检查点上的中断 ID 则各不相同。

# 中断 ID 的确定性生成
Interrupt.from_ns(
    value="请确认操作",
    ns=conf[CONFIG_KEY_CHECKPOINT_NS],  # 例如 "agent:task-id-123"
)
# id = xxh3_128_hexdigest(b"agent:task-id-123")

这个设计选择服务于多个任务同时中断时的精确恢复：客户端可以通过中断 ID 指定要恢复哪个任务上的中断。由于 xxh3_128_hexdigest 是一个非密码学的高速哈希函数，ID 的计算几乎没有性能开销。128 位的输出空间足够大，碰撞概率可以忽略不计。

@final 装饰器和 slots=True 的使用也值得注意。这里的 final 来自 typing，只对类型检查器声明"不应被子类化"，运行时并不阻止继承（pin 版实测 class X(Interrupt): pass 可以正常定义）。slots=True 则通过使用 __slots__ 而非 __dict__ 来存储属性，减少了内存开销并略微提升了属性访问速度。

9.2.2 GraphInterrupt 异常

GraphInterrupt 是中断机制的传播载体：

# langgraph/errors.py

class GraphBubbleUp(Exception):  # 源码无 docstring：需要向上冒泡的异常的基类
    pass

class GraphInterrupt(GraphBubbleUp):
    """Raised when a subgraph is interrupted, suppressed by the root graph.
    Never raised directly, or surfaced to the user."""

    def __init__(self, interrupts: Sequence[Interrupt] = ()) -> None:
        super().__init__(interrupts)

GraphInterrupt 继承自 GraphBubbleUp，这是 LangGraph 中所有需要跨层传播的异常的基类。关键设计：GraphInterrupt 在根图中被抑制，不会抛到调用 invoke/stream 的代码里。它只是一种内部的控制流机制。

graph TB
    subgraph "异常层级"
        BE[Exception]
        BE --> GBU[GraphBubbleUp]
        GBU --> GI[GraphInterrupt]
        GBU --> PC[ParentCommand]
        GI --> NI["NodeInterrupt (deprecated)"]
    end
    subgraph "传播行为"
        N[Node] -->|raise GraphInterrupt| RN["runner.commit 写入 INTERRUPT<br/>本步结束后汇总重新抛出"]
        RN --> PL[PregelLoop]
        PL -->|is_nested?| SUB{子图?}
        SUB -->|Yes| PARENT[父图捕获并记录]
        SUB -->|No| SUPPRESS[根图抑制<br/>invoke/stream 照常返回]
    end

9.3 interrupt() 函数深度解析

9.3.1 核心实现

interrupt() 函数是用户在节点中触发中断的推荐入口（直接抛出 NodeInterrupt 是已弃用的旧路径）：

# langgraph/types.py（节选，省略 docstring）

def interrupt(value: Any) -> Any:
    from langgraph._internal._constants import (
        CONFIG_KEY_CHECKPOINT_NS, CONFIG_KEY_SCRATCHPAD,
        CONFIG_KEY_SEND, RESUME,
    )
    from langgraph.config import get_config
    from langgraph.errors import GraphInterrupt

    conf = get_config()["configurable"]
    # 1. 追踪中断索引
    scratchpad = conf[CONFIG_KEY_SCRATCHPAD]
    idx = scratchpad.interrupt_counter()

    # 2. 查找之前保存的恢复值
    if scratchpad.resume:
        if idx < len(scratchpad.resume):
            conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)])
            return scratchpad.resume[idx]

    # 3. 查找当前的恢复值（来自 Command(resume=...)）
    v = scratchpad.get_null_resume(True)
    if v is not None:
        assert len(scratchpad.resume) == idx
        scratchpad.resume.append(v)
        conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)])
        return v

    # 4. 没有恢复值，抛出中断
    raise GraphInterrupt((
        Interrupt.from_ns(
            value=value,
            ns=conf[CONFIG_KEY_CHECKPOINT_NS],
        ),
    ))

这个函数的精妙之处在于它的双重身份：第一次调用时它是一个"暂停器"（抛出异常），恢复后重新执行时它是一个"值提供器"（返回恢复值）。从用户的角度看，interrupt() 就像一个普通的阻塞式输入函数——调用它时程序暂停，收到输入后继续。但在底层，这两次调用分属两次独立的图调用，甚至可能发生在不同的进程中，中间可能经历了数分钟甚至数天的等待。Checkpoint 机制弥合了这种时间断裂，使得节点函数的编写者不需要关心持久化的细节。

理解这个函数的关键在于认识到它的四个步骤：步骤 1 通过 scratchpad 的 interrupt_counter 追踪当前是第几个中断调用；第一个分支（步骤 2）检查是否有之前保存的恢复值列表，如果当前索引在范围内则直接返回；第二个分支（步骤 3）尝试获取新提供的恢复值（Command(resume=...) 写入的那一个，取用后即从本次调用的 pending writes 中移除，只能被消费一次）；步骤 4 在没有任何恢复值时抛出异常。这种分层的查找策略使得多中断场景可以正确工作——前面的中断返回已缓存的值，最新的中断返回新提供的值，再后面的中断抛出异常等待用户输入。

9.3.2 PregelScratchpad 的角色

PregelScratchpad 是每个任务执行期间的临时工作区：

# langgraph/_internal/_scratchpad.py（行尾注释为本专栏所加）
@dataclasses.dataclass(**_DC_KWARGS)
class PregelScratchpad:
    step: int
    stop: int
    call_counter: Callable[[], int]
    interrupt_counter: Callable[[], int]  # 追踪当前任务的中断索引
    get_null_resume: Callable[[bool], Any]
    resume: list[Any]                     # 之前保存的恢复值列表
    subgraph_counter: Callable[[], int]

interrupt_counter 是一个 LazyAtomicCounter 实例（pregel/_algo.py:1224，首次调用时才创建 itertools.count(0).__next__），每次调用返回当前索引并自增，第一次返回 0。每个任务在准备阶段由 _scratchpad()（pregel/_algo.py:1078）创建自己的 scratchpad，计数器因此是任务独享的——在并行执行的场景中，多个任务可能同时包含 interrupt() 调用，它们的索引互不干扰。resume 列表同样按任务取：_scratchpad() 从 pending writes 里找出本任务 ID 的 RESUME 写入，按 ID 恢复时还会把 resume map 里命中本任务命名空间哈希的值追加进来。这使得同一个节点中的多个 interrupt() 调用可以按顺序匹配恢复值：

# 节点中的多个中断
def review_node(state):
    # 第一次 interrupt: idx=0
    name = interrupt("请输入您的姓名")
    # 第二次 interrupt: idx=1
    age = interrupt("请输入您的年龄")
    return {"name": name, "age": age}

9.3.3 执行流程的四个阶段

sequenceDiagram
    participant U as 用户
    participant G as Graph
    participant N as 节点
    participant S as Scratchpad
    participant CP as Checkpoint
    Note over U,CP: 阶段1: 首次执行（中断）
    U->>G: invoke(input, config)
    G->>N: 执行节点
    N->>S: interrupt("请确认")
    S-->>S: idx=0, resume=[]
    S->>N: raise GraphInterrupt
    N-->>G: 异常冒泡
    G->>CP: put_writes(INTERRUPT, [Interrupt(...)])
    G-->>U: {..., "__interrupt__": [Interrupt(value="请确认", id=...)]}
    Note over U,CP: 阶段2: 恢复执行
    U->>G: invoke(Command(resume="确认"), config)
    G->>CP: 加载 checkpoint + pending_writes
    G->>N: 重新执行节点
    N->>S: interrupt("请确认")
    S-->>S: idx=0, resume=[], 取出 null resume "确认"
    S-->>N: return "确认"
    N-->>G: 返回更新
    G->>CP: put(new_checkpoint)
    G-->>U: 最终结果

让我们详细追踪每个阶段：

阶段 1 - 首次执行：节点调用 interrupt(value)，scratchpad.resume 为空列表，idx 为 0。没有找到恢复值，抛出 GraphInterrupt。

阶段 2 - 异常处理：GraphInterrupt 先被 runner 捕获：commit() 把中断信息作为 INTERRUPT 类型的 pending write 保存到检查点中，等本步所有任务结束后，runner.tick 把各任务的中断汇总成一个 GraphInterrupt 重新抛出。根图在 PregelLoop 退出时由 _suppress_interrupt 抑制异常，中断信息经输出返回给用户。

阶段 3 - 恢复输入：用户通过 Command(resume=value) 提供恢复值。_first() 把它作为 NULL_TASK_ID 名下的 RESUME 写入存进 pending writes，准备任务时 _scratchpad() 让 get_null_resume 能取到它。

阶段 4 - 重新执行：节点从头开始重新执行。再次遇到 interrupt() 调用时，scratchpad.resume 仍为空，于是走步骤 3：get_null_resume(True) 取出恢复值、追加进 resume 并返回，不抛出异常。节点正常完成执行。

9.4 interrupt_before 与 interrupt_after

9.4.1 声明式中断配置

除了在节点内部调用 interrupt() 的命令式中断，LangGraph 还支持在编译时声明式地配置中断：

# 编译时配置
graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["human_review"],    # 执行前中断
    interrupt_after=["tool_call"],        # 执行后中断
)

# 也支持通配符
graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before="*",  # 所有节点执行前中断
)

9.4.2 调度时的中断判断

在 PregelLoop.tick() 方法中，中断判断发生在任务准备之后、实际执行之前（interrupt_before）和执行之后（interrupt_after）：

# langgraph/pregel/_loop.py（简化）

def tick(self) -> bool:
    # 准备下一步任务
    self.tasks = prepare_next_tasks(...)

    # 执行前中断检查
    if self.interrupt_before and should_interrupt(
        self.checkpoint, self.interrupt_before, self.tasks.values()
    ):
        self.status = "interrupt_before"
        raise GraphInterrupt()
    # ...
    return True  # 任务随后由 runner.tick 执行，结束后再调用 after_tick()

def after_tick(self) -> None:
    # 应用写入
    self.updated_channels = apply_writes(...)

    # 保存检查点
    self._put_checkpoint({"source": "loop"})

    # 执行后中断检查
    if self.interrupt_after and should_interrupt(
        self.checkpoint, self.interrupt_after, self.tasks.values()
    ):
        self.status = "interrupt_after"
        raise GraphInterrupt()

should_interrupt（pregel/_algo.py:141-171）的判断其实是两个条件相与：

1. 自上次中断以来有 Channel 被更新过——把 checkpoint["channel_versions"] 逐个和 checkpoint["versions_seen"][INTERRUPT] 比较，只要有一个更大就成立。恢复执行时 _first() 会把 versions_seen[INTERRUPT] 刷成当前全量版本（见第 6 章 §6.4），于是这个条件为假——这正是"恢复后不会在同一个点再次中断"的机制。
2. 本步任务里有节点命中中断配置——配置为 "*" 时匹配所有任务但跳过带 TAG_HIDDEN 标签的（比如 __start__），否则按 task.name 精确匹配。

这个检查在任务准备完成后进行，意味着 interrupt_before 不会阻止任务的创建，只是阻止它们的执行——get_state() 的 next 和 tasks 里能看到这些待执行的任务。恢复时新的循环会在 tick() 里对同一个检查点重新调用 prepare_next_tasks，因为任务 ID 是确定性计算的，得到的还是同一批任务，再照常执行。

9.4.3 声明式 vs 命令式中断的对比

graph TB
    subgraph "声明式中断 (interrupt_before/after)"
        C[编译时配置] --> BEFORE["interrupt_before=['review']"]
        C --> AFTER["interrupt_after=['tool']"]
        BEFORE --> B_CHECK{step 开始前检查}
        AFTER --> A_CHECK{step 完成后检查}
        B_CHECK -->|匹配| B_INT[中断：任务尚未执行]
        A_CHECK -->|匹配| A_INT[中断：任务已完成]
    end
    subgraph "命令式中断 (interrupt())"
        NODE[节点代码] --> CALL["answer = interrupt('确认?')"]
        CALL --> FIRST{首次执行?}
        FIRST -->|Yes| RAISE[抛出 GraphInterrupt]
        FIRST -->|No| RETURN[返回恢复值]
    end

特性	声明式	命令式
定义位置	compile() 参数	节点函数内部
粒度	节点级别	代码行级别
携带数据	无自定义数据	可携带任意值
恢复方式	invoke(None, config)	Command(resume=value)
条件中断	不支持（始终中断匹配的节点）	支持（通过条件判断控制是否调用）
多次中断	每个步骤最多一次	同一节点内可调用多次
恢复数据	不携带恢复数据	恢复值作为返回值传递
适用场景	调试审查和安全审批	交互式数据收集和条件审批

9.5 恢复机制的完整链路

9.5.1 Command(resume=...) 的处理

当用户通过 Command(resume=value) 恢复执行时，PregelLoop._first() 方法负责处理恢复逻辑：

# langgraph/pregel/_loop.py - _first() 方法

if input_is_command:
    if (resume := cast(Command, self.input).resume) is not None:
        if not self.checkpointer:
            raise RuntimeError("Cannot use Command(resume=...) without checkpointer")

        # 判断是否是按 ID 恢复（dict 映射）
        if resume_is_map := (
            isinstance(resume, dict)
            and all(is_xxh3_128_hexdigest(k) for k in resume)
        ):
            self.config[CONF][CONFIG_KEY_RESUME_MAP] = resume
        else:
            # 简单恢复：检查是否有多个待处理中断
            if len(self._pending_interrupts()) > 1:
                raise RuntimeError(
                    "When there are multiple pending interrupts, you must specify the interrupt id when resuming. "
                    "Docs: https://docs.langchain.com/oss/python/langgraph/add-human-in-the-loop#resume-multiple-interrupts-with-one-invocation."
                )

恢复支持两种模式：

简单恢复：Command(resume=value)，将值传递给下一个未处理的中断。只在待处理中断不超过一个时有效（同一节点里先后出现的多个中断是逐个挂起的，每次只有一个待处理，仍可用简单恢复）。

按 ID 恢复：Command(resume={interrupt_id: value})，通过中断 ID 精确指定要恢复哪个任务上的中断。多个任务同时挂起时必须用这种方式。只有字典的所有键都是 32 位十六进制的 xxh3 摘要时才按 ID 解释，否则整个字典被当作一个普通恢复值。

9.5.2 pending_interrupts 的追踪

_pending_interrupts() 方法通过分析 checkpoint_pending_writes 来确定哪些中断尚未被恢复：

# libs/langgraph/langgraph/pregel/_loop.py:590
def _pending_interrupts(self) -> set[str]:
    pending_interrupts: dict[str, str] = {}
    pending_resumes: set[str] = set()

    for task_id, write_type, value in self.checkpoint_pending_writes:
        if write_type == INTERRUPT:
            pending_interrupts[task_id] = value[0].id
        elif write_type == RESUME:
            pending_resumes.add(task_id)

    # 已恢复的中断 ID
    resumed_interrupt_ids = {
        pending_interrupts[task_id]
        for task_id in pending_resumes
        if task_id in pending_interrupts
    }

    # 返回尚未恢复的中断 ID
    return {
        interrupt_id
        for interrupt_id in pending_interrupts.values()
        if interrupt_id not in resumed_interrupt_ids
    }

这个方法的逻辑是：遍历所有 pending writes，找出所有 INTERRUPT 类型的写入和所有 RESUME 类型的写入，然后通过 task_id 关联，排除已经有对应 RESUME 的 INTERRUPT。

9.5.3 RESUME 写入的保存

当 interrupt() 函数成功找到恢复值时，它不仅返回值，还通过 CONFIG_KEY_SEND 将 RESUME 写入发送到 Pregel 循环：

# interrupt() 函数中
if scratchpad.resume:
    if idx < len(scratchpad.resume):
        # 写入 RESUME 标记，表示此中断已处理
        conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)])
        return scratchpad.resume[idx]

这个 RESUME 写入被持久化到检查点中，确保即使在恢复过程中再次中断，之前已处理的中断值也不会丢失。

stateDiagram-v2
    [*] --> Running: invoke(input, config)
    Running --> Interrupted: interrupt() 抛出异常
    Interrupted --> WritingCP: 保存中断到 pending_writes
    WritingCP --> Suspended: checkpoint 已保存
    Suspended --> Resuming: Command(resume=value)
    Resuming --> LoadingCP: 加载 checkpoint
    LoadingCP --> ReExecuting: 重新执行节点
    ReExecuting --> Running: interrupt() 返回恢复值
    ReExecuting --> Interrupted: 遇到新的 interrupt()
    Running --> [*]: 执行完成

9.6 中断的抑制与传播

9.6.1 根图的抑制逻辑

GraphInterrupt 在根图中被抑制，这发生在 PregelLoop._suppress_interrupt 方法中：

# langgraph/pregel/_loop.py（简化）

def _suppress_interrupt(self, exc_type, exc_value, traceback) -> bool | None:
    # 持久化当前状态（exit 模式）
    if self.durability == "exit" and (
        not self.is_nested or exc_value is not None
        or all(NS_END not in part for part in self.checkpoint_ns)
    ):
        self._put_checkpoint(self.checkpoint_metadata)
        self._put_pending_writes()

    # 只有根图才抑制中断
    suppress = isinstance(exc_value, GraphInterrupt) and not self.is_nested

    if suppress:
        # 把 pending writes 应用到通道上，补发最后一个 "values" 事件
        ...
        # GraphInterrupt 不带中断对象时（静态断点），补发 {"__interrupt__": ()}；
        # interrupt() 产生的中断已在 put_writes → output_writes 时发出
        if exc_value is not None and (not exc_value.args or not exc_value.args[0]):
            self._emit("updates", lambda: iter([{INTERRUPT: exc_value.args[0]}]))
        # 保存最终输出
        self.output = read_channels(self.channels, self.output_keys)
        return True
    elif exc_type is None:
        self.output = read_channels(self.channels, self.output_keys)

关键设计：子图中的 GraphInterrupt 不会被抑制，它会向上冒泡到父图。父图负责决定如何处理子图的中断。这种分层处理策略确保了嵌套图中的中断不会被子图默默吞掉——只有最外层的根图有权决定是否将中断暴露给用户。在 exit 持久化模式下，根图总会在退出时保存检查点和 pending writes，子图则在出错或中断时才保存（另有一种情况：检查点命名空间各段都不含 :任务ID，源码注释说明对应 checkpointer=True 的子图），确保即使在最激进的延迟持久化策略下，中断状态也不会丢失。

抑制逻辑中的 not self.is_nested 检查至关重要。is_nested 通过检查配置中是否存在 CONFIG_KEY_TASK_ID 来判断——只有作为父图任务一部分执行的图才是嵌套的。独立执行的根图（即使配置中携带了 checkpoint_ns）不被视为嵌套的，因为 __init__ 中会清理外部传入的命名空间。

9.6.2 中断信息的输出

中断信息最终以 __interrupt__ 键出现在输出中：

# 用户视角
for chunk in graph.stream(input, config):
    print(chunk)
# 输出:
# {'__interrupt__': (Interrupt(value='请确认操作', id='45fda847...'),)}

需要注意的是，stream() 的 updates 块里，中断信息以元组形式包装在 __interrupt__ 键中；静态断点（interrupt_before/interrupt_after）触发时这里是空元组 {'__interrupt__': ()}。invoke() 则从 updates 块收集中断，拼进最终状态返回，此时是列表：{..., '__interrupt__': [Interrupt(...)]}（pin 版实跑）。每个元素是一个 Interrupt 对象，包含 value（中断传递的值）和 id（用于精确恢复的标识符）两个属性。

中断挂起期间，graph.get_state(config) 也能查到它们：StateSnapshot.interrupts 汇总当前检查点上各任务记录的中断，tasks 里每个 PregelTask 的 interrupts 字段则标明中断属于哪个任务。注意按 ID 部分恢复（§9.7.4）之后，已经完成的任务仍会带着原来的中断出现在这两处，要看 PregelTask.result 是否为空才能区分（pin 版实跑）。pin 版实跑 §9.7.3 的表单节点，第一次中断后：

# next: ('form',)
# interrupts: (Interrupt(value='请输入您的姓名', id='f6d9191560fb259bb818a83aab2c83cd'),)
# tasks: (PregelTask(id='2f765606-...', name='form', path=('__pregel_pull', 'form'), error=None,
#         interrupts=(Interrupt(value='请输入您的姓名', id='f6d9191560fb259bb818a83aab2c83cd'),),
#         state=None, result=None),)

在 v2 版本的流式接口中，中断信息也会出现在 ValuesStreamPart 的 interrupts 字段中，提供了更加结构化的访问方式：

# v2 stream（需指定 stream_mode="values"，默认模式下只有 updates 块）
for part in graph.stream(input, config, stream_mode="values", version="v2"):
    if part["type"] == "values":
        if part["interrupts"]:
            print("中断:", part["interrupts"])

9.7 人机协作模式

9.7.1 审批模式（Approval）

最经典的人机协作模式。Agent 在执行高风险操作前请求人类审批：

from langgraph.types import interrupt, Command

def tool_executor(state):
    tool_call = state["pending_tool_call"]

    # 请求人类审批
    approval = interrupt({
        "action": tool_call["name"],
        "args": tool_call["args"],
        "question": "是否允许执行此操作？"
    })

    if approval == "approved":
        result = execute_tool(tool_call)
        return {"result": result}
    else:
        return {"result": "操作已被用户拒绝"}

# 使用
config = {"configurable": {"thread_id": "thread-1"}}
graph.invoke({"query": "删除所有记录"}, config)
# -> 中断: {'action': 'delete_all', 'question': '是否允许执行此操作？'}

graph.invoke(Command(resume="approved"), config)
# -> 执行操作并返回结果

9.7.2 信息补充模式（Information Gathering）

当 Agent 需要用户提供额外信息时：

def gather_info(state):
    if not state.get("user_preference"):
        preference = interrupt("请选择您偏好的方案: A) 快速 B) 经济 C) 平衡")
        return {"user_preference": preference}

    # 基于用户偏好继续处理
    return process_with_preference(state)

9.7.3 多步交互模式

同一节点中可以有多个中断点，实现多轮对话：

def multi_step_form(state):
    name = interrupt("请输入您的姓名")
    email = interrupt("请输入您的邮箱")
    confirm = interrupt(f"确认信息: {name}, {email}. 是否正确？")

    if confirm == "yes":
        return {"name": name, "email": email, "confirmed": True}
    else:
        return {"confirmed": False}

恢复时每次只解决一个新的中断，但已解决的中断的值会被缓存在 scratchpad 中自动返回。这意味着节点的执行看起来是"渐进式推进"的——每次恢复都让节点向前推进一步，直到所有中断都被解决。以下是完整的交互流程：

# 第一次中断
graph.invoke(input, config)  # -> 请输入姓名

# 恢复第一个中断，触发第二个
graph.invoke(Command(resume="Alice"), config)  # -> 请输入邮箱

# 恢复第二个中断，触发第三个
graph.invoke(Command(resume="alice@example.com"), config)  # -> 确认信息...

# 最终确认
graph.invoke(Command(resume="yes"), config)  # -> 完成

9.7.4 并行中断与按 ID 恢复

当图中有多个并行执行的节点同时中断时，需要使用按 ID 恢复：

# 多个节点同时中断
# node_a: interrupt("需要 A 的输入")  -> id="abc123..."
# node_b: interrupt("需要 B 的输入")  -> id="def456..."
# （示意：真实 id 是 32 位十六进制串，从返回值 "__interrupt__" 里各 Interrupt 的 .id 取）

# 一次性恢复所有中断
graph.invoke(
    Command(resume={
        "abc123...": "A 的值",
        "def456...": "B 的值",
    }),
    config,
)

按 ID 恢复的机制支持"部分恢复"——你可以选择只恢复一部分中断，让其余的中断继续保持挂起状态（pin 版实跑：只恢复 A 时，返回值里 out 已含 A 的结果，__interrupt__ 只剩 B 的中断）。这在需要渐进式审批的场景中很有用：例如，一个并行执行的风控系统中有三个独立的检查节点同时中断，风控人员可以先处理最紧急的一个，然后再逐一处理其余的。

如果有多个待处理中断时仍用简单恢复，_first() 会直接抛出 RuntimeError：

if len(self._pending_interrupts()) > 1:
    raise RuntimeError(
        "When there are multiple pending interrupts, you must specify the interrupt id when resuming. "
        "Docs: https://docs.langchain.com/oss/python/langgraph/add-human-in-the-loop#resume-multiple-interrupts-with-one-invocation."
    )

graph TB
    subgraph "并行中断场景"
        START((开始)) --> A[Node A]
        START --> B[Node B]
        A -->|"interrupt('输入A')<br/>id=abc123"| INT_A[中断 A]
        B -->|"interrupt('输入B')<br/>id=def456"| INT_B[中断 B]
        INT_A -.->|等待| RESUME
        INT_B -.->|等待| RESUME
        RESUME["Command(resume={<br/>'abc123': 'val_a',<br/>'def456': 'val_b'<br/>})"]
        RESUME --> A2[Node A 恢复]
        RESUME --> B2[Node B 恢复]
        A2 --> END((结束))
        B2 --> END
    end

9.8 时间旅行与 Checkpoint 的配合

9.8.1 时间旅行回退

中断机制与 Checkpoint 的配合不仅限于暂停/恢复，还支持"时间旅行"——回退到任意历史检查点并从那里重新执行：

# 查看历史状态
states = list(graph.get_state_history(config))
for state in states:
    print(f"Step {state.metadata['step']}: {state.values}")

# 回退到特定检查点
target_config = states[2].config  # 历史按时间倒序，取倒数第 3 个检查点
graph.invoke(None, target_config)  # 从该点恢复执行

9.8.2 RESUME 写入的清理

在时间旅行场景下，存在一个微妙的问题：如果回退到一个已经有 RESUME 写入的检查点，这些旧的恢复值应该被清理，否则 interrupt() 会错误地返回旧值而不是重新中断。

# PregelLoop._first()
if self.is_replaying and (
    (self.is_nested and configurable.get(CONFIG_KEY_CHECKPOINT_NS, "")
     in configurable.get(CONFIG_KEY_CHECKPOINT_MAP, {}))
    or not (
        (input_is_command and cast(Command, self.input).resume is not None)
        or configurable.get(CONFIG_KEY_RESUMING, False)
    )
):
    # 时间旅行时清理旧的 RESUME 写入
    self.checkpoint_pending_writes = [
        w for w in self.checkpoint_pending_writes if w[1] != RESUME
    ]

条件判断非常精细：只有在真正的时间旅行（而非正常恢复）时才清理 RESUME 写入。正常恢复场景下，之前的 RESUME 值必须保留，以支持多中断的顺序恢复。这个区分的逻辑是相当复杂的：对于外层图，通过检查输入是否为 Command(resume=...) 来判断是否是恢复操作；对于子图，通过 CONFIG_KEY_RESUMING 标志来判断；对于时间旅行到子图检查点的情况，通过检查子图自身的命名空间是否出现在 checkpoint_map 中来识别。这些条件组合起来，让恢复时不会误删 RESUME 写入导致重复中断，用 invoke(None, 历史 config) 时间旅行时也不会残留旧值导致跳过中断。但有一个边角：在一个已经恢复过的历史检查点上直接发 Command(resume=新值)，因为输入是 resume 命令，旧的 RESUME 写入不会被清理——pin 版实跑中先以 "A" 恢复，再对中断所在的历史检查点发 Command(resume="B")，节点拿到的仍是 "A"。

flowchart TD
    IS_REPLAY{is_replaying?}
    IS_REPLAY -->|No| KEEP[保留所有 RESUME 写入]
    IS_REPLAY -->|Yes| CHECK_RESUME{正在恢复?}
    CHECK_RESUME -->|"Command(resume=...)"| KEEP
    CHECK_RESUME -->|"CONFIG_KEY_RESUMING=True"| KEEP
    CHECK_RESUME -->|否| CLEAN[清理旧 RESUME 写入]
    CLEAN --> READY[准备重新执行]
    KEEP --> READY

9.9 中断与错误处理的交互

9.9.1 中断期间的异常处理

在复杂的工作流中，中断和错误处理可能交织在一起。LangGraph 对这种交互有明确的处理策略。

当一个节点在中断之前抛出了异常，异常通过正常的重试机制处理（如果配置了 RetryPolicy，langgraph-1.1.6/libs/langgraph/langgraph/types.py:404）。如果重试耗尽，错误被记录为 ERROR 类型的 pending write（在 WRITES_IDX_MAP 里占 -1 这个保留下标，langgraph-1.1.6/libs/checkpoint/langgraph/checkpoint/base/__init__.py:563），与中断信息共存在检查点中。再次调用时，_match_writes 会跳过 ERROR/INTERRUPT/RESUME 这几类写入，出错的任务因此拿不到任何已有写入、会被重新执行；它这次的写入经 put_writes 替换掉内存中该任务名下的旧记录（pregel/_loop.py:323-327），本步完成后生成新检查点，旧检查点上的 ERROR 写入不再参与后续执行。

更微妙的情况是：如果一个节点在第一个 interrupt()（langgraph-1.1.6/libs/langgraph/langgraph/types.py:705）返回恢复值之后、但在到达第二个 interrupt() 之前抛出了异常。在这种情况下，第一个中断的恢复值已经被缓存在 scratchpad 中。如果用户重新发起恢复（可能修复了导致异常的外部条件），节点会重新执行，第一个 interrupt() 会从缓存中返回之前的值，然后代码继续执行到异常点。这种行为确保了恢复值的幂等性——一旦用户提供了某个中断的恢复值，它就会被持久化，后续的重新执行不需要用户再次提供。但要当心另一面：导致这次失败的那条 Command(resume=...) 写入（NULL_TASK_ID 名下的 RESUME）仍留在该检查点的 pending writes 里——get_null_resume(True) 只从内存列表里移除它。修好外部条件后用 invoke(None, config) 重试，第二个 interrupt() 会直接拿到这个旧值而不再暂停；pin 版实跑中，第一次恢复传 "A" 后节点报错，重试时第二个 interrupt() 返回的也是 "A"。若改用 Command(resume=新值) 重试，新值则会顶替旧值交给第二个 interrupt()。

9.9.2 超时与中断的关系

LangGraph 的图执行有 recursion_limit 限制（由 RunnableConfig 提供，1.1.6 的默认值是 DEFAULT_RECURSION_LIMIT = 10007，见 langgraph-1.1.6/libs/langgraph/langgraph/_internal/_config.py:31——它还允许用环境变量 LANGGRAPH_DEFAULT_RECURSION_LIMIT 覆盖；另见第 6 章 §6.10）。步数一旦越过 PregelLoop.stop，tick() 会把 status 置为 "out_of_steps" 并直接返回 False（langgraph-1.1.6/libs/langgraph/langgraph/pregel/_loop.py:469），stream() 随后抛出 GraphRecursionError（langgraph-1.1.6/libs/langgraph/langgraph/errors.py:45）——这条路径和中断路径互斥，也就是说撞到步数上限时你拿到的是异常而不是 __interrupt__ 输出。这是一个安全措施——防止单次调用内的无限循环消耗过多资源。注意步数上限按每次调用计算：stop = step + recursion_limit + 1，其中 step 取自当前检查点（pregel/_loop.py:1194-1195），每次 Command(resume=...) 恢复都会重新获得完整额度。所以设计多步骤的人机协作流程时，要让 recursion_limit 容纳的是两次中断之间单次调用里跑的超步数，而不是整个交互流程的总步数。

9.9.3 中断的幂等性保证

LangGraph 的中断机制提供了一定程度的幂等性保证。具体来说：

对于同一个检查点，多次发送相同的 Command(resume=value) 不会导致异常或不一致的状态。第一次恢复时，值被写入 RESUME pending write 并驱动节点执行；如果由于网络问题导致客户端没有收到响应而重试，第二次写入的是同一个 (task_id, idx=-4) 主键（Command(resume=...) 的 task_id 是 NULL_TASK_ID）——RESUME 在 WRITES_IDX_MAP 里正是 -4（langgraph-1.1.6/libs/checkpoint/langgraph/checkpoint/base/__init__.py:563），走的是 UPSERT_CHECKPOINT_WRITES_SQL 的 ON CONFLICT ... DO UPDATE（langgraph-1.1.6/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py:140），用相同的值覆盖相同的行，结果与只写一次等价。

然而，需要注意的是，这种幂等性仅限于检查点级别。如果第一次恢复已经成功推进了图的状态（创建了新的检查点），第二次使用同一个配置发送恢复命令实际上是在最新的检查点上操作，而该检查点可能已经没有待处理的中断了——这时 pin 版不会报错，只是原样返回当前状态（实跑：图已结束后再发一次 Command(resume="A")，返回值与第一次相同）。客户端应该通过 get_state(config) 查看最新检查点 ID 以及 tasks/interrupts 来判断操作是否实际生效。

9.10 设计决策分析

9.10.1 为什么选择异常机制而非回调？

LangGraph 选择通过抛出异常来实现中断，而非传统的回调或 Promise 模式。这个选择有深层原因：

1. 保持节点函数的简单性：节点函数可以是普通的同步函数，不需要感知异步框架
2. 可重放的恢复：重新执行整个节点函数，按调用顺序把已提供的恢复值依次交还给各个 interrupt()；只要中断前的逻辑是确定的，就能走回同一个暂停点
3. 与 Checkpoint 天然配合：异常中断时，执行栈被丢弃，所有状态通过 Checkpoint 保存，实现了真正的持久化暂停

9.10.2 为什么重新执行整个节点？

恢复时不是从中断点继续执行，而是重新执行整个节点。这个设计的考量：

1. 简化实现：不需要保存 Python 执行栈的状态（这在 CPython 中几乎不可能可靠地实现）
2. 确保一致性：重新执行可以拾取可能已变化的外部状态
3. 多中断支持：通过 scratchpad 的索引追踪，已解决的中断直接返回缓存值，新的中断正常触发

代价是：中断前的副作用（如 API 调用、数据库写入、文件操作等）会被重复执行。这意味着在包含 interrupt() 的节点中，所有位于 interrupt() 之前的操作都必须是幂等的，或者需要通过额外的逻辑来检测和跳过已执行的操作。这是使用 interrupt() 时最重要的注意事项之一。在实践中，一个常见的模式是将非幂等的操作放在 interrupt() 之后——这样它们只在恢复后执行一次。或者，使用检查点中的状态来记录已完成的操作，在重新执行时根据状态跳过已完成的步骤。

9.10.3 为什么中断 ID 是确定性的？

中断 ID 通过对检查点命名空间哈希生成，而非随机 UUID。这带来两个好处：

1. 稳定性：同一任务被中断、恢复、重新执行时 ID 不变，客户端拿到的 ID 在恢复前一直有效
2. 可定位：resume map 以 ID 为键，_scratchpad() 用任务命名空间的哈希去查，恰好对上同一个任务（注意是任务级而非调用级：同一节点里的多个 interrupt() 共用一个 ID）

9.10.4 WRITES_IDX_MAP 的负索引设计

特殊写入（ERROR=-1, SCHEDULED=-2, INTERRUPT=-3, RESUME=-4）使用负索引，这是一个精巧的设计：

WRITES_IDX_MAP = {ERROR: -1, SCHEDULED: -2, INTERRUPT: -3, RESUME: -4}

普通写入从零开始正向编号，特殊写入使用固定的负数索引。在数据库的主键约束 (thread_id, checkpoint_ns, checkpoint_id, task_id, idx) 下，这保证了普通写入可以有任意数量且索引自增不冲突，而特殊写入每种类型最多一个且通过 UPSERT 语义保证最新值覆盖旧值。这种正负分区的索引设计避免了为特殊写入单独创建表或列的复杂性，在同一个统一的写入表中容纳了两种不同语义的数据。

9.10.5 声明式中断的局限性与适用场景

interrupt_before 和 interrupt_after 看似简单，但在设计复杂工作流时需要理解它们的局限性。声明式中断是无条件的——只要配置了节点名，每次执行到该节点都会中断。这意味着它不适用于"只在特定条件下才需要人工审核"的场景。对于条件性中断，应该在节点内部使用 interrupt() 函数，并用 if 语句控制是否调用。

声明式中断也不携带任何上下文信息。当客户端收到由 interrupt_before 触发的中断时，它只知道"某个节点即将执行"，但不知道为什么需要暂停或者用户应该做什么。这与 interrupt(value) 形成对比——后者可以传递一个描述性的消息来指导用户的操作。

那么声明式中断适用于什么场景？它最适合以下情况：在开发和调试阶段，需要在每个关键节点前后暂停以检查状态；在安全敏感的流程中，需要对所有外部操作进行人工审批；在演示和教学场景中，需要单步执行图以展示每个步骤的行为。在这些场景下，无条件暂停和无需携带数据恰恰是优势而非限制。

9.11 小结

本章深入分析了 LangGraph 的中断与人机协作机制。我们看到，这套机制远非简单的"暂停/继续"，而是一个经过精心设计的、可靠的持久化交互框架：

• interrupt() 函数通过异常机制实现暂停，通过 scratchpad 的索引追踪实现恢复时的精确值匹配。这个函数巧妙地让同一个函数调用在不同的执行生命周期中扮演截然不同的角色——首次调用时作为暂停器抛出异常，恢复后重新执行时作为值提供器返回缓存数据
• Interrupt 数据类型使用基于命名空间哈希的确定性标识符，支持精确的按标识符恢复，解决了多个任务同时中断时的歧义问题；同一节点内的多个中断则靠调用顺序匹配
• interrupt_before 和 interrupt_after 提供了编译时的声明式中断配置，适用于不需要携带自定义数据的简单审查和调试场景
• 与 Checkpoint 的深度集成确保中断状态可以跨进程、跨机器、跨时间持久化，基于 pending writes 的 RESUME 写入管理机制支持多中断的顺序恢复和并行中断的按标识符精确恢复
• 时间旅行中对恢复写入的精细清理逻辑体现了系统在复杂边界条件和极端场景下的健壮性设计

在下一章中，我们将看到 Command 类型如何提供更强大的高级控制流能力——不仅能恢复中断，还能在节点内部直接控制图的执行路径。