LangGraph 源码深潜：Interrupt 与 Human-in-the-Loop 机制解剖

1
def compile(
2
    self,
3
    checkpointer: Checkpointer = None,
4
    *,
5
    cache: BaseCache | None = None,
6
    store: BaseStore | None = None,
7
    interrupt_before: All | list[str] | None = None,
8
    interrupt_after: All | list[str] | None = None,
9
    debug: bool = False,
10
    name: str | None = None,
11
) -> CompiledStateGraph:
12
    ...

1
业务代码
2
  ↓
3
builder.compile(checkpointer=InMemorySaver())
4
  ↓
5
StateGraph.compile
6
  ↓
7
CompiledStateGraph / Pregel 构造

1
def compile(self, checkpointer=None, **options):
2
    # 1. 校验 graph 结构
3
    self.validate(
4
        interrupt_before=options.get("interrupt_before"),
5
        interrupt_after=options.get("interrupt_after"),
6
    )
7

8
    # 2. 规范化 checkpointer
9
    checkpointer = ensure_valid_checkpointer(checkpointer)
10

11
    # 3. 构造 Pregel 节点、channel、边、branch
12
    compiled = CompiledStateGraph(
13
        builder=self,
14
        channels=self.channels,
15
        nodes={},
16
        checkpointer=checkpointer,
17
        store=options.get("store"),
18
        cache=options.get("cache"),
19
        interrupt_before=options.get("interrupt_before"),
20
        interrupt_after=options.get("interrupt_after"),
21
        debug=options.get("debug"),
22
        name=options.get("name"),
23
    )
24

25
    # 4. 把 builder 中的 nodes / edges / branches 附加到 compiled graph
26
    for node_name, node_spec in self.nodes.items():
27
        compiled.attach_node(node_name, node_spec)
28

29
    for start, end in self.edges:
30
        compiled.attach_edge(start, end)
31

32
    for source, branches in self.branches.items():
33
        for branch_name, branch_spec in branches.items():
34
            compiled.attach_branch(source, branch_name, branch_spec)
35

36
    # 5. 校验 compiled graph
37
    return compiled.validate()

1
checkpointer 合法
2
  ↓
3
compiled graph 构造完成
4
  ↓
5
运行时可保存 checkpoint

1
@dataclass
2
class Command(Generic[N], ToolOutputMixin):
3
    graph: str | None = None
4
    update: Any | None = None
5
    resume: dict[str, Any] | Any | None = None
6
    goto: Send | Sequence[Send | N] | N = ()

1
外部调用方
2
  ↓
3
Command(resume=True)
4
  ↓
5
graph.invoke(Command(resume=True), config)
6
  ↓
7
Pregel runtime
8
  ↓
9
恢复 pending interrupt

1
@dataclass
2
class Command:
3
    graph: str | None = None
4
    update: Any | None = None
5
    resume: dict[str, Any] | Any | None = None
6
    goto: Send | Sequence[Send | str] | str = ()
7

8
    def _update_as_tuples(self):
9
        if isinstance(self.update, dict):
10
            return list(self.update.items())
11

12
        if is_list_of_key_value_tuples(self.update):
13
            return self.update
14

15
        if is_annotated_dataclass_or_model(self.update):
16
            return extract_update_fields(self.update)
17

18
        if self.update is not None:
19
            return [("__root__", self.update)]
20

21
        return []

1
Command(resume=True)
2
  ↓
3
同一个 thread_id 调用 graph
4
  ↓
5
resume value 进入 pending interrupt
6
  ↓
7
interrupt() 返回 True

1
@dataclass
2
class Interrupt:
3
    value: Any
4
    id: str
5

6
class StateSnapshot(NamedTuple):
7
    values: dict[str, Any] | Any
8
    next: tuple[str, ...]
9
    config: RunnableConfig
10
    metadata: CheckpointMetadata | None
11
    created_at: str | None
12
    parent_config: RunnableConfig | None
13
    tasks: tuple[PregelTask, ...]
