LangGraph 并行分支与人工介入中断

本文档系统梳理 LangGraph 中控制流的核心机制：add_edge、Send、Command的差异与用法，并行分支的多种实现方式，以及人工介入中断（interrupt）的完整恢复逻辑。所有结论均基于实际运行测试。

第一部分：控制流三大机制

在 LangGraph 中，add_edge、Send、Command都可以控制节点之间的流转，甚至都能产生并行分支，但设计目标和适用场景截然不同。

1.1 add_edge：静态边

add_edge是图的声明式结构，在编译期就固定了节点之间的连接关系。

builder.add_edge("node_a", "node_b")

• 返回者：无返回值，纯粹声明。

• 决定时机：编译期（compile()时）。

• 拓扑：固定的一对一或多对一。

• 节点职责：节点函数只返回dict（状态更新），不关心下一步去哪。

适用场景：流程固定、分支明确、无需运行时动态决策。

1.2 Send：路由中的动态扇出

Send用于在条件边的路由函数中动态生成多个并行任务，特别适合 map-reduce 模式。

from langgraph.types import Send def fanout_router(state) -> list[Send]: return [ Send("worker", {"item": "A"}), Send("worker", {"item": "B"}), ] builder.add_conditional_edges("entry", fanout_router)

• 返回者：路由函数。

• 决定时机：运行时，根据当前 state 动态计算。

• 关键特征：

  • 同一个目标节点可以被并行调用多次，每次携带独立的输入状态（Send的第二个参数）。

  • 每个Send的arg会替换目标节点收到的 state，而不是合并进主图状态。

  • 只有目标节点返回值中、且主图State里定义了 reducer 的键，才会合并回主图。

• 必须配合 reducer：多个并行分支写入同一个键时，该键必须用Annotated指定 reducer（如operator.add），否则会抛出InvalidUpdateError。

import operator from typing import Annotated, TypedDict class State(TypedDict): results: Annotated[list[str], operator.add]

适用场景：动态数量的并行任务、每个任务需要不同输入、结果需要合并。

1.3 Command：节点内的动态跳转

Command允许节点函数在运行时决定下一步去哪个节点，甚至形成循环。

from langgraph.types import Command def node_a(state) -> Command: if state["condition"]: return Command(goto="node_x", update={"key": "value"}) else: return Command(goto="node_y")

• 返回者：普通节点函数。

• 决定时机：运行时，节点内部自行决策。

• 关键特征：

  • 可以跳转到任意节点（包括之前执行过的节点），形成循环。

  • 可以同时携带状态更新（update参数）。

  • Command(goto=[...])也可以接受节点列表，实现并行跳转。

• 特殊形式：Command(resume=...)专用于恢复interrupt()造成的中断。

适用场景：Agent 路由、条件循环、需要根据推理结果动态选择下一步。

1.4 三者对比

选择原则：

• 流程固定 →add_edge

• 动态数量的并行任务 →Send

• 运行时决定下一步（含循环） →Command

第二部分：并行分支的实现方式

2.1 add_edge：静态并行

从同一个源节点添加多条边：

builder.add_edge("entry", "node_a") builder.add_edge("entry", "node_b") # entry 执行完后，node_a 和 node_b 并行

原理：entry节点完成后，引擎发现它有两条出边，就把node_a和node_b都加入同一超步（superstep）并行执行。

限制：

• 编译期固定，运行时无法改变。

• 并行的是不同的节点，不能让同一个节点并行跑多次。

• 多个并行分支写入同一个无 reducer 的键时，会抛InvalidUpdateError。

2.2 路由返回节点名列表：动态并行

路由函数可以返回节点名列表，实现运行时动态决定并行哪些节点：

def router(state) -> list[str]: return ["node_a", "node_b"] builder.add_conditional_edges("entry", router, {...})

原理：与Command(goto=[...])一致，引擎将列表中的节点加入同一超步并行执行。此写法已通过实际测试验证可行。

2.3 Command(goto=[...])：节点内动态并行

return Command(goto=["node_a", "node_b"])

原理：Command的goto参数类型定义为Send | Sequence[Send | N] | N，支持节点名列表，引擎将列表中的节点加入同一超步并行执行。

2.4 Send：动态并行 + 同节点多实例

