☰
LangGraph Router工程化实战:从状态设计到可靠性兜底
2026/10/20 11:40:30 网站建设 项目流程

先说我遇到的一个真实现场:某个客服Agent上线不到一周,用户大量反馈“我问发票,怎么给我推荐商品?”查trace才发现,所有含“发票”的会话全部走进了售前节点。问题不在模型,而在路由——当时只写了一个“让LLM输出一个词”的简易Router,既不校验、也没兜底,更没人记录决策依据。这件事之后,我把LangGraph工程里最不被重视、却最影响大局的Router环节彻底重做了一遍,也决定把这套经验完整写出来。这篇文章不讲概念,直接按工程落地的顺序,把Router的选型逻辑、状态设计、条件边实现、可靠性兜底、性能调优和真实深坑全部拆开讲明白。

有一点先说清楚:Route不是“用大模型选个分支”那么简单。LangGraph把Agent流程建模成节点加边的有向图,Router就是图上的决策点。这个点分错了,后面再好的工具、再强的模型都白搭。文章里我会用一个实际能跑通的客服工单分诊场景贯穿全程,把代码、设计思路和排错链路都摆出来,适合正在用LangGraph做正式项目的团队,也适合刚接触LangGraph、想理解Router工程化边界的读者。

1. Router在LangGraph里的真实定位:决策点即架构

1.1 为什么说Agent的能力上限由路由决定

Agent本身做的事情其实是“把输入调度到正确的处理逻辑”。用户一句话进来,系统需要判断这句话应该交给哪个工具、哪个子Agent、哪一段业务流程。这个判断动作就是Router。判断错了,后续环节再精密也只会沿着错误方向越跑越远。

拿快递分拣来类比最直观:Router就是传送带上的那个岔口。包裹在岔口被拨错方向,后面所有运输和投递流程都会跟着错,而且越错越深。很多团队花大量时间调优某个业务节点的prompt,却忽略了入口处那个看起来“很简单”的Router,结果是业务节点做得再完美,线上数据依然一塌糊涂。

从LangGraph的实现视角看,路径选择就是一条“条件边”(conditional edge)。条件边的函数接收当前状态,返回一个字符串,图根据这个字符串决定跳到哪里。所以Router的输入输出就是状态的变更和方向的选择,它的工程质量会直接决定整条链路的稳定性和可调试性。我见过很多Agent项目最初只有一条线性链路,跑着跑着需求变复杂,才开始加入各种分支判断——这时候如果没有从一开始就按工程化方式设计Router,重构成本会非常高。

1.2 路由节点的职责边界:只做决策,不做业务

这是我在实战里最想强调的一点:Router节点应该只负责“决策”,不负责“执行业务”。所谓业务,指的是真正处理用户问题的那部分逻辑,比如生成回复、调用API、写工单。Router的职责是分析当前状态,算出“下一步该走哪条路”,然后把这个决策结果放回状态里,让条件边去跳转。

这样说听起来很简单,但实际项目里有个极常见的反模式:Router函数里顺手把业务字段改了。比方说,Router判断出用户意图是“售后”,就直接把state["topic"]改成了“售后”,然后在后面的业务节点里,topic被当作真实用户诉求来使用。看起来没什么问题,可是当业务节点本身的处理逻辑依赖topic做更细的分类时,这个字段已经被Router污染了,业务节点拿到的就不是原始信息,而是Router的“二手判断”。我之前提到的发票事故就是这个原因——Router把“售后”写进了topic,售前节点又把topic当成用户意图,自然就把“发票询问”导向了商品推荐。

所以我的规矩是:路由产生的所有信息,一律放进router_meta这个专用字段里。业务字段保持“原始输入”语义,路由字段单独存放。这样即使路由判断出错,业务节点拿到的信息也是原始可追溯的,排错时一眼就能看出问题出在哪个环节。

