2.3 边 (Edge) 的类型与控制流

2.3 边 (Edge) 的类型与控制流

边决定图往哪走

本节在全册位置：控制流是图的灵魂。普通边表达固定顺序；条件边用路由函数按状态选下一个节点；START/END 是出入口；Send 用于把一个状态分发给多个并行节点。没有边，图只是一堆孤立节点。我们见过最隐蔽的 bug，是把分支逻辑写进节点内部 if，而图上还是一条直线，结果完全不可视、不可调。

类型：固定、条件与分发

固定边 add_edge(u,v) 永远从 u 到 v。条件边 add_conditional_edges(u, route) 中 route 返回节点名或 END，也可返回列表实现一对多。Send 则把一个「任务对象」发给指定节点，常用于扇出。路由函数必须是纯函数，依据状态判断。类比交通：固定边是单行道，条件边是红绿灯，Send 是分流匝道。控制流写在哪里，决定了你日后能不能改。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class S(TypedDict):
    score: int

def judge(s: S):
    # 条件边路由：返回下一个节点名
    return "pass" if s["score"] >= 60 else "retry"

def passed(s: S):
    return {"score": s["score"]}
def retry(s: S):
    return {"score": s["score"] + 1}

b = StateGraph(S)
b.add_node("passed", passed)
b.add_node("retry", retry)
b.add_edge(START, "judge") if False else None
b.add_node("judge", lambda s: {})
b.add_conditional_edges("judge", judge, {"pass": "passed", "retry": "retry"})
b.add_edge("passed", END)
b.add_edge("retry", "judge")
g = b.compile()
print(g.invoke({"score": 55}))
​

#  2.3 边 (Edge) 的类型与控制流
​

案例：评分重试环

背景：模型生成内容需达到质量线，不达标就重写，轮数不固定。

操作：judge 节点只评分，条件边按分数回到 retry 或去 passed。

结果：自动重试直到达标，最多由另一个条件限制轮数。

解读：控制流与业务逻辑分离：judge 纯判断，retry 纯生成，各节点职责单一。

变式：在 retry 前插入中断（interrupt）即可让人来定是否继续，控制流不用改，只多一个节点。

工程清单

• 固定边永远从 u 到 v，条件边按路由函数选下一个节点，Send 用于一对多扇出。

• 路由函数必须是纯函数，只依据状态判断，在里面调模型会破坏静态分析与可调试性。

• 控制流写在哪里决定你日后能不能改，把分支藏进节点内部 if 等于放弃了图的可视化。

常见误区

在条件边里调用模型做路由，破坏纯函数约定，导致图不可静态分析、难调试；路由要读已算好的状态字段，模型判断留在节点内。

条件边的三种 path 写法

add_conditional_edges 的第三个参数控制「路由返回值和去向」的对应方式，三种写法适用场景不同。

# 写法一：字典映射，返回键必须都在 dict 里，编译期就能校验拼写
b.add_conditional_edges("judge", judge, {"pass": "passed", "retry": "retry"})

# 写法二：列表，路由函数直接返回目标节点名，框架校验返回值在列表中
b.add_conditional_edges("judge", judge, ["passed", "retry"])

# 写法三：不传第三个参数，路由函数返回值就是节点名或 END
b.add_conditional_edges("judge", judge)
​

三种写法对应三种契约强度：字典映射最严，返回值必须精确命中键，miss 会抛 KeyError；列表次之，要求返回值在列表内，适合去向固定的多分支；不传映射最松，路由函数直接返回节点名，适合把路由逻辑完全交给函数、且返回节点集动态的场景。生产里建议优先用字典映射——它的校验最早，能在编译期把「路由值拼错」拦下来。

路由键匹配的排查清单

条件边「不生效」是出现频率最高的控制流问题，八成出在路由值本身：

1. 返回值大小写与映射键不一致，Python 字符串匹配严格区分大小写。
2. 返回值带了多余空格或换行，打印出来肉眼看不出来。
3. 映射键与节点名只差一个字符，例如 retry 与 rety。
4. 路由函数返回了 None，而映射里没有 None 这个键。
5. 条件边挂在错误的节点上，路由函数读的状态字段被别的节点覆盖。

排查时先在路由函数里打印返回值，对照映射键逐字符比对，比盯着图结构猜快得多。

分支汇合与扇出的取舍

图里表达「多个分支聚到一处」用普通边即可——所有分支都 add_edge 到同一个汇合节点，汇合节点在所有入边来源完成后才执行。这与 Send 的扇出语义互补：

汇合节点有一个常见陷阱：它会在「每条入边到达后」被调度一次，如果多条边到达同一轮，框架会自动合并成一次执行，但前提是这些入边确实在同一超级步进内就绪。若你发现汇合节点被重复执行，多半是有节点在循环中多次触发了入边——把入边改成条件边，或调整结构让汇合只在主路径上触发。

普通边与条件边并存的优先级

同一节点上同时画了普通出边和条件边时，执行顺序遵循两条规则：普通边只在编译时固定，条件边在运行时求值；两者都指向的节点，条件边优先决定去向。这意味着「想用条件边覆盖默认路径」时，直接把默认走向写进条件边即可，不必删普通边——但删除会更清晰。

b.add_edge("judge", "fallback")            # 默认兜底
b.add_conditional_edges("judge", judge,
    {"pass": "passed", "retry": "retry"})  # 运行时优先
# 命中条件键走对应分支，否则走 fallback
​

这条规则解释了为什么「我加了条件边但没生效」往往是把条件边挂在错误节点上，或条件边返回值没命中任何键——此时执行器会回落到普通边，看起来就像条件边不存在。定位时先确认返回值在映射中，再看是否被普通边兜了底。