LangGraph 源码深潜：Conditional Edge 与 Router 路由机制解剖

1
def add_conditional_edges(
2
    self,
3
    source: str,
4
    path: Callable[..., Hashable | Sequence[Hashable]]
5
        | Callable[..., Awaitable[Hashable | Sequence[Hashable]]]
6
        | Runnable[Any, Hashable | Sequence[Hashable]],
7
    path_map: dict[Hashable, str] | list[str] | None = None,
8
) -> Self:
9
    ...

1
用户代码
2
  ↓
3
StateGraph.add_conditional_edges
4
  ↓
5
coerce_to_runnable
6
  ↓
7
BranchSpec.from_path
8
  ↓
9
self._add_schema(optional)

1
def add_conditional_edges(self, source, path, path_map=None):
2
    # 1. 编译后修改 builder 只给 warning
3
    if self.compiled:
4
        logger.warning("compiled graph will not be changed")
5

6
    # 2. 把普通函数 / async 函数 / Runnable 统一成 Runnable
7
    path_runnable = coerce_to_runnable(
8
        path,
9
        name=None,
10
        trace=True,
11
    )
12

13
    # 3. 给这个 branch 找一个名字
14
    branch_name = path_runnable.name or "condition"
15

16
    # 4. 同一个 source 下，不允许同名 branch 重复注册
17
    if branch_name in self.branches[source]:
18
        raise ValueError("Branch already exists for source")
19

20
    # 5. 构造 BranchSpec：保存 path、ends、input_schema
21
    branch_spec = BranchSpec.from_path(
22
        path=path_runnable,
23
        path_map=path_map,
24
        infer_schema=True,
25
    )
26

27
    # 6. 保存到 builder 的 branch registry
28
    self.branches[source][branch_name] = branch_spec
29

30
    # 7. 如果 branch path 有独立 input schema，则加入 schemas
31
    if branch_spec.input_schema is not None:
32
        self._add_schema(branch_spec.input_schema)
33

34
    return self

1
用户注册 route_by_intent
2
  ↓
3
route_by_intent 被转成 Runnable
4
  ↓
5
BranchSpec 保存 path 和 ends
6
  ↓
7
self.branches["classify_intent"]["route_by_intent"] = BranchSpec(...)

1
不要在 add_conditional_edges() 里期待 router 立即运行。
2
不要把 router 写成依赖外部可变全局状态的函数。
3
应该让 router 只依赖当前 state / context。

1
@classmethod
2
def from_path(
3
    cls,
4
    path: Runnable[Any, Hashable | list[Hashable]],
5
    path_map: dict[Hashable, str] | list[str] | None,
6
    infer_schema: bool = False,
7
) -> BranchSpec:
8
    ...

1
StateGraph.add_conditional_edges
2
  ↓
3
BranchSpec.from_path
4
  ├── path_map dict/list 归一化
5
  ├── Literal return type 推断
6
  └── _get_branch_path_input_schema

1
@classmethod
2
def from_path(cls, path, path_map, infer_schema=False):
3
    path_map_ = None
4

5
    # 1. 用户传 dict：逻辑 label -> 实际节点名
6
    if isinstance(path_map, dict):
7
        path_map_ = path_map.copy()
8

9
    # 2. 用户传 list：每个名字映射到自己
10
    elif isinstance(path_map, list):
11
        path_map_ = {name: name for name in path_map}
12

13
    # 3. 用户未传 path_map：尝试从 Literal 返回类型推断
14
    else:
15
        func = extract_callable_from_runnable(path)
16

17
        if func is not None:
18
            return_type = get_type_hints(func).get("return")
19

20
            if get_origin(return_type) is Literal:
21
                path_map_ = {
22
                    name: name
23
                    for name in get_args(return_type)
24
                }
25

26
    # 4. 可选推断 router 输入 schema
27
    if infer_schema:
28
        input_schema = _get_branch_path_input_schema(path)
29
    else:
30
        input_schema = None
31

32
    # 5. 构造 BranchSpec
33
    return BranchSpec(
34
        path=path,
35
        ends=path_map_,
36
        input_schema=input_schema,
37
    )

1
path_map={
2
    "budget": "budget_node",
3
    "attraction": "attraction_node",
4
    "general": "general_node",
5
}

1
path_map=["budget_node", "attraction_node"]

1
{"budget_node": "budget_node", "attraction_node": "attraction_node"}

1
def route(state) -> Literal["budget_node", "general_node"]:
2
    ...

