Day 13 / 共 60 天 · 阶段3 控制流
普通边 vs 条件边
阶段 2 我们把"状态怎么存、怎么合并"讲透了。可一张图光有节点还不会动——边(edge)决定"下一步走谁"。今天开阶段 3 控制流:先分清两种边——写死的普通边 add_edge 和 跑时才决定的条件边 add_conditional_edges,并看它们在 StateGraph 里到底存进了哪个数据结构。
📍 你在 60 天里的位置(阶段3 · 控制流 D13-18)
阶段2 状态收尾→
D13 边→
D14 Branch→
D15 Command→
D16 Send→
D17 递归上限→
D18 START/END
💡 先用一个类比兜住今天(地铁世界观,本阶段沿用)
把图想成地铁网:节点是站台,边是轨道。普通边=固定轨道,A 站发车必到 B 站,写死了、不看乘客;条件边=一个道岔(扳道员),列车到了岔口,扳道员看一眼"车上装的是什么货"(当前 state),临时决定拨去 X 线还是 Y 线。多入边=两条轨道并进一站,要等两趟车都到齐才发下一班(汇合等待)。记住"固定轨道 / 扳道员 / 汇合等待"这三样,今天全通。
L01
图为什么非要有"边"?
🤔 痛点:我有一堆节点函数,谁先跑、跑完去哪?Day 03~12 我们会写节点了,可如果只有节点、没有边,引擎根本不知道执行顺序和分支逻辑。真实 Agent 里到处是"如果工具调用成功就去汇总、失败就重试""置信度不够就打回专家再查一遍"——这些"下一步走谁"的判断,必须有地方描述。
💡 本质:边 = 把节点之间的"跳转关系"声明成数据LangGraph 不让你在节点函数里写
if ...: goto B(那样跳转逻辑散落各处、没法画图、没法校验)。而是让你在建图阶段把跳转关系登记进 builder 的几个集合/字典里,编译时统一编译成 Pregel 的触发通道。这样跳转关系是"可检视的数据",能画流程图、能静态校验"有没有孤儿节点"。LangGraph 提供三类"连线"能力,今天全部覆盖:
| API | 决定下一步的时机 | 典型用途 | 存进哪 |
|---|---|---|---|
add_edge(A, B) | 建图时写死 | A 完了固定去 B | self.edges(集合) |
add_edge([A,B], C) | 写死 + 汇合等待 | A、B 都完了才去 C | self.waiting_edges |
add_conditional_edges(A, fn) | 运行时算 | 看 state 决定去 X/Y/END | self.branches(字典) |
说明:这三者不是互斥的——同一个节点可以既有普通出边、又挂条件边。它们最终都会在
compile() 时统一转成 Pregel 节点的"触发订阅关系"(阶段 4 会拆)。今天先看"建图时它们各自被存到哪、怎么存"。L02
add_edge:最简单的固定轨道
先看普通边。源码在 graph/state.py:915,这里裁掉文档字符串看核心:
# graph/state.py:915
def add_edge(self, start_key: str | list[str], end_key: str) -> Self:
if self.compiled:
logger.warning("Adding an edge to a graph that has already been compiled. ...")
if isinstance(start_key, str): # 单个起点
if start_key == END:
raise ValueError("END cannot be a start node") # ← 边界①
if end_key == START:
raise ValueError("START cannot be an end node") # ← 边界②
# 只对非 StateGraph 生效的一条老校验(这里略)
self.edges.add((start_key, end_key)) # ← 核心:存进 set
return self
# ...(start_key 是 list 的情况见 L03)
self.compiled 警告图一旦 compile() 过,再加边不会生效。它不报错、只 warning——避免你以为改了其实没改。这是"编译快照"式设计:编译那一刻把 builder 冻结成运行时对象。start_key == END边界①:END 是"终点虚拟节点",不能当起点——从终点还往外发车没有意义。end_key == START边界②:START 是"起点虚拟节点",不能当别人的终点——没人能"回到起点之前"。self.edges.add((start,end))核心就一行:把 (起点, 终点) 这个二元组塞进一个 set。edges 的类型在 state.py:201 声明为 set[tuple[str, str]]。📝 真实值
builder.add_edge("retrieve", "generate") 执行后,self.edges = {("retrieve","generate")}。再来一句 add_edge(START,"retrieve"),就变成 {("retrieve","generate"), ("__start__","retrieve")}——注意 START 的真身是字符串 "__start__"(Day 18 细讲)。💐 设计取舍①:边为什么用
set 而不是 list?
因为边是无序且不该重复的关系。用 set:① 重复 add_edge("a","b") 两次自动去重,不会出现两条一模一样的轨道;② 成员判断 O(1),编译时校验"这条边在不在"很快;③ 语义上"边的集合"本就是数学里的集合。代价是丢失插入顺序——但边本来就不该依赖声明顺序执行(顺序由 state 依赖决定,不由你写代码的先后决定),所以这个代价正好不痛。L03
多入边:waiting_edges 与"汇合等待"
add_edge 的第一个参数还能是一个列表——表示"这些节点全部完成后,才执行终点"。看 graph/state.py:956:
# graph/state.py:956
for start in start_key: # start_key 是 list
if start == END:
raise ValueError("END cannot be a start node")
if start not in self.nodes:
raise ValueError(f"Need to add_node `{start}` first") # ← 边界:节点得先存在
if end_key == START:
raise ValueError("START cannot be an end node")
if end_key != END and end_key not in self.nodes:
raise ValueError(f"Need to add_node `{end_key}` first")
self.waiting_edges.add((tuple(start_key), end_key)) # ← 存进另一个 set
start not in self.nodes多入边校验更严:每个起点必须已经 add_node。为什么单起点那版不强制?因为单边常用来接 START(虚拟节点、不在 nodes 里),列表版则明确是"真实节点汇合"。tuple(start_key)把起点列表转成元组再存——因为要放进 set,而 set 的元素必须可哈希,list 不可哈希、tuple 可以。一个很典型的"可哈希化"技巧。self.waiting_edges和普通 edges 分开存:普通边是"或"关系(任一上游好了就能触发下游?不——普通边是一对一),而 waiting_edges 是"与"关系(全部上游完成才触发)。语义不同,所以两个容器。图注:普通边是"点到点直达";多入边是"汇合屏障",等齐所有上游才放行——底层靠 Day 32 的 NamedBarrierValue 通道实现。
🚫 坑:多入边不是"任一到就走",是"全部到才走"新手常把
add_edge(["a","b"], "c") 误当成"a 或 b 完成就去 c"。实际是与(AND)语义:只要 b 这一路没被触发(比如条件边没选中它),c 会永远等下去,表现为"图卡住不动"。要"任一到就走",得用普通边分别连、或用条件边。L04
add_conditional_edges:运行时才决定的扳道员
条件边是控制流的灵魂。源码 graph/state.py:969,核心逻辑其实只有最后几行:
# graph/state.py:969
def add_conditional_edges(self, source: str, path, path_map=None) -> Self:
if self.compiled:
logger.warning("Adding an edge to a graph that has already been compiled. ...")
# find a name for the condition
path = coerce_to_runnable(path, name=None, trace=True) # ① 把普通函数包成 Runnable
name = path.name or "condition" # ② 给这条分支起个名
if name in self.branches[source]: # ③ 同名分支不能重复挂
raise ValueError(
f"Branch with name `{path.name}` already exists for node `{source}`"
)
self.branches[source][name] = BranchSpec.from_path(path, path_map, True) # ④ 存!
if schema := self.branches[source][name].input_schema: # ⑤ 顺手登记推断出的输入 schema
self._add_schema(schema)
return self
coerce_to_runnable(path)你传进来的 path 可能是普通函数、lambda、也可能已是 Runnable。这里统一"强制转成 Runnable",后面就能一视同仁地 .invoke() 它。归一化是框架内部反复出现的手法。name = path.name or "condition"每条分支要有名字,默认取函数名,取不到就叫 "condition"。名字用来防重复和画图标注。if name in self.branches[source]边界:同一个源节点挂两条同名条件边会报错。因为 branches[source] 是以名字为 key 的字典,同名会互相覆盖——干脆早失败。想挂多条,函数得起不同名字。BranchSpec.from_path(path, path_map, True)把"路由函数 + 路径映射表"打包成一个 BranchSpec(Day 14 主角)。第三个参数 True = 要顺便推断输入 schema。self._add_schema(schema)如果路由函数标注了它想读哪种状态,就把这个 schema 也登记进图——保证路由函数拿得到它声明的字段。💡 本质:条件边存的是"一个会算下一步的函数",不是"一条固定线"普通边存的是死数据
("a","b");条件边存的是一段可执行逻辑(path 函数)+ 一张返回值→节点名的翻译表(path_map)。运行到 source 时,引擎调用 path(state),拿返回值去表里查该去哪个节点。这就是"运行时决定"的由来。📝 真实值:一个典型条件边
def route(state) -> str:
return "tools" if state["messages"][-1].tool_calls else "end"
builder.add_conditional_edges("agent", route, {"tools": "tools", "end": END})
执行后 self.branches["agent"]["route"] = 一个 BranchSpec,其 ends = {"tools":"tools", "end":"__end__"}。运行时若模型这轮发起了工具调用,route 返回 "tools",查表得去 tools 节点;否则返回 "end",查表得去 __end__ 结束。L05
branches 存哪 & BranchSpec 长啥样
条件边不像普通边塞进一个扁平 set,而是存进一个两层字典。声明在 graph/state.py:203:
# graph/state.py:201-203
edges: set[tuple[str, str]] # 普通边:扁平集合
nodes: dict[str, StateNodeSpec[Any, ContextT]]
branches: defaultdict[str, dict[str, BranchSpec]] # 条件边:源节点 → {分支名 → BranchSpec}
defaultdict[str, dict]外层 key 是源节点名,value 是"该节点上所有分支"的字典。用 defaultdict 是为了 self.branches[source][name]=... 时,即使 source 从没出现过也不用先手动建空字典——第一次访问自动生成,代码干净。内层 dict[str, BranchSpec]一个源节点可以挂多条条件边(不同名字),所以内层还是字典。再看被存进去的 BranchSpec 本体,它是个 NamedTuple,graph/_branch.py:83:
# graph/_branch.py:83
class BranchSpec(NamedTuple):
path: Runnable[Any, Hashable | list[Hashable]] # 路由函数(已归一成 Runnable)
ends: dict[Hashable, str] | None # 返回值 → 目标节点名 的翻译表
input_schema: type[Any] | None = None # 路由函数想读的状态 schema
path那段"算下一步走谁"的可执行逻辑。ends翻译表。你传的 path_map={"tools":"tools","end":END} 会被规整进这里;不传的话它可能是 None(那时 path 必须直接返回真实节点名,见 Day 14)。input_schema从路由函数的类型标注推断出来的输入类型,可空。它让路由函数能声明"我只关心状态里这几个字段"。图注:建图阶段三种边各自落进独立容器;真正"变成能跑的跳转"是在
compile()(阶段 4 Pregel)。💐 设计取舍②:为什么条件边要独立成 branches,而不塞进 edges?
因为二者信息量根本不同。普通边只需存"从哪到哪"两个字符串;条件边要存一个函数 + 一张映射表 + 一个 schema,是个复合对象。硬塞进
set[tuple] 表达不了。分开存还有个好处:编译和画图时能分别处理——普通边画实线箭头,条件边画虚线菱形岔口。用一个统一容器反而会让两种截然不同的语义纠缠不清。L06
三种边的控制流全景对比
把今天三种连线放一起,从"何时决定、几个下游、等不等"三个维度对照:
| 维度 | add_edge 单 | add_edge 列表 | add_conditional_edges |
|---|---|---|---|
| 下一步谁决定 | 建图写死 | 建图写死 | 运行时函数算 |
| 下游数量 | 1 个 | 1 个 | 1 或多个(可扇出) |
| 是否等待 | 否 | 是,等齐全部上游 | 否(各自触发) |
| 能去 END 吗 | 能 | 能 | 能(返回值映射到 END) |
| 存储容器 | edges | waiting_edges | branches |
👶 小白:既然条件边最灵活,那我全用条件边不就行了?
👨🏫 老师:不建议。① 可读性:固定顺序用普通边,一眼看懂"A→B→C";全写成条件边,读者得去读每个路由函数才知道流程,图退化成"随便跳"。② 可视化:普通边能画确定的实线;条件边没有类型标注时(Day 14 会讲),画图工具只能画"可能去任意节点"的乱麻。③ 校验:普通边能静态查孤儿节点,条件边的目标是运行时才知道的。原则:能写死就写死,只在真需要"看数据分流"时才用条件边。
📝 真实值:一个 ReAct 图同时用到三种
builder.add_edge(START, "agent") # 固定入口(普通边)
builder.add_conditional_edges("agent", route, { # 看有没有工具调用(条件边)
"tools": "tools", "end": END,
})
builder.add_edge("tools", "agent") # 工具跑完固定回 agent(普通边)
这就是最经典的 ReAct 循环骨架:固定进 agent;agent 用条件边决定"还要调工具 or 收工";工具跑完固定回 agent 再想。循环正是靠 tools→agent 这条回边形成的(Day 17 会讲怎么防它转成死循环)。L07
今日小结 + 动手 + 明日预告
🧠 今天你应该能回答
- 为什么跳转逻辑不写在节点里,而要登记成"边数据"?(可检视、可画图、可静态校验)
- 三种边分别存进哪个容器?(
edges/waiting_edges/branches) - 普通边为什么用
set?(无序、去重、O(1) 判断,且边不该依赖声明顺序) - 多入边是"或"还是"与"语义?(与:等齐所有上游才发;少一路就永远卡住)
add_conditional_edges核心做了什么?(归一化 path→BranchSpec,按branches[源][名]存)- BranchSpec 三个字段?(
path路由函数 /ends翻译表 /input_schema)
✋ 10 分钟动手
# 1. 读 add_edge 与 add_conditional_edges 全文
sed -n '915,1017p' libs/langgraph/langgraph/graph/state.py
# 2. 看三个容器的类型声明
sed -n '200,210p' libs/langgraph/langgraph/graph/state.py
# 3. 看 BranchSpec 定义
sed -n '83,120p' libs/langgraph/langgraph/graph/_branch.py
# 4. 亲手建个带三种边的小图,打印它的容器
python - <<'PY'
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class S(TypedDict): x: int
g = StateGraph(S)
g.add_node("a", lambda s: s); g.add_node("b", lambda s: s)
g.add_edge(START, "a")
g.add_conditional_edges("a", lambda s: "b" if s["x"]>0 else "end", {"b":"b","end":END})
print("edges:", g.edges)
print("branches:", dict(g.branches))
PY
明天预告 · Day 14:今天我们只看到条件边被存成
BranchSpec,还没看它运行时到底怎么把返回值变成"写哪个通道"。Day 14 钻进 _branch.py 的 run / _route / _finish,逐行看扳道员如何工作、如何做类型推断、如何处理"返回列表就一次扇出多个下游"。