LangGraph Router工程实战：从规则、向量到LLM的组合路由方案

做LLM应用开发的人，基本都绕不开LangGraph。我大概在两年前开始在生产流程里用LangGraph，说实话，一开始觉得它不过是有向图加状态机，用起来并不复杂。直到我真正在流程中间塞了一个Router，也就是根据用户输入决定"下一步走哪个节点"的判断逻辑，才发现这里面的坑比想象中多。

这篇就把LangGraph Router的工程实战完整梳理一遍。从Router解决的核心问题、三种常见的实现方式、一个能直接复用的客服工单分流案例，到实际排查中的高频坑，尽量写成一篇可以当工具书收藏的参考。适合那种已经在写Agent流程、想把路由这部分做得更稳的同学，也适合刚开始接触LangGraph但不想走弯路的新手。

1. 先把路由想明白：LangGraph Router解决的到底是哪个问题

1.1 从单线流程到状态机，为什么必须引入路由

一个最朴素的LLM应用，往往是"用户输入一次，模型回复一次"的单线流程。你用一个prompt，把用户问题交给大模型，拿到回复就直接返回。这种链路写起来简单，但一旦涉及到工具调用、多步骤处理或者不同的业务分支，就完全不够用。比如客服机器人，用户说"我的快递到哪了"和"我要退货"，背后需要查询的订单系统、需要调用的政策规则、需要走的审批流程完全是两套东西。把这些分支都塞进一个大prompt里，模型会越来越糊涂，你的prompt也会越来越难维护。

这就是LangGraph这一类状态图框架要解决的根问题：把一个本来线性的对话/任务处理流程，拆成有向图上的若干个节点，每个节点负责一块明确的工作，节点和节点之间通过状态传递数据进行协作。而Router就是图中负责"分支选择"的节点：它根据当前状态（比如用户输入、历史记忆、工具返回结果）计算出一个路由结果，告诉状态机下一步到底跳转到哪个节点。本质上，它是把"业务决策"从执行逻辑里单独抽离出来，让决策方和执行方各司其职。

1.2 Router不是一个函数，而是一组工程决策

很多人第一次看到LangGraph的条件边（conditional edges），会觉得这不就是if/else换了个马甲。但实际上，Router在工程上需要考虑的东西远不止一个判断函数。你至少需要决策：路由的输入信号是什么，是直接用原始文本，还是先用意图分类模型输出一个意图再作为信号；路由的目标集合怎么定义，意图之间是有重叠还是互斥；判断结果不准确时，怎么降级和兜底。这些决策直接影响后续的可维护性，并不只是写一个函数的事。

我见过不少早期项目，为了图方便，把路由判断直接写进某个节点的prompt里，让大模型返回一个分支标记。刚开始一两个分支还能跑，等到分支超过四个，prompt里的判断逻辑和业务逻辑缠在一起，调试的时候只能靠打印日志猜模型在想什么。后来把它拆成独立的Router节点，状态里专门记录意图和置信度，再难看清楚全流程每个步骤做了什么。所以Router在我的理解里，不是"一个放在图上的判断框"，而是"一套让决策可见、可测、可回退的机制"。

1.3 判断你的项目到底需不需要Router

不是所有LangGraph项目都必须上Router。如果整个流程永远是固定的顺序，比如"接收输入-调用一个工具-生成回复-结束"，你直接用一个顺序图就够了，加Router反而增加了维护成本。Router真正带来价值的场景，是存在两个以上互斥的业务分支，而且分支选择不能靠简单关键字确定，或者在运行过程中分支可能会被反复访问。所以做设计时，先画一张流程草图：看是否存在"根据状态不同，下一步不同"的箭头。存在，就需要Router；不存在，就别硬凑。

还有一种情况是，虽然当前只有一个分支，但你预计未来一定会有多分支。这种我一般建议先画好扩展点，而不是马上把路由逻辑写进去。原因很简单：需求还没成形时写的路由规则，大概率会在业务落地前被推翻；不如等分支真的出现，再花半天时间重构，比你维护一版想当然的路由规则要划算得多。

2. 路由怎么写法：规则、向量、LLM三种落地方式

2.1 规则路由：零成本上线，但别指望它扛住一切

