1. 本章目标 #

前三章的图都是一条直线：START → A → B → END。但真实业务几乎没有直线：退货和查物流要走不同流程，金额超阈值要多一道审批，模型判断不出意图就得转人工。

这一章让图分叉：

根据当前状态决定下一步走哪条路。

学完这一章，你应该能做到：

• 用 add_conditional_edges 写出分支，并说清路由函数的三条规矩
• 理解声明出口为什么值得写：它的价值不在改名，而在把静默失败变成大声失败
• 掌握几种分支写法：直接结束、并行扇出、多级路由、Send 动态并行
• 用 Command 在节点里「边改状态边跳转」，并知道它为什么比条件边更容易静默出错
• 避开五个只在分支场景才有的坑，其中三个不报错
• 产出：一个意图路由示例，模型判意图 + 结构化约束 + 置信度兜底

参考文档：

• Graph API - Conditional edges
• Command

1.1. 分支为什么值得单独讲一章 #

第 20 章说过，节点是「干活的地方」，边是「谁接着谁」。到目前为止，边都是死的：写死在代码里，跑一百次走一百次同样的路。

分支引入了第一个运行时才确定的结构：这一次走哪条边，取决于这一次的状态。这带来两样新东西：

1. 一个新的可写位置。以前只在节点里写逻辑，现在边上也有（路由函数）。它有自己的规矩，违反了不一定报错。
2. 一批新的失败模式。「节点没跑」在直线图里几乎不会发生，在分支图里却最常见，而且多数时候没有任何异常。

所以本章重点不是「怎么写出分支」（API 就一个），而是「怎么写出不会静默出错的分支」。

1.2. 本章统一环境 #

本章前半部分共用这份状态、分类节点，以及一个造初始状态的小工具：

# operator 提供 add，用来给列表字段挂「累加」reducer
import operator
# Annotated 用来给字段附加 reducer，TypedDict 用来声明状态结构
from typing import Annotated, TypedDict

# END/START 是两个特殊节点名，StateGraph 是图的构建器
from langgraph.graph import END, START, StateGraph

# 本章共用的状态类型
class S(TypedDict):
    # 用户原始输入，普通字段（后写的覆盖先写的）
    text: str
    # 分类结果，普通字段
    kind: str
    # 执行轨迹，挂了 operator.add，所有节点的写入会累加起来
    log: Annotated[list[str], operator.add]

# 普通节点：判断类别，把结果写进状态
def classify(state: S) -> dict:
    """普通节点：判断类别，写进状态。"""
    # 只要输入里带「退」字就算退货，否则算其他（故意做成确定性逻辑，方便复现）
    kind = "退货" if "退" in state["text"] else "其他"
    # 返回部分更新：kind 被覆盖写，log 被累加
    return {"kind": kind, "log": [f"classify -> {kind}"]}

# 造一份完整初始状态，避免每个例子都手写一遍这个字典
def new_input(text: str) -> S:
    """造初始状态：三个字段都给出零值，后面例子直接复用。"""
    # log 挂了 reducer，其实可以省略，但显式写全更清楚（见第 22 章 §3.2）
    return {"text": text, "kind": "", "log": []}

为什么 classify 用「有没有『退』字」这种土办法，而不是调模型？因为本章要看的是分支机制本身。分类一旦引入模型，输出每次都可能变，你就分不清「路由走错」是机制问题还是模型问题。真正调模型的版本留到 §8。

2. 为什么固定边不够用 #

第 20 章的 add_edge 是无条件的：连上了就一定走。想表达「如果……就……」，只有两个办法。

办法一：把判断塞进节点内部。

import operator
from typing import Annotated, TypedDict

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

# 一个节点内部同时干了「判断」和「处理」两件事
def handle(state: S) -> dict:
    # 判断逻辑藏在节点内部，图上完全看不到
    if "退" in state["text"]:
        # 退货的处理逻辑
        return {"log": ["处理退货"]}
    # 其他情况的处理逻辑
    return {"log": ["处理其他"]}

# 直接当普通函数调用，看两种输入分别走到哪个 return
print("退货输入:", handle({"text": "我要退货", "kind": "", "log": []}))
# 换一个不带「退」字的输入
print("其他输入:", handle({"text": "你好", "kind": "", "log": []}))

退货输入: {'log': ['处理退货']}
其他输入: {'log': ['处理其他']}

能跑，但把两件事揉成一坨：

• 图上看不出有分支，draw_mermaid() 画出来仍是一条直线
• 退货流程和通用流程无法各自演进（想给退货加一步审批？只能继续往 if 里塞）
• 无法只对退货分支做中断、重试、审批
• 第 22 章的 stream 和 get_state_history 只能看到「handle 跑了」，看不到「这次走的是退货」

最后一条最要紧：藏在节点里的分支，审计不到。

办法二：用条件边，让分支体现在图的结构上。

              ┌──> refund ──┐
START → classify            ├──> END
              └──> general ─┘

这才是第 19 章说的「用图换确定性」：分支画在图上，才能被看见、被审计、被单独干预。

判断标准很简单：两个分支后续步骤不一样，就该用条件边；只是同一步里算法不同，塞节点里就行。

3. add_conditional_edges 三件套 #

3.1. 签名 #

# inspect 用来在运行时查看函数签名
import inspect

# 被检查的对象是 StateGraph 上的方法
from langgraph.graph import StateGraph

# 只打印参数名列表：真实签名的类型注解很长，直接打印会糊成一片
print(list(inspect.signature(StateGraph.add_conditional_edges).parameters))

['self', 'source', 'path', 'path_map']

三个参数：

参数	类型	含义
source	str	从哪个节点出发（也可以是 START）
path	函数 / async 函数 / Runnable	路由函数：读状态，返回「下一步是谁」
path_map	dict[Hashable, str] / list[str] / None	可选，声明这个分支所有可能的出口

两个容易忽略的细节：

• path 的返回值类型是 Hashable | Sequence[Hashable]，也就是返回一个名字，或者一串名字。返回一串就是并行扇出（§5.2）。
• 这个方法返回 Self，所以支持链式调用：builder.add_node(...).add_conditional_edges(...)。

3.2. 最小例子 #

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 路由函数：读状态，返回下一个节点的名字
def route(state: S) -> str:
    """路由函数：读状态，返回节点名。"""
    # 打印一行，方便观察路由函数确实执行了、看到了什么
    print(f"   [route] 看到 kind={state['kind']}")
    # 只做一件事：把状态映射成「去哪」
    return "refund" if state["kind"] == "退货" else "general"

# 新建图构建器，绑定状态类型 S；未编译的图纸统一叫 builder
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册退货分支节点（用 lambda 简化，只往 log 里记一笔）
builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
# 注册通用分支节点
builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
# 入口：START 无条件连到 classify
builder.add_edge(START, "classify")
# 关键这一行：classify 跑完后，由 route 决定去 refund 还是 general
builder.add_conditional_edges("classify", route)
# 两个分支各自连到 END
builder.add_edge("refund", END)
# 通用分支也连到 END
builder.add_edge("general", END)
# 编译成可执行图，编译后的对象统一叫 graph
graph = builder.compile()

# 输入带「退」字，应该走 refund
print(graph.invoke(new_input("我要退货")))
# 输入不带「退」字，应该走 general
print(graph.invoke(new_input("你好")))

   [route] 看到 kind=退货
{'text': '我要退货', 'kind': '退货', 'log': ['classify -> 退货', '走了退货分支']}
   [route] 看到 kind=其他
{'text': '你好', 'kind': '其他', 'log': ['classify -> 其他', '走了通用分支']}

同一张图，两种输入走出两条路。注意 log 里能看到完整路径，这正是办法一做不到的。

执行顺序要说清楚：classify 跑完、写入合并进状态之后，路由函数才执行。所以它看到的 kind 已是 classify 写进去的新值，不是初始空串。这和第 20 章 §6.4 的「超步」模型一致：一个超步内先跑节点，再算边。

3.3. 路由函数的三条规矩 #

3.3.1 规矩一：只读状态，不写状态。 #

路由函数返回的东西会被当成「节点名」，不是状态更新。返回一个字典试试：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 错误示范：路由函数试图返回状态更新
def route_side(state: S):
    # 这个字典不会被当成状态更新，而会被当成「节点名」
    return {"log": ["路由也想写状态"]}

# 搭一张最小的图来观察后果
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册一个目标节点
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 入口边
builder.add_edge(START, "classify")
# 挂上这个「想写状态」的路由函数，先不写 path_map
builder.add_conditional_edges("classify", route_side)
# 出口边
builder.add_edge("refund", END)
# 编译并运行
graph = builder.compile()
# 观察最终状态里有没有那句「路由也想写状态」
print(graph.invoke(new_input("x")))

{'text': 'x', 'kind': '其他', 'log': ['classify -> 其他']}
Task classify with path ('__pregel_pull', 'classify') wrote to unknown channel branch:to:{'log': ['路由也想写状态']}, ignoring it.

log 里没有那句话，refund 也没跑：字典被当成节点名了，找不到对应节点，于是什么都没发生。

第二行那句警告把这件事说得很直白：branch:to: 后面跟的就是你返回的整个字典，它被当成了目标节点的名字。这行字来自标准 logging，程序不会因此中断，很容易被淹没在其它日志里，§4.5 会讲怎么主动把它捞出来。

同样的代码，若声明了出口（path_map），就会当场报错：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def route_side(state: S):
    return {"log": ["路由也想写状态"]}

# 同一个错误示范，这次把出口声明出来
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册目标节点
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 入口边
builder.add_edge(START, "classify")
# 第三个参数声明「出口只可能是 refund」
builder.add_conditional_edges("classify", route_side, ["refund"])
# 出口边
builder.add_edge("refund", END)
# 编译
graph = builder.compile()
# 运行并捕获异常，看看报什么
try:
    # 路由函数返回了字典，而字典不可哈希，无法在出口表里查找
    print(graph.invoke(new_input("x")))
except Exception as e:
    # 打印异常类型和信息
    print("报错:", type(e).__name__, e)

报错: TypeError unhashable type: 'dict'

想在分支时顺便改状态，用 §6 的 Command。

3.3.2 规矩二：返回节点名字符串（或它们的列表）。 #

「节点名」有三种合法值：已注册的节点名、END（§5.1）、以及它们组成的列表（§5.2）。

返回 None 也算一个值，会被当成名字去查表：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 观察路由函数返回 None 会怎样
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册目标节点
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 入口边
builder.add_edge(START, "classify")
# 路由函数忘了 return（等价于 return None），并声明了出口
builder.add_conditional_edges("classify", lambda s: None, ["refund"])
# 出口边
builder.add_edge("refund", END)
# 编译
graph = builder.compile()
# 运行并捕获异常
try:
    # None 会被当成节点名去出口表里查
    print(graph.invoke(new_input("x")))
except Exception as e:
    # 打印异常类型和信息
    print("报错:", type(e).__name__, e)

报错: KeyError None

KeyError None 几乎总意味着路由函数里有一条分支忘了 return，比如只写了 if、没写 else。

3.3.3 规矩三：可以接第二个参数 config。 #

import operator
from typing import Annotated, TypedDict

# RunnableConfig 是 config 的类型注解
from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 路由函数的第二个参数会拿到本次调用的 config
def route_cfg(state: S, config: RunnableConfig) -> str:
    # 从 config.configurable 里读业务开关，注意用 get 防止键不存在
    print("   [route] config 里的 vip:", config.get("configurable", {}).get("vip"))
    # 这里演示用，固定走 refund
    return "refund"

# 搭图
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册目标节点
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 入口边
builder.add_edge(START, "classify")
# 挂上带 config 参数的路由函数，并声明出口
builder.add_conditional_edges("classify", route_cfg, ["refund"])
# 出口边
builder.add_edge("refund", END)
# 编译
graph = builder.compile()
# 调用时通过 configurable 传入业务开关（第 22 章 §5）
print(graph.invoke(new_input("x"), {"configurable": {"vip": True}}))

   [route] config 里的 vip: True
{'text': 'x', 'kind': '其他', 'log': ['classify -> 其他', '退货分支']}