1
path_map 显式 dict
2
  ↓
3
复制成 ends
4
  ↓
5
BranchSpec.ends 保存 label → node 映射

1
path_map None + router return Literal
2
  ↓
3
从 Literal 提取候选目标
4
  ↓
5
BranchSpec.ends 保存目标集合

1
1. 把 router 的逻辑返回值映射到真实节点名。
2
2. 让框架在编译、校验、可视化时知道可能的目标集合。

1
生产级 Router 尽量使用 Literal 或 path_map。
2
如果 router 返回业务枚举 label，则必须使用 path_map。
3
如果 router 直接返回节点名，也建议加 Literal，便于可视化和维护。

1
def _get_branch_path_input_schema(
2
    path: Callable[..., Hashable | Sequence[Hashable]]
3
        | Callable[..., Awaitable[Hashable | Sequence[Hashable]]]
4
        | Runnable[Any, Hashable | Sequence[Hashable]],
5
) -> type[Any] | None:
6
    ...

1
BranchSpec.from_path(infer_schema=True)
2
  ↓
3
_get_branch_path_input_schema
4
  ↓
5
get_type_hints / inspect.signature

1
def _get_branch_path_input_schema(path):
2
    input_schema = None
3

4
    try:
5
        callable_ = None
6

7
        # 1. 如果 path 是 RunnableCallable / RunnableLambda，
8
        #    尝试找到内部同步函数或异步函数
9
        if isinstance(path, (RunnableCallable, RunnableLambda)):
10
            callable_ = find_inner_func_or_afunc(path)
11

12
        # 2. 如果 path 本身就是 callable
13
        elif callable(path):
14
            callable_ = path
15

16
        # 3. 读取类型提示
17
        if callable_ is not None:
18
            hints = get_type_hints(callable_)
19
            first_param = first_parameter_name(signature(callable_))
20
            input_hint = hints.get(first_param)
21

22
            # 4. 只有 input_hint 是可展开类型时才返回
23
            if isinstance(input_hint, type) and get_type_hints(input_hint):
24
                input_schema = input_hint
25

26
    except (TypeError, StopIteration):
27
        pass
28

29
    return input_schema

1
def route_by_intent(state: TravelState) -> Literal[...]:
2
    ...

1
route_by_intent(state: TravelState)
2
  ↓
3
get_type_hints(route_by_intent)
4
  ↓
5
first parameter = state
6
  ↓
7
input_hint = TravelState
8
  ↓
9
BranchSpec.input_schema = TravelState

1
给 router 补全 state 类型提示，主要提升可读性、校验和可视化质量。
2
不要依赖类型提示承担业务安全校验。

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
    transformers: Sequence[Callable[[tuple[str, ...]], Any]] | None = None,
12
) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:
13
    ...

1
用户 graph = builder.compile()
2
  ↓
3
StateGraph.compile
4
  ↓
5
self.validate
6
  ↓
7
CompiledStateGraph(...)
8
  ↓
9
compiled.attach_node(...)
10
  ↓
11
compiled.attach_edge(...)
12
  ↓
13
compiled.attach_branch(...)
14
  ↓
15
compiled.validate()

1
def compile(self, ...):
2
    # 1. 校验 graph 结构
3
    self.validate(interrupt=...)
4

5
    # 2. 构造 CompiledStateGraph / Pregel app
6
    compiled = CompiledStateGraph(
7
        builder=self,
8
        channels={
9
            **self.channels,
10
            **self.managed,
11
            START: EphemeralValue(self.input_schema),
12
        },
13
        input_channels=START,
14
        output_channels=resolve_output_channels(...),
15
        stream_channels=resolve_stream_channels(...),
16
        ...
17
    )
18

19
    # 3. 挂 START 和普通节点
20
    compiled.attach_node(START, None)
21
    for key, node in self.nodes.items():
22
        compiled.attach_node(key, node)
23

24
    # 4. 挂固定边
25
    for start, end in self.edges:
26
        compiled.attach_edge(start, end)
27

28
    # 5. 挂条件边
29
    for start, branches in self.branches.items():
30
        for name, branch in branches.items():
31
            compiled.attach_branch(start, name, branch)
32

33
    return compiled.validate()

1
self.branches["classify_intent"]["route_by_intent"]
2
  ↓
3
compile()
4
  ↓
5
compiled.attach_branch("classify_intent", "route_by_intent", BranchSpec)
6
  ↓
7
source node 出口获得 branch writer

1
add_edge:
2
  构建期已经知道 start → end