def router(state) -> list[Send]: return [Send("worker", {"item": item}) for item in state["items"]]

原理：每个Send创建一个独立的并行任务，目标节点可以相同，但arg不同。worker 节点收到的是Send传递的独立状态，而不是主图状态。

限制：结果合并回主图时，只有主图State中定义了 reducer 的键才会被合并。

2.5 关键限制：哪些写法不能用节点名列表

并非所有构件都支持列表形式的节点名。经过实际验证：

• add_edge** 不能用节点名列表**。它只接受单个节点名作为参数，想并行多条边，只能多次调用：

  builder.add_edge("entry", "node_a") builder.add_edge("entry", "node_b") # ✅ 正确 # builder.add_edge("entry", ["node_a", "node_b"]) ❌ 报错

• Send** 的第一个参数不能用节点名列表**。Send(node, arg)的node必须是单个字符串：

  return [Send("node_a", {...}), Send("node_b", {...})] # ✅ 正确 # Send(["node_a", "node_b"], {...}) ❌ 报错

• 只有路由函数和Command(goto=...)支持节点名列表：

  # 路由函数 def router(state) -> list[str]: return ["node_a", "node_b"] # ✅ # Command return Command(goto=["node_a", "node_b"]) # ✅ # 如果要写path_map，应该写成这样 path_map={"node_a": "node_a", "node_b": "node_b"}

原因：add_edge和Send是“单目标”构件，设计假设就是“一个目标”；而路由函数和Command是“调度决策”构件，返回值本身是路由决策，列表自然表示“去多个地方”。

2.6 关键问题：能否在节点内直接返回 Send？

不能直接return [Send(...)]，但可以通过Command间接使用。

普通节点函数的返回值被 LangGraph 严格限制为dict。如果在节点里直接返回Send列表：

def my_node(state) -> list[Send]: return [Send("worker", {"item": "A"})] # ❌ 报错

运行时会抛出：

InvalidUpdateError: Expected dict, got [Send(...)]

正确做法：通过Command的goto参数传递。Command的goto类型定义为Send | Sequence[Send | N] | N，明确支持Send对象或Send序列：

from langgraph.types import Command, Send def fanout_node(state) -> Command: sends = [Send("worker", {"item": item}) for item in state["items"]] return Command(goto=sends) # ✅ 合法

2.7 并行方式对比

选择原则：

• 并行的是不同节点，且数量固定 →add_edge

• 并行的是不同节点，但运行时才知道是哪些 →Command(goto=[...])或路由函数返回节点名列表

• 需要同一个节点并行多次，每次输入不同 →Send（路由或Command(goto=[Send...])）

• 所有并行场景，共享键必须配置 reducer（如Annotated[list, operator.add]）

第三部分：人工介入中断详解

interrupt()用于在节点执行过程中暂停图，等待外部输入（如人工审批），然后通过Command(resume=...)恢复执行。

3.1 基本用法

from langgraph.types import interrupt def approval_node(state): answer = interrupt({"question": "是否批准？"}) return {"result": answer}

首次调用图时，执行到interrupt()会暂停，返回结果包含__interrupt__字段：

result = graph.invoke({"request": "..."}, config) # result["__interrupt__"] 是 Interrupt 对象列表 # 每个 Interrupt 有 id、value 等属性

恢复执行：

from langgraph.types import Command graph.invoke(Command(resume=恢复值), config) # 必须使用相同的 thread_id

3.2 恢复机制与 ID 映射

Command(resume=...)的行为取决于当前待处理中断的数量。

单个待处理中断

LangGraph 对单中断做宽松处理：

• 传裸值：Command(resume="批准")→interrupt()返回"批准"。

• 传字典，键不是真实 ****interrupt_id：整个字典作为值传递。

  Command(resume={"id": "fake"}) # interrupt() 收到 {'id': 'fake'}

• 传字典，键是真实 ****interrupt_id：按 ID 映射，字典的值传递给中断。

  Command(resume={真实id: "fake"}) # interrupt() 收到 "fake"

多个待处理中断

必须使用{interrupt_id: value}字典，否则抛出RuntimeError：

RuntimeError: When there are multiple pending interrupts, you must specify the interrupt id when resuming.

正确写法：