14
    interrupts: tuple[Interrupt, ...]

1
interrupt(value)
2
  ↓
3
Interrupt(value, id)
4
  ↓
5
result["__interrupt__"] 或 stream.interrupts
6
  ↓
7
调用方展示 payload
8

9
graph.get_state(config)
10
  ↓
11
StateSnapshot
12
  ↓
13
snapshot.interrupts

1
class Interrupt:
2
    def __init__(self, value, id=None, ns=None):
3
        self.value = value
4

5
        if id is not None:
6
            self.id = id
7
        elif ns is not None:
8
            self.id = hash_namespace(ns)
9
        else:
10
            self.id = placeholder_or_generated_id
11

12

13
class StateSnapshot(NamedTuple):
14
    values: dict
15
    next: tuple[str, ...]
16
    config: RunnableConfig
17
    metadata: dict | None
18
    created_at: str | None
19
    parent_config: RunnableConfig | None
20
    tasks: tuple[PregelTask, ...]
21
    interrupts: tuple[Interrupt, ...]

1
节点 interrupt
2
  ↓
3
runtime 生成 Interrupt
4
  ↓
5
snapshot.interrupts 包含 pending interrupt
6
  ↓
7
外部系统读取并展示

1
def interrupt(value: Any) -> Any:
2
    ...

1
approval_node
2
  ↓
3
interrupt(payload)
4
  ↓
5
scratchpad / runtime interrupt state
6
  ↓
7
GraphInterrupt 或 resume value

1
def interrupt(value):
2
    # 1. 获取当前 task 的运行时 scratchpad
3
    scratchpad = get_current_task_scratchpad()
4

5
    # 2. 计算当前节点内第几次 interrupt 调用
6
    index = scratchpad.interrupt_counter()
7

8
    # 3. 如果当前 task 已经有对应 index 的 resume 值
9
    if index < len(scratchpad.resume_values):
10
        return scratchpad.resume_values[index]
11

12
    # 4. 如果 resume 是按 interrupt_id 映射传入，也尝试用当前 interrupt id 匹配
13
    interrupt_id = make_interrupt_id_from_task_path_and_index(
14
        scratchpad.task_path,
15
        index,
16
    )
17

18
    if interrupt_id in scratchpad.resume_map:
19
        value = scratchpad.resume_map[interrupt_id]
20
        scratchpad.resume_values.append(value)
21
        return value
22

23
    # 5. 没有 resume 值，说明这是首次执行到这个 interrupt
24
    interrupt_obj = Interrupt(
25
        value=value,
26
        id=interrupt_id,
27
    )
28

29
    # 6. 抛出可恢复中断，让 Pregel task 停止
30
    raise GraphInterrupt((interrupt_obj,))

1
首次执行：
2
interrupt(payload)
3
  ↓
4
没有 resume 值
5
  ↓
6
raise GraphInterrupt
7
  ↓
8
图暂停
9

10
恢复执行：
11
interrupt(payload)
12
  ↓
13
发现 resume 值
14
  ↓
15
return resume_value
16
  ↓
17
节点继续执行

1
graph.invoke(input: InputT, config: RunnableConfig | None = None, **kwargs) -> OutputT

1
graph.invoke(initial_state, config)
2
  ↓
3
Pregel.stream / loop
4
  ↓
5
PregelRunner.tick
6
  ↓
7
approval_node
8
  ↓
9
interrupt()
10
  ↓
11
GraphInterrupt
12
  ↓
13
checkpoint save

1
def invoke(input_state, config):
2
    # 1. 规范化 config，读取 thread_id
3
    config = ensure_config(config)
4
    thread_id = config["configurable"]["thread_id"]
5

6
    # 2. 从 checkpointer 读取当前 thread 的 checkpoint
7
    checkpoint = checkpointer.get_tuple(config)
8

9
    # 3. 如果没有 checkpoint，创建初始 checkpoint
10
    if checkpoint is None:
11
        state = initialize_state(input_state)