规则路由是所有路由方案里最基础、最稳的一类。它不做任何语义理解，就是靠正则、关键词、甚至简单的条件判断，把输入映射到某个出口。优势很明显：确定性高——同样的一句话，今天判断是这个结果，明天也是；零成本——不需要额外模型推理，不需要等待网络；调试方便——出问题看一眼正则就知道错在哪。适合做第一版MVP，或者给后续复杂路由做前置快速通道。

比如客服场景，我通常会先抓一些高频强特征词。用户提到"快递、物流、订单、没收到"，大概率是订单物流类；提到"退货、退款、换货、质量问题"，基本是售后类。用正则匹配，命中即返回对应分支。具体实现可以是这样的：

python复制import re

ORDER_PATTERNS = re.compile(r"(订单|快递|物流|包裹|配送|签收|没收到)")
RETURN_PATTERNS = re.compile(r"(退货|退款|换货|质量问题|错发|破损)")

def rule_route(state: dict) -> str:
    text = state["user_input"]
    if ORDER_PATTERNS.search(text):
        return "order_status"
    if RETURN_PATTERNS.search(text):
        return "return_process"
    return "human_agent"

看起来很简单，但有两个雷。一个是关键词顺序：如果用户说"我不退货了，帮我看看订单"，正则一旦先命中退货就会走错分支；所以规则路由里一定要做优先级排序，或者做否定词检测。另一个是覆盖度：口语表达千变万化，"帮我催一下"这种没有强特征词的话，规则几乎是失效的。所以规则路由通常不能是最后一道关卡，它只适合做最前置的快速判断。

2.2 向量路由：用语义相似度处理口语化表达

规则路由最大的短板是不会联想。用户说"东西不合适"，你很难用关键词规则命中"退货"分支。这时候可以把文本映射到向量空间，拿用户输入和每个意图的示例句子做相似度比较。这个方案我们内部叫"给规则路由加一层模糊匹配"，它不依赖精确词，靠语义拉近距离。具体做的时候，需要选一个文本Embedding模型，并且为每个意图准备几条代表性示例，算相似度后取最高分。

下面是一个简化实现思路：

python复制from sentence_transformers import SentenceTransformer

encoder = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")

intent_examples = {
    "order_status": ["订单到哪了", "我的快递一直没收到", "帮我查下物流"],
    "return_process": ["我要退货", "这个衣服不合适", "东西坏了能换吗"],
    "human_agent": ["转人工", "找客服聊聊", "我想打电话投诉"],
}

def vector_route(state: dict) -> str:
    input_vec = encoder.encode(state["user_input"])
    best_intent, best_score = "human_agent", -1
    for intent, samples in intent_examples.items():
        sample_vecs = encoder.encode(samples)
        score = float((input_vec @ sample_vecs.T).max())
        if score > best_score:
            best_intent, best_score = intent, score
    return best_intent

向量路由的好处是能处理用户没说出关键词的情况，但也有新问题：相似度阈值怎么定。不同Encoder输出范围不一样，不能拿一个固定阈值套所有场景。我的习惯是先抽出几百条真实历史对话，分别算每个意图的最高相似度，画出分布后选一个能区分的阈值，低于这个阈值不许放行，统一走兜底。同时，示例句子的质量比数量重要，每个意图不要贪多，5到10条高质量示例就够。

2.3 LLM路由：灵活度最高，却也最容易失控

如果业务意图特别细、边界特别模糊，规则和向量都顶不住，最后就得请大模型来判断。LLM路由的原理很简单：把用户输入连同系统指令交给大模型，让它输出结构化的路由结果，比如意图标签加置信度。我实践下来，LLM路由在理解复杂语义上的确远超规则和向量，但它在工程上也是最脆弱的：需要提示词工程，需要结构化输出解析，大模型偶尔会飘，输出格式偶尔会错，推理延迟也不可忽略。

用LangChain的with_structured_output可以比较优雅地拿结构化结果。示例：

python复制from typing import Literal
from pydantic import BaseModel

class RouteResult(BaseModel):
    intent: Literal["order_status", "return_process", "human_agent"]
    confidence: float = 0.0

def llm_route(state: dict) -> dict:
    messages = [
        {"role": "system", "content": "你是智能客服的路由判断器。只能输出JSON，包含intent和confidence。"},
        {"role": "user", "content": f"用户输入：{state['user_input']}"},
    ]
    result = llm.with_structured_output(RouteResult).invoke(messages)
    return {"intent": result.intent, "confidence": result.confidence}

