条件分支：add_conditional_edges 与动态路由决策

本文是「LangGraph 核心与实战」系列专栏的第 4 篇。专栏总览参见：《LangGraph 核心架构全景：用状态图构建可控 Agent 工作流》。

在顺序图中，所有节点都由固定的 add_edge 单向连接，每一步的去向在编写代码时就已被锁死。但在真实的智能体应用中，系统必须根据运行期的动态数据做出选择：用户提问是常规咨询还是退款投诉？代码测试通过了还是报错了？信息充分还是需要再次补充检索？

LangGraph 通过 条件边（Conditional Edge） 来承载这些动态分流。它将“业务逻辑计算”与“下一步去哪里”彻底解耦，使控制流的流转规则成为系统架构中的一等公民。

核心 API：add_conditional_edges 机制

LangGraph 定义条件路由的标准接口如下：

1
2
3
4
5
6
7
8
9
graph.add_conditional_edges(
    source="analyze_node",       # 1. 触发决策的源节点
    path=routing_function,       # 2. 路由函数：接收 State，返回分支标识
    path_map={                   # 3. 映射字典：将标识路由到目标节点
        "praise": "thank_node",
        "bug": "escalate_node",
        "end": END
    }
)

这三个参数的分工必须明确：

  • source：当前刚执行完毕的工位节点；
  • routing_function：一个普通的 Python 函数，签名是 (state: State) -> str。它只读当前状态，推断出代表分支意图的短标签（如 "praise"、"bug"）；
  • path_map：显式映射表，将路由函数返回的字符串标签与图中的下一个具体 Node（或内置的 END）关联起来。

如果路由函数直接返回图中的节点名，path_map 参数可以省略；但在生产代码中，强烈建议始终显式传入 path_map。这样不仅可以在 compile() 编译阶段触发严格的静态校验（如果映射指向了未注册的节点会直接报错），而且便于对路由函数进行独立的断言测试。

条件分支与代码内 if-else 的区别

为什么不直接在节点内部写 if-else 调用下游函数？

维度 在节点内部写 if-else 使用 add_conditional_edges
职责边界 节点既要处理业务数据，又要感知下游拓扑，职责严重耦合 节点保持纯粹：节点只负责计算，不关心下一个处理者是谁
拓扑可见度 分支隐藏在深层代码中，从图的全局视角无法看出存在分流 拓扑全局透明：分支规则作为图的边声明，直接参与可视化与审计
单元测试难度 必须 Mock 整个下游链条才能测试分支逻辑 路由函数是纯函数，只需构造 State 字典即可独立单测
环形回路支持 命令式循环容易引入死锁 天然支持环（Cycle）：条件边可以将目标指向前序节点，形成可控自纠环

实战：用户反馈自适应分流系统

构建一个实际的工单客服分流系统，包含情感与意图分析、差异化响应路径以及短路退出机制。

1. 状态契约建模

1
2
3
4
5
6
7
from typing import TypedDict, Optional

class FeedbackState(TypedDict):
    feedback_text: str          # 原始用户反馈
    category: Optional[str]     # 分类标签: "praise" | "complaint" | "inquiry" | "spam"
    urgency_level: int          # 紧急程度: 1 ~ 5
    response: Optional[str]     # 最终输出文本

2. 节点与独立路由函数实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# 阶段 1: 文本分析节点
def analyze_feedback_node(state: FeedbackState) -> dict:
    text = state["feedback_text"].lower()
    
    # 模拟规则或轻量模型分类
    if any(w in text for w in ["垃圾", "退钱", "崩溃", "投诉"]):
        return {"category": "complaint", "urgency_level": 5}
    elif any(w in text for w in ["太棒了", "好评", "感谢", "喜欢"]):
        return {"category": "praise", "urgency_level": 1}
    elif any(w in text for w in ["怎么用", "如何", "在哪里"]):
        return {"category": "inquiry", "urgency_level": 2}
    else:
        return {"category": "spam", "urgency_level": 0}

# 阶段 2: 差异化业务处理节点
def handle_complaint_node(state: FeedbackState) -> dict:
    return {"response": f"【加急通道】：已为您创建 P0 级工单，客服主管将在 10 分钟内主动联系您。"}

def handle_praise_node(state: FeedbackState) -> dict:
    return {"response": "【感谢反馈】：非常感谢您的认可与支持，已为您派发一张体验券！"}

def handle_inquiry_node(state: FeedbackState) -> dict:
    return {"response": "【智能客服】：正在为您检索知识库使用手册，请稍候..."}

# 阶段 3: 专职路由函数 (只读状态，计算决策)
def feedback_router(state: FeedbackState) -> str:
    cat = state.get("category")
    if cat == "complaint":
        return "route_complaint"
    elif cat == "praise":
        return "route_praise"
    elif cat == "inquiry":
        return "route_inquiry"
    else:
        return "route_spam"  # 垃圾信息直接短路结束

