第 4 章 让流程学会决策

本章目标：掌握 add_conditional_edges——让"下一步去哪"由运行时的状态决定。分支、循环、打回重做，都由这一个 API 承担。同时建立两条安全意识：循环必须有出口，出口要有安全阀。

4.1 静态边的局限

到目前为止，我们的图全是"静态边"：add_edge("A", "B") 在编译时就写死了 A 之后必是 B。但真实流程很少这么听话：

简历解析完成后，如果关键字段缺失（比如没识别出技能清单），应该打回重新解析；解析完整了才继续评估。

"下一步去哪"取决于运行时的状态（这次解析得完不完整），而不是编译时的结构。静态边表达不了这件事。

你可能会想：把检查逻辑放进节点，节点内部自己调解析函数不就行了？可以，但这样"重试"就变成了节点内部的私有循环——从图上看不见。图的价值恰恰在于把所有控制流显式化：能看见，才能观测、才能重试单步、才能在持久化时精确恢复。凡是控制流，尽量画进图里，这是使用 LangGraph 的一条心法。

4.2 条件边：add_conditional_edges

条件边把"边"从一个固定连接升级为"路由函数 + 映射表"：

builder.add_conditional_edges(
    "extract",            # 源节点：从谁出发
    route_after_extract,  # 路由函数：读状态，返回"去向标签"
    {                     # 映射表（可选）：标签 -> 目标节点
        "complete": "evaluate",
        "incomplete": "extract",
    },
)

三个参数各自的角色：

• 源节点：条件边挂在哪个节点后面
• 路由函数：签名与节点类似（入参是状态），返回值是"去向标签"——一个字符串，或一个字符串列表（多路并行，4.5 节）
• 映射表：把标签翻译成真正的节点名。省略时，返回值直接就是节点名

配上 END 还能表达"条件结束"。整体效果：extract 每次执行完，引擎调用 route_after_extract，按返回标签查表决定下一个超步执行谁。

4.3 完整 demo：解析不完整就打回重做

下面是可运行的完整例子。extract 负责解析简历；每次解析后由路由函数检查关键字段是否齐备，不齐备就回到 extract 再来一次（演示中模拟"第二次才能解析成功"）：

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

class ScreenState(TypedDict):
    resume_text: str
    candidate: dict          # 解析结果
    parse_attempts: Annotated[int, operator.add]   # 解析次数：累加
    ready: bool              # 是否解析完整
    score: int               # 最终得分

def extract(state: ScreenState) -> dict:
    attempts = state.get("parse_attempts", 0)
    if attempts < 1:
        # 模拟第一次解析缺字段
        return {"parse_attempts": 1, "candidate": {"name": "李思远"}, "ready": False}
    # 第二次成功解析出完整字段
    return {
        "parse_attempts": 1,
        "candidate": {
            "name": "李思远",
            "skills": ["Python", "FastAPI", "PostgreSQL"],
            "years": 6,
        },
        "ready": True,
    }

def evaluate(state: ScreenState) -> dict:
    c = state["candidate"]
    score = min(100, 20 + len(c["skills"]) * 10 + c["years"] * 3)
    return {"score": score}

def route_after_extract(state: ScreenState) -> str:
    return "evaluate" if state["ready"] else "extract"

builder = StateGraph(ScreenState)
builder.add_node("extract", extract)
builder.add_node("evaluate", evaluate)
builder.add_edge(START, "extract")
builder.add_conditional_edges("extract", route_after_extract, {
    "evaluate": "evaluate",
    "extract": "extract",     # 注意：指回自己，形成循环
})
builder.add_edge("evaluate", END)
graph = builder.compile()

result = graph.invoke({"resume_text": "李思远 ..."})
print(result["parse_attempts"])   # 2 —— 解析执行了两轮
print(result["candidate"])        # 完整字段

注意几个细节：

1. 循环的实现没有任何特殊 API——一条指回上游节点的边就是循环。图从不禁止环，终止靠路由函数的逻辑。
2. parse_attempts 用了 Annotated[int, operator.add]：每执行一次 extract 就 +1，天然是"计数器"。
3. 路由函数读的是合并后的最新状态（extract 的补丁已生效），所以它能立即看到 ready 的值。

4.4 循环的安全阀：recursion_limit 与主动计数

有循环就有死循环的风险。假如 extract 永远解析不完整，上面的图就会永远转下去。LangGraph 给了两道防线。

第一道：引擎级安全阀 recursion_limit。引擎限制一次 invoke 最多执行的超步数（可以理解为"图推进的轮次"，第 5 章精确定义），默认 25。超过即抛出 GraphRecursionError，防止进程挂死：

from langgraph.errors import GraphRecursionError

try:
    graph.invoke({"resume_text": "永远解析不出来的简历..."})
except GraphRecursionError:
    print("超过最大轮次，强制终止")

也可以在调用时显式调整：

config = {"recursion_limit": 10}
graph.invoke({"resume_text": "..."}, config)

第二道：业务级出口。安全阀是保险丝，不是业务逻辑。"最多重试几次"属于业务规则，应该自己数：

MAX_PARSE_ATTEMPTS = 3