这里有几个经验：第一，路由判断的system prompt里尽量给出每个意图的定义和正反示例，尤其是容易混淆的两类意图；第二，让模型输出confidence字段，并且设置一个"低置信度走兜底"的规则；第三，最好加一次格式校验和意图白名单校验，防止模型输出一个你图里根本不存在的节点名——这种情况一旦发生，LangGraph会在运行时直接报错，把整个流程打崩。

2.4 组合路由：工程上真正可用的完整链路

现在把三种方式拼起来，才是生产环境里比较完整的路由链路。我的建议是分层：第一层，规则路由，用最强特征词命中明显意图，直接返回，不浪费后面两层；第二层，向量路由，对未命中的文本做语义相似度，给一个候选意图；第三层，LLM路由，对向量层置信度不高的输入做精细化判断；最后加一层兜底，所有层都无法给出高置信度结果时，走人工或兜底节点。

关于每一层的优先级，不同业务可以微调，但核心原则是"先便宜后贵，先确定后模糊"。分层的好处不只在于省钱，还在于可以单独调优：规则面有问题就加正则，向量面有问题就换Encoder或调阈值，LLM面有问题就优化提示词。哪一层出问题，从日志里一眼就能定位。这个设计也为长期迭代留好了空间。

3. 完整工程实战：手写一个带Router的客服工单分流系统

3.1 业务设定与状态图设计

这里用一个具体的案例把上面的路由方式串起来。假设某电商平台要做一个客服工单分流系统，目标是先识别用户意图，再分流到对应处理节点。业务上一开始定义了三个意图：订单查询、售后退货、人工客服。后来发现用户经常表达不清楚，干脆再加一个"其他"兜底意图。整个图的流程可以设计成：用户输入进入入口节点，经过路由节点判断意图，然后各自进入对应处理节点，最终统一汇总生成回复，走向结束。

在图上，这些节点我都单独定义成函数。入口节点只做基本清洗和必要上下文准备；路由节点只做意图判断并写入状态；各处理节点只处理各自业务；汇总节点把结果整合成用户能看懂的回复。这样每个节点都能单测，路由节点还能单独做准确率评估，这是单先用一个大prompt包揽所有逻辑的写法比不了的。

状态（State）我定义为这样一份数据结构：

python复制from typing import TypedDict, Literal, List

class AgentState(TypedDict):
    user_input: str
    intent: Literal["order_status", "return_process", "human_agent", "fallback"]
    confidence: float
    route_trace: List[str]
    final_answer: str

route_trace用来记录每次路由的路径，方便后面对线排查时知道这条工单到底经过了哪些节点。

3.2 实现路由节点：组合链路怎么落到代码里

路由节点我按组合链路来写。先跑规则，命中就返回；没命中就抽向量算相似度，得分超过当前意图的自适应阈值才采用；再不行喊LLM判断，最后所有办法都用完还是低置信度，就走fallback。代码里我尽量把每一层的判断都记录下来，写进route_trace，这比只返回一个最终意图要实用得多。

python复制def route_node(state: AgentState) -> dict:
    text = state["user_input"]
    trace = state.get("route_trace", [])

    # 第一层：规则
    intent = rule_route(text)
    trace.append(f"rule:{intent}")
    if intent in ("order_status", "return_process"):
        return {"intent": intent, "confidence": 0.99, "route_trace": trace}

    # 第二层：向量
    intent, score = vector_route(text)
    trace.append(f"vector:{intent}:{score:.2f}")
    if score >= 0.6:
        return {"intent": intent, "confidence": score, "route_trace": trace}

    # 第三层：LLM
    llm_result = llm_route(text)
    trace.append(f"llm:{llm_result['intent']}:{llm_result['confidence']:.2f}")
    if llm_result["confidence"] >= 0.5:
        return {"intent": llm_result["intent"], "confidence": llm_result["confidence"], "route_trace": trace}

    # 兜底
    trace.append("fallback:默认转人工")
    return {"intent": "human_agent", "confidence": 0.0, "route_trace": trace}

明确说一下这里的vector_route返回的分值，不同Embedding模型分布不同，不能直接照抄0.6这个阈值。我的做法是先拿一批标注好的线下一数据，算所有真阳性和真阴性样本的分值分布，选一个误判可接受的切分点，写进配置。这个动作虽然一开始多花点时间，但在线上能少挨很多骂。