还有一个工程细节值得提:LangGraph条件边的目标节点,可以是普通节点,也可以是子图。当多个分支都需要走同一套“信息收集-确认-完成”流程时,别把流程复制到每个分支里,直接定义一个子图,然后把路由目标指向子图入口。子图内部可以再做二次路由,这样整个图结构不会因为分支增多而爆炸,每个分支的可维护性也高得多。

2. 三类常见Router的选型逻辑与工程取舍

2.1 提示词Router:原型期够用,生产期别依赖

最原始、也最常被采用的Router实现方式是提示词Router:给LLM一段指令,比如“你是路由助手,只能输出一个词:售后、售前、物流、人工”,然后把LLM返回的文本拿去匹配条件边映射表。它的优点非常明显:实现快、思路直观、任何自然语言模型都能跑,用于Demo和原型验证效率极高。

但它的缺点也足够致命。第一,输出不稳定,模型可能输出“售后。”、“售后\n\n(用户想申请退款)”,甚至输出一段解释文字;第二,没有置信度,你无法知道模型做这个判断时到底是笃定还是犹豫;第三,输出没有结构,后续想做拦截、记录、归因分析非常困难。我见过不少团队在原型阶段用这套方案跑得很开心,上线第一周就被真实流量打回原形——真实用户的话术远比测试集丰富,稍微一模糊,模型就开始“自由发挥”。

如果你只是在本地跑通一个LangGraph Demo,用提示词Router没问题。但只要你打算把Agent部署到正式环境,建议直接跳过它做结构化方案,不要在它上面投入任何调优时间。

2.2 结构化输出Router:生产环境的首选

生产环境里我优先推荐结构化输出Router。核心做法是用Pydantic定义一个结果模型,让LLM直接输出结构化的对象,而不是纯文本。这样后续代码可以直接读取字段做校验和阈值判断。

举一个具体的模型定义:

from typing import Literal from pydantic import BaseModel, Field class IntentResult(BaseModel): intent: Literal["售后", "售前", "物流", "人工", "闲聊"] confidence: float = Field(ge=0, le=1, description="模型对意图判断的置信度") reason: str = Field(description="用一句话解释判断依据")

定义好之后,配合with_structured_output使用:

llm = ChatOpenAI( model="route-model-v2", base_url="http://llm-gateway.internal/v1", temperature=0 ) result = llm.with_structured_output(IntentResult).invoke(user_input)

结构化输出的价值不在于“格式好看”,而在于它把不可控的文本变成了可控的编程对象。你可以做三件之前做不到的事:第一,用confidence做低置信度拦截,不确定时转人工;第二,把reason和intent一起记录下来,方便后续分析判断逻辑;第三,模型输出的字段严格受Schema约束,不存在“多出一个没注册的意图”这种问题。整套链路下来,Router就从“碰运气”变成了“可观测、可拦截、可审计”。

2.3 向量语义Router:高频固定意图的低成本选择

如果你的业务意图域非常固定,比如客服场景里的“改地址”“查物流”“退款”,而且希望路由延迟降到几十毫秒、成本趋近于零,那就不需要用大模型做分类,直接用向量语义Router更合适。

做法也不复杂:把每个意图准备若干条示例话术,通过Embedding模型转换成向量存起来;用户输入进来时同样转成向量,和所有意图示例向量算余弦相似度,取最高分对应的意图作为路由结果。如果所有相似度都很低,说明命中不了任何已知意图,就走默认兜底。

下面这张表是我在实际项目里做技术选型时会参考的对比:

维度提示词Router结构化Router向量语义Router
延迟高,一次大模型调用中,一次大模型结构化调用最低,Embedding+相似度计算
成本大模型Token大模型TokenEmbedding调用,成本极低
灵活度高,意图域可随时改高,字段可扩展低,新增意图需要准备新样本
可解释性中高,有意图与置信度低,只有相似度分数
输出稳定性低,依赖模型发挥高,受Schema校验高,取决于样本质量
适用阶段原型验证生产标配高频固定场景

