LangGraph 条件边与路由

第 3 篇的多轮翻译实战收尾时，route 函数已经亮过一次相：审校节点跑完，它读一眼黑板上的 done，决定把流程送回 draft 回炉，还是直达 END。当时一句带过，本篇把它背后的机制正面讲透——条件边就是图上的 if，是 LangGraph 全部控制流的灵魂，分支、回环、提前收束全由它承载。

版本口径与全系列一致：langgraph 1.2.x，PyPI 独立发布，示例只依赖 langgraph 一个包，条件边的规则以Graph API 文档的 Edges 一节为准。全文按"固定边的局限 → 基本用法 → 路径映射 → 多路分发 → 设计准则 → 与 reducer 联动 → 客服分流器实战 → 总结"展开。

1、固定边的局限

第 2 篇的五步搭图里，流程全靠 add_edge 串起：从 START 到写提纲，再到写成稿，一条边一个脚印。这种普通边是固定转移——下一站在建图时就已写死，源节点跑完必去，不管黑板上写的是什么。流水线业务这样最省事；流程一旦需要"看情况"，固定边立刻露怯。

三类典型诉求：

• 分支：稿件达标走发布，不达标回炉重写
• 回环：信息不够，退回去补一轮检索再来
• 提前收束：前置校验失败，直接带着错误出图

用固定边硬拗这些形状只有两条歪路。要么把 if 塞进节点函数体里手工调用下一步，控制流又缩回代码里，draw_mermaid 打印的拓扑与真实路径对不上；要么为每种组合各建中转节点，节点数随分支数爆炸。

LangGraph 给的是第三条路：条件边。给源节点挂一个路由函数，框架在该节点跑完、增量合入黑板之后调用它，按当前 State 现场决定下一站。两种边对比如下：

一句话立住定位：普通边表达"接下来一定是它"，条件边表达"接下来看情况"。

2、第一个条件边

沿用写作流水线的语境，把流程升级成"写作—审校—按判决处置"：初稿交审校，通过则直达终点，不通过回炉修订后再审。State、两个节点与路由函数如下：

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

class State(TypedDict):
    topic: str
    article: str
    passed: bool

def write(state: State) -> dict:
    if state.get("article"):                      # 回炉：在原稿上修订
        return {"article": state["article"] + "（按意见修订）"}
    return {"article": f"围绕[{state['topic']}]的初稿"}

def review(state: State) -> dict:
    return {"passed": "修订" in state["article"]}  # 演示口径：修订过才放行

def after_review(state: State) -> str:
    return END if state["passed"] else "write"    # 通过收工，否则回炉

解读：after_review 就是路由函数——入参是当前 State，返回值是下一站的节点名。它不干活、不返回增量，只回答"去哪"；通过时返回 END 常量（值为字符串 "__end__"，与 START 同在 langgraph.graph 导入），图到此收束。建图只比第 2 篇多一行：

builder = StateGraph(State)
builder.add_node("write", write)
builder.add_node("review", review)
builder.add_edge(START, "write")
builder.add_edge("write", "review")                    # 固定边：写完必审
builder.add_conditional_edges("review", after_review)  # 条件边：审完看判决
print(builder.compile().invoke({"topic": "条件边"})["article"])

输出如下：

围绕[条件边]的初稿（按意见修订）

走读执行轨迹：write 产出初稿，review 发现稿中没有"修订"字样判不通过，after_review 返回 write，流程沿回边回到写作节点；write 在原稿上补修订标记，二审放行，after_review 返回 END。首次上手记三件事：去向不必逐条 add_edge 声明，路由返回什么框架就找同名节点；返回未注册的名字会在运行期报错；条件写反造成死循环时，第 2 篇的 GraphRecursionError 会兜底。第 1 篇 Agent 循环里判断"还有没有 tool call"的 route，用的正是同一机制。

3、显式路径映射

第 2 节的路由函数直接返回节点名，函数与图拓扑焊死在一起——节点一改名，路由得跟着改。更讲究的写法是让它返回业务标签，翻译交给 add_conditional_edges 的第三个参数 path_map：一个字典，键是路由返回值，值是真正的目标节点。以工单分级为例，进件节点写好等级后分流：

def by_level(state: State) -> str:
    return "vip" if state["level"] >= 5 else "normal"

builder.add_conditional_edges(
    "intake", by_level,
    {"vip": "expert_reply", "normal": "auto_reply"},  # 标签到节点的翻译表
)