这样一来，「同一张图，VIP 用户走不同分支」就不必污染状态。判断依据是业务数据就放状态，是调用方身份/开关就放 config。

参数最多就这两个，写第三个会当场报错：

import operator
from typing import Annotated, TypedDict

from langchain_core.runnables import RunnableConfig
from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 错误示范：路由函数写了三个参数
def route_3(state: S, config: RunnableConfig, extra) -> str:
    # 函数体本身没问题，问题在于 LangGraph 只会传两个参数
    return "refund"

# 搭图
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册目标节点
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 入口边
builder.add_edge(START, "classify")
# 挂上三参数路由函数
builder.add_conditional_edges("classify", route_3, ["refund"])
# 出口边
builder.add_edge("refund", END)
# 编译
graph = builder.compile()
# 运行并捕获异常
try:
    # 调用时 extra 拿不到值
    print(graph.invoke(new_input("x")))
except Exception as e:
    # 打印异常类型和信息
    print("报错:", type(e).__name__, e)

报错: TypeError route_3() missing 1 required positional argument: 'extra'

路由函数也可以是 async def，但那样这张图就只能用 ainvoke / astream 跑：

# asyncio 用来运行协程
import asyncio
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 异步路由函数：真实场景里可能要 await 一次远程配置查询
async def aroute(state: S) -> str:
    # 让出一次控制权，模拟异步 I/O
    await asyncio.sleep(0)
    # 返回节点名
    return "refund"

# 搭图
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册目标节点
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 入口边
builder.add_edge(START, "classify")
# 挂上异步路由函数
builder.add_conditional_edges("classify", aroute, ["refund"])
# 出口边
builder.add_edge("refund", END)
# 编译
graph = builder.compile()
# 先试同步 invoke，预期会报错
try:
    # 同步入口拿不到协程的结果
    print(graph.invoke(new_input("x")))
except Exception as e:
    # 打印异常类型和信息的第一行
    print("同步 invoke 报错:", type(e).__name__, str(e).splitlines()[0])
# 换成异步入口就正常
print("异步 ainvoke:", asyncio.run(graph.ainvoke(new_input("x"))))

同步 invoke 报错: TypeError No synchronous function provided to "aroute".
异步 ainvoke: {'text': 'x', 'kind': '其他', 'log': ['classify -> 其他', '退货分支']}

一句话记住路由函数：它是纯函数，输入状态，输出「去哪」。 别在里面调 API、别改状态、别有副作用（原因见 §7.5）。

3.4. 路由函数什么时候执行、执行几次 #

这一小节回答一个实际问题：路由函数会不会被重复执行？ 答案直接决定你要不要在里面写副作用。

import operator
from typing import Annotated, TypedDict

# 需要检查点才能演示恢复和重放
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 用一个列表记录路由函数被调用了几次
CALLS: list[str] = []
# 用一个可变字典当开关，控制目标节点要不要抛异常
BOOM = {"on": True}

# 被计数的路由函数
def route_counted(state: S) -> str:
    # 每次执行都记一笔
    CALLS.append(state["kind"])
    # 固定返回目标节点
    return "target"

# 目标节点：开关打开时故意失败
def target(state: S) -> dict:
    # 模拟「下游节点因为脏数据/网络问题失败」
    if BOOM["on"]:
        # 抛出业务异常
        raise ValueError("目标节点炸了")
    # 开关关闭后正常返回
    return {"log": ["target 成功"]}

# 搭图：classify → (route_counted) → target
builder = StateGraph(S)
# 分类节点固定写死 kind，减少干扰
builder.add_node("classify", lambda s: {"kind": "退货", "log": ["classify"]})
# 注册可能失败的目标节点
builder.add_node("target", target)
# 入口边
builder.add_edge(START, "classify")
# 条件边，出口只有 target
builder.add_conditional_edges("classify", route_counted, ["target"])
# 出口边
builder.add_edge("target", END)
# 编译时挂上检查点，这样才能恢复和重放
graph = builder.compile(checkpointer=InMemorySaver())
# 固定一个会话 id
cfg = {"configurable": {"thread_id": "t1"}}

# 第一次运行：target 会失败
try:
    # 异常会原样抛出（第 22 章 §7.1）
    graph.invoke(new_input("x"), cfg)
except Exception as e:
    # 打印失败信息
    print("第一次失败:", type(e).__name__, e)
# 此时路由函数执行过一次
print("路由执行次数:", len(CALLS))

# 关掉开关，模拟「问题修好了」
BOOM["on"] = False
# 用 invoke(None) 从断点恢复（第 22 章 §7.3）
print("恢复结果:", graph.invoke(None, cfg)["log"])
# 关键观察：恢复没有重算路由
print("路由执行次数:", len(CALLS))

# 再看时间旅行：找到 classify 还没跑的那个检查点
snap = [h for h in graph.get_state_history(cfg) if h.next == ("classify",)][0]
# 从那里重放，classify 会重跑
print("重放结果:", graph.invoke(None, snap.config)["log"])
# 关键观察：这次路由被重算了
print("路由执行次数:", len(CALLS))

第一次失败: ValueError 目标节点炸了
路由执行次数: 1
恢复结果: ['classify', 'target 成功']
路由执行次数: 1
重放结果: ['classify', 'target 成功']
路由执行次数: 2

三条结论：

1. 正常一次 invoke 里，一个条件边的路由函数只执行一次。
2. 下游失败后 invoke(None) 恢复，路由函数不重算。 因为「去哪」这个结论是跟着 classify 的写入一起存进检查点的，恢复时直接读结果。
3. 从源节点之前的检查点重放，路由函数会重算。 源节点重跑了，边自然要重算。

所以路由函数的执行次数不由你控制：可能 1 次，也可能因一次重放变成 2 次。这就是 §7.5 说「别在路由函数里写副作用」的硬理由。

4. path_map：本章最重要的一个建议 #

path_map 看起来只是个可选的「改名表」，其实它是分支代码可靠性的关键分界。

先说结论，后面三小节逐条证明：

add_conditional_edges 的第三个参数虽是可选，请当成必填。 不写它，你会同时丢掉编译期校验、运行期报错和正确的图纸，而这三样恰好是排查分支问题的全部手段。

后面的例子都会在同一张「半成品图」上补一条条件边，所以先做一个工厂函数：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

# 工厂函数：造一张只差条件边的半成品图
def new_builder() -> StateGraph:
    """造一个「只差条件边」的半成品图，后面每个例子在它上面补一条边。"""
    # 新建构建器，未编译的图纸统一叫 builder
    builder = StateGraph(S)
    # 注册分类节点
    builder.add_node("classify", classify)
    # 注册退货分支节点
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    # 注册通用分支节点
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    # 入口边
    builder.add_edge(START, "classify")
    # 退货分支出口
    builder.add_edge("refund", END)
    # 通用分支出口
    builder.add_edge("general", END)
    # 返回构建器本身（还没编译），留给调用方补条件边
    return builder

这段代码只有定义、没有输出，下面每个例子都会在它后面接上自己的条件边。new_builder() 会在后续代码块里重复出现。

4.1. 两种写法 #

4.1.1 字典形式 #

把路由函数的返回值映射到节点名：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 取一张半成品图
builder = new_builder()
# 路由函数返回 "a"/"b" 这类抽象标签，由字典翻译成真实节点名
builder.add_conditional_edges(
    # 从 classify 出发
    "classify",
    # 路由函数只负责给出标签
    lambda s: "a" if s["kind"] == "退货" else "b",
    # 出口表：返回 "a" 就去 refund，返回 "b" 就去 general
    {"a": "refund", "b": "general"},
)
# 编译
graph = builder.compile()
# 运行，验证走了退货分支
print("字典形式:", graph.invoke(new_input("我要退货")))

字典形式: {'text': '我要退货', 'kind': '退货', 'log': ['classify -> 退货', '走了退货分支']}

4.1.2 列表形式 #

声明「出口只可能是这几个」（路由函数直接返回节点名）：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 取一张半成品图
builder = new_builder()
# 路由函数直接返回节点名，列表只是把可能的出口声明一遍
builder.add_conditional_edges(
    # 从 classify 出发
    "classify",
    # 直接返回真实节点名
    lambda s: "refund" if s["kind"] == "退货" else "general",
    # 出口声明：只可能是这两个
    ["refund", "general"],
)
# 编译
graph = builder.compile()
# 运行
print("列表形式:", graph.invoke(new_input("我要退货")))

列表形式: {'text': '我要退货', 'kind': '退货', 'log': ['classify -> 退货', '走了退货分支']}

列表形式最常用：不必改名，只是把出口声明清楚。什么时候用字典形式？两种情况：

• 路由依据是业务枚举而不是节点名，比如模型给的 intent 是 "refund"，处理节点叫 handle_refund，字典正好当适配层；
• 想让多个标签指向同一节点，比如 {"unknown": "handle_other", "low_conf": "handle_other"}，图上会有两条虚线、标签不同，一眼看出「兜底有两种触发原因」。

4.2. 理由一：把静默失败变成大声失败 #

这是关键。假设路由函数把节点名拼错了：实际节点叫 refund，路由却返回 handle_refund。

不写 path_map 时：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 取一张半成品图
builder = new_builder()
# 路由函数返回一个不存在的节点名，并且故意不声明出口
builder.add_conditional_edges("classify", lambda s: "handle_refund")
# 编译
graph = builder.compile()
# 编译居然通过了
print("compile 通过了")
# 运行也不报错
print("invoke 结果:", graph.invoke(new_input("x")))

compile 通过了
invoke 结果: {'text': 'x', 'kind': '其他', 'log': ['classify -> 其他']}
Task classify with path ('__pregel_pull', 'classify') wrote to unknown channel branch:to:handle_refund, ignoring it.

编译通过，运行也不报错，图就这么悄悄结束了，refund 节点从未跑过。唯一痕迹是最后那行警告，而它既不是异常也不打断执行，在真实项目的日志里极容易被淹没。

写了 path_map 时：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 取一张半成品图
builder = new_builder()
# 同样拼错节点名，但这次声明了出口
builder.add_conditional_edges("classify", lambda s: "handle_refund", ["refund"])
# 编译仍然通过（出口表里的 refund 确实存在）
graph = builder.compile()
# 打印一行，说明问题不在编译期
print("compile 通过")
# 运行时才发现路由返回值不在出口表里
try:
    # 这次会抛异常
    graph.invoke(new_input("x"))
except Exception as e:
    # 打印异常类型和信息
    print("报错:", type(e).__name__, e)

compile 通过
报错: KeyError 'handle_refund'

直接抛异常。 同样的 bug，一个让你排查半天，一个当场点破。

顺便一提，如果是 path_map 本身写错了（映射到不存在的节点），连编译都过不去：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 取一张半成品图
builder = new_builder()
# 出口表指向一个根本没注册的节点
builder.add_conditional_edges("classify", lambda s: "a", {"a": "幽灵节点"})
# 编译时就会校验出口表里的节点是否都存在
try:
    # 这行会抛异常
    builder.compile()
except Exception as e:
    # 打印异常类型和信息
    print("报错:", type(e).__name__, e)

报错: ValueError At 'classify' node, 'condition' branch found unknown target '幽灵节点'

列表形式同理：写了不存在的节点，也是编译期就拦住。报错信息里的 'condition' 是这条分支的名字：用 lambda 当路由函数时分支叫 condition，用具名函数时分支就叫函数名。 这个细节在 §7.3 会变得很重要。

三种情况对照：

写法	路由返回未知值	path_map 指向未知节点
无 path_map	静默结束（最危险）	—
有 path_map	运行时 KeyError	编译时 ValueError