实际工程里,这三种Router并不是互斥选项。我现在的项目里经常把它们组合成“分级路由”,先向量粗分,再结构化精判。这个放在后面的性能调优章节再详细展开。

3. 一个可落地的客服路由Agent:状态设计与条件边实现

3.1 场景定义与状态字段拆分

为了把Router的工程链路讲透,我用一个具体的客服工单分诊Agent作为贯穿案例。假设用户输入是一句话,Agent需要判断这句话应该进入售后工单、售前咨询、物流查询、人工客服,还是闲聊。目标不是实现完整客服系统,而是把路由环节完整跑通,并且暴露足够的工程细节。

先定义状态结构。这是整个LangGraph工程的地基,定义不好后面所有节点都会跟着遭殃:

from typing import TypedDict class AgentState(TypedDict): user_input: str router_meta: dict final_answer: str

可以看到,我刻意把状态拆成了三个区域:

  • user_input:原始用户输入,业务节点需要时可以读取,但它永远不会被Router改写。
  • router_meta:路由决策产生的所有元信息,包括意图、置信度、理由、模型版本、耗时等。
  • final_answer:最终要返回给用户的回复内容。

你可能会觉得“多拆一个字段很麻烦”,但正是这一个小小的隔离动作,能避免掉我在前面提到的“发票事故”一整类问题。路由字段和业务字段混在一起,短期看省事,长期看是给未来排错埋雷。

3.2 条件边与路由函数完整实现

接下来是完整的核心实现。先写路由节点,它负责调用模型、解析结果、把决策信息写入router_meta:

from langgraph.graph import StateGraph, START, END def intent_router_node(state: AgentState) -> AgentState: llm = ChatOpenAI( model="route-model-v2", base_url="http://llm-gateway.internal/v1", temperature=0 ) result = llm.with_structured_output(IntentResult).invoke(state["user_input"]) state["router_meta"] = { "intent": result.intent, "confidence": result.confidence, "reason": result.reason, "raw": result.model_dump(), } return state

然后写路由函数,它是条件边的核心,决定“返回什么字符串”:

def route_by_intent(state: AgentState) -> str: intent = state["router_meta"]["intent"] confidence = state["router_meta"]["confidence"] if confidence < 0.6: return "human" return intent

最后把它们装配成图:

builder = StateGraph(AgentState) builder.add_node("intent_router", intent_router_node) builder.add_node("after_sales", after_sales_node) builder.add_node("pre_sales", pre_sales_node) builder.add_node("logistics", logistics_node) builder.add_node("human_handoff", human_handoff_node) builder.add_node("smalltalk", smalltalk_node) builder.add_node("fallback", fallback_node) builder.add_edge(START, "intent_router") builder.add_conditional_edges( "intent_router", route_by_intent, { "售后": "after_sales", "售前": "pre_sales", "物流": "logistics", "人工": "human_handoff", "闲聊": "smalltalk", "human": "human_handoff", "__default__": "fallback", } ) builder.add_edge("after_sales", END) builder.add_edge("pre_sales", END) builder.add_edge("logistics", END) builder.add_edge("human_handoff", END) builder.add_edge("smalltalk", END) builder.add_edge("fallback", END) graph = builder.compile()

这段代码里有几个细节需要特别说明。

第一,条件边返回的字符串必须是映射表的key。如果route_by_intent返回了一个没有在映射表里注册的值,比如“退款”,而映射表里没有“退款”这个key,LangGraph在运行时就会抛KeyError。这是个很隐蔽的坑,后面专门讲。

第二,我在映射表里同时注册了“人工”和“human”,因为低置信度时route_by_intent返回的是"human",而正常意图识别结果里的“人工”中文会被直接返回。两个都注册,才能保证无论走哪条路径都不会漏配。

