摘要：路由（Routing）为智能体引入条件分支：先对输入或当前状态做判断，再决定下一步走哪条处理路径，从而从「单一路径执行」变为「按条件选路径」。本文说明路由的动机、典型实现方式（LLM / 规则 / 嵌入 / 小模型），并重点介绍在 LangGraph 中如何用状态图 + 条件边实现「先分类、再分发」的流程，配套示例为基于 LLM 的意图分类（booker / info / unclear）驱动三条分支。

关键词：路由；Routing；条件分支；LangGraph；条件边；StateGraph；意图分类；LLM 路由；Agentic Design Patterns

源代码链接：Langgragh 2. 路由 Routing 源代码
。

---

1 为什么需要「路由」？

在只用顺序链（Prompt Chaining） 时，流程是固定的：步骤 A → B → C，每次请求都走同一条路。但现实里的智能体往往要根据当前情况做选择：用户是在问订单、问产品，还是在找技术支持？不同意图应当走不同的处理流程，而不是「一刀切」地走同一条流水线。

路由（Routing） 就是在智能体里引入条件分支：先对输入（或当前状态）做判断，再决定下一步走哪条路——是调用某个工具、交给某个子智能体，还是进入某段子流程。这样系统就从「单一路径执行」变成「按条件选路径」，更贴近真实业务需求。

💡 理解要点：路由 = 先判断，再选路；没有路由时只有一条固定路径，有了路由才能「因请求而异」地走不同分支。

---

2 路由在解决什么问题？

想象一个客服场景：

• 用户问「订单到哪了」→ 应走订单查询流程（查库、返回物流）。
• 用户问「这款有没有货」→ 应走商品信息流程（查库存/目录）。
• 用户说「设备坏了要保修」→ 应走技术支持/升级流程（工单或转人工）。

若没有路由，要么所有话都进同一套逻辑（难以兼顾三种需求），要么写死大量 if-else（难维护、难扩展）。路由模式把「该走哪条路」抽象成一个独立决策步骤，后续每条路可以各自实现、替换或扩展。

🔍 实际例子：就像公司前台：先听你说「找谁/办什么事」，再把你指到对应部门（销售、技术、人事），而不是所有人都进同一间办公室。

---

3 路由的几种实现方式

路由的「判断」可以用不同方式来做，常见有四类：

方式	简要说明	特点
基于 LLM 的路由	用大模型分析输入，输出一个类别或标签（如 booker / info / unclear），再根据该标签选分支	灵活、能理解语义，适合意图复杂、表述多样的场景
基于规则的路由	用关键词、正则或结构化规则（if-else / switch）决定分支	实现简单、可解释、确定性高，但对表述变化敏感
基于嵌入的路由	把输入和若干「路由选项描述」都变成向量，用相似度选最接近的那条路	语义相似即可命中，适合选项固定、表述多样的场景
基于 ML 模型的路由	用单独训练的分类器（小模型）做路由决策	可离线优化、推理快，需要标注数据与训练流程

本系列以 LangGraph 为核心，示例代码采用 LLM 路由：由 LLM 对用户请求做意图分类，再用分类结果驱动图上的条件边，把请求送到对应处理节点。其他方式只需把「路由节点」的实现换成规则/嵌入/小模型即可，图结构不变。

💡 理解要点：路由的「决策逻辑」可以换（LLM / 规则 / 嵌入 / 小模型），但「先决策、再按结果选分支」这一模式是统一的；LangGraph 用条件边（conditional edges） 把这一模式表达得很直观。

---

4 LangGraph 中的路由：状态图 + 条件边

LangGraph 用状态图（State Graph） 描述流程：节点是处理步骤，边是步骤之间的转移。条件边表示：从某节点出来后，根据当前状态决定下一个进入的节点，而不是固定到唯一后继。

• 状态（State）：在图中流转的共享数据结构，例如 request、decision、response。
• 路由节点：负责根据输入（或状态）做分类，并把结果写回状态（如 decision）。
• 路由函数：供条件边使用，读状态并返回「下一节点名」。
• 条件边：从路由节点出发，根据路由函数的返回值，映射到不同的下游节点。

这样就把「分析请求 → 选路 → 进入对应处理器」清晰拆成：一个路由节点 + 一条条件边 + 多个处理节点。