解读：路由函数只认业务语义，布线全部集中在 path_map——节点改名、调整去向动一行字典即可，路由一行不改；布尔路由同样常见，官方文档示例正是把 True 与 False 分别映射到两个节点。它还送一份防御红利：路由返回的标签若不在映射表里，运行期立刻报错，配置遗漏当场暴露，而不是静默走错路。完整签名为 add_conditional_edges(source, path, path_map, then)，逐参数说明见 add_conditional_edges API 参考；then 参数可为这批分支统一指定跑完之后的下一站，是扇入的速记写法。

4、多路分发与汇入

路由函数不必只返回一个名字——返回节点名列表，一次判断能同时点亮多条支路。官方规则说得很直白：列表里的目标节点将作为下一个超步并行执行。以风控审核为例，高风险账户双线处置，人工介入与自动冻结同时进行；低风险单线归档：

def after_check(state: State) -> list[str]:
    if state["risk"] == "high":
        return ["alert_human", "freeze_account"]   # 高风险：双线并行
    return ["auto_note"]                           # 低风险：单线归档

builder.add_conditional_edges("check", after_check)
for name in ("alert_human", "freeze_account", "auto_note"):
    builder.add_edge(name, END)                    # 三条支路全部汇入终点

拓扑画出来是这样：

解读：一次判断、两条箭头同时激活，alert_human 与 freeze_account 在同一超步并行执行，两条都结束图才继续，最后全部汇入 END——这是扇出扇入的最小完整形态。分支节点本身依旧是普通节点，还可以继续挂自己的条件边，嵌套分支不过是条件边套条件边，结构在图上始终一目了然。两个衔接点。其一，并行分支写同一个没有 reducer 的字段会撞上第 3 篇的 InvalidUpdateError，各写各的键或挂合并函数即可化解。其二，条件边也能挂在 START 上做条件入口，图一开始就按输入分路；至于运行前不知开几路、每路还要各带一份 State 的 map-reduce 场景，路由函数可返回 Send 对象，第 10 篇多 Agent 组网再展开。

5、路由函数设计准则

条件边好不好用，九成看路由函数干不干净。四条准则，按踩坑频率排。

其一，只读不写，保持纯函数。路由函数读 State、返回去向，此外什么都不做：不改字段、不发请求、不写日志。苛刻的原因是两个收益都值钱。可恢复性：第 7 篇的持久化会在超步边界打快照、必要时重放，带副作用的路由会让重放不可复现。可测试性：纯函数一行就能断言，不需要建图：

assert after_review({"topic": "t", "article": "初稿",
                     "passed": False}) == "write"
assert after_review({"topic": "t", "article": "初稿（修订）",
                     "passed": True}) == END

解读：两行断言就是路由逻辑的全部测试，图与函数各自独立验证。真需要"边改状态边路由"，官方另备 Command 原语——一个返回值同时携带增量与去向，后续实战篇用到再展开。

其二，返回值必须是注册过的节点名或 END。返回不存在的名字运行期直接报错，这是特性不是缺陷：拼错的节点名当场暴露。

其三，同一节点的出边别混用。官方明确警告：固定边与条件边挂在同一节点上时，两条路径都会执行，图的行为将难以推理。要么固定、要么动态，二选一。

其四，判断放对位置。"去哪"的判断属于路由函数，"干什么"的判断属于节点；把业务逻辑堆进路由，它就成了图上看不见的隐形节点。

节点回答"做什么"，路由回答"去哪"。两件事分开，图才看得懂、测得动。

6、条件分支与 reducer 联动

第 3 篇讲并发写冲突时留了两条路：给字段挂 reducer 汇合，或让各分支写不同的键。后者与条件边天然般配——分支各写各的字段互不干扰，下游节点按需取用。看一张修稿分诊小图：先给稿件打分（演示口径按长度），过线走轻润色，不过线走重写，最后统一装订：

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

class State(TypedDict):
    text: str
    score: int
    light: str        # 轻润色分支专属
    heavy: str        # 重写分支专属
    final: str

def grade(state: State) -> dict:
    return {"score": len(state["text"])}           # 演示口径：按长度打分

def route(state: State) -> str:
    return "light_polish" if state["score"] >= 10 else "heavy_rewrite"

def light_polish(state: State) -> dict:
    return {"light": state["text"] + "（已润色）"}

def heavy_rewrite(state: State) -> dict:
    return {"heavy": state["text"] + "（已重写）"}

def assemble(state: State) -> dict:
    return {"final": state.get("light") or state.get("heavy")}