一句话：path_map 把「出口清单」从路由函数的脑子里搬到图上，LangGraph 才有东西可校验。

4.3. 理由二：不声明出口，图就画错 #

第 22 章说 draw_mermaid() 是排查结构问题的主力工具。但不写 path_map 时，它画出来的图是错的：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 取一张半成品图
builder = new_builder()
# 路由函数写得完全正确，只是没声明出口
builder.add_conditional_edges("classify", lambda s: "refund" if s["kind"] == "退货" else "general")
# 编译
graph = builder.compile()
# 打印 mermaid 图源码
print(graph.get_graph().draw_mermaid())

---
config:
  flowchart:
    curve: linear
---
graph TD;
    __start__(<p>__start__</p>)
    classify(classify)
    refund(refund)
    general(general)
    __end__(<p>__end__</p>)
    __start__ --> classify;
    classify --> __end__;
    classDef default fill:#f2f0ff,line-height:1.2
    classDef first fill-opacity:0
    classDef last fill:#bfb6fc

refund 和 general 两个节点孤零零挂着，classify 直接连到 __end__。图和实际行为完全对不上。

原因不难理解：路由函数是运行时才执行的普通 Python 函数，LangGraph 静态分析不出它可能返回什么，只能保守画成「跑完就结束」。

这里还藏着一个诊断信号：注意 __start__ 这一行是 __start__(<p>__start__</p>)，正常图里应是 __start__([<p>__start__</p>]):::first。:::first 和 :::last 这两个样式标记，只有在「入边为空的节点恰好一个、出边为空的节点也恰好一个」时才会打上。孤立节点一出现，标记就没了。

看到 draw_mermaid() 输出里没有 :::first / :::last，基本可以断定图里有孤立节点。

加上 path_map 之后：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 取一张半成品图
builder = new_builder()
# 这次用字典形式声明出口（写法同 §4.1，压成一行）
builder.add_conditional_edges("classify", lambda s: "a" if s["kind"] == "退货" else "b", {"a": "refund", "b": "general"})
# 编译
graph = builder.compile()
# 只打印图里跟边有关的行，去掉固定的头尾噪音
print("\n".join(l for l in graph.get_graph().draw_mermaid().splitlines() if "-->" in l or "-." in l or ":::" in l))

    __start__([<p>__start__</p>]):::first
    __end__([<p>__end__</p>]):::last
    __start__ --> classify;
    classify -. &nbsp;b&nbsp; .-> general;
    classify -. &nbsp;a&nbsp; .-> refund;
    general --> __end__;
    refund --> __end__;

虚线（-.->）表示条件边，上面还标了触发它的返回值（a / b），:::first 和 :::last 也回来了。用列表形式时虚线照样有，只是没有标签：

    classify -.-> general;
    classify -.-> refund;

如果不想靠肉眼看 mermaid，还有更适合写断言的办法：直接读结构对象。

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 取一张半成品图
builder = new_builder()
# 用字典形式声明出口，方便看到标签
builder.add_conditional_edges("classify", lambda s: "a" if s["kind"] == "退货" else "b", {"a": "refund", "b": "general"})
# 编译
graph = builder.compile()
# get_graph() 返回可编程访问的图结构（第 19 章 §2.2）
for e in graph.get_graph().edges:
    # conditional 标出这条边是不是条件边，data 是虚线上的标签
    print(f"  {e.source} -> {e.target}  conditional={e.conditional} label={e.data}")

  __start__ -> classify  conditional=False label=None
  classify -> general  conditional=True label=b
  classify -> refund  conditional=True label=a
  general -> __end__  conditional=False label=None
  refund -> __end__  conditional=False label=None

conditional=True 就是虚线。在单元测试里断言「classify 有且只有 2 个条件出口」，比人肉看图靠谱得多。

4.4. 第三种办法：Literal 类型注解 #

不想写 path_map，也可以用返回值的类型注解告诉 LangGraph 可能的出口：

import operator

# Literal 用来把返回值限定在几个字面量里
from typing import Annotated, Literal, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 返回类型注解写成 Literal，等价于声明出口
def route_typed(state: S) -> Literal["refund", "general"]:
    # 函数体和普通路由函数没区别
    return "refund" if state["kind"] == "退货" else "general"

# 取一张半成品图
builder = new_builder()
# 注意：第三个参数没写，出口信息全靠上面的 Literal 注解
builder.add_conditional_edges("classify", route_typed)
# 编译
graph = builder.compile()
# 正常运行
print("运行:", graph.invoke(new_input("我要退货"))["log"])
# 只打印带边的行，看看图画对了没
print("\n".join(l for l in graph.get_graph().draw_mermaid().splitlines() if "-." in l))

运行: ['classify -> 退货', '走了退货分支']
    classify -.-> general;
    classify -.-> refund;

图画对了（虚线无标签）。那它做不做校验？很多资料说「Literal 只影响画图」，实测不是这样：

import operator
from typing import Annotated, Literal, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 注解声明只会返回 refund/general，但函数体返回了别的
def route_typed_bad(state: S) -> Literal["refund", "general"]:
    # 故意返回一个注解之外、也不存在的节点名
    return "handle_refund"

# 取一张半成品图
builder = new_builder()
# 挂上这个「注解和实现不一致」的路由函数
builder.add_conditional_edges("classify", route_typed_bad)
# 编译
graph = builder.compile()
# 运行并捕获异常
try:
    # 如果 Literal 只管画图，这里应该静默结束
    print("结果:", graph.invoke(new_input("x")))
except Exception as e:
    # 实际会当场报错
    print("报错:", type(e).__name__, e)

报错: KeyError 'handle_refund'

和 path_map 一模一样的运行时报错。 编译期校验也有：

import operator
from typing import Annotated, Literal, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 注解里写了一个不存在的节点
def route_ghost(state: S) -> Literal["幽灵节点"]:
    # 函数体老实返回注解里的名字
    return "幽灵节点"

# 取一张半成品图
builder = new_builder()
# 挂上它
builder.add_conditional_edges("classify", route_ghost)
# 编译时就会校验注解里的节点是否存在
try:
    # 这行会抛异常
    builder.compile()
except Exception as e:
    # 报错信息里的分支名就是函数名 route_ghost
    print("报错:", type(e).__name__, e)

报错: ValueError At 'classify' node, 'route_ghost' branch found unknown target '幽灵节点'

所以实际情况是：Literal 注解和列表形式的 path_map 效果等价，LangGraph 会从注解里把出口清单读出来当 path_map 用。三种方案的正确对比是：

方案	画图正确	运行时校验	编译时校验	出口清单能否运行时算出
什么都不写	×	×	×	—
Literal 注解	√（无标签）	√ KeyError	√ ValueError	×（必须是字面量）
path_map 列表	√（无标签）	√ KeyError	√ ValueError	√
path_map 字典	√（有标签）	√ KeyError	√ ValueError	√

选哪个？

• 用 lambda 当路由函数时只能用 path_map（lambda 写不了返回注解）。
• 出口清单要动态生成时只能用 path_map，比如 §8 里的 list(HANDLERS.values())；Literal 必须是写死的字面量。
• 想在图上看到「哪个返回值触发哪条边」时用字典形式的 path_map，这是唯一能带标签的写法。
• 具名路由函数 + 固定出口，用 Literal 最省事，顺便还能被 mypy / Pyright 静态检查出「函数体返回了注解外的值」。

结论：三者选一都行，唯一不可接受的是「什么都不写」。

4.5. 那条被忽略的警告：怎么把它捞出来 #

万一接手的是已经写满「什么都不写」的老项目，来不及全部补上出口声明，至少可以把那行警告变得显眼。它来自标准 logging，logger 名叫 langgraph，级别 WARNING：

# 标准库 logging
import logging
import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 用一个列表收集「路由被丢弃」的警告
LOST: list[str] = []

# 自定义 Handler：只挑我们关心的那条消息
class CollectLostRoute(logging.Handler):
    """把 langgraph 那条『wrote to unknown channel』警告收集起来。"""

    # 每条日志都会走到这里
    def emit(self, record: logging.LogRecord) -> None:
        # 取出格式化后的消息文本
        msg = record.getMessage()
        # 只保留路由丢弃这一类
        if "wrote to unknown channel branch:to:" in msg:
            # 收进列表，留给调用方判断
            LOST.append(msg)

# 拿到 langgraph 的 logger（只挂它，不影响其它日志）
lg_logger = logging.getLogger("langgraph")
# 挂上收集器
handler = CollectLostRoute()
# 注册
lg_logger.addHandler(handler)

# 造一张「路由名拼错且没声明出口」的图
builder = new_builder()
# 返回不存在的节点名
builder.add_conditional_edges("classify", lambda s: "handle_refund")
# 编译
graph = builder.compile()
# 运行：本身不会报错
out = graph.invoke(new_input("x"))
# 用完摘掉 handler，避免影响后面的例子
lg_logger.removeHandler(handler)

# 事后检查：捞到了就说明有路由被静默丢弃
print("捞到条数:", len(LOST))
# 打印内容
print("内容:", LOST[0] if LOST else "-")
# 在测试/启动自检里可以据此直接失败
print("判定:", "有路由被丢弃" if LOST else "正常")

捞到条数: 1
内容: Task classify with path ('__pregel_pull', 'classify') wrote to unknown channel branch:to:handle_refund, ignoring it.
判定: 有路由被丢弃

两点提醒：

• 它不是 Python 的 warnings，warnings.catch_warnings() 抓不到（实测抓到 0 条），只有 logging 能拿到。
• 别在 emit 里直接 raise。异常确实会从图里抛出来，但抛出位置在日志系统内部，堆栈不指向你的路由函数，反而更难排查。像上面这样「先收集、跑完再判断」更可控。

这只是给老项目兜底的手段。新代码请直接声明出口。

5. 分支的几种花样 #

路由函数返回值的花样比「一个节点名」多得多。这一节把四种常用形态过一遍。

5.1. 直接结束 #

路由函数可以返回 END，表示这一支到此为止：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 新建一张图：这次只有 refund 一个处理节点
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册退货处理节点
builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
# 入口边
builder.add_edge(START, "classify")
# 是退货就去 refund，否则直接结束；注意 END 也要写进出口表
builder.add_conditional_edges("classify", lambda s: "refund" if s["kind"] == "退货" else END, ["refund", END])
# refund 跑完也去 END
builder.add_edge("refund", END)
# 编译
graph = builder.compile()
# 走分支的情况
print("  走分支:", graph.invoke(new_input("退货")))
# 直接结束的情况
print("  直接结束:", graph.invoke(new_input("你好")))

  走分支: {'text': '退货', 'kind': '退货', 'log': ['classify -> 退货', '走了退货分支']}
  直接结束: {'text': '你好', 'kind': '其他', 'log': ['classify -> 其他']}

「命中黑名单直接拒绝」「无需处理直接返回」这类需求就是这么写的。图上会看到一条从 classify 直接虚线连到 __end__ 的边：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

builder = StateGraph(S)
builder.add_node("classify", classify)
builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", lambda s: "refund" if s["kind"] == "退货" else END, ["refund", END])
builder.add_edge("refund", END)
graph = builder.compile()

# 只打印带边的行
print("\n".join(l for l in graph.get_graph().draw_mermaid().splitlines() if "-->" in l or "-." in l))

    __start__ --> classify;
    classify -.-> __end__;
    classify -.-> refund;
    refund --> __end__;

END 必须显式声明为出口。 忘了写进 path_map，后果和拼错节点名一样：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 新建一张图
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册退货处理节点
builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
# 入口边
builder.add_edge(START, "classify")
# 路由函数会返回 END，但出口表里只写了 refund
builder.add_conditional_edges("classify", lambda s: "refund" if s["kind"] == "退货" else END, ["refund"])
# 出口边
builder.add_edge("refund", END)
# 编译
graph = builder.compile()
# 走「其他」这一支，触发返回 END
try:
    # 预期报错
    print("  结果:", graph.invoke(new_input("你好")))