第三,__default__是条件边的兜底分支。当返回值不在任何映射key里时,会走到__default__指向的节点。它很重要,但不能作为唯一的兜底防线,因为本质上你还是依赖“返回了一个错误值”才触发它。更可靠的方案是让路由函数自己把返回值约束在白名单内,这一点后面排错章节会详细说。

3.3 为什么用条件边而不是节点内手动跳转

经常有读者问:既然intent_router_node里都已经拿到了意图,为什么不直接在节点内部调用after_sales_node或其他节点函数,而非要通过条件边跳转?

这个问题的答案是工程上的,不是写法上的。

LangGraph的图结构能做的事情远不止“按代码顺序执行”。它支持断点重放、时间旅行、并发分支、可视化调试,这些能力全部依赖于“显式的边”。如果你在节点内部手动调用业务函数,等于绕过了图引擎,把控制流藏进了业务代码里。一旦线上出现问题,你想在某个分支的中间位置暂停、修改状态、重新执行,根本无从下手。

使用显式条件边,路由决策的结果清晰可见:一个节点负责计算方向,一条边负责跳转,下一个节点只处理自己的业务。这种单一职责的拆分,在Agent逻辑越来越复杂的今天,不是“优雅”,而是“保命”。

我自己的习惯是:Router节点里的代码量应该非常少,最好只是一个“识别意图并写入状态”的壳子。如果这个节点里开始出现大量分支处理逻辑,那一定是设计出了问题,该考虑拆子图了。

4. 路由结果不可靠怎么办:兜底、重试与可观测性

4.1 路由错误的三种类型与应对方向

即使你用了结构化输出Router,路由环节依然会出错。我把实际生产中遇到的路由错误归纳成三类,方便你对照处理。

第一类是格式性失误。模型输出“售后,”或者“售后\n”,甚至带有额外解释文字。这类问题最轻,用结构化输出基本能解决,万一底层模型网关返回异常,再做一层字符串清理和类型判断也能兜住。

第二类是语义性模糊。用户说“我说了不算bug,你要怎么赔礼道歉?”这到底是投诉、售后还是售前咨询?人都不一定能分清,模型更不可能稳定判断。这类问题没有绝对正确答案,核心策略是识别出低置信度,把这单转给人工,而不是硬着头皮走某个业务分支。

第三类是系统性超时。模型网关抖动、服务过载、网络问题导致路由节点没有返回结果。这个问题不在“判断对错”,而在“有没有结果”。处理方式要围绕超时和重试来做。

三种错误对应三套策略:格式问题用结构化与校验;语义问题用置信度阈值;系统问题用超时控制与兜底链路。

4.2 三层兜底链路与重试上限

我给Router设计了一个三层兜底链路,每一层解决不同的问题。

第一层在路由函数内部:把模型输出做归一化处理。字符串去空格、去换行、转意图别名,比如把五花八门的表达映射到白名单意图集。同时做白名单校验,一旦结果不在允许集合里,直接强转“人工”。这层解决的是“模型输出了合法但不是预期值”的情况。

第二层在条件边映射表:注册__default__,让任何未注册的返回值都走兜底节点。这层解决的是“路由函数有疏漏、有值漏配”的情况。兜底节点可以做成一个安全策略节点,不做任何后续业务,直接返回转人工话术。

第三层在业务节点本身:如果某个业务节点发现router_meta缺失或异常,它不应该继续执行原有业务逻辑,而应该返回安全话术或转人工。这层解决的是“上游节点提前崩溃或状态被外部覆盖”的情况。

三层都做完,Router才算有了基本的防御能力。

重试方面有一条铁律:路由入口的重试次数一定要严格控制。路由是链路第一跳,它每多卡一秒,整个用户的等待时间就被放大一秒。我给路由调用默认配置是timeout=5、max_retries=1,超过就直接转人工。不要做无限重试,也不要因为“这个请求很简单,应该能成功”而放开超时限制。入口的鲁棒性,永远比入口的准确率优先级更高。

4.3 路由决策的完整记录位设计

