10.3 条件路由与动态决策

上一节的图是直来直去的。真实业务里，岔路口要多得多：按用户意图分成三四个分支，或者让模型判断“这份回答够不够好，要不要重写”。

官方把职责拆得很干净：节点负责干活，边负责决定下一步去哪。 条件路由，就是那张最关键的指路牌。

---

1. 本质：一个函数 + 一张路径表

条件路由只有两个要素：

1. 路由函数：读当前 State，返回一个“逻辑地名”；
2. 路径表（可选）：把逻辑地名翻译成图里真实存在的节点名。

from langgraph.graph import StateGraph, START, END

def route_logic(state: State) -> str:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "call_tool"
    return "finish"

builder.add_conditional_edges(
    "assistant",
    route_logic,
    {
        "call_tool": "tools",
        "finish": END,
    },
)

路径表像一道过滤器。模型常会吐出各种措辞，路径表把不稳定的字符串映射到稳定的节点名。省略路径表时，LangGraph 会把返回值直接当节点名去查找。

给返回值加 Literal，能让 IDE 和 mypy 在写错节点名时当场标红。它只作用于静态检查，运行时仍以真正返回的字符串为准：

from typing import Literal

def route_logic(state: State) -> Literal["search_node", "answer_node"]:
    if state.get("needs_search"):
        return "search_node"
    return "answer_node"

---

2. 三岔路口：先分类，再分发

最常见的用法是意图分流。例如一个内容处理工作流：先判断用户想审稿、翻译、摘要还是闲聊，再走进对应节点。

这很像医院分诊，或客服系统里的一级分类：先打标签，再进专员。没有这一步，所有请求都挤在同一个模型节点里，Prompt 会迅速膨胀，工具也更容易用错。

from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    input: str
    category: str

llm = init_chat_model("openai:gpt-4o-mini", temperature=0)

def classify(state: State):
    """先把意图写进 State，路由函数稍后只负责读字段"""
    response = llm.invoke(
        f"请判断这句话的意图，只能回答 review / translate / summarize / chat 之一：{state['input']}"
    )
    return {"category": response.content.strip().lower()}

def handle(state: State):
    return {"input": f"[{state['category']}] 已处理：{state['input']}"}

builder = StateGraph(State)
builder.add_node("classify", classify)
for name in ["review", "translate", "summarize", "chat"]:
    builder.add_node(f"{name}_node", handle)
    builder.add_edge(f"{name}_node", END)

builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    lambda state: state["category"],
    {
        "review": "review_node",
        "translate": "translate_node",
        "summarize": "summarize_node",
        "chat": "chat_node",
    },
)

graph = builder.compile()
print(graph.invoke({"input": "帮我把这段话翻成英文：你好，世界"}))

分支超过三条时，官方和社区都推荐同一条纪律：决策与路由分离。普通节点负责算出 category 并写入 State，路由函数只读字段、返回节点名。决策逻辑可以单独单测，流式输出里也能直接看到分类字段。

---

3. 让模型当裁判，而不是当交通警

有些规则写死在 Python 里会太僵硬，比如“这段回答有没有真正解决问题”。这时让模型输出一个被约束过的小结构，路由函数再读字段：

from pydantic import BaseModel, Field
from typing import Literal

class GradeAnswer(BaseModel):
    binary_score: Literal["yes", "no"] = Field(
        description="已经解决问题填 yes，否则填 no"
    )

grader = llm.with_structured_output(GradeAnswer)

def decide_retry(state: State) -> str:
    result = grader.invoke(state["messages"])
    return "rewrite" if result.binary_score == "no" else END

这就是一棵最小决策树：不合格重写，合格结束。循环必须另加保险丝——recursion_limit，或在 State 里记 revision 次数，超过上限强制放行。09 节的 Evaluator-Optimizer 就是这个模式的完整版。

---

4. 一条条件边可以指向多个目的地

路由函数除了返回单个字符串，还可以返回字符串列表。LangGraph 会在下一个超步里同时激活这些节点：

def fan_out_route(state: State) -> list[str]:
    return ["search_flights", "search_hotels"]

builder.add_conditional_edges(
    "planner",
    fan_out_route,
    ["search_flights", "search_hotels"],
)

条件路由负责“分岔”，并行节点负责“同时跑”。数量运行时才知道时，要用下一节的 Send。

---

5. Command：改状态和跳转一次完成

LangGraph 1.x 的 Command 让节点在返回更新的同时声明下一步。它适合“算完就知道该去哪”，或从子图跳回父图：

from typing import Literal
from langgraph.types import Command

def smart_node(state: State) -> Command[Literal["next_node"]]:
    decision = decide_next(state)
    return Command(goto="next_node", update={"last_decision": decision})

怎么选：

• 能用 add_conditional_edges 说清的，优先用条件边——图画出来更直观；
• 节点必须“改状态 + 跳转”一起做，或 goto 要指向父图 / 子图里的节点，再用 Command；
• 恢复中断时，调用方传入的是 Command(resume=...)（13 节）。不要把 Command(update=...) 当成下一次用户输入。

同一节点不要既连固定边、又返回 Command，两条路可能同时走。

---

6. 路由也需要失败路径

条件边常见的事故不是“不会分支”，而是模型给出未知标签、状态缺字段、或几条规则同时命中。路由函数应先做校验，并为未知结果安排出口：回到澄清节点，或交给人工，而不是随便选一条“看起来接近”的路。

测试至少覆盖：正常分支、边界输入、未知分类、缺失字段。还要确认每条支路最终能结束或回到预期节点，避免某条路进了没有出口的循环。

其他高频坑：

• 返回了不存在的节点 → 运行时报 ValueError（有时包成 GraphRuntimeError）；
• 路由函数里改 State → 和 reducer 打架，排查极痛，保持纯函数；
• 用 print(app.get_graph().draw_mermaid()) 一眼核对条件边连到了哪（05 节细讲）。

📁 本节示例代码：code/examples/03_conditional_routing_demo.py —— 规则版分流，无需 API Key。

---

扩展阅读

官方文档

• 图 API 概念（条件边、Send、Command、递归上限）：Graph API
• 使用图 API（条件分支、并行、Map-Reduce）：How-to：Graph API
• 常见工作流模式：Workflows and agents

社区教程

• PocketFlow《Control Flow Primitives：Branch / Send / Interrupt》：GitHub
• LangGraph Basics《Conditional Edges & Routing Logic》：shafiqulai.github.io

---

下一节： 当一条边同时指向多个节点，它们会并行吗？结果又如何合并回状态？我们来看 Fan-out / Fan-in 与 Send 动态分发。