except Exception as e:
    # END 的真实值是字符串 "__end__"
    print("  报错:", type(e).__name__, e)

  报错: KeyError '__end__'

看到 KeyError '__end__' 就是这个原因：路由函数返回了 END，你却没把它列进出口。

5.2. 返回列表 = 并行扇出 #

路由函数返回列表时，列表里的节点会同时执行：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 新建一张图，准备三个并行节点
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 循环注册 n1/n2/n3，用默认参数固化闭包变量
for n in ("n1", "n2", "n3"):
    # 每个节点只往 log 里记自己的名字
    builder.add_node(n, lambda s, _n=n: {"log": [_n]})
    # 每个节点跑完都去 END
    builder.add_edge(n, END)
# 入口边
builder.add_edge(START, "classify")
# 路由函数一次返回三个节点名 → 三个节点并行
builder.add_conditional_edges("classify", lambda s: ["n1", "n2", "n3"], ["n1", "n2", "n3"])
# 编译
graph = builder.compile()
# 运行，观察三个节点的写入都进了 log
print("结果:", graph.invoke(new_input("x")))

结果: {'text': 'x', 'kind': '其他', 'log': ['classify -> 其他', 'n1', 'n2', 'n3']}

怎么确认它们真是「同时」而不是「依次」？看第 22 章 debug 流里的 step（超步序号）：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

builder = StateGraph(S)
builder.add_node("classify", classify)
for n in ("n1", "n2", "n3"):
    builder.add_node(n, lambda s, _n=n: {"log": [_n]})
    builder.add_edge(n, END)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", lambda s: ["n1", "n2", "n3"], ["n1", "n2", "n3"])
graph = builder.compile()

# 用 debug 模式流式执行，能看到每个任务属于第几个超步
for chunk in graph.stream(new_input("x"), stream_mode="debug"):
    # 只关心 task 类型的事件
    if chunk["type"] == "task":
        # 打印超步号和节点名
        print(f"   step={chunk['step']} node={chunk['payload']['name']}")

   step=1 node=classify
   step=2 node=n1
   step=2 node=n2
   step=2 node=n3

三个节点的 step 都是 2，同一个超步，确实是并行。

这和第 20 章「一个节点连出多条固定边」效果一样，区别在于列表内容是运行时算出来的，可以根据状态决定这次并行跑几个：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 同一张图结构，但路由函数这次只挑两个节点
builder = StateGraph(S)
# 注册分类节点
builder.add_node("classify", classify)
# 注册三个候选节点
for n in ("n1", "n2", "n3"):
    # 节点体同上
    builder.add_node(n, lambda s, _n=n: {"log": [_n]})
    # 出口边
    builder.add_edge(n, END)
# 入口边
builder.add_edge(START, "classify")
# 出口表声明三个，但这次只返回其中两个
builder.add_conditional_edges("classify", lambda s: ["n1", "n3"], ["n1", "n2", "n3"])
# 编译
graph = builder.compile()
# 运行：n2 不会执行
print("结果:", graph.invoke(new_input("x")))

结果: {'text': 'x', 'kind': '其他', 'log': ['classify -> 其他', 'n1', 'n3']}

出口表是「可能的全集」，返回值是「这次的子集」。 这也说明 path_map 为什么不能省：它是唯一能表达「全集」的地方。

既然是并行，第 21 章 §5 的规则全部适用：共同写入的字段必须有 reducer：

from typing import TypedDict

from langgraph.graph import END, START, StateGraph

# 定义一个没有任何 reducer 的状态
class Plain(TypedDict):
    # 普通字段
    text: str
    # 也是普通字段，下面会被两个并行节点同时写
    kind: str

# 搭一张并行图
builder = StateGraph(Plain)
# 起点节点
builder.add_node("classify", lambda s: {"kind": "x"})
# 两个并行节点都写 kind
builder.add_node("n1", lambda s: {"kind": "来自n1"})
# 第二个并行节点
builder.add_node("n2", lambda s: {"kind": "来自n2"})
# 入口边
builder.add_edge(START, "classify")
# 扇出到两个节点
builder.add_conditional_edges("classify", lambda s: ["n1", "n2"], ["n1", "n2"])
# 出口边
builder.add_edge("n1", END)
# 出口边
builder.add_edge("n2", END)
# 编译
graph = builder.compile()
# 运行并捕获异常
try:
    # 同一超步内两个值写同一个无 reducer 字段
    print("结果:", graph.invoke({"text": "x", "kind": ""}))
except Exception as e:
    # 打印异常类型和第一行信息
    print("报错:", type(e).__name__, str(e).splitlines()[0])

报错: InvalidUpdateError At key 'kind': Can receive only one value per step. Use an Annotated key to handle multiple values.

返回空列表则什么都不做，图直接结束：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 取一张半成品图
builder = new_builder()
# 路由函数返回空列表：一个出口都不触发
builder.add_conditional_edges("classify", lambda s: [], ["refund", "general"])
# 编译
graph = builder.compile()
# 运行：不报错，也没有任何分支节点执行
print("结果:", graph.invoke(new_input("x")))

结果: {'text': 'x', 'kind': '其他', 'log': ['classify -> 其他']}

这也是一种静默行为。写「按条件过滤出要处理的分支」这类逻辑时要留意：过滤到一个都不剩，图就悄悄结束了。 稳妥写法是空列表时回退到一个兜底节点：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

def new_builder() -> StateGraph:
    builder = StateGraph(S)
    builder.add_node("classify", classify)
    builder.add_node("refund", lambda s: {"log": ["走了退货分支"]})
    builder.add_node("general", lambda s: {"log": ["走了通用分支"]})
    builder.add_edge(START, "classify")
    builder.add_edge("refund", END)
    builder.add_edge("general", END)
    return builder

# 先按业务规则算出要触发的节点
def route_filtered(state: S) -> list[str] | str:
    # 这里用一个明显会过滤空的条件来演示
    targets = [n for n in ("refund", "general") if n in state["text"]]
    # 一个都没剩就走兜底，而不是返回空列表
    return targets or "general"

# 挂上这个带兜底的路由函数
builder = new_builder()
# 出口表把两个候选节点都声明上
builder.add_conditional_edges("classify", route_filtered, ["refund", "general"])
# 编译
graph = builder.compile()
# 输入里含 refund，过滤后还剩一个目标
print("过滤有结果:", graph.invoke(new_input("refund"))["log"])
# 输入里什么都不含，过滤为空，靠 or 兜到 general
print("过滤为空:", graph.invoke(new_input("x"))["log"])

过滤有结果: ['classify -> 其他', '走了退货分支']
过滤为空: ['classify -> 其他', '走了通用分支']

5.3. 多级路由 #

分支里还能再分支，条件边可以串起来：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

# 这个例子用自己的状态类型，只留 text 和 log
class M(TypedDict):
    # 用户输入
    text: str
    # 执行轨迹，累加
    log: Annotated[list[str], operator.add]

# 第一级路由：先分「跟钱有关」和「其他」
def lvl1(state: M) -> str:
    # 出现「钱」字就进入金额分支
    return "money" if "钱" in state["text"] else "other"

# 第二级路由：金额分支内部再分大小额
def lvl2(state: M) -> str:
    # 出现「万」字算大额
    return "big" if "万" in state["text"] else "small"

# 搭图
builder = StateGraph(M)
# 起点节点
builder.add_node("start_n", lambda s: {"log": ["start"]})
# 一级分支：金额相关
builder.add_node("money", lambda s: {"log": ["money"]})
# 一级分支：其他
builder.add_node("other", lambda s: {"log": ["other"]})
# 二级分支：大额
builder.add_node("big", lambda s: {"log": ["big"]})
# 二级分支：小额
builder.add_node("small", lambda s: {"log": ["small"]})
# 入口边
builder.add_edge(START, "start_n")
# 第一级条件边
builder.add_conditional_edges("start_n", lvl1, {"money": "money", "other": "other"})
# 第二级条件边：挂在 money 节点上
builder.add_conditional_edges("money", lvl2, {"big": "big", "small": "small"})
# 三个叶子节点各自结束
builder.add_edge("other", END)
# 大额结束
builder.add_edge("big", END)
# 小额结束
builder.add_edge("small", END)
# 编译
graph = builder.compile()
# 三种输入分别走三条不同的路径
for q in ("给我一万块钱", "给我点钱", "你好"):
    # 只打印 log，路径一目了然
    print(f"  {q}: {graph.invoke({'text': q, 'log': []})['log']}")

  给我一万块钱: ['start', 'money', 'big']
  给我点钱: ['start', 'money', 'small']
  你好: ['start', 'other']

图上是两层虚线：

    __start__ --> start_n;
    money -.-> big;
    money -.-> small;
    start_n -.-> money;
    start_n -.-> other;

层级不要超过两三层，否则图会变成一团意大利面。超过了，就该把子流程抽成独立的图（第 25 章）。另外注意：money 这个节点既是「一级分支的目的地」，又是「二级分支的出发点」。这种既处理又转发的节点很常见，但会让 money 职责变重，代码 review 时要留意。

5.4. Send：动态并行，每份带不同输入 #

前面的并行扇出有个限制：所有分支收到的是同一份状态。但「把 10 个文档分给 10 个 worker，每人处理一个」这种 map-reduce 需求，要给每个分支不同的输入。这就是 Send：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

# Send 用来「给某个节点投递一份专属输入」
from langgraph.types import Send

# 主状态：待处理清单 + 汇总结果
class MapState(TypedDict):
    # 待处理的条目
    items: list[str]
    # 结果收集区，挂 reducer 才能汇总多个 worker 的输出
    results: Annotated[list[str], operator.add]

# worker 看到的状态：只有自己那一份
class WorkerState(TypedDict):
    # 单条待处理数据
    item: str

# 分发函数：本质还是路由函数，只是返回的是一串 Send
def dispatch(state: MapState) -> list[Send]:
    # 每个 Send 是一个 (目标节点, 专属输入) 组合
    return [Send("worker", {"item": it}) for it in state["items"]]

# worker 节点：处理自己那一份，把结果写回汇总字段
def worker(state: WorkerState) -> dict:
    # 注意这里读的是 item，而不是 items
    return {"results": [f"处理了 {state['item']}"]}

# 搭图：只有一个 worker 节点，份数由运行时数据决定
builder = StateGraph(MapState)
# 注册 worker
builder.add_node("worker", worker)
# 条件边可以直接从 START 出发，省掉一个空转节点
builder.add_conditional_edges(START, dispatch, ["worker"])
# worker 跑完就结束
builder.add_edge("worker", END)
# 编译
graph = builder.compile()
# 三个条目 → 三份并行任务
print(graph.invoke({"items": ["a", "b", "c"], "results": []}))

{'items': ['a', 'b', 'c'], 'results': ['处理了 a', '处理了 b', '处理了 c']}

三个要点：

1. Send(节点名, 输入) 的第二个参数会成为那个节点看到的完整状态：不是「在主状态上叠加」，而是整个替换。
2. 并行份数由运行时数据决定，items 有几个就跑几份。
3. 结果靠 reducer 汇总，results 挂了 operator.add 才能收集全部输出。

第 1 点最容易踩坑，实测一下 worker 到底能看到什么：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

class MapState(TypedDict):
    items: list[str]
    results: Annotated[list[str], operator.add]

def dispatch(state: MapState) -> list[Send]:
    return [Send("worker", {"item": it}) for it in state["items"]]

# 一个只报告自己看到哪些键的 worker
def worker_probe(state) -> dict:
    # 把看到的键名排序后写进结果
    return {"results": [f"看到的键: {sorted(state.keys())}"]}