resume_map = { interrupt_1.id: "批准-财务", interrupt_2.id: "批准-法务", } graph.invoke(Command(resume=resume_map), config)

实测结论

3.3 Checkpoint 与恢复流程

LangGraph 使用 checkpoint 保存执行状态。每次超步结束或中断时写入 checkpoint，包含：

• values：当前已合并的状态。

• next：下一步计划执行的节点元组。中断发生时，next仍包含该超步计划执行的所有节点，而不是只包含未完成的节点。

• interrupts：待处理的中断列表。

恢复时，引擎加载最新 checkpoint，根据超步内任务的完成标记：

• 跳过已完成的节点（不重放，但保留其写入结果）。

• 重放被中断的节点（从头执行，interrupt()返回恢复值）。

• 所有写入合并后，继续推进后续超步。

3.4 注意事项

• 节点重放：恢复时，包含interrupt()的节点会从函数开头重新执行。interrupt()之前的代码（如 API 调用、数据库写入）会再次运行。必须确保这些操作幂等，或将副作用移到interrupt()之后。

• 相同 ****thread_id：恢复时必须使用与中断时相同的thread_id，否则会启动全新线程。

• 一次性恢复所有中断：多个中断待处理时，必须为所有中断提供恢复值，只提供部分值会导致图再次暂停。

• 同一节点内多个 ****interrupt()：恢复值按调用顺序匹配，而不是按 ID。例如节点内有两个interrupt()，Command(resume=[v1, v2])会依次返回。

3.5 单中断完整示例

from typing import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import interrupt, Command from langgraph.checkpoint.memory import InMemorySaver class State(TypedDict): result: str def only_node(state: State) -> dict: answer = interrupt({"question": "审批?"}) return {"result": f"interrupt 收到: {answer}"} builder = StateGraph(State) builder.add_node("only", only_node) builder.add_edge(START, "only") builder.add_edge("only", END) graph = builder.compile(checkpointer=InMemorySaver()) config = {"configurable": {"thread_id": "t1"}} r1 = graph.invoke({"result": ""}, config) real_id = r1["__interrupt__"][0].id r1_resume = graph.invoke(Command(resume="批准"), config) print(r1_resume["result"]) # interrupt 收到: 批准

3.6 多中断完整示例

import operator from typing import Annotated, TypedDict from langgraph.graph import StateGraph, START, END from langgraph.types import interrupt, Command, Send from langgraph.checkpoint.memory import InMemorySaver class State2(TypedDict): results: Annotated[list[str], operator.add] def entry(state: State2) -> dict: return {} def a(state: dict) -> dict: ans = interrupt({"node": "a"}) return {"results": [f"a: {ans}"]} def b(state: dict) -> dict: ans = interrupt({"node": "b"}) return {"results": [f"b: {ans}"]} def fanout_router(state: State2) -> list[Send]: return [Send("a", {}), Send("b", {})] builder2 = StateGraph(State2) builder2.add_node("entry", entry) builder2.add_node("a", a) builder2.add_node("b", b) builder2.add_edge(START, "entry") builder2.add_conditional_edges("entry", fanout_router) builder2.add_edge("a", END) builder2.add_edge("b", END) graph2 = builder2.compile(checkpointer=InMemorySaver()) config2 = {"configurable": {"thread_id": "t2"}} r2 = graph2.invoke({"results": []}, config2) resume_map = {} for intr in r2["__interrupt__"]: node = intr.value.get("node") resume_map[intr.id] = f"批准-{node}" r2_resume = graph2.invoke(Command(resume=resume_map), config2) print(r2_resume["results"]) # ['a: 批准-a', 'b: 批准-b']

第四部分：总结速查

控制流机制

并行分支

中断恢复

核心陷阱

1. 节点函数不能返回Send列表，必须通过路由函数或Command(goto=...)返回。

2. add_edge** 和Send不接受节点名列表**，只有路由函数和Command(goto=[...])可以。

3. 并行写入同一键必须配置 reducer，否则抛InvalidUpdateError。

4. 多中断必须用interrupt_id映射，单中断的宽松处理不适用于多中断。

5. 恢复时节点从头重放，interrupt()之前的副作用必须幂等。

6. Send** 传递的arg替换目标节点 state**，而非合并进主图，只有节点返回值中带 reducer 的键才会合并回主图。