建图连边与运行如下：

builder = StateGraph(State)
builder.add_node("grade", grade)
builder.add_node("light_polish", light_polish)
builder.add_node("heavy_rewrite", heavy_rewrite)
builder.add_node("assemble", assemble)
builder.add_edge(START, "grade")
builder.add_conditional_edges("grade", route)    # 打完分当场分流
builder.add_edge("light_polish", "assemble")     # 两分支殊途同归
builder.add_edge("heavy_rewrite", "assemble")
builder.add_edge("assemble", END)
print(builder.compile().invoke({"text": "短句"})["final"])

输出为 短句（已重写）。三个要点：路由读的 score 由上游 grade 写入，"上游产判断依据、路由做裁判"是条件边最常见的配套；两分支各写各的键，assemble 用 get 判空、各取所需，黑板上还能追溯走了哪条路；本例分支互斥，其实写同一个键也不会撞——写冲突只发生在并行扇出，互斥分流与并行汇合的合并策略要分开设计。这是第 3 篇与本篇的合流点：并行汇合交给 reducer，互斥分流用不同的键。

7、客服分流器实战

把前几节合起来做一个够用的真东西：客服分流器。进线先分类，退款类走退款专席，技术类走技术支持，其余走通用答复；分类用关键词规则演示，骨架与真模型完全同构。State 与全部节点如下：

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

class State(TypedDict):
    question: str
    category: str
    answer: str

KEYWORDS = {"退款": "refund", "账号": "tech", "报错": "tech"}

def classify(state: State) -> dict:
    for word, label in KEYWORDS.items():
        if word in state["question"]:
            return {"category": label}
    return {"category": "general"}                # 兜底：认不出的走通用

def refund_desk(state: State) -> dict:
    return {"answer": f"退款已受理：{state['question']}"}

def tech_support(state: State) -> dict:
    return {"answer": f"技术排查中：{state['question']}"}

def general_desk(state: State) -> dict:
    return {"answer": f"常见问题答复：{state['question']}"}

路由与建图如下，path_map 让布线一目了然：

def route(state: State) -> str:
    return state["category"]                      # 路由只认业务标签

builder = StateGraph(State)
builder.add_node("classify", classify)
for name, fn in [("refund_desk", refund_desk),
                 ("tech_support", tech_support),
                 ("general_desk", general_desk)]:
    builder.add_node(name, fn)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, {
    "refund": "refund_desk", "tech": "tech_support",
    "general": "general_desk"})                   # 标签到节点的翻译表
for name in ("refund_desk", "tech_support", "general_desk"):
    builder.add_edge(name, END)                   # 三席全部汇入终点
graph = builder.compile()

for q in ("我要退款", "登录报错怎么办", "你们几点上班"):
    print(graph.invoke({"question": q})["answer"])

输出如下：

退款已受理：我要退款
技术排查中：登录报错怎么办
常见问题答复：你们几点上班

解读：classify 产标签、route 传标签、path_map 翻译标签，三段各司其职；三个处理节点都写 answer 键，而每轮 invoke 只有一个分支执行，互斥路径写同键天然安全，印证第 6 节的分界。想确认某条进线实际走了哪条路，用第 2 篇的 stream 配 updates 模式回放，每个片段的键就是实际执行的节点名。升级成真分流器只动一处：把 classify 换成一次模型调用，让它结构化输出 refund、tech、general 三选一，图结构一字不改——条件边把"怎么分流"与"谁来分流"彻底解耦。

8、总结

条件边给图装上了 if：add_conditional_edges(源, 路由函数) 两参数起步，路由读 State 返回节点名，返回 END 收束、返回列表扇出并行；path_map 把业务标签翻译成节点名，拓扑集中一处可维护；纪律三条——路由函数只读不写的纯函数、返回值必须是注册名或 END、同一节点固定边与动态边不混用；合并策略分辖区，互斥分流各写各的键，并行汇合交给 reducer。

练习一道：给客服分流器加第四条"投诉"路径——KEYWORDS 加一个关键词映射到 complaint，path_map 加一条指向新建的 complaint_desk，三处改动跑通新链路。进一步，让投诉命中时返回 ["complaint_desk", "refund_desk"] 双线并行，注意让两个节点写不同的键或给 answer 挂 reducer，正好检验第 3 篇与第 4 节的功课。对于准备继续深入的读者，建议下一站进第 5 篇——create_agent 把本篇的路由机制预制成官方 Agent 图，模型节点与工具循环即将登场。