# 搭一张同结构的图
builder = StateGraph(MapState)
# 换成探针 worker
builder.add_node("worker", worker_probe)
# 同样从 START 分发
builder.add_conditional_edges(START, dispatch, ["worker"])
# 出口边
builder.add_edge("worker", END)
# 编译
graph = builder.compile()
# 只发一份，输出更清楚
print(graph.invoke({"items": ["a"], "results": []}))

{'items': ['a'], 'results': ["看到的键: ['item']"]}

只有 item：主状态里的 items 和 results 都看不见。所以 worker 若需要主状态里的公共配置（比如 language、user_id），必须在 Send 里手动带上：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

class MapState(TypedDict):
    items: list[str]
    results: Annotated[list[str], operator.add]

# 正确做法：把 worker 需要的公共字段一并塞进 Send
def dispatch_with_ctx(state: MapState) -> list[Send]:
    # 每份输入 = 自己那条数据 + 需要的公共上下文
    return [Send("worker", {"item": it, "total": len(state["items"])}) for it in state["items"]]

# 这个 worker 既要读自己那条数据，也要读公共的总数
def worker_ctx(state) -> dict:
    # total 不在主状态里，是 dispatch_with_ctx 手动带进来的
    return {"results": [f"处理了 {state['item']}（共 {state['total']} 条）"]}

# 搭图
builder = StateGraph(MapState)
# 注册这个需要公共上下文的 worker
builder.add_node("worker", worker_ctx)
# 用带上下文的分发函数
builder.add_conditional_edges(START, dispatch_with_ctx, ["worker"])
# 出口边
builder.add_edge("worker", END)
# 编译
graph = builder.compile()
# 两个条目，每份输入里都带上了 total
print(graph.invoke({"items": ["a", "b"], "results": []}))

{'items': ['a', 'b'], 'results': ['处理了 a（共 2 条）', '处理了 b（共 2 条）']}

漏带字段的后果是大声报错（这一点比其它坑友好）：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

class MapState(TypedDict):
    items: list[str]
    results: Annotated[list[str], operator.add]

class WorkerState(TypedDict):
    item: str

def worker(state: WorkerState) -> dict:
    return {"results": [f"处理了 {state['item']}"]}

# 故意投递一个空输入
builder = StateGraph(MapState)
# 用最开始那个要读 item 的 worker
builder.add_node("worker", worker)
# Send 的输入是空字典
builder.add_conditional_edges(START, lambda s: [Send("worker", {})], ["worker"])
# 出口边
builder.add_edge("worker", END)
# 编译
graph = builder.compile()
# 运行并捕获异常
try:
    # worker 里 state['item'] 会失败
    print("结果:", graph.invoke({"items": ["a"], "results": []}))
except Exception as e:
    # 打印异常类型和信息
    print("报错:", type(e).__name__, e)

报错: KeyError 'item'

Send 还能和普通节点名混在同一个返回列表里，这是实现「分发 + 同时启动一个汇总/统计节点」的常用手法：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

class MapState(TypedDict):
    items: list[str]
    results: Annotated[list[str], operator.add]

class WorkerState(TypedDict):
    item: str

def worker(state: WorkerState) -> dict:
    return {"results": [f"处理了 {state['item']}"]}

# 一个不需要专属输入的普通节点
def summary(state: MapState) -> dict:
    # 它读的是主状态
    return {"results": [f"共 {len(state['items'])} 条"]}

# 搭图
builder = StateGraph(MapState)
# 注册 worker
builder.add_node("worker", worker)
# 注册普通节点
builder.add_node("summary", summary)
# 返回值里既有 Send 又有普通节点名，出口表把两类目标都声明上
builder.add_conditional_edges(START, lambda s: [Send("worker", {"item": i}) for i in s["items"]] + ["summary"], ["worker", "summary"])
# worker 出口
builder.add_edge("worker", END)
# summary 出口
builder.add_edge("summary", END)
# 编译
graph = builder.compile()
# 运行
print(graph.invoke({"items": ["a", "b"], "results": []}))

{'items': ['a', 'b'], 'results': ['共 2 条', '处理了 a', '处理了 b']}

Send 是 map-reduce 模式的标准做法，比如「把长文档切成 N 段分别摘要，再合并」。它的坑见 §7.4。

6. Command：在节点里边改状态边跳转 #

6.1. 另一种写法 #

条件边把「判断」和「跳转」拆成节点 + 路由函数两部分。Command 让节点自己两件事一起做：

import operator
from typing import Annotated, Literal, TypedDict

from langgraph.graph import END, START, StateGraph

# Command 是「状态更新 + 跳转」的组合返回值
from langgraph.types import Command

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 返回类型注解里的 Literal 就是这个节点声明的出口
def classify_cmd(state: S) -> Command[Literal["refund", "general"]]:
    # 判断逻辑和普通 classify 一样
    kind = "退货" if "退" in state["text"] else "其他"
    # 顺手算出该去哪
    target = "refund" if kind == "退货" else "general"
    # update 改状态，goto 决定去哪，一次返回
    return Command(update={"kind": kind, "log": [f"classify -> {kind}"]}, goto=target)

# 搭图
builder = StateGraph(S)
# 注册这个「自己决定去哪」的节点
builder.add_node("classify", classify_cmd)
# 退货分支
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 通用分支
builder.add_node("general", lambda s: {"log": ["通用分支"]})
# 入口边
builder.add_edge(START, "classify")
# 注意：这里没有 add_conditional_edges，跳转信息全在节点返回值里
builder.add_edge("refund", END)
# 通用分支出口
builder.add_edge("general", END)
# 编译
graph = builder.compile()
# 退货输入
print(graph.invoke(new_input("我要退货")))
# 其他输入
print(graph.invoke(new_input("你好")))

{'text': '我要退货', 'kind': '退货', 'log': ['classify -> 退货', '退货分支']}
{'text': '你好', 'kind': '其他', 'log': ['classify -> 其他', '通用分支']}

效果和条件边完全一样，但少了一个路由函数，也不用 add_conditional_edges。

goto 支持的值和路由函数的返回值一致：

# END 是「跳到终点」的特殊名字
from langgraph.graph import END

# Command 是跳转指令，Send 用来投递专属输入
from langgraph.types import Command, Send

# 直接结束
print(Command(update={"kind": "x"}, goto=END))
# 并行触发多个节点
print(Command(update={"kind": "x"}, goto=["n1", "n2"]))
# 也可以投递 Send（配合 §5.4 的 map-reduce）
print(Command(goto=[Send("worker", {"item": "a"})]))
# 只改状态、不跳转：后续走这个节点的固定边
print(Command(update={"kind": "x"}))

Command(update={'kind': 'x'}, goto='__end__')
Command(update={'kind': 'x'}, goto=['n1', 'n2'])
Command(goto=[Send(node='worker', arg={'item': 'a'})])
Command(update={'kind': 'x'})

最后一种（只给 update）要注意：它退化成普通节点，跳转靠图上的固定边。所以 Command 节点不一定要有 goto，但你得保证「没有 goto 时有固定边接着」，否则图就到此为止了。

6.2. Literal 注解在这里更重要 #

用 Command 时没有 path_map 可写，出口信息只能靠返回类型注解：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 同样的逻辑，但去掉了返回类型注解
def classify_cmd_bare(state: S):
    # 判断类别
    kind = "退货" if "退" in state["text"] else "其他"
    # 算出目标
    target = "refund" if kind == "退货" else "general"
    # 返回 Command，但没有任何地方声明出口
    return Command(update={"kind": kind, "log": [f"classify -> {kind}"]}, goto=target)

# 搭图
builder = StateGraph(S)
# 注册无注解版本
builder.add_node("classify", classify_cmd_bare)
# 退货分支
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 通用分支
builder.add_node("general", lambda s: {"log": ["通用分支"]})
# 入口边
builder.add_edge(START, "classify")
# 退货出口
builder.add_edge("refund", END)
# 通用出口
builder.add_edge("general", END)
# 编译
graph = builder.compile()
# 运行完全正常
print("能跑:", graph.invoke(new_input("我要退货"))["log"])
# 但图画错了：只打印带边的行
print("\n".join(l for l in graph.get_graph().draw_mermaid().splitlines() if "-->" in l or "-." in l))

能跑: ['classify -> 退货', '退货分支']
    __start__ --> classify;
    classify --> __end__;

分支又没了，和 §4.3 一模一样的问题。写上注解就正常：

import operator
from typing import Annotated, Literal, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify_cmd(state: S) -> Command[Literal["refund", "general"]]:
    kind = "退货" if "退" in state["text"] else "其他"
    target = "refund" if kind == "退货" else "general"
    return Command(update={"kind": kind, "log": [f"classify -> {kind}"]}, goto=target)

builder = StateGraph(S)
builder.add_node("classify", classify_cmd)
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
builder.add_node("general", lambda s: {"log": ["通用分支"]})
builder.add_edge(START, "classify")
builder.add_edge("refund", END)
builder.add_edge("general", END)
graph = builder.compile()

# 有注解的版本（§6.1 的 classify_cmd），图上就有虚线了
print("\n".join(l for l in graph.get_graph().draw_mermaid().splitlines() if "-." in l))

    classify -.-> general;
    classify -.-> refund;

注解还能带来编译期校验：

import operator
from typing import Annotated, Literal, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

# 注解里写了一个不存在的节点
def cmd_ghost(state: S) -> Command[Literal["幽灵"]]:
    # 函数体老实按注解跳
    return Command(update={"log": ["c"]}, goto="幽灵")

# 搭图
builder = StateGraph(S)
# 注册它
builder.add_node("classify", cmd_ghost)
# 再放一个正常节点
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 入口边
builder.add_edge(START, "classify")
# 出口边
builder.add_edge("refund", END)
# 编译时校验注解里的节点
try:
    # 这行会抛异常
    builder.compile()
except Exception as e:
    # 打印异常类型和信息
    print("报错:", type(e).__name__, e)

报错: ValueError Found edge ending at unknown node `幽灵`

但校验止步于此：Command 的 goto 没有运行时校验。 这是 Command 和条件边最重要的区别，实测两种情况：

import operator
from typing import Annotated, Literal, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 情况一：goto 一个既不在注解里、也不存在的节点
def cmd_typo(state: S) -> Command[Literal["refund", "general"]]:
    # 模拟把节点名拼错
    return Command(update={"log": ["c"]}, goto="handle_refund")

# 搭图
builder = StateGraph(S)
# 注册它
builder.add_node("classify", cmd_typo)
# 两个正常分支
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 通用分支
builder.add_node("general", lambda s: {"log": ["通用分支"]})
# 入口边
builder.add_edge(START, "classify")
# 退货出口
builder.add_edge("refund", END)
# 通用出口
builder.add_edge("general", END)
# 编译
graph = builder.compile()
# 运行：条件边遇到这种情况会 KeyError，Command 呢？
print("情况一:", graph.invoke(new_input("x")))

# 情况二：goto 一个存在、但没写进注解的节点
def cmd_undeclared(state: S) -> Command[Literal["refund"]]:
    # 注解只声明了 refund，却跳去了 general
    return Command(update={"log": ["c"]}, goto="general")

# 重新搭一张图，上一张已经用完，这里复用同样的变量名
builder = StateGraph(S)
# 注册它
builder.add_node("classify", cmd_undeclared)
# 退货分支
builder.add_node("refund", lambda s: {"log": ["退货分支"]})
# 通用分支
builder.add_node("general", lambda s: {"log": ["通用分支"]})
# 入口边
builder.add_edge(START, "classify")
# 退货出口
builder.add_edge("refund", END)
# 通用出口
builder.add_edge("general", END)
# 编译
graph = builder.compile()
# 运行
print("情况二:", graph.invoke(new_input("x")))

情况一: {'text': 'x', 'kind': '', 'log': ['c']}
情况二: {'text': 'x', 'kind': '', 'log': ['c', '通用分支']}
Task classify with path ('__pregel_pull', 'classify') wrote to unknown channel branch:to:handle_refund, ignoring it.