12
        next_tasks = ("approval_node",)
13
    else:
14
        state = read_state_from_checkpoint(checkpoint)
15
        next_tasks = checkpoint.pending_tasks
16

17
    # 4. 执行当前 super-step 中的 tasks
18
    task_results = []
19
    task_interrupts = []
20

21
    for task in next_tasks:
22
        try:
23
            result = run_node(task, state, config)
24
            task_results.append(result)
25

26
        except GraphInterrupt as interrupt_exc:
27
            task_interrupts.extend(interrupt_exc.interrupts)
28

29
    # 5. 如果出现 interrupt，不继续调度后续节点
30
    if task_interrupts:
31
        snapshot = build_state_snapshot(
32
            values=state,
33
            next=next_tasks,
34
            tasks=mark_tasks_interrupted(next_tasks, task_interrupts),
35
            interrupts=tuple(task_interrupts),
36
        )
37

38
        # 6. 保存 checkpoint
39
        checkpointer.put(
40
            config=config,
41
            checkpoint=snapshot_to_checkpoint(snapshot),
42
            metadata=build_metadata(source="loop", writes=None),
43
        )
44

45
        # 7. 返回 interrupt 给调用方
46
        return {
47
            "__interrupt__": tuple(task_interrupts)
48
        }
49

50
    # 8. 没有 interrupt，应用节点写入并继续调度
51
    state = apply_writes(state, task_results)
52
    next_tasks = compute_next_tasks(state)
53

54
    return continue_until_end(state, next_tasks)

1
initial invoke
2
  ↓
3
approval_node
4
  ↓
5
interrupt
6
  ↓
7
checkpoint saved
8
  ↓
9
caller receives Interrupt

1
Command(resume: dict[str, Any] | Any | None = None)

1
graph.invoke(Command(resume=True), same config)
2
  ↓
3
Pregel runtime
4
  ↓
5
读取 checkpoint
6
  ↓
7
把 resume value 注入 task scratchpad
8
  ↓
9
重新执行 approval_node
10
  ↓
11
interrupt() 返回 True

1
def invoke(input_value, config):
2
    config = ensure_config(config)
3

4
    if isinstance(input_value, Command) and input_value.resume is not None:
5
        return resume_from_interrupt(
6
            command=input_value,
7
            config=config,
8
        )
9

10
    return start_or_continue_normal_input(input_value, config)
11

12

13
def resume_from_interrupt(command, config):
14
    # 1. 用同一个 thread_id 读取暂停 checkpoint
15
    checkpoint = checkpointer.get_tuple(config)
16

17
    if checkpoint is None:
18
        raise ValueError("No checkpoint found for resume")
19

20
    # 2. 读取 pending interrupts 与 next tasks
21
    pending_tasks = checkpoint.pending_tasks
22
    pending_interrupts = extract_interrupts(checkpoint)
23

24
    # 3. 把 resume 值转换成 task 可读取的 resume 数据
25
    if isinstance(command.resume, dict):
26
        resume_map = command.resume
27
    else:
28
        resume_map = assign_to_next_interrupt(
29
            pending_interrupts,
30
            command.resume,
31
        )
32

33
    # 4. 将 resume_map 注入 task scratchpad
34
    runtime = build_runtime_from_checkpoint(
35
        checkpoint=checkpoint,
36
        resume_map=resume_map,
37
    )
38

39
    # 5. 重新执行被中断的节点
40
    task_results = []
41

42
    for task in pending_tasks:
43
        result = run_node_again(
44
            task=task,
45
            state=checkpoint.state,
46
            runtime=runtime,
47
        )
48
        task_results.append(result)
49

50
    # 6. 此次 interrupt() 不再抛中断，而是返回 resume value
51
    # approval_node 得到 approved=True 后正常返回 {"approved": True}
52

53
    # 7. 应用写入，继续后续节点
54
    state = apply_writes(
55
        checkpoint.state,
56
        task_results,
57
    )
58

