建图时 add_conditional_edges 调了 BranchSpec.from_path。它最烧脑的一段，是努力搞清楚"这条分支到底可能去哪几个节点"（ends）。_branch.py:88：

# graph/_branch.py:88
@classmethod
def from_path(cls, path, path_map, infer_schema=False) -> BranchSpec:
    path_map_: dict[Hashable, str] | None = None
    try:
        if isinstance(path_map, dict):
            path_map_ = path_map.copy()                       # ① 你直接给了字典 → 拷一份
        elif isinstance(path_map, list):
            path_map_ = {name: name for name in path_map}     # ② 你给了列表 → 名字既当 key 又当值
        else:
            # ③ 啥都没给：尝试从函数返回类型标注里"猜"
            func = None
            if isinstance(path, (RunnableCallable, RunnableLambda)):
                func = path.func or path.afunc
            if func is not None:
                if (cal := getattr(path, "__call__", None)) and ismethod(cal):
                    func = cal
                if rtn_type := get_type_hints(func).get("return"):
                    if get_origin(rtn_type) is Literal:                       # ← 关键
                        path_map_ = {name: name for name in get_args(rtn_type)}
    except Exception:
        pass
    input_schema = _get_branch_path_input_schema(path) if infer_schema else None
    return cls(path=path, ends=path_map_, input_schema=input_schema)

编译时调 run，把 BranchSpec 变成一个真正能塞进图去执行的 RunnableCallable。_branch.py:122：

# graph/_branch.py:122
def run(self, writer, reader=None) -> RunnableCallable:
    return ChannelWrite.register_writer(
        RunnableCallable(
            func=self._route,          # 同步路由
            afunc=self._aroute,        # 异步路由
            writer=writer,             # ← 注入"怎么把目标变成通道写入"
            reader=reader,             # ← 注入"怎么读当前状态"
            name=None, trace=False,
        ),
        list(zip_longest(
            writer([e for e in self.ends.values()], True),
            [str(la) for la, e in self.ends.items()],
        )) if self.ends else None,     # ← 预先登记"我可能写哪些通道"（给画图/校验用）
    )

运行时，源节点一跑完，这个包好的节点被触发，执行 _route。_branch.py:146：

# graph/_branch.py:146
def _route(self, input, config, *, reader, writer) -> Runnable:
    if reader:
        value = reader(config)                       # ① 用 reader 读出当前状态
        # 把节点输出的额外字段"透传"给路由函数（仅 dict 状态、且没指定 input_schema 时）
        if (isinstance(value, dict) and isinstance(input, dict)
                and self.input_schema is None):
            value = {**input, **value}               # ② 合并：节点刚写的 + 通道里的
    else:
        value = input
    result = self.path.invoke(value, config)         # ③ 真正调用你的路由函数！
    return self._finish(writer, input, result, config)

整条 Branch 的高潮在 _finish——它把路由函数的返回值变成"往哪些通道写"。_branch.py:192：

# graph/_branch.py:192
def _finish(self, writer, input, result, config):
    if not isinstance(result, (list, tuple)):
        result = [result]                              # ① 单值也统一成列表 → 支持"扇出多个"
    if self.ends:
        destinations = [                               # ② 有翻译表：查表
            r if isinstance(r, Send) else self.ends[r] for r in result
        ]
    else:
        destinations = result                          # ② 无表：返回值本身就是节点名
    if any(dest is None or dest == START for dest in destinations):
        raise ValueError("Branch did not return a valid destination")   # ③ 边界
    if any(p.node == END for p in destinations if isinstance(p, Send)):
        raise InvalidUpdateError("Cannot send a packet to the END node") # ④ 边界
    entries = writer(destinations, False)              # ⑤ 目标 → 通道写入项
    if not entries:
        return input
    # ...（PASSTHROUGH 处理，见下）
    ChannelWrite.do_write(config, entries)             # ⑥ 真正落笔写通道
    return input

Day 13 我们看到 from_path 第三参 infer_schema=True 时会推断输入 schema。它靠 _get_branch_path_input_schema（_branch.py:41）读函数第一个参数的类型标注：

# graph/_branch.py:41（裁剪核心）
def _get_branch_path_input_schema(path):
    input = None
    try:
        callable_ = None
        if isinstance(path, (RunnableCallable, RunnableLambda)):
            if isfunction(path.func) or ismethod(path.func):
                callable_ = path.func            # 找到真正的底层函数
            # ...（还会尝试 __call__、afunc 等多种情况）
        elif callable(path):
            callable_ = path
        if callable_ is not None and (hints := get_type_hints(callable_)):
            first_parameter_name = next(
                iter(signature(callable_).parameters.keys())   # 第一个参数名
            )
            if input_hint := hints.get(first_parameter_name):
                if isinstance(input_hint, type) and get_type_hints(input_hint):
                    input = input_hint           # 它自己也是个 TypedDict/类 → 采纳
    except (TypeError, StopIteration):
        pass
    return input

✋ 10 分钟动手

# 1. 通读 _branch.py（才 226 行，值得整读一遍）
sed -n '83,226p' libs/langgraph/langgraph/graph/_branch.py

# 2. 观察"返回列表 = 扇出"：让路由函数返回两个目标
python - <<'PY'
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
import operator
class S(TypedDict): out: Annotated[list, operator.add]
g = StateGraph(S)
g.add_node("a", lambda s: {"out": ["A"]})
g.add_node("b", lambda s: {"out": ["B"]})
g.add_conditional_edges(START, lambda s: ["a","b"], ["a","b"])  # 一次去两个！
g.add_edge("a", END); g.add_edge("b", END)
print(g.compile().invoke({"out": []}))   # {'out': ['A','B']}
PY

# 3. 故意返回没登记的目标，看边界报错
#    add_conditional_edges(START, lambda s:"zzz", {"a":"a"}) → 运行时 ValueError