• 情况一：静默结束。 Command 里的名字拼错了，没有 KeyError，log 里只有 ['c']，只留下最后那行 logging 警告。
• 情况二：照跳。 注解没声明的节点也能跳过去，注解在运行时完全不参与判断。

结论有点反直觉，但很重要：

Command[Literal[...]] 只管画图和编译期校验，不管运行时。 同样是「节点名拼错」这个 bug，条件边 + path_map 会当场 KeyError，Command 会静默结束。

6.3. 怎么选 #

	条件边	Command
判断逻辑放在	独立的路由函数	节点内部
能同时改状态吗	不能	能
出口声明方式	path_map 或 Literal	只有 Literal
编译期校验声明的出口	√	√（写了注解才有）
运行时校验实际跳转	√ KeyError	× 静默结束
图上的可读性	好，路由是独立环节	稍差，得读节点代码
适合	判断依据已在状态里、多个节点复用同一套路由	判断和状态更新是一回事、多智能体交接

实际建议：

• 默认用条件边。 它把路由显式化，还有运行时校验兜底。「跳转目标由模型输出决定」时，这一条几乎是刚需。
• 两种情况用 Command：一是节点算出结果的同时就知道该去哪（比如模型判完意图顺手写进状态，省掉「写状态 → 再读状态」这一圈）；二是多智能体互相「交接」（第 25 章会用到）。
• 用 Command 时，自己在节点里把 goto 兜住，比如 goto=HANDLERS.get(intent, "handle_other")。因为 LangGraph 不会替你检查。

7. 五个只在分支场景才有的坑 #

第 20 章列过四种静默失败，那些是「图结构」层面的。分支又带来五个新坑，其中三个不报错。

坑	表现	报错吗
§7.1 路由返回未知节点名	图静默结束	不报（除非声明了出口）
§7.2 固定边 + 条件边并存	两条都走	不报
§7.3 同一节点两次条件边	具名函数：两条都走；lambda：报错	看写法
§7.4 Send 目标名写错	任务被丢弃	不报
§7.5 路由函数里有副作用	副作用次数不确定	不报

7.1. 路由返回未知节点名 → 静默结束 #

§4.2 已经演示过，这里再强调危险性：这是本章最容易踩、后果最严重的坑。

典型触发场景是模型输出直接当节点名：

# 危险写法：模型给什么就当节点名用
def route_unsafe(state: dict) -> str:
    # intent 是模型给的，可能是任何字符串
    return state["intent"]

模型某次返回了 "Refund"（大写 R）或者 "退货"（中文），路由就废了，日志里却什么都看不到。

7.2. 固定边 + 条件边 = 两条都走 #

这个坑很反直觉。同一个节点既有固定边又有条件边时，它们不是「二选一」，而是都生效：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 搭一张 a → b（固定）+ a → c（条件）的图
builder = StateGraph(S)
# 节点 a
builder.add_node("a", lambda s: {"log": ["a"]})
# 节点 b
builder.add_node("b", lambda s: {"log": ["b"]})
# 节点 c
builder.add_node("c", lambda s: {"log": ["c"]})
# 入口边
builder.add_edge(START, "a")
# 固定边：a 跑完去 b
builder.add_edge("a", "b")
# 条件边：a 跑完也可能去 c
builder.add_conditional_edges("a", lambda s: "c", ["c"])
# b 出口
builder.add_edge("b", END)
# c 出口
builder.add_edge("c", END)
# 编译
graph = builder.compile()
# 用 stream 逐节点观察（第 22 章 §4.1）
print("=== stream ===")
# 每个 chunk 就是一个节点的写入
for chunk in graph.stream(new_input("x")):
    # 打印节点名和它写了什么
    print("  ", chunk)
# 再看最终状态
print("最终:", graph.invoke(new_input("x")))

=== stream ===
   {'a': {'log': ['a']}}
   {'b': {'log': ['b']}}
   {'c': {'log': ['c']}}
最终: {'text': 'x', 'kind': '', 'log': ['a', 'b', 'c']}

b 和 c 都跑了。

道理其实自洽：两种边都表示「触发」，a 跑完就把两条边都点亮了，等价于并行扇出。但若本意是「默认走 b，特殊情况走 c」，那就完全错了。

正确写法是把默认分支也放进条件边：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 正确写法：所有出口都由同一个路由函数决定
builder = StateGraph(S)
# 节点 a
builder.add_node("a", lambda s: {"log": ["a"]})
# 节点 b（默认分支）
builder.add_node("b", lambda s: {"log": ["b"]})
# 节点 c（特殊分支）
builder.add_node("c", lambda s: {"log": ["c"]})
# 入口边
builder.add_edge(START, "a")
# 只有一条条件边，默认分支写在 else 里
builder.add_conditional_edges("a", lambda s: "c" if s["kind"] == "退货" else "b", ["b", "c"])
# b 出口
builder.add_edge("b", END)
# c 出口
builder.add_edge("c", END)
# 编译
graph = builder.compile()
# 运行：只走一条
print("修正后:", graph.invoke(new_input("x"))["log"])

修正后: ['a', 'b']

好消息是这个坑能被看出来：实线和虚线同时从 a 出发。

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

builder = StateGraph(S)
builder.add_node("a", lambda s: {"log": ["a"]})
builder.add_node("b", lambda s: {"log": ["b"]})
builder.add_node("c", lambda s: {"log": ["c"]})
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_conditional_edges("a", lambda s: "c", ["c"])
builder.add_edge("b", END)
builder.add_edge("c", END)
graph = builder.compile()

# 打印错误版本里跟 a 有关的边
print("\n".join(l for l in graph.get_graph().draw_mermaid().splitlines() if l.strip().startswith("a ")))
# 或者直接读结构，写成断言更方便
for e in graph.get_graph().edges:
    # 找出所有从 a 出发的边
    if e.source == "a":
        # 打印目标和边的类型
        print(f"  a -> {e.target}  conditional={e.conditional}")

    a --> b;
    a -.-> c;
  a -> b  conditional=False
  a -> c  conditional=True

同一个节点同时出现实线和虚线出边，几乎总是 bug。 这条规则可以直接写进 CI 检查。

顺便一提，用 Command 也躲不开这个坑：节点返回 goto="refund"，同时图上又有一条 add_conditional_edges("classify", ...)，结果是两个目标都会执行。跳转来源只能有一处。

7.3. 同一节点加两次条件边 #

先看用 lambda 的情况：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

# 搭图
builder = StateGraph(S)
# 分类节点
builder.add_node("classify", classify)
# 两个候选节点
builder.add_node("a", lambda s: {"log": ["a"]})
# 第二个候选节点
builder.add_node("b", lambda s: {"log": ["b"]})
# 入口边
builder.add_edge(START, "classify")
# 第一条条件边，路由函数是 lambda
builder.add_conditional_edges("classify", lambda s: "a", ["a"])
# 再加一条，路由函数还是 lambda
try:
    # 预期报错
    builder.add_conditional_edges("classify", lambda s: "b", ["b"])
    # 如果没报错会打印这行
    print("居然没报错")
except Exception as e:
    # 打印异常类型和信息
    print("报错:", type(e).__name__, e)

报错: ValueError Branch with name `None` already exists for node `classify`

看到这个报错，很多人会总结成「一个节点只能有一个路由函数」。但换成具名函数，结果完全不同：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph

class S(TypedDict):
    text: str
    kind: str
    log: Annotated[list[str], operator.add]

def classify(state: S) -> dict:
    kind = "退货" if "退" in state["text"] else "其他"
    return {"kind": kind, "log": [f"classify -> {kind}"]}

def new_input(text: str) -> S:
    return {"text": text, "kind": "", "log": []}

# 两个具名路由函数
def r1(state: S) -> str:
    # 第一个路由：固定去 a
    return "a"

# 第二个路由：固定去 b
def r2(state: S) -> str:
    # 固定去 b
    return "b"

# 搭图
builder = StateGraph(S)
# 分类节点
builder.add_node("classify", classify)
# 候选节点 a
builder.add_node("a", lambda s: {"log": ["a"]})
# 候选节点 b
builder.add_node("b", lambda s: {"log": ["b"]})
# 入口边
builder.add_edge(START, "classify")
# 第一条条件边
builder.add_conditional_edges("classify", r1, ["a"])
# 第二条条件边：这次不报错了
builder.add_conditional_edges("classify", r2, ["b"])
# a 出口
builder.add_edge("a", END)
# b 出口
builder.add_edge("b", END)
# 编译
graph = builder.compile()
# 运行看看到底走了哪条
print("结果:", graph.invoke(new_input("x"))["log"])

结果: ['classify -> 其他', 'a', 'b']

两条都走了。 原因在 §4.2 提过：每条条件边在内部有个「分支名」，具名函数用函数名当分支名，lambda 没有可用名字（记作 None）。所以：

• 两个 lambda：分支名都是 None，撞名，于是大声报错。
• 两个具名函数：分支名不同，和平共存，于是两条都触发，退化成 §7.2 那个坑。

这比「只能有一个」的说法危险得多：把 lambda 改成具名函数这种纯重构，可能把编译期错误变成静默的并行执行。 想表达多个维度的判断，就在一个路由函数里写完，或拆成多级路由（§5.3）。

7.4. Send 目标名写错 → 任务被丢弃 #

Send 里的节点名同样不做运行时校验，而且连 path_map 都救不了：

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

class MapState(TypedDict):
    items: list[str]
    results: Annotated[list[str], operator.add]

class WorkerState(TypedDict):
    item: str

def worker(state: WorkerState) -> dict:
    return {"results": [f"处理了 {state['item']}"]}

# 搭一张 map-reduce 图
builder = StateGraph(MapState)
# 注册 worker
builder.add_node("worker", worker)
# Send 的目标名故意写错，但出口表写的是正确的 worker
builder.add_conditional_edges(START, lambda s: [Send("worker_x", {"item": "a"})], ["worker"])
# 出口边
builder.add_edge("worker", END)
# 编译
graph = builder.compile()
# 运行：既不报错，也没有任何 worker 执行
print("结果:", graph.invoke({"items": ["a"], "results": []}))

结果: {'items': ['a'], 'results': []}
Ignoring unknown node name worker_x in pending sends

results 是空的，任务凭空消失了，只留下第二行那句 logging 警告。

注意这里和 §4.2 的区别：普通路由返回值会拿去和 path_map 比对，Send 的目标名不会。 所以写 Send 时的自我保护手段只有一个：别手写字符串。

import operator
from typing import Annotated, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Send

class MapState(TypedDict):
    items: list[str]
    results: Annotated[list[str], operator.add]

class WorkerState(TypedDict):
    item: str

def worker(state: WorkerState) -> dict:
    return {"results": [f"处理了 {state['item']}"]}

# 把节点名定义成常量，注册和投递都用它
WORKER = "worker"

# 分发函数里引用常量，拼错会变成 NameError（大声失败）
def dispatch_safe(state: MapState) -> list[Send]:
    # 用常量，不用字面量
    return [Send(WORKER, {"item": it}) for it in state["items"]]

# 搭图：注册节点时也用同一个常量，名字就不可能对不上
builder = StateGraph(MapState)
# 注册 worker
builder.add_node(WORKER, worker)
# 分发，出口表同样引用常量
builder.add_conditional_edges(START, dispatch_safe, [WORKER])
# 出口边
builder.add_edge(WORKER, END)
# 编译
graph = builder.compile()
# 运行：两份任务都正常落地
print("结果:", graph.invoke({"items": ["a", "b"], "results": []}))

结果: {'items': ['a', 'b'], 'results': ['处理了 a', '处理了 b']}

这条建议对普通路由函数同样适用，只是在 Send 场景下，它从「好习惯」变成了「唯一防线」。

7.5. 路由函数里做了副作用 #

路由函数看起来像普通函数，容易顺手写点别的：