3.3 把路由挂到LangGraph状态图上

接下来就是把节点和边组装起来。用StateGraph定义图，用add_node注册节点，再用add_conditional_edges把路由节点的条件和目标节点关联起来。这里关键的工程细节是：条件映射表的键，必须和路由函数实际返回的路由名一一对应，一旦多返回一个不在映射表里的名字，LangGraph会直接抛异常。

python复制from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)

graph.add_node("entry", entry_node)
graph.add_node("route", route_node)
graph.add_node("order_status", order_node)
graph.add_node("return_process", return_node)
graph.add_node("human_agent", human_node)
graph.add_node("fallback", fallback_node)
graph.add_node("finalize", finalize_node)

graph.add_edge("entry", "route")
graph.add_conditional_edges(
    "route",
    lambda state: state["intent"],
    {
        "order_status": "order_status",
        "return_process": "return_process",
        "human_agent": "human_agent",
        "fallback": "fallback",
    },
)

for node in ["order_status", "return_process", "human_agent", "fallback"]:
    graph.add_edge(node, "finalize")

graph.add_edge("finalize", END)

app = graph.compile()

这里有一个小坑，条件边映射表里的值看着是字符串，但实际上它必须指向通过add_node注册过的节点名。我一开始在这一步写错过几次，每次都是运行时才发现，报错信息描述得比较抽象，排查半天才意识到是节点名拼写的问题。所以建议把节点名定义成模块级常量，至少能靠IDE提示挡掉一部分低级错误。

注意：所有条件边映射的节点名必须存在于图中，否则LangGraph在运行时直接报错。写代码时最好给节点名定义成常量，不要手写字符串。

3.4 给Router加日志、Trace与可观测性

生产环境里，路由节点发生误判时，最怕的是你完全不知道它为什么走错了。所以我在设计里加了route_trace，同时在图编译后会接入一个简单的日志回调：每次节点执行完，把当前状态的关键字段打出来。这样线上收到一条工单，你就能清楚看到它进了入口、被路由到订单节点、最后生成了什么样的回复。可观测不等于复杂的监控平台，很多时候一个结构化日志就够。

python复制import json

class TraceLogger:
    def __call__(self, config, node_name, state):
        print(json.dumps({
            "node": node_name,
            "user_input": state.get("user_input"),
            "intent": state.get("intent"),
            "confidence": state.get("confidence"),
            "trace": state.get("route_trace"),
        }, ensure_ascii=False))

这种日志最大的好处是能把"路由决策过程"拆穿。比如用户说"我不退货了，帮我查订单"，规则层如果因为"退货"两个字直接走了售后退货节点，看日志就会发现trace里记录的是rule:return_process。这时候你要做的不是骂规则简单，而是把"否定词检测"加进规则路由的优先级里，让"不退货"不触发退货分支。没有这条trace，这类问题基本只能靠用户投诉才暴露。

3.5 从单轮路由扩展到多轮上下文

上面的例子是单轮路由：只看当前这一条输入。但真实客服往往是多轮对话，用户可能先说"我要退货"，后面又说"但是运费谁出"。单轮路由只看后一句，很可能把它当成一个咨询问题，这就有问题了。所以我在这个项目第二版里，把对话历史也加进了State，让路由节点在判断之前先拼接最近的用户消息和系统上下文。

实现上不复杂，无非在State里增加一个messages字段，用Annotated和operator.add做累加，然后路由节点构造判断文本时把最近几轮拼进去。这里要特别注意的是判断粒度：并不是每轮用户输入都要重新路由，很多情况下当前分支已经确定，只有用户主动提出切换意图时才需要重新走路由。因此我通常会设一个needs_routing标记，只有标记为True时才执行Router节点，避免频繁做高成本的大模型路由判断。

4. 实战中的坑：Router相关的常见问题与排查技巧

4.1 路由结果越改越乱，准确率扛不住

最直接的问题是准确率。规则路由覆盖不了口语，向量路由阈值难调，LLM路由偶尔会飘，组合路由虽然可靠，但每一层都在做贡献，一起出错也不是没见过。我的排查顺序是：先看route_trace，确认是哪一层给出的错误判断；如果是规则层，就加否定词或提高优先级；如果是向量层，就检查示例句和阈值；如果是LLM层，优先考虑把容易混淆的意图在prompt里给反例。