🔍 实际例子：本示例中的三类意图——预订（booker）、一般信息（info）、未识别（unclear）——分别映射到三个节点 booking、info、unclear，每个节点只做一件事，结构一目了然。

5 配套代码结构概览

代码建议与源代码的 README 对照阅读。

5.1 状态定义

图的状态用 TypedDict 定义，贯穿路由与各处理节点：

# 摘自 demo_codes/routing_graph.py
class RoutingState(TypedDict):
    request: str        # 用户原始请求
    decision: str       # 路由节点输出的分类：booker | info | unclear
    response: str       # 当前分支处理节点的输出

5.2 路由节点（LLM 分类）

路由节点只做一件事：用 LLM 对 state["request"] 做意图分类，并把结果写入 state["decision"]（这里规范为 booker / info / unclear）：

# 摘自 demo_codes/routing_graph.py（片段）
ROUTER_PROMPT = """分析用户请求，判断应由哪个专门处理器处理。
- 若与预订机票/酒店相关，只输出：booker
- 若为一般信息类问题，只输出：info
- 若无法归类或不清楚，只输出：unclear
只输出一个词：booker、info 或 unclear。
用户请求：{request}"""

def node_router(state: RoutingState) -> dict:
    chain = _router_chain()  # prompt | llm | StrOutputParser
    raw = chain.invoke({"request": state["request"]})
    decision = (raw or "").strip().lower()
    if decision not in ("booker", "info", "unclear"):
        decision = "unclear"
    return {"decision": decision}

💡 理解要点：路由节点不直接「调下一个节点」，只负责更新状态里的决策字段；真正选路由后面的条件边 + 路由函数完成。

5.3 路由函数与条件边

路由函数读当前状态，返回下一节点名；条件边根据该返回值做映射：

# 摘自 demo_codes/routing_graph.py
def route_by_decision(state: RoutingState) -> Literal["booking", "info", "unclear"]:
    d = (state.get("decision") or "unclear").strip().lower()
    if d == "booker":
        return "booking"
    if d == "info":
        return "info"
    return "unclear"

# 图中：从 "router" 出发的条件边
workflow.add_conditional_edges(
    "router",
    route_by_decision,
    {"booking": "booking", "info": "info", "unclear": "unclear"},
)

这样，LLM 输出 booker 会进入 booking 节点，info 进入 info 节点，其余（含解析异常）进入 unclear 节点。

5.4 处理节点与图的入口/出口

三个处理节点仅根据当前状态生成回复并写回 response，然后从各自节点连到 END：

# 摘自 demo_codes/routing_graph.py（片段）
def node_booking(state: RoutingState) -> dict:
    return {"response": f"Booking Handler 已处理请求：'{state['request']}'。结果：模拟完成预订操作。"}
# node_info、node_unclear 同理

workflow.add_edge(START, "router")
workflow.add_edge("booking", END)
workflow.add_edge("info", END)
workflow.add_edge("unclear", END)

整体流程：START → router →（条件边）→ booking / info / unclear → END。

5.5 完整代码

如下，也可以下载完整代码然后直接运行：

"""
LangGraph Routing 示例：基于意图的路由图。

流程：用户请求 → [路由节点：LLM 分类] → 条件边 → [预订/信息/未识别] 处理节点 → 响应

运行前请配置环境变量 OPENAI_API_KEY 或 DASHSCOPE_API_KEY，或在项目根目录放置 .env 文件。
"""
import os
from typing import Literal, TypedDict

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langgraph.graph import StateGraph, START, END

from config_parser import routing_config

# ---------------------------------------------------------------------------
# 状态定义：图的状态，贯穿路由与各处理节点
# ---------------------------------------------------------------------------


class RoutingState(TypedDict):
    """图的状态：请求、路由决策与最终响应。"""
    request: str        # 用户原始请求
    decision: str       # 路由节点输出的分类：booker | info | unclear
    response: str      # 当前分支处理节点的输出


# ---------------------------------------------------------------------------
# LLM 与路由链
# ---------------------------------------------------------------------------


def _build_llm():
    """构建 LLM（从 config 读取 API Key 与 base_url）。"""
    cfg = routing_config
    llm = ChatOpenAI(
        model=cfg.model,
        api_key=cfg.api_key,
        base_url=cfg.base_url if cfg.base_url else None,
        temperature=0,
    )
    return llm