# 错误示范：路由函数里干了不该干的事
def route_dirty(state: dict) -> str:
    # 千万别：写库这类副作用，执行次数不受你控制（见 §3.4）
    # db.save(state)
    # 也别：模型调用的结果无处可存，路由函数只能返回节点名
    # result = model.invoke(...)
    # 只做映射，这才是它该干的
    return "refund"

两个问题：

1. 执行次数不确定。 §3.4 实测过：正常跑一次；从源节点之前重放会再跑一次。副作用于是可能重复。
2. 结果无处可存。 路由函数的返回值只能是节点名；模型调用的结果、算出来的中间量一律没地方放，下一个节点也读不到。

要调模型就单独开一个节点，把结果写进状态，路由函数只读那个字段。§8 的实战就是这个结构，也是本章推荐的标准分层：

节点：干活、调模型、写状态（可以有副作用，有检查点保护）
边：只读状态、只回答「去哪」（必须是纯函数）

8. 实战：意图路由 #

8.1. 设计 #

START → classify（模型判意图，结构化输出约束到四个枚举值）
      → route（纯函数：置信度不足 → 转人工；否则查表）
         ├─ handle_refund     退货售后
         ├─ handle_shipping   物流查询
         ├─ handle_tech       技术支持
         └─ handle_other      兜底 / 转人工
      → END

三个设计要点，每一个都对应前面讲过的坑：

设计	防住哪个坑
classify 用 with_structured_output 把意图钉死在 Literal 枚举里	§7.1 模型自由发挥
route 是纯函数，用 HANDLERS.get(..., "handle_other") 兜底	§7.1 枚举被绕过时静默走丢
add_conditional_edges 显式传 path_map	§4.2 编译期校验 + §4.3 图画得对
调模型的活全在节点里，route 一行 I/O 都没有	§7.5 副作用重复执行

另外加了一个置信度阈值：模型自己说不准时直接转人工，比强行分类更可靠。注意置信度是模型自己报的，不是概率意义上的严格数值，但作为「模型有没有把握」的粗粒度信号，足够好用。

为什么不用 Command 把 classify 和 route 合成一个节点？可以，但这里故意分开，理由见 §6.3 那张表的第五行：跳转目标来自模型输出时，条件边的运行时校验是最后一道保险。（练习 4 会让你改成 Command 版，对比一下。）

8.2. intent_router.py #

"""实战：意图路由图"""

# 让文件里的类型注解延迟求值，避免前向引用问题
from __future__ import annotations

# operator.add 给 log 字段当 reducer
import operator
# lru_cache 用来让模型客户端只构造一次
from functools import lru_cache
# 类型工具：Annotated 挂 reducer，Literal 约束枚举，TypedDict 定义状态
from typing import Annotated, Literal, TypedDict

# 从 .env 读 API Key
from dotenv import load_dotenv
# 统一的模型入口
from langchain.chat_models import init_chat_model
# 图的三个基本构件
from langgraph.graph import END, START, StateGraph
# Pydantic 用来定义结构化输出
from pydantic import BaseModel, Field

# 加载环境变量，override=True 让 .env 覆盖已有的同名变量
load_dotenv(override=True)

# 四个意图枚举，模型和路由表共用这一个定义
IntentName = Literal["refund", "shipping", "tech", "other"]

# 意图 -> 处理节点名。路由函数查这张表，图也用它声明出口
HANDLERS: dict[str, str] = {
    # 退货退款换货
    "refund": "handle_refund",
    # 发货物流配送
    "shipping": "handle_shipping",
    # 产品使用与故障
    "tech": "handle_tech",
    # 其他 / 兜底
    "other": "handle_other",
}

# 置信度阈值：低于这个值就不相信模型的分类
MIN_CONFIDENCE = 0.6

# 系统提示单独抽出来：把分类标准和「拿不准怎么办」都写清楚
SYSTEM_PROMPT = (
    # 先说角色和任务
    "你是客服工单分类器。把用户问题归到四类之一："
    # 给每个枚举值配一句解释，分类准确率会明显提升
    "refund（退货退款换货）、shipping（发货物流配送）、"
    # 剩下两类的解释
    "tech（产品使用与故障）、other（其他）。"
    # 明确「拿不准怎么办」，这是置信度兜底能生效的前提
    "拿不准就选 other 并给低置信度。"
)

# 结构化输出的模式：模型必须按这个结构回答
class Intent(BaseModel):
    """约束模型只能在四个意图里选，顺便要一个置信度。"""

    # 意图名，Literal 会变成 JSON Schema 里的 enum
    name: IntentName = Field(description="用户意图分类")
    # 置信度，ge/le 会变成 schema 里的取值范围约束
    confidence: float = Field(description="0~1 的置信度", ge=0.0, le=1.0)
    # 要一句理由，既方便人看，也能让模型「想一下再答」
    reason: str = Field(description="一句话理由")

# 图的状态：一次问答的全部上下文
class RouterState(TypedDict):
    # 用户问题
    question: str
    # 模型判出的意图名
    intent: str
    # 模型给的置信度
    confidence: float
    # 模型给的理由
    reason: str
    # 最终回复
    answer: str
    # 执行轨迹，累加
    log: Annotated[list[str], operator.add]

# 模型客户端只构造一次，避免每次 invoke 都重新建连接
@lru_cache(maxsize=1)
def get_model():
    """构造带结构化输出的分类模型（缓存，只构造一次）。"""
    # thinking 模式不支持 tool_choice，而 with_structured_output 默认走 function_calling
    model = init_chat_model(
        # 用 deepseek 的 flash 模型，够快够便宜
        "deepseek:deepseek-v4-flash",
        # 分类任务要稳定，温度调到 0
        temperature=0,
        # 显式关掉思考模式
        extra_body={"thinking": {"type": "disabled"}},
    )
    # 绑定输出模式，返回值会是 Intent 实例
    return model.with_structured_output(Intent)

# 节点：调模型判意图
def classify(state: RouterState) -> dict:
    """用模型判意图，并用结构化输出把结果钉死在枚举里。"""
    # 取缓存好的模型客户端
    model = get_model()
    # 一条 system 定规则 + 一条 user 给内容
    result = model.invoke(
        [
            # 系统消息用上面抽出来的常量
            {"role": "system", "content": SYSTEM_PROMPT},
            # 用户消息就是原始问题
            {"role": "user", "content": state["question"]},
        ]
    )
    # 把模型输出的三个字段都写进状态，路由函数只读其中两个
    return {
        # 意图名
        "intent": result.name,
        # 置信度
        "confidence": result.confidence,
        # 理由
        "reason": result.reason,
        # 记一笔轨迹，带上置信度方便复盘
        "log": [f"classify -> {result.name} ({result.confidence:.2f})"],
    }

# 路由函数：整个图里唯一决定「去哪」的地方
def route(state: RouterState) -> str:
    """路由函数：只读状态、只返回节点名，不改状态。"""
    # 第一道判断：模型自己没把握就直接转人工
    if state["confidence"] < MIN_CONFIDENCE:
        # 走兜底节点
        return "handle_other"
    # 第二道判断：查表，并用 get 兜底
    # 模型万一返回了枚举外的值，也不会静默走丢（§7.1）
    return HANDLERS.get(state["intent"], "handle_other")

# 处理节点：退货售后
def handle_refund(state: RouterState) -> dict:
    # 真实项目里这里会查订单、算退款金额
    return {"answer": "已为您转接售后：签收 7 天内可无理由退货。", "log": ["refund"]}

# 处理节点：物流查询
def handle_shipping(state: RouterState) -> dict:
    # 真实项目里这里会查物流接口
    return {"answer": "现货 48 小时内发出，物流单号会短信通知。", "log": ["shipping"]}

# 处理节点：技术支持
def handle_tech(state: RouterState) -> dict:
    # 真实项目里这里会走知识库检索（第 15 章的 RAG）
    return {"answer": "请描述具体故障现象，我帮您排查。", "log": ["tech"]}

# 处理节点：兜底 / 转人工
def handle_other(state: RouterState) -> dict:
    # 区分两种情况：模型确定是「其他类」，还是模型自己弃权
    low = state["confidence"] < MIN_CONFIDENCE
    # 只有弃权时才提示转人工
    tip = "（置信度不足，转人工）" if low else ""
    # 拼出最终回复
    return {"answer": f"已记录您的问题，稍后由人工回复。{tip}", "log": ["other"]}

# 把图组装起来
def build_graph():
    """组装意图路由图。"""
    # 绑定状态类型，未编译的图纸叫 builder
    builder = StateGraph(RouterState)
    # 注册分类节点
    builder.add_node("classify", classify)
    # 按表批量注册四个处理节点，避免手写四遍导致名字不一致
    for name in HANDLERS.values():
        # globals() 按名字取到同名函数
        builder.add_node(name, globals()[name])
    # 入口边
    builder.add_edge(START, "classify")
    # path_map 显式列出全部出口：既能让 compile 校验，也能让 mermaid 画对
    builder.add_conditional_edges("classify", route, list(HANDLERS.values()))
    # 四个处理节点跑完都结束
    for name in HANDLERS.values():
        # 每个处理节点连到 END
        builder.add_edge(name, END)
    # 起个名字，LangSmith 里好认（第 22 章 §2.6）
    return builder.compile(name="intent-router")

# 造一份完整初始状态
def new_input(question: str) -> RouterState:
    """造初始状态：非 reducer 字段必须给零值，否则下游读取会 KeyError。"""
    # 六个字段一个都不能少（第 22 章 §3.2）
    return {
        # 用户问题
        "question": question,
        # 意图待填
        "intent": "",
        # 置信度待填
        "confidence": 0.0,
        # 理由待填
        "reason": "",
        # 回复待填
        "answer": "",
        # 轨迹从空开始
        "log": [],
    }

# 直接运行本文件时的演示入口
if __name__ == "__main__":
    # 组装图
    graph = build_graph()

    # 先把图纸打出来，确认四条虚线都在
    print(graph.get_graph().draw_mermaid())

    # 五个问题：四类各一个，最后一个是模糊输入
    QUESTIONS = [
        # 应该判成 refund
        "买的杯子有裂缝，我想退掉",
        # 应该判成 shipping
        "下单三天了还没发货，什么时候能到",
        # 应该判成 tech
        "净水器滤芯指示灯一直闪红灯怎么办",
        # 应该判成 other（高置信度）
        "你们公司在哪个城市",
        # 应该判成 other（低置信度，触发转人工）
        "嗯",
    ]
    # 逐个跑
    for q in QUESTIONS:
        # 每次都用全新的初始状态，避免串味
        out = graph.invoke(new_input(q))
        # 打印问题
        print(f"\n问：{q}")
        # 打印意图和置信度
        print(f"  意图：{out['intent']}  置信度：{out['confidence']:.2f}")
        # 打印模型给的理由
        print(f"  理由：{out['reason']}")
        # 打印实际走过的路径
        print(f"  路径：{out['log']}")
        # 打印最终回复
        print(f"  回复：{out['answer']}")

8.3. 图长这样 #

graph TD;
    __start__([<p>__start__</p>]):::first
    classify(classify)
    handle_refund(handle_refund)
    handle_shipping(handle_shipping)
    handle_tech(handle_tech)
    handle_other(handle_other)
    __end__([<p>__end__</p>]):::last
    __start__ --> classify;
    classify -.-> handle_other;
    classify -.-> handle_refund;
    classify -.-> handle_shipping;
    classify -.-> handle_tech;
    handle_other --> __end__;
    handle_refund --> __end__;
    handle_shipping --> __end__;
    handle_tech --> __end__;

四条虚线、四个出口，:::first / :::last 也都在，这就是声明出口换来的。把 list(HANDLERS.values()) 那个参数删掉再打印一次，你会看到 classify --> __end__ 和四个孤立节点。

8.4. 运行结果 #