路由决策一定要记录,这直接决定了你能不能对线上路由质量做闭环分析。我在router_meta里固定记录四类信息:

  • 决策信息:意图、置信度、判断理由。
  • 模型信息:使用的模型版本、提示词版本或Schema版本。
  • 性能信息:从发起调用到拿到结果花了多少毫秒。
  • 原始信息:模型未经过任何处理的完整返回内容。

这样设计的理由是:后续做路由归因分析时,你能回答四类关键问题。比如“上线新版提示词之后,售后意图的命中率为什么掉了10个百分点”,你只需要对比prompt_version和intent的分布变化就能看出来。“某些模糊表达一直在误判”,你需要看raw和reason才能定位到具体是哪类话术带偏了模型。“最近系统变慢,是不是路由拖了后腿”,你需要latency_ms来验证。

没有这些记录,路由调优就只能靠猜。有了这些记录,一次线上事故调优可能只需要十几分钟。可观测性做在路由环节,比做在任何其他环节都值得。

5. 性能调优:从串行路由到并行分发

5.1 路由延迟是整条链路的放大器

很多Agent项目做性能优化时,注意力都放在业务节点上,却忘了用户请求的第一跳就是路由。如果路由做一次大模型调用需要1到2秒,那整条链路不管后面多快,用户感知到的延迟都已经加上这1到2秒了。所以在做延迟预算时,我通常会单独给路由划一块时间。

举个例子,如果目标是接口P95延迟小于3秒,路由这段我一般只给它500毫秒以内的预算。超出这个预算,就要考虑换更快的模型、做缓存、或者用向量语义Router替代大模型分类。这不是苛刻,是因为路由之后往往还有好几层业务节点要用大模型,你不可能把时间全部预留在最前面。

5.2 并行扇出:一次入口同时做多路判断

有些场景下,路由不只是要做意图判断,还想同时做情绪识别、实体抽取、语种检测。如果串行处理,一次判断结束再做下一次,延迟就是线性叠加的。LangGraph支持多个节点之间的并行分发,你可以从入口节点同时连到多个分析节点,等它们都返回后,再进入汇聚节点统一处理。

这种并行扇出能显著降低整体延迟,但不是没有代价。每个并行分支都会产生一次模型调用,都会写入状态,最后汇聚时还要处理字段合并和冲突。我的建议是:如果另外几个分析结果在后续节点里根本用不上,就别并行;只有当多个分析结果都是后续链路必需输入时,并行才有正收益。

在实际的客服场景里,我见过一个很好的实践:入口节点同时跑“意图识别”和“紧急程度判断”两个轻量任务,等两个结果都出来后,再决定走普通售后节点还是紧急应对节点。这样既没有牺牲整体延迟,又避免了“紧急用户被普通流程耽误”的问题。

5.3 分级路由:粗分快判+困难样本精判

性能优化里性价比最高的一招,是分级路由。思路很简单:先用便宜的、低延迟的模型做粗分,粗分结果置信度高就直接走对应分支;粗分结果模糊,再升级到大模型精判。

具体落地时,我通常用向量语义Router做第一级粗分:

  • 计算用户输入和所有意图示例的相似度。
  • 如果最高相似度大于0.85,直接采用该意图,整个过程只需一次Embedding计算,延迟很低,成本几乎为零。
  • 如果最高相似度介于0.6到0.85之间,说明表达比较模糊,进入第二级,用结构化LLM Router做精判。
  • 如果最高相似度低于0.6,说明完全命中不了已知意图,直接转人工,避免浪费大模型调用。

这套方案上线后,大概能覆盖60%到70%的简单请求在粗分阶段快速返回,只有模糊样本才需要消耗更大模型的Token。整体路由延迟的中位数下降非常明显,同时模型成本也控制住了。工程里永远不要追求“所有请求都用最强模型处理”,按难度分级,才是长期可维护的架构。