59
    next_tasks = compute_next_tasks(state)
60

61
    return continue_until_end(state, next_tasks)

1
Command(resume=True)
2
  ↓
3
读取 checkpoint
4
  ↓
5
注入 resume value
6
  ↓
7
重放 approval_node
8
  ↓
9
interrupt() 返回 True
10
  ↓
11
return {"approved": True}
12
  ↓
13
final_node
14
  ↓
15
END

1
def get_state(
2
    self,
3
    config: RunnableConfig,
4
    *,
5
    subgraphs: bool = False,
6
) -> StateSnapshot:
7
    ...

1
审批系统
2
  ↓
3
graph.get_state(config)
4
  ↓
5
checkpointer.get_tuple(config)
6
  ↓
7
StateSnapshot(values, next, tasks, interrupts)

1
def get_state(config, subgraphs=False):
2
    # 1. 校验 checkpointer
3
    if self.checkpointer is None:
4
        raise ValueError("No checkpointer set")
5

6
    # 2. 用 thread_id / checkpoint_id 读取 checkpoint tuple
7
    checkpoint_tuple = self.checkpointer.get_tuple(config)
8

9
    if checkpoint_tuple is None:
10
        return empty_state_snapshot(config)
11

12
    # 3. 从 checkpoint 读取 channel values
13
    values = read_channels(
14
        checkpoint_tuple.checkpoint,
15
        self.output_channels,
16
    )
17

18
    # 4. 根据 pending sends / tasks 推断 next
19
    next_nodes = infer_next_tasks(checkpoint_tuple)
20

21
    # 5. 构造 tasks，包含 error / interrupts / subgraph state
22
    tasks = build_pregel_tasks(
23
        checkpoint_tuple,
24
        include_subgraphs=subgraphs,
25
    )
26

27
    # 6. 从 tasks 中收集 interrupts
28
    interrupts = tuple(
29
        interrupt
30
        for task in tasks
31
        for interrupt in task.interrupts
32
    )
33

34
    # 7. 返回 StateSnapshot
35
    return StateSnapshot(
36
        values=values,
37
        next=next_nodes,
38
        config=checkpoint_tuple.config,
39
        metadata=checkpoint_tuple.metadata,
40
        created_at=checkpoint_tuple.checkpoint["ts"],
41
        parent_config=checkpoint_tuple.parent_config,
42
        tasks=tasks,
43
        interrupts=interrupts,
44
    )

1
graph.get_state(config)
2
  ↓
3
读取最新 checkpoint
4
  ↓
5
构造 StateSnapshot
6
  ↓
7
snapshot.interrupts 展示给外部系统

1
def approval_node(state):
2
    # 错误示例：不要这样包住 interrupt
3
    try:
4
        approved = interrupt({"plan": state["plan"]})
5
    except Exception as exc:
6
        # 这会吞掉 GraphInterrupt，runtime 看不到中断
7
        return {"approved": False}
8

9
    return {"approved": approved}
10

11

12
def safe_approval_node(state):
13
    # 正确示例：interrupt 与易出错业务逻辑分离
14
    approved = interrupt({
15
        "question": "是否确认？",
16
        "plan": state["plan"],
17
    })
18

19
    try:
20
        validate_approval(approved)
21
    except ValueError:
22
        return {"approved": False}
23

24
    return {"approved": approved}

1
# 初次执行后，外部系统拿到多个 pending interrupt
2
snapshot = graph.get_state(config)
3
pending = snapshot.interrupts
4

5
# pending 可能类似：
6
# [
7
#   Interrupt(id="int_a", value={"question": "审批 A"}),
8
#   Interrupt(id="int_b", value={"question": "审批 B"}),
9
# ]
10

11
# 外部系统分别收集审批结果
12
resume_map = {
13
    pending[0].id: True,
14
    pending[1].id: False,
15
}
16

17
# 一次性恢复多个 interrupt
18
result = graph.invoke(
19
    Command(resume=resume_map),
20
    config=config,
21
)