问：买的杯子有裂缝，我想退掉
  意图：refund  置信度：0.95
  理由：用户购买的商品有质量问题（裂缝），明确表示想退货，属于退货退款类。
  路径：['classify -> refund (0.95)', 'refund']
  回复：已为您转接售后：签收 7 天内可无理由退货。

问：下单三天了还没发货，什么时候能到
  意图：shipping  置信度：0.95
  理由：用户询问发货时间和到货时间，属于物流配送问题。
  路径：['classify -> shipping (0.95)', 'shipping']
  回复：现货 48 小时内发出，物流单号会短信通知。

问：净水器滤芯指示灯一直闪红灯怎么办
  意图：tech  置信度：0.95
  理由：用户询问净水器滤芯指示灯闪红灯的故障处理方法，属于产品使用与故障类问题。
  路径：['classify -> tech (0.95)', 'tech']
  回复：请描述具体故障现象，我帮您排查。

问：你们公司在哪个城市
  意图：other  置信度：0.95
  理由：用户询问公司所在城市，与退货、物流、产品故障均无关，属于其他类问题。
  路径：['classify -> other (0.95)', 'other']
  回复：已记录您的问题，稍后由人工回复。

问：嗯
  意图：other  置信度：0.30
  理由：用户只发了"嗯"，没有提供任何具体问题信息，无法判断意图。
  路径：['classify -> other (0.30)', 'other']
  回复：已记录您的问题，稍后由人工回复。（置信度不足，转人工）

（置信度的具体数值和理由措辞每次跑会有出入；「嗯」这类模糊输入可能给 0.1 也可能给 0.3，但都远低于 0.6 的阈值，路由结果是稳定的。）

最后两条对比着看很有意思：都路由到了 handle_other，原因却完全不同。

• 「你们公司在哪个城市」→ 置信度 0.95，模型很确定这就是「其他类」
• 「嗯」→ 置信度 0.30，模型明确表示自己判断不了

前者是分类结果，后者是弃权。因为把置信度也放进了状态，handle_other 能区分这两种情况，给出不同话术。如果只用意图名做路由，这两种就混为一谈了。这也是「路由依据要写进状态」的一个额外好处：下游节点能看到路由是怎么做出来的。

8.5. 验收清单 #

1. 四类问题各自路由正确，log 里能看到完整路径
2. 低置信度触发兜底，handle_other 的回复带「转人工」提示
3. draw_mermaid() 画出四条虚线，且有 :::first / :::last，图和行为一致
4. route 是纯函数，不调模型、不改状态、不打日志到外部系统
5. 意图被 Literal 约束，模型不可能返回枚举外的值
6. HANDLERS.get 有默认值，即使意图异常也不会静默结束
7. 把 path_map 参数删掉后图纸立刻变错，说明它确实在起作用

9. 实用约定与坑 #

约定	说明
出口一定要声明	path_map 或 Literal，换来编译期校验 + 运行期报错 + 图画对（§4）
出口清单要动态生成时用 path_map	Literal 只能写字面量（§4.4）
想在图上看到「哪个返回值走哪条边」用字典 path_map	只有字典形式的虚线带标签（§4.1、§4.3）
用 Command 就写 Command[Literal[...]]	否则图画错；但它不做运行时校验，goto 要自己兜（§6.2）
路由函数保持纯净	只读状态、只返回名字、不调 API（§3.3、§7.5）
模型输出当路由依据时必须约束枚举	with_structured_output + Literal（§7.1）
查表路由一律用 get(x, 默认值)	别用 [x]（§7.1）
END 也要写进 path_map	漏了就是 KeyError '__end__'（§5.1）
默认分支也放进条件边	别用固定边当「默认路径」（§7.2）
一个节点只由一处决定跳转	条件边、固定边、Command 三者别混用（§7.2、§7.3）
节点名定义成常量再引用	Send 场景下这是唯一防线（§7.4）
分支层级控制在两三层内	再深就抽子图（§5.3）
改完分支跑一次 draw_mermaid()	能看出双边并存、孤立节点（§4.3、§7.2）
断言用 get_graph().edges 而不是看图	conditional 和 data 字段可编程（§4.3）

会当场报错的坑：

现象	原因	处理
KeyError: 'xxx'	路由返回值不在出口清单里	这是好事，说明校验生效了（§4.2）
KeyError: None	路由函数有分支忘了 return	补上 else（§3.3）
KeyError: '__end__'	路由返回了 END 但没写进 path_map	把 END 加进出口清单（§5.1）
TypeError: unhashable type: 'dict'	路由函数返回了状态更新字典	路由不能改状态，改用 Command（§3.3）
ValueError: ... found unknown target	path_map / Literal 指向不存在的节点	编译期就拦住了，改名字（§4.2、§4.4）
ValueError: Found edge ending at unknown node	Command[Literal[...]] 里的节点不存在	同上（§6.2）
ValueError: Branch with name None already exists	给同一节点加了两次 lambda 条件边	合并成一个路由函数（§7.3）
TypeError: ... missing 1 required positional argument	路由函数写了第三个参数	最多两个：state、config（§3.3）
TypeError: No synchronous function provided	路由函数是 async def，却用 invoke 跑	改用 ainvoke（§3.3）
InvalidUpdateError: ... only one value per step	返回列表扇出后共写一个字段	加 reducer（§5.2、第 21 章）
KeyError: 'item'（Send 场景）	Send 的输入没包含节点需要的字段	该带的公共字段一起塞进 Send（§5.4）

不报错但结果不对的坑：

现象	原因	处理
分支节点从没执行，也不报错	路由返回了未知节点名，且没声明出口	声明出口（§4.2）；老项目可先捞日志（§4.5）
draw_mermaid() 里分支消失，直连 __end__	没声明出口	补 path_map 或 Literal（§4.3）
mermaid 里没有 :::first / :::last	图里有孤立节点	顺着找哪个节点没入边（§4.3）
两个分支都执行了	同一节点同时有固定边和条件边	默认分支也写进条件边（§7.2）
两个分支都执行了，且都是条件边	同一节点挂了两个具名路由函数	合并成一个（§7.3）
Command 的 goto 拼错却没报错	Command 不做运行时校验	自己在节点里 get(..., 兜底)（§6.2）
Send 的任务凭空消失	Send 目标名写错，且不受 path_map 约束	节点名用常量（§7.4）
路由函数里返回的字典没生效	路由函数不能改状态	改用 Command（§3.3、§6）
路由返回空列表后图直接结束	空列表 = 不触发任何节点	用 targets or "兜底节点"（§5.2）
路由函数里的副作用执行了两次	从源节点之前重放会重算路由	副作用搬进节点（§3.4、§7.5）

口诀：

判断放节点，路由放边上。 路由函数只回答「去哪」，不做别的；而「去哪」的所有可能，都要在图上声明清楚。

10. 练习 #

机制验证（不调模型）：

1. 感受静默失败：故意把路由函数里的节点名拼错，先不声明出口跑一次，再加上 path_map 跑一次，对比两次的表现。
2. 画图验证：对同一张分支图，分别在「什么都不写」「Literal 注解」「列表 path_map」「字典 path_map」四种情况下打印 draw_mermaid()，重点观察虚线标签和 :::first 标记。
3. 推翻一个说法：很多资料说「Literal 注解只影响画图」。用 §4.4 的方法验证它其实也做运行时和编译期校验，然后找出 Literal 和 path_map 的真正差异。
4. 双边陷阱：复现 §7.2，用 stream 观察 b 和 c 都执行；然后写一个函数，输入编译后的图，输出所有「同时有实线和虚线出边」的节点。
5. 具名 vs lambda：复现 §7.3 的两种情况，体会「把 lambda 重构成具名函数」如何把报错变成静默双跑。
6. Command 的短板：给一个 Command 节点的 goto 写一个不存在的节点名，确认它静默结束；再换成条件边，确认它 KeyError。
7. 路由执行次数：复现 §3.4，把恢复和重放两种情况的计数都跑出来。

能力构建：

8. 改用 Command：把 §8 的 classify + route 合并成一个返回 Command 的节点，自己在节点里做兜底，对比两版代码的可读性和安全性。
9. 加一层分支：给 handle_refund 后面接一个条件边，金额大于 1000 走 manual_review，否则直接 END，并保证 END 写进了出口清单。
10. 动态并行：用 Send 实现「把一段长文本按段落分给多个 worker 统计字数，最后汇总」，worker 需要知道总段数，想清楚这个字段怎么传。
11. 置信度调参：把 MIN_CONFIDENCE 改成 0.99，看哪些问题会掉进兜底分支，思考阈值该怎么定。
12. 绕过枚举：在 route 前面加一个节点，手动把 state["intent"] 改成 "Refund"（大写），验证 HANDLERS.get 的兜底确实生效了；再把 get 换成 []，看看变成什么错误。
13. 写进 CI：把「所有条件边都声明了出口」「没有节点同时有实线和虚线出边」这两条写成 pytest 断言。

11. 本章小结 #

1. add_conditional_edges(source, path, path_map) 让图分叉：路由函数读状态，返回「下一步是谁」。它在源节点的写入合并进状态之后执行。
2. 路由函数是纯函数：只读状态、只返回节点名（或列表）、不改状态、不做副作用；最多两个参数（state、config）；可以是 async def，但那样整张图只能异步跑。
3. 出口必须声明。 path_map（列表或字典）和 Literal 返回注解效果等价，都同时带来三件事：编译期校验声明的节点存在、运行期对未知返回值抛 KeyError、draw_mermaid() 画得对。
4. 什么都不写时，路由返回未知节点名会静默结束：不抛异常，warnings 也抓不到，只有一条 logging 警告（logger 名 langgraph，级别 WARNING，可以捞出来做自检）。这是本章最危险的坑。
5. Literal 和 path_map 的真正差异只有两点：Literal 必须是写死的字面量（出口清单不能运行时生成），只有字典形式的 path_map 能在虚线上带标签。
6. draw_mermaid() 里 :::first / :::last 消失，说明图里有孤立节点；要写断言就读 get_graph().edges 的 conditional 和 data 字段。
7. 路由函数可以返回 END（直接结束，END 也要写进出口清单）、列表（并行扇出，同一超步）、空列表（什么都不做，静默结束）。
8. Send(节点名, 专属输入) 实现 map-reduce 式的动态并行：目标节点只能看到 Send 里给的那份输入，公共字段要手动带上；结果靠 reducer 汇总；Send 可以和普通节点名混在同一个返回列表里。
9. Command(update=..., goto=...) 让节点边改状态边跳转，省掉路由函数；goto 支持节点名、END、列表、Send；只给 update 不给 goto 时靠固定边继续。
10. Command[Literal[...]] 只管画图和编译期校验，不做运行时校验：goto 拼错是静默结束，跳到注解外的已有节点也照跳。所以默认用条件边，Command 留给「判断和状态更新本就是一回事」的场景，并自己兜底。
11. 固定边和条件边并存时两条都走，不是二选一；Command 和条件边并存同样如此。跳转来源只能有一处。
12. 同一节点加两次条件边：两个 lambda 会撞名报错（Branch with name None already exists），两个具名函数却能共存并两条都走，重构时特别容易踩。
13. Send 的目标名不受 path_map 约束，写错就是任务被静默丢弃。节点名请定义成常量。
14. 路由函数的执行次数不由你控制：正常一次，下游失败恢复不重算，从源节点之前重放会重算。这就是不能有副作用的硬理由。
15. 模型输出当路由依据时，三重保险缺一不可：Literal 约束枚举、get 兜底默认值、path_map 声明出口。
16. 本章产出：意图路由图，模型判意图 + 结构化输出约束 + 置信度阈值兜底；handle_other 还能区分「确定是其他类」和「模型弃权」两种情况。

到这里图能分叉了，但还是「一去不回头」。下一章让图回头：循环与终止，模型 ↔ 工具环怎么转、怎么保证它一定会停。