还有一条被很多人忽视的规律：路由意图集合不是越大越好。粒度越细，互相之间的边界越容易模糊，判断准确率直线下降。我在项目里就踩过，原先把"查物流"和"改地址"分为两个意图，结果大模型经常分不清，后来合并成"订单相关"，准确率一下上去了。业务上需要区分，可以在处理节点内部再细分，不要让Router一开始就背太多的决策压力。

4.2 状态字段没有传下去，节点拿到的是空值

我在LangGraph里遇到最多的运行时问题，就是某个节点返回了State里没有定义的新字段，或者只返回了局部字段，导致下游节点取不到值。LangGraph的机制是：每个节点函数返回的字典会合并到全局State，但如果在State定义里不存在这个字段，某些版本会直接报错；如果字段存在但某次没返回，下游取出来就是None，进而触发各种诡异行为。排查方法很简单，在每个节点入口打一行日志，打印完整State的key集合，立刻能看到谁丢了字段。

另一个常见原因是可变对象被覆盖。比如route_trace用普通列表，每次返回整个列表，前一个节点记录的路径很容易被覆盖。我建议这类"只增不改"的状态字段，在定义State时使用带Reducer的类型，比如Annotated[list, operator.add]，让LangGraph自动合并列表而不是整体覆盖。用对了之后，状态传递的Bug明显少了一大截。

4.3 条件分支走进死循环或者莫名提前结束

条件边如果设计得不严密，可能会出现两个节点互相来回跳的循环，最常见的是"判断节点-执行节点-再判断-再执行"死循环。LangGraph本身有recursion_limit，默认次数是25次，超过会抛GraphRecursionError，但这不代表业务上正常。更危险的是提前结束：路由节点某次意外返回了一个不在映射表里的键，图在运行时直接报错，或者某条边被漏加，节点执行完没有连到任何下一跳。

我自己的防御做法是两件事。一是在路由函数末尾强制一个兜底分支，除了白名单四类意图之外，任何情况都返回"human_agent"；二是为每个处理节点都保证有一条到汇聚节点或END的边，绝不允许某个分支执行完悬空。检查的时候，把图上所有分支过一遍，确认没有断头路，再进行联调。

4.4 并发上来了，路由节点成了瓶颈

路由判断不是免费的，尤其是向量编码和LLM调用都涉及外部资源。当流量上来以后，规则路由还好，向量和LLM路由的耗时和成本都会成为瓶颈。我遇到过一种情况：一个客服入口日请求量几万，路由节点里同步调用Embedding模型，导致同一时间大量请求排队，整体P95延迟直接飙到好几秒。后来把规则路由调前置，能挡掉一半流量，才把后面的压力降下来。

更进一步的优化思路是批量化和缓存。向量路由可以按批次编码，把并发请求合并成一次模型调用；对于短时间内重复的问题，可以按文本hash做结果缓存。LLM路由则尽量只在向量层无法判断时才调用，同时为LLM设置合理的超时和重试策略。这些优化不需要改业务代码，但要把路由放在整个系统性能模型里统一考虑，不能当它不存在。

4.5 几条值得长期遵守的经验

最后分享几条我从这个项目里沉淀下来的经验。第一条，Router的输出必须是结构化数据，最好包含意图和置信度，不要把判断结果混在文本回复里。第二条，路由意图集合要扁平，尽量控制在5个以内，超过5个就要考虑先粗后细的二级路由。第三条，每一层路由都要有降级出口，宁可直接转人工，也不要让系统硬编一个根本不存在的意图。

还有一条关于代码组织：路由节点不要什么都做。它只负责"判断去哪"，不负责"去到那里之后做什么"。判断逻辑和执行逻辑放在一起，短时间看着方便，等节点一多就会变成一坨不可维护的状态机。保持节点职责单一，配合trace日志，这套系统才能真正长期稳定跑下去。

我个人在把这些路由方案从实验搬到生产的过程中，最大的体会是：技术路线本身并不难，难的是你愿不愿意把路由当作一个独立模块来长期经营。规则、向量、大模型，说到底都是实现手段，真正重要的是你给Router设计了清晰的输入信号、明确的输出协议、可靠的兜底路径。如果你刚准备动手，别急着一次上满三种路由，先用规则路由把全链路跑通，再一层层把语义和智能判断加上去，你会发现后面每一个问题都变得特别容易定位。