6. 我踩过的五个Router实战深坑与修复记录

6.1 映射表不完整导致KeyError

我踩的第一个坑,是条件边映射表漏配,导致线上直接报KeyError。当时路由函数返回了一个“退款”意图,但映射表里只注册了“售后”“售前”“物流”,LangGraph发现返回值没有对应key,立刻抛异常,整个请求直接失败。

这个坑的麻烦之处在于,开发和测试阶段很难触发。开发用的测试集覆盖不到“退款”这种边缘意图,只有真实用户的表达多样性才能把它逼出来。修复方法有两个:第一,在映射表里把实际业务中可能碰到的意图域都注册上;第二,也是最有效的,在路由函数里做白名单强制归一化:

allowed = {"售后", "售前", "物流", "人工", "闲聊"} def normalize_intent(intent: str) -> str: if intent in allowed: return intent return "人工"

现在我的所有路由函数都有这行白名单校验,宁可多转到人工,也不能让整个图因为一个未注册值直接崩溃。

6.2 路由结果污染业务状态

这就是文章开头那个发票案例的根因。当时Router判断出用户意图是售后,就直接写进了state["topic"],后续售前节点把topic当作用户真实诉求,给出了“推荐商品”的答案。整个排查过程花了两个多小时,最后在trace里看到topic被Router覆盖的那一刻,真的很想骂人。

修复方案前面已经说了:所有路由决策统一放进router_meta,业务字段保持原始输入语义。这个规矩我在团队里定成了代码评审的红线——凡是发现Router节点写业务字段的,一律打回。别觉得这是小题大做,线上这种坑踩一次就一定够长了。

6.3 结构化输出偶尔仍是字符串

你以为使用了with_structured_output就安全了,但真实世界没那么理想。有一次底层模型网关出了异常,返回了一个纯字符串而不是结构化对象,结果在解析阶段直接抛异常,路由节点崩溃。

从那以后,我给所有结构化输出都包了一层安全解析函数:

from typing import Optional def safe_parse_intent(raw) -> Optional[IntentResult]: if isinstance(raw, IntentResult): return raw if isinstance(raw, dict): try: return IntentResult(**raw) except Exception: return None return None

解析失败时的行为很明确:放弃本次路由判断,转人工。同时把异常信息写进router_meta,方便事后定位为什么底层网关会返回异常格式。安全解包这个动作看似多余,但它在关键时刻能避免一个全局崩溃。

6.4 单轮输入误判率高

有一段时间我发现一个诡异的现象:单独测试每一轮对话都能正确分类,但线上数据一统计,误判率接近30%。后来排查发现,问题出在输入上——我当时只传了用户最后一句话去做路由。

客服场景经常会有“多轮才说清楚”的情况。用户先问“我有一个订单”,再追问“发货了没”,单独拿后面这句话去路由,缺乏上下文,几乎所有模型都会懵。修复方法是在路由前增加一个轻量的上下文摘要节点,把最近几轮对话压缩成一段摘要,和当前问题一起传给Router。这样一个轻微的成本增加,换来的是意图判断准确率明显提升。

要说清楚的是,上下文摘要会增加一次小模型调用。但相比错误路由带来的后续无效链路消耗,这笔开销通常划算得多。如果你做的场景里用户表达高度依赖上下文语境,这个节点基本是必需的。

6.5 超时默认值导致整链路卡死

最后一个坑来自调用侧的默认配置。有一次内部模型网关配置的超时上限是120秒,路由调用也没有额外设置超时参数,结果一个简单的用户请求在路由节点整整卡了两分钟。用户早就走了,Agent还在等模型返回。

这个问题的本质是:路由入口的鲁棒性比准确率更重要。我后来给所有路由调用统一加上了timeout=5和max_retries=1,超时直接按转人工处理。不管后续业务节点有多复杂,用户在入口处绝不能因为路由等待而失去耐心。路由快不一定是好事,但路由卡死一定是灾难。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询