def route_after_extract(state: ScreenState) -> str:
    if state["ready"]:
        return "evaluate"
    if state["parse_attempts"] >= MAX_PARSE_ATTEMPTS:
        return "reject"          # 业务出口：放弃解析
    return "extract"

def reject(state: ScreenState) -> dict:
    return {"score": -1, "candidate": {"error": "简历无法解析"}}

两个实践原则：

• 依赖业务出口，兜底依赖安全阀。让正常流程永远走业务出口结束，GraphRecursionError 只应在你写出 bug 时出现。
• recursion_limit 调大要谨慎。它同时放大了最坏情况下的 token 消耗与执行时长，先问自己是不是该加业务出口。

4.5 多路分支：路由函数返回列表

路由函数返回单个标签是"单选"，返回列表就是"多选"——多个目标节点会在下一个超步并行执行（并行语义的细节第 5 章展开，这里先建立用法直觉）。

需求：简历解析完成后，技术评估和背景核查同时进行：

def route_after_extract(state: ScreenState) -> list[str]:
    targets = ["tech_review"]
    if state["candidate"].get("company"):   # 有公司信息才做背景核查
        targets.append("background_check")
    return targets

builder.add_conditional_edges("extract", route_after_extract, {
    "tech_review": "tech_review",
    "background_check": "background_check",
})

路由函数返回 ["tech_review", "background_check"] 时，两个节点并行跑；只返回 ["tech_review"] 时就只有一个。扇出数量由运行时状态决定——这是条件边比静态边强大的地方，也是第 7 章 Send（更彻底的动态扇出）的思想前身。

4.6 从 START 分流：入口路由

条件边不仅能挂在普通节点后面，也能直接挂在 START 上。典型用法是入口分发：按请求类型决定从哪个节点开始。

builder.add_conditional_edges(
    START,
    route_by_request,
    {"normal": "extract", "vip": "fast_track"},
)

比如 VIP 候选人走快速通道，普通候选人走完整流程。这让"从哪开始"也成为了状态驱动的一部分，而不是写死的第一节点。

4.7 条件边与 Command 的分界线（预告）

条件边把路由逻辑放在节点外面：节点只产出数据，路由函数独立存在。这保证了节点本身的纯粹，但当路由规则越来越复杂、甚至需要节点"边干活边决定去向"时，代码会开始别扭。

LangGraph 为此准备了另一套机制：节点直接返回 Command 对象，同时携带"状态更新 + 去向指令"。两套机制怎么选、怎么混用，第 6 章专门讨论。现阶段记住一点：简单的、基于结果的分流用条件边；与节点内部逻辑强耦合的跳转，留给 Command。

常见坑与排错

1. 路由函数返回了映射表里没有的标签。比如 route_after_extract 返回了 "evalute"（拼错）。若提供了映射表，未命中的标签会在运行时抛 KeyError 类错误；若没提供映射表（返回值即节点名），拼错会在运行时报找不到节点。对策：把标签定义成模块级常量或 Literal 类型，让类型检查器帮你盯。

2. 条件边和普通出边同时挂在同一个源节点上。这是合法的：静态边的目标一定会执行，条件边的目标按路由结果执行，二者会合并成下一超步的节点集合。但这个组合很容易让人误读流程图。除非明确需要"固定 + 条件"混合扇出，否则一个源节点的出边尽量只用一种方式表达。

3. 把 recursion_limit 当业务重试上限。recursion_limit 数的是超步不是"某个节点的重试次数"，把它当重试上限会在图结构变化（比如插入一个新节点）后悄悄改变语义。重试上限永远用业务字段自己数（如 parse_attempts）。

4. 路由函数里做重活。路由函数每个超步都会被调用，且应该廉价、可重放。在里面调 LLM、查数据库都是反模式——需要"智能路由"时，把决策做成前一个节点的输出（节点里调 LLM 得到标签写入状态，路由函数只读状态），路由函数保持哑的。

5. 循环里忘记累加计数。parse_attempts 若忘了加 reducer（用了默认覆盖语义且每次返回 1），计数永远停在 1，业务出口永远不触发，直到安全阀引爆。循环 + 计数器是固定搭配，计数器字段几乎必然要 operator.add。

6. 条件边指向的节点未注册。映射表的目标节点必须已 add_node 注册，否则编译期报错。这个错倒是好事——编译期暴露优于运行期爆炸。

动手练习

1. 给 4.3 节的图加一个 MAX_PARSE_ATTEMPTS = 3 的业务出口：解析三次仍不完整时走 reject 节点，输出 {"score": -1}。验证：把 extract 改成永远返回 ready=False，确认最终走的是 reject 而不是 GraphRecursionError。

2. 把 4.5 节的多路分支补全成可运行程序：tech_review 与 background_check 各自给 score 累加分数（别忘了 reducer），两个节点都汇合到 summarize。先猜一下两个并行节点是谁先执行、breakdown 顺序如何，再运行验证（答案在第 5 章）。

3. 用 START 条件边实现入口分流：状态里加 is_vip 字段，VIP 走 fast_track（直接打 85 分），非 VIP 走完整流程。思考：这种入口分流和把判断塞进第一个节点内部相比，可观测性上的差别在哪。

本章速查