# 路由提示：让 LLM 仅输出一个分类词
ROUTER_PROMPT = """分析用户请求，判断应由哪个专门处理器处理。
- 若与预订机票/酒店相关，只输出：booker
- 若为一般信息类问题，只输出：info
- 若无法归类或不清楚，只输出：unclear

只输出一个词：booker、info 或 unclear。

用户请求：
{request}"""


def _router_chain():
    """路由链：request -> LLM -> decision 字符串。"""
    prompt = ChatPromptTemplate.from_messages([
        ("user", ROUTER_PROMPT),
    ])
    llm = _build_llm()
    return prompt | llm | StrOutputParser()


# ---------------------------------------------------------------------------
# 图节点
# ---------------------------------------------------------------------------


def node_router(state: RoutingState) -> dict:
    """
    路由节点：调用 LLM 对用户请求做意图分类，写入 state["decision"]。
    """
    chain = _router_chain()
    raw = chain.invoke({"request": state["request"]})
    decision = (raw or "").strip().lower()
    if decision not in ("booker", "info", "unclear"):
        decision = "unclear"
    return {"decision": decision}


def node_booking(state: RoutingState) -> dict:
    """预订处理器：模拟处理机票/酒店预订类请求。"""
    msg = (
        f"Booking Handler 已处理请求：'{state['request']}'。"
        "结果：模拟完成预订操作。"
    )
    return {"response": msg}


def node_info(state: RoutingState) -> dict:
    """信息处理器：模拟处理一般信息类请求。"""
    msg = (
        f"Info Handler 已处理请求：'{state['request']}'。"
        "结果：模拟信息检索与回答。"
    )
    return {"response": msg}


def node_unclear(state: RoutingState) -> dict:
    """未识别处理器：无法归类时的兜底回复。"""
    msg = (
        f"无法将请求归类：'{state['request']}'。"
        "请补充说明是「预订」还是「一般信息」类问题。"
    )
    return {"response": msg}


# ---------------------------------------------------------------------------
# 路由函数：供条件边使用，根据 state["decision"] 返回下一节点名
# ---------------------------------------------------------------------------


def route_by_decision(state: RoutingState) -> Literal["booking", "info", "unclear"]:
    """根据路由节点的分类结果，返回下一节点名称。"""
    d = (state.get("decision") or "unclear").strip().lower()
    if d == "booker":
        return "booking"
    if d == "info":
        return "info"
    return "unclear"


# ---------------------------------------------------------------------------
# 构建图：START -> router -> [booking | info | unclear] -> END
# ---------------------------------------------------------------------------


def build_routing_graph():
    """构建并编译 Routing 图。"""
    workflow = StateGraph(RoutingState)

    workflow.add_node("router", node_router)
    workflow.add_node("booking", node_booking)
    workflow.add_node("info", node_info)
    workflow.add_node("unclear", node_unclear)

    workflow.add_edge(START, "router")
    workflow.add_conditional_edges(
        "router",
        route_by_decision,
        {
            "booking": "booking",
            "info": "info",
            "unclear": "unclear",
        },
    )
    workflow.add_edge("booking", END)
    workflow.add_edge("info", END)
    workflow.add_edge("unclear", END)

    return workflow.compile()

---

6 如何运行示例

1. 进入 demo_codes 目录，创建并激活虚拟环境，安装依赖：

   pip install -r requirements.txt
   

2. 在目录下配置 .env（如 OPENAI_API_KEY 或 DASHSCOPE_API_KEY，可选 BASE_URL、MODEL）。
3. 运行：

   python main.py
   

4. 终端会依次对三条示例请求（预订、信息、未识别）做路由并打印决策与响应。

---

7 小结与延伸

• 路由让智能体根据输入或状态选择不同执行路径，是超越「单一路径链式调用」的关键一步。
• 实现方式可以是 LLM、规则、嵌入或小模型；示例采用 LLM + 条件边，便于理解与修改。
• 在 LangGraph 中：状态承载数据，路由节点写决策，条件边 + 路由函数根据状态选下一节点，结构清晰、易扩展。

若要把路由节点改成规则（如关键词匹配）或嵌入相似度，只需替换 node_router 的实现与 decision 的取值约定，图的其余部分（条件边、处理节点）可保持不变。