3

4
add_conditional_edges:
5
  构建期只知道 source → branch，
6
  运行时才知道 branch → destination

1
如果想查图的静态边，只看 edges 不够，还要看 branches。
2
如果画图不准确，通常是 path_map / Literal 信息不足。

1
def attach_branch(
2
    self,
3
    start: str,
4
    name: str,
5
    branch: BranchSpec,
6
    *,
7
    with_reader: bool = True,
8
) -> None:
9
    ...

1
StateGraph.compile
2
  ↓
3
CompiledStateGraph.attach_branch
4
  ├── get_writes
5
  ├── ChannelRead.do_read
6
  ├── branch.run
7
  └── source node writer append

1
def attach_branch(self, start, name, branch, with_reader=True):
2
    # 1. 定义 destinations -> channel writes 的转换函数
3
    def get_writes(packets, static=False):
4
        writes = []
5

6
        for packet in packets:
7
            if isinstance(packet, Send):
8
                writes.append(packet)
9
            else:
10
                if static or packet != END:
11
                    channel = (
12
                        packet
13
                        if packet == END
14
                        else f"branch:to:{packet}"
15
                    )
16
                    writes.append(ChannelWriteEntry(channel, None))
17

18
        return writes
19

20
    # 2. 为 branch 准备 reader：运行时读取最新 state
21
    if with_reader:
22
        schema = branch.input_schema or source_node_input_schema(start)
23
        channels = channels_for_schema(schema)
24
        mapper = mapper_for_schema(schema)
25

26
        reader = partial(
27
            ChannelRead.do_read,
28
            select=channels,
29
            fresh=True,
30
            mapper=mapper,
31
        )
32
    else:
33
        reader = None
34

35
    # 3. 把 branch.run(...) 包成可写 channel 的 Runnable
36
    branch_runnable = branch.run(
37
        writer=get_writes,
38
        reader=reader,
39
    )
40

41
    # 4. 将 branch_runnable 挂到 source node 的 writers
42
    self.nodes[start].writers.append(
43
        ChannelWriteTupleEntry(
44
            mapper=branch_runnable,
45
        )
46
    )

1
branch:to:{target_node}

1
source node 完成
2
  ↓
3
source node writers 被执行
4
  ↓
5
branch.run 包装的 writer 被调用
6
  ↓
7
_route 读取 fresh state 并运行 router
8
  ↓
9
write 写入 branch:to:target

1
1. source node 仍只负责返回 partial update。
2
2. router 永远在 source update 合并之后读取 state。

1
router 可以安全读取 source 节点刚写入的字段。
2
但是 router 不应该返回 state update，因为它的职责是返回 destination。

1
def run(
2
    self,
3
    writer: _Writer,
4
    reader: Callable[[RunnableConfig], Any] | None = None,
5
) -> RunnableCallable:
6
    ...

1
CompiledStateGraph.attach_branch
2
  ↓
3
BranchSpec.run
4
  ↓
5
ChannelWrite.register_writer
6
  ↓
7
RunnableCallable(func=_route, afunc=_aroute)

1
def run(self, writer, reader=None):
2
    # 1. 创建 RunnableCallable，运行时调用 _route/_aroute
3
    branch_callable = RunnableCallable(
4
        func=self._route,
5
        afunc=self._aroute,
6
        writer=writer,
7
        reader=reader,
8
        name=None,
9
        trace=False,
10
    )
11

12
    # 2. 如果 ends 已知，注册静态 writer 信息
13
    if self.ends:
14
        static_writes = writer(
15
            [end for end in self.ends.values()],
16
            static=True,
17
        )
18
        labels = [str(label) for label in self.ends.keys()]
19
        writer_metadata = zip_longest(static_writes, labels)
20
    else:
21
        writer_metadata = None
22

23
    # 3. 返回注册 writer 后的 RunnableCallable
24
    return ChannelWrite.register_writer(
25
        branch_callable,
26
        writer_metadata,
27
    )

1
BranchSpec(path=route_by_intent, ends={...})
2
  ↓
3
run(writer, reader)
4
  ↓
5
RunnableCallable(_route/_aroute)
6
  ↓
7
source node writer 阶段被执行

1
def _route(
2
    self,
3
    input: Any,
4
    config: RunnableConfig,
5
    *,
6
    reader: Callable[[RunnableConfig], Any] | None,
7
    writer: _Writer,
8
) -> Runnable:
9
    ...
10