3. 编排图拓扑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from langgraph.graph import StateGraph, START, END

builder = StateGraph(FeedbackState)

# 注册所有节点
builder.add_node("analyze", analyze_feedback_node)
builder.add_node("complaint_handler", handle_complaint_node)
builder.add_node("praise_handler", handle_praise_node)
builder.add_node("inquiry_handler", handle_inquiry_node)

# 入口连接到分析节点
builder.add_edge(START, "analyze")

# 核心：装配条件边
builder.add_conditional_edges(
    source="analyze",
    path=feedback_router,
    path_map={
        "route_complaint": "complaint_handler",
        "route_praise": "praise_handler",
        "route_inquiry": "inquiry_handler",
        "route_spam": END  # 垃圾信息不经过任何处理节点，直接结束
    }
)

# 各业务处理分支汇聚到终点
builder.add_edge("complaint_handler", END)
builder.add_edge("praise_handler", END)
builder.add_edge("inquiry_handler", END)

app = builder.compile()

4. 运行验证与分支路径测试

1
2
3
4
5
6
7
8
9
10
11
test_cases = [
    {"feedback_text": "你们这个软件频繁崩溃，赶快给我退钱！"},
    {"feedback_text": "界面很简洁，非常好用，感谢团队！"},
    {"feedback_text": "请问导出 PDF 的按钮在哪里？"},
    {"feedback_text": "代开各类正规发票，微信联系 138xxxx"}
]

for item in test_cases:
    res = app.invoke(item)
    print(f"用户输入: {item['feedback_text']}")
    print(f"分流类别: {res.get('category')} | 响应: {res.get('response')}\n")

输出：

1
2
3
4
5
6
7
8
9
10
11
用户输入: 你们这个软件频繁崩溃，赶快给我退钱！
分流类别: complaint | 响应: 【加急通道】：已为您创建 P0 级工单，客服主管将在 10 分钟内主动联系您。

用户输入: 界面很简洁，非常好用，感谢团队！
分流类别: praise | 响应: 【感谢反馈】：非常感谢您的认可与支持，已为您派发一张体验券！

用户输入: 请问导出 PDF 的按钮在哪里？
分流类别: inquiry | 响应: 【智能客服】：正在为您检索知识库使用手册，请稍候...

用户输入: 代开各类正规发票，微信联系 138xxxx
分流类别: spam | 响应: None

结合大模型：基于 Pydantic 的强类型结构化路由

在生产环境中，意图分类往往由大语言模型执行。严禁直接让模型自由发挥输出字符串（例如返回“应该是咨询”或“投诉吧”），这会导致 path_map 发生未命中错误（KeyError）。

工程最佳实践是结合 Pydantic 的 with_structured_output 强制约束模型输出为受控枚举：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from pydantic import BaseModel, Field
from typing import Literal

# 1. 约束输出结构
class RouteDecision(BaseModel):
    intent: Literal["refund", "tech_support", "general"] = Field(
        ..., description="严格从给定的三类意图中选择最匹配的一项"
    )
    confidence: float = Field(..., ge=0.0, le=1.0, description="分类置信度")

# 2. 节点内部通过结构化输出提取分类结果
def llm_classifier_node(state: MyState) -> dict:
    structured_llm = llm.with_structured_output(RouteDecision)
    decision = structured_llm.invoke(f"分析以下输入意图: {state['user_text']}")
    return {"classified_intent": decision.intent}

# 3. 确定性路由函数
def route_intent(state: MyState) -> str:
    # 状态中的字段已经受 Pydantic 枚举保证，绝无格式幻觉
    return state.get("classified_intent", "general")

进阶技巧：构建自纠循环图（Cyclic Graph）

条件边的强大之处在于，它的目标不仅可以是下游节点，还可以指回上游节点。

例如在代码生成工作流中：

  1. generate_code 生成代码；
  2. verify_code 运行测试；
  3. route_by_test_result 进行判断：
    • 若测试通过 走 END；
    • 若测试失败且重试次数 重新指向 generate_code；
    • 若重试次数已达上限 走向 escalate_human。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
def check_test_status(state: CodeState) -> str:
    if state["tests_passed"]:
        return "success"
    if state["retry_count"] >= 3:
        return "give_up"
    return "retry"

builder.add_conditional_edges(
    "verify_code",
    check_test_status,
    {
        "success": END,
        "retry": "generate_code",      # 环形回退边
        "give_up": "alert_engineer"
    }
)

这种机制让 LangGraph 原生拥有了闭环自愈能力，而无需在外层编写易损的 while 循环。

系列导航与参考