11
async def _aroute(
12
    self,
13
    input: Any,
14
    config: RunnableConfig,
15
    *,
16
    reader: Callable[[RunnableConfig], Any] | None,
17
    writer: _Writer,
18
) -> Runnable:
19
    ...

1
source node writer
2
  ↓
3
RunnableCallable
4
  ↓
5
BranchSpec._route / _aroute
6
  ↓
7
reader(config)
8
  ↓
9
self.path.invoke / ainvoke
10
  ↓
11
self._finish

1
def _route(self, input, config, *, reader, writer):
2
    # 1. 如果有 reader，优先读取最新 state
3
    if reader is not None:
4
        value = reader(config)
5

6
        # 2. dict state 场景下，透传 input 的额外 key
7
        if (
8
            isinstance(value, dict)
9
            and isinstance(input, dict)
10
            and self.input_schema is None
11
        ):
12
            value = {**input, **value}
13
    else:
14
        # 3. 没有 reader 时，直接用 input 调用 router
15
        value = input
16

17
    # 4. 调用 router path
18
    result = self.path.invoke(value, config)
19

20
    # 5. 处理 router 返回值并写入目标 channel
21
    return self._finish(
22
        writer=writer,
23
        input=input,
24
        result=result,
25
        config=config,
26
    )

1
async def _aroute(self, input, config, *, reader, writer):
2
    if reader is not None:
3
        value = reader(config)
4
        if isinstance(value, dict) and isinstance(input, dict) and self.input_schema is None:
5
            value = {**input, **value}
6
    else:
7
        value = input
8

9
    result = await self.path.ainvoke(value, config)
10
    return self._finish(writer, input, result, config)

1
reader 读取最新 TravelState
2
  ↓
3
route_by_intent.invoke(state)
4
  ↓
5
返回 "budget_node"
6
  ↓
7
_finish 规范化并写入目标 channel

1
router 应该保持纯函数风格：state -> destination。
2
不要在 router 中调用外部副作用操作。
3
复杂判断可以在上一个节点写入结构化字段，再由 router 读取字段。

1
def _finish(
2
    self,
3
    writer: _Writer,
4
    input: Any,
5
    result: Any,
6
    config: RunnableConfig,
7
) -> Runnable | Any:
8
    ...

1
BranchSpec._route / _aroute
2
  ↓
3
BranchSpec._finish
4
  ├── normalize result to list
5
  ├── map labels through self.ends
6
  ├── validate destinations
7
  └── ChannelWrite.do_write / ChannelWrite(entries)

1
def _finish(self, writer, input, result, config):
2
    # 1. router 可以返回单个目标，也可以返回 list/tuple
3
    if not isinstance(result, (list, tuple)):
4
        result = [result]
5

6
    # 2. 如果 ends 存在，先把逻辑 label 映射为真实 destination
7
    if self.ends:
8
        destinations = []
9
        for r in result:
10
            if isinstance(r, Send):
11
                destinations.append(r)
12
            else:
13
                destinations.append(self.ends[r])
14
    else:
15
        destinations = result
16

17
    # 3. 目标合法性校验：不能是 None，不能是 START
18
    for dest in destinations:
19
        if dest is None or dest == START:
20
            raise ValueError(
21
                "Branch did not return a valid destination"
22
            )
23

24
    # 4. Send 不能发送到 END
25
    for dest in destinations:
26
        if isinstance(dest, Send) and dest.node == END:
27
            raise InvalidUpdateError(
28
                "Cannot send a packet to the END node"
29
            )
30

31
    # 5. 把 destinations 转换为 channel writes
32
    entries = writer(destinations, static=False)
33

34
    # 6. 如果没有可写 entry，返回原 input
35
    if not entries:
36
        return input
37

38
    # 7. 如果 entries 需要 passthrough，返回 ChannelWrite 对象
39
    if any(entry.value is PASSTHROUGH for entry in entries):
40
        return ChannelWrite(entries)
41

42
    # 8. 否则直接执行 channel write
43
    ChannelWrite.do_write(config, entries)
44

45
    return input

1
self.ends[r]

1
path_map={"budget": "budget_node"}
2
route() -> "budget"
3
最终 destination = "budget_node"

1
route_by_intent -> "budget_node"
2
  ↓
3
result = ["budget_node"]
4
  ↓
5
destinations = ["budget_node"]
6
  ↓
7
writer(destinations)
8
  ↓
9
ChannelWriteEntry("branch:to:budget_node", None)
10
  ↓
11
Pregel 触发 budget_node