做 AI Agent 的开发者，迟早都会遇到同一个问题：没有任何一个模型能同时满足“便宜、快、准、稳”这四个要求。我在做一个客服支持类 Agent 项目的时候，初期图省事，把所有请求都丢给当时能力最强的商业闭源模型。结果半个月后一看账单，token 消耗比预期翻了快三倍；更糟的是某天下午该模型接口限流，整个 Agent 直接瘫痪，用户的问题全部卡在排队队列里。这时候我才认真思考一件事：能不能在 LangChain 里加一层“模型路由”，让不同的请求走不同的模型，并且在一个模型出问题的时候自动切换？

这篇是 LangChain 入门系列的第十一篇，主题就是多模型路由策略以及 LLM 智能切换。我会先讲清楚为什么需要路由，再拆解四种常见的路由策略，然后给出基于 LangChain 的完整实现方案，最后分享一次线上事故的排查复盘。适合正在做 Agent 落地、想控制成本、想提升系统稳定性的开发者阅读。

1. 一个模型撑不住整个 Agent：成本、能力与稳定性的三角困局

很多人的直觉是：既然最强模型的效果最好，那就所有请求都走它。这个思路在小规模 Demo 里没有任何问题，但一旦流量上来，你会发现三个因素会互相拉扯。

1.1 四个真实痛点：成本、能力错配、限流雪崩与数据边界

第一个痛点是成本。Agent 场景和普通对话不一样，它内部会有多次模型调用：规划任务、调用工具、总结回复、判断是否继续。一次用户请求可能隐含着三到五轮 LLM 调用。如果全部用最强模型，成本是线性的，流量一涨，账单就跟着涨。而且很多调用根本用不到那么强的能力——比如让模型做“判断用户是否在打招呼”这种二分类任务，和能力最强的模型相比，一个轻量模型的效果几乎没有差别，价格却能差出几十倍。

第二个痛点是能力错配。不同的模型擅长的事情是不一样的。有些模型代码生成和复杂推理很强，但日常闲聊风格生硬；有些模型中文表达更自然，但逻辑推演会漏步骤；有些开源模型在私有化部署场景下效果反而更可控。把所有请求都固定到一个模型上，等于放弃了这种差异化优势。

第三个痛点是可用性风险。任何一家模型服务都不敢保证永远不限流、不出故障。如果 Agent 只有一个模型依赖，那这个模型一抖，整个业务就跟着抖。你在外面看到的所有“高可用 Agent 架构”，本质上都在模型之上加了一层路由和容灾机制。

第四个痛点是数据边界。有些业务数据不适合发往外部模型，有些任务必须走私有化部署的模型。如果你不在路由层做管控，数据流向就是失控的。

我在项目里最直观的感受是：当你把 Agent 当成一个“系统”来做，而不是一个“调用脚本”来做，路由就成了刚需。它解决的不是“哪个模型更好”的问题，而是“在什么场景下用哪个模型更合适”的问题。

1.2 路由层和“散落的 if else”有什么本质区别

有人会说：这有什么难的？我在业务代码里判断一下关键词，if xxx 就走模型 A，else 就走模型 B，不就叫路由了吗？

表面上差不多，但工程上有本质区别。散落的 if else 是面向过程的，每次新增一个模型，你都要翻遍业务代码去改判断逻辑，路由决策和业务逻辑完全耦合在一起。而一个真正的“路由层”应该做到四件事：

• 一处定义，全局生效。所有模型的路由规则、备选关系、切换阈值都集中在一个配置文件或注册表里。
• 策略可动态调整。不需要发版就能调整某个任务类型的默认模型，或者调整熔断阈值。
• 失败有回退路径。主模型不可用时，自动走备用模型，而不是直接把错误抛给用户。
• 调用方能感知路由结果。日志里能清楚地看到这次请求分给了谁、为什么分给它、切换过程是否发生。

LangChain 在这方面的价值在于，它本身就是一个编排框架，路由逻辑在链路层实现，业务逻辑只关心最终拿到一个Runnable。你在代码里写的是“我要得到一个回复”，至于这个回复来自哪个模型、中间有没有切换，业务层可以不感知，但日志里每一跳都看得清清楚楚。

2. 从规则到自评判：四类路由策略的适用边界与选择逻辑

模型路由在 LangChain 生态里其实是一个很成熟的话题，社区里有不少现成方案，但底层策略可以归纳为四类。

2.1 规则路由：最朴素的确定性分发

规则路由不依赖任何模型判断，直接通过输入内容的显性特征来决定走向。比如：

• 用户问题长度超过 2000 字，走强推理模型；
• 问题里包含“投诉”“退款”等关键词，走专门处理复杂工单的模型；
• 问题以“你好”“在吗”开头，走轻量对话模型；
• 用户 ID 属于付费用户，走效果更好的模型。

我见过很多团队觉得这种路由太土，但实际生产环境里，它是最可靠、最不容易出岔子的方案。原因很简单：它没有额外调用，没有模型幻觉，也不会因为提示词写得不清楚而把请求分错。我自己的经验是，能走规则覆盖的场景，优先用规则；只有当规则覆盖不住的时候，才引入更高阶的路由方式。

2.2 提示路由：让一个便宜模型当“分拣员”

提示路由的思路是：不写规则，而是直接把候选模型的描述写进一个提示词模板，然后让一个便宜、快速的小模型来阅读用户请求，从候选中选一个最合适的模型。这个过程很像快递分拣中心的传送带：你不会让一辆大货车去挨家挨户敲门送快递，而是让它在分拣中心停下，由分拣员决定这件包裹去哪个区、上哪辆车。

LangChain 里有专门的LLMRouterChain实现这种策略，底层就是构造一个路由提示模板，模型输出“应选择的模型编号”。这个方案的好处是能处理语义模糊的请求，坏处是增加了一次模型调用，而且调度模型本身也可能判断错误。我后面会在落地部分给出更工程可用的写法。

2.3 语义路由：把问题嵌入到向量空间再比较

语义路由的思路是：把你可能遇到的典型问题都预先写成示例，给每个示例算好 Embedding，存成一张“问题地图”。当新请求到来时，先算它的 Embedding，再在地图里找最相似的示例，命中哪个分类就走哪个模型。

LangChain 里有一个EmbeddingRouterChain，核心逻辑就是这么做的。这个方案适合意图模式相对固定的场景。比如你的 Agent 只处理“售前咨询”“售后维修”“退换货申请”三种意图，每种意图各找 20 个典型问题做 Embedding，效果通常比纯关键词规则好，比 LLM 路由稳定且便宜。它的局限在于：如果用户提问的方式多种多样，你要不断补充示例库，维护成本会上升。

2.4 反馈路由：用历史表现给模型打分

反馈路由是最进阶的一类：不是根据请求内容来路由，而是根据“每个模型的近期表现”来路由。你可以统计每个模型的成功率、平均响应时长、用户点赞率、推理质量得分等指标，然后把这些指标换算成分数，请求到来时挑选分数最高的模型。

它特别适合那种线上运行了很久、积累了丰富反馈数据的 Agent。但冷启动阶段通常要和规则路由配合使用——没有历史数据时用规则兜底，等数据积累到一定量再让反馈路由接管。

3. RunnableBranch 与 RouterChain：LangChain 路由实现的两条路线

聊完策略，进入实现。LangChain 生态里目前实现路由主要有两条路线：一类是早期的RouterChain系列，另一类是 LCEL 体系里的RunnableBranch。

3.1 为什么我最终选择了 RunnableBranch 而不是 RouterChain

RouterChain这套体系在上古版本里很流行，有LLMRouterChain、EmbeddingRouterChain等，但它们的问题是和后来的LangChain Expression Language（也就是 LCEL）集成得不太好，很难做流式输出、并行执行和统一的重试。现在的项目里，如果你用ChatOpenAI这类新模型接口，官方也更推荐用 LCEL 组合构造链路。

RunnableBranch是一种结构化的条件分支构造器，它的核心逻辑是：按顺序执行一组(condition, runnable)对，只要某个条件为真，就执行对应的 runnable；如果所有条件都不满足，就走一个 default runnable。这个模型天然契合上面的规则路由，也天然支持流式调用，因为整个链路都基于 LCEL。

更关键的一点：RunnableBranch可以嵌套。你完全可以在某个分支里再挂一个小的RunnableBranch，做成二级路由。这比强行用一套 RouterChain 处理所有情况要灵活得多。

3.2 基于规则的 RunnableBranch 代码实现

先看最基础的规则路由实现。我用的是模拟项目 X 里的一段简化代码，模型名用了虚构代号：thinker-x代表强推理模型，general-7b代表通用模型，lite-3b代表轻量模型。

from langchain_core.runnables import RunnableBranch, RunnableLambda from langchain_openai import ChatOpenAI # 模拟模型工厂，实际项目中这里会对接不同的模型服务 def create_model(name: str): # 这里只是示意，不同模型有不同的 base_url、model_name、api_key return ChatOpenAI(model=name, temperature=0.1) thinker_x = create_model("thinker-x") general_7b = create_model("general-7b") lite_3b = create_model("lite-3b") # 条件函数：判断是否为复杂工单 def is_complex_query(query: str) -> bool: # 这里用关键词和长度做启发式判断，实际项目中可以加更多特征 risk_words = ["退款", "投诉", "故障", "升级", "合同", "法律"] if any(word in query for word in risk_words): return True return len(query) > 500 def is_simple_greeting(query: str) -> bool: greeting_words = ["你好", "在吗", "hi", "hello", "早"] return any(word in query.lower() for word in greeting_words) # 构造路由分支 router = RunnableBranch( # 按顺序匹配：先处理复杂工单，再处理简单问候，其余走通用模型 (lambda x: is_complex_query(x["query"]), thinker_x), (lambda x: is_simple_greeting(x["query"]), lite_3b), general_7b # 默认分支 ) # 调用时传入包含 query 的字典 result = router.invoke({"query": "我想投诉，你们的产品出现严重故障"}) print(result.content)

这里有个关键点：RunnableBranch的条件函数接收的不是原始字符串，而是invoke传进来的整个字典对象。所以实际项目里我会在字典里多放一些字段，比如user_level、request_type、session_id，这样条件函数就不用写死在字符串匹配里，而是直接读字典里的结构化信息。

3.3 让模型当调度员：基于 LLM 的路由实现

规则路由覆盖不了的情况，我会上一个轻量模型做调度。核心思路是：先用一个提示词把候选模型列出来，让调度模型输出一个决定，再把这个决定映射到具体模型。

这里我建议不要直接用LLMRouterChain的旧式用法，而是自己构造一个两步链路：第一步调用调度模型得到路由结果，第二步根据结果选择目标模型。

from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate router_prompt = ChatPromptTemplate.from_template( """你是一个模型路由调度员。根据用户请求，从下面的候选中选择一个最合适的模型，只输出模型编号。 1. thinker-x：擅长复杂推理、代码生成、长文本分析，适用于投诉、技术问题、合同纠纷等复杂场景。 2. lite-3b：擅长简短对话、打招呼、情绪安抚，适用于便宜快速的轻量交互。 3. general-7b：通用对话、常见问题解答，适用于大部分日常请求。 用户请求：{query} 请只输出数字编号：""" ) # 调度模型本身要足够便宜快，一般用轻量模型 dispatcher_model = ChatOpenAI(model="lite-3b", temperature=0) router_chain = router_prompt | dispatcher_model | StrOutputParser() # 将模型编号映射到具体 runnable def route_by_model_id(model_id: str): model_map = { "1": thinker_x, "2": lite_3b, "3": general_7b, } return model_map.get(model_id.strip(), general_7b)

然后我会把规则路由和这个 LLM 路由组合在一起：

from langchain_core.runnables import RunnableLambda def smart_router(inputs: dict): query = inputs["query"] # 规则优先 if is_complex_query(query): return "1" if is_simple_greeting(query): return "2" # 规则覆盖不了，交给调度模型 return router_chain.invoke({"query": query}) final_router = ( RunnableLambda(smart_router) | RunnableLambda(route_by_model_id) )

RunnableLambda的连续管道写法非常直观：第一步是路由决策，第二步是根据决策拿到对应模型。这样写的好处是决策逻辑可以随时替换，你可以从“纯规则”平滑过渡到“规则+LLM 调度”的混合模式，对线上系统非常友好。

4. 智能切换不只是换模型：健康检查、熔断与上下文保真的工程细节

真正在线上跑过 Agent 的人都清楚，路由只是第一层，第二层是“切换”。很多开发者把路有理解成“选模型”，但实际更重要的是“在模型异常时如何优雅地换一个”。

4.1 模型健康注册表：记录每个模型的实时状态

我会维护一个简单的健康注册表，用一个 Python 字典保存每个模型的累计调用次数、失败次数、平均响应时间、最近错误信息。条件分支在做路由时，不只是看请求内容，还会看这个模型的健康分：如果某个模型最近失败率超过阈值，就临时把它从候选列表里摘掉。

import time class ModelHealthRegistry: def __init__(self): self._stats = {} def record_success(self, model_name: str, latency_ms: float): stat = self._stats.setdefault(model_name, {"success": 0, "fail": 0, "total_latency": 0, "last_fail_time": 0}) stat["success"] += 1 stat["total_latency"] += latency_ms def record_fail(self, model_name: str, error: str): stat = self._stats.setdefault(model_name, {"success": 0, "fail": 0, "total_latency": 0, "last_fail_time": 0}) stat["fail"] += 1 stat["last_fail_time"] = time.time() stat["last_error"] = error def health_score(self, model_name: str) -> float: stat = self._stats.get(model_name) if not stat: return 1.0 total = stat["success"] + stat["fail"] if total == 0: return 1.0 recent_fail = stat["fail"] / total # 也可以用滑动窗口，这里简化为总体失败率 return 1.0 - recent_fail

然后在路由条件里增加健康分判断。比如thinker-x虽然是最适合复杂工单的模型，但如果它最近失败率超过 0.2，我就临时把它替换成general-7b。这个逻辑相当于给路由加了一个熔断开关。

4.2 超时、重试和熔断参数的设计经验

模型调用超时和重试参数看起来是小事，但设置错了往往会在高并发下把故障放大。针对 LangChain 的模型接口，我目前的推荐值如下：

特别提醒：重试不能盲目。如果模型服务已经限流，重试只会放大压力。我自己的做法是区分错误类型——限流类错误（通常有特定状态码）只做有限重试，认证类错误不重试直接报错，超时类错误可以多试一次。

4.3 切换模型时不丢上下文的小技巧

模型切换最容易出问题的点并不是切换本身，而是切换后的上下文。因为不同模型的输入格式、System Prompt 的敏感程度、对历史消息的依赖程度都不一样。

我踩过几次坑之后总结了一套操作：

• 每次切换前，把历史对话先压缩成一段摘要，保留最近三轮的完整消息原文。
• 对新的模型统一格式化 System Prompt，不要直接复用旧模型的提示词，否则新模型会因为指令冲突而表现异常。
• 如果两个模型使用不同的 tokenizer，上下文长度单位不要按字符算，按模型各自的窗口比例算。我给每个模型预设了max_context_ratio，比如强推理模型允许塞入 70% 的窗口，轻量模型只塞入 40%，留出输出空间。

这套逻辑我会封装成一个context_adapter，在路由链路的入口处做统一处理。这样下游模型拿到的输入永远是“摘要+最近几轮原文”的紧凑结构，不会因为切换模型而把整个历史全量重放一遍。

5. 一个带路由的客服 Agent 完整示例：注册、分流与调度

理论说了这么多，是时候看一个能跑通的最小系统了。我模拟了一个客服 Agent 的结构，核心思想是完全可复现。

5.1 模型注册表与配置中心

第一步是把所有模型的接入信息集中管理。实际项目中这些配置应该放在配置中心或环境变量里，这里我用一个简单字典示意。

MODEL_REGISTRY = { "thinker-x": { "provider": "openai-like", "model_name": "thinker-x", "max_context_ratio": 0.7, "healthy": True, "description": "复杂推理、投诉、技术工单", }, "general-7b": { "provider": "openai-like", "model_name": "general-7b", "max_context_ratio": 0.6, "healthy": True, "description": "通用问答、日常对话", }, "lite-3b": { "provider": "openai-like", "model_name": "lite-3b", "max_context_ratio": 0.4, "healthy": True, "description": "打招呼、简短回复、分类", }, }

5.2 组装路由链路与 Agent 工具

接下来构造一个带工具的客服 Agent。假设这个 Agent 有一个查订单状态的外部工具和一个查售后政策的工具。路由层的输出会作为 Agent 的可选模型之一。

from langchain.tools import tool from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain_core.prompts import ChatPromptTemplate @tool def query_order_status(order_id: str) -> str: """查询指定订单的当前状态。""" # 模拟调用外部订单系统 return f"订单 {order_id} 状态：已发货，预计三天内送达。" @tool def query_after_sales_policy() -> str: """查询售后政策说明。""" return "支持七天无理由退货，质量问题可免费换新。" tools = [query_order_status, query_after_sales_policy] agent_prompt = ChatPromptTemplate.from_messages([ ("system", "你是客服助手，请根据用户请求使用工具并给出简洁回答。"), ("placeholder", "{chat_history}"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}"), ])

然后我把路由结果嵌入到 Agent 的模型选择中。基本思路是：先用一个route_query函数决定走哪个模型，再用这个模型创建 Agent。

def route_query(inputs: dict) -> ChatOpenAI: query = inputs.get("input", "") health = ModelHealthRegistry() # 规则路由 if is_complex_query(query) and health.health_score("thinker-x") > 0.8: model = create_model("thinker-x") elif is_simple_greeting(query) and health.health_score("lite-3b") > 0.8: model = create_model("lite-3b") else: model = create_model("general-7b") return model def build_agent_for_query(inputs: dict): model = route_query(inputs) agent = create_tool_calling_agent(model, tools, agent_prompt) return AgentExecutor(agent=agent, tools=tools, verbose=True)

这个写法虽然简单，但有一个很实用的点：每次请求都会根据路由结果新建一个 AgentExecutor，实际上是把“路由决策”和“Agent 执行”解耦了。你在日志里看到的是route_query的输入输出，以及 Agent 执行详情，两条记录合起来就能完整还原一次请求的流向。

5.3 实测效果与成本对比

我在这套模拟环境里跑了三组典型请求，结果如下：

如果所有请求都走 thinker-x，第三类请求的体验确实没变化，但第一类请求的成本会直接放大 15 倍，而且延迟还会高出不少。路由的价值在这里体现得很直观：把最便宜的资源留给最简单的请求，把最强的资源留给最值得的场景。

6. 一次线上事故的排查复盘：路由顺序写错之后

代码写得再漂亮，线上还是会出幺蛾子。下面这个事故让我对RunnableBranch的匹配顺序有了更深的理解，也希望对你们有启发。

6.1 现象：某天下午，Agent 突然变慢且输出质量下降

模拟项目 X 上线了路由策略后的第三周，有用户反馈 Agent 的回复速度明显变慢，而且一些简单问题的回答风格也变得冗长。我第一时间去看模型健康注册表，所有模型的失败率都正常，没有限流记录。这排除了模型服务本身出问题的可能性。

6.2 排查过程：从日志路由分布到代码逻辑

我去翻路由日志，发现一个反常现象：大量简单问候类的请求被路由到了thinker-x，而不是lite-3b。按照我的规则，is_simple_greeting应该优先匹配，为什么会走到强推理模型？

打开代码一看，问题出在RunnableBranch的分支顺序上。我在调整策略时，把thinker-x的复杂工单判断写在了lite-3b的问候判断之前，本意是可以的——因为复杂工单和简单问候在逻辑上是互斥的。但实际生效的代码里，我那个is_complex_query的判断条件写得太宽：只要请求里包含“投诉”“退款”等词就返回 True。有个用户发的是“你好，我想咨询一下退款政策，但是不投诉，请简单介绍”，这句话同时命中了is_complex_query和is_simple_greeting。

而RunnableBranch的语义是优先匹配第一个为 True 的条件，所以这条请求直接进了thinker-x。更强的模型处理一个简单咨询请求当然也能完成，但风格冗余、延迟偏高，给用户的感觉就是“变慢了、变啰嗦了”。

6.3 根因与修复：条件互斥分析和默认分支的兜底语义

根因有两个层面。第一层是条件函数设计不当：is_complex_query用纯关键词匹配，没有考虑否定语境；“退款”出现在咨询句和投诉句里，语义完全不同。第二层是我对RunnableBranch的顺序敏感度认识不足：它天然是“第一个命中就短路”，所以条件的互斥性必须保证。

修复也比较直接。我先把条件函数改成支持否定语境，比如检测“不投诉”“不申请”等否定短语，将对应关键词排除；然后调整了分支顺序，把问候类判断放到最前面；最后在 default 分支里做了一个保护——如果模型健康分过低，即使被路由到该分支，也会立刻被替换成备用模型。

def is_complex_query(query: str) -> bool: risk_words = ["退款", "投诉", "故障", "升级", "合同", "法律"] # 简单否定检测：如果风险词前面有“不”“没”“无需”等否定词，则忽略该词 negative_prefixes = ["不", "没", "无", "别", "不用"] for word in risk_words: if word in query: idx = query.index(word) prefix = query[max(0, idx - 2):idx] if any(neg in prefix for neg in negative_prefixes): continue return True return len(query) > 500

经过这次事故，我总结出一个排查顺序的模板，希望对你们有用：

• 第一步：看路由日志，确认请求实际分给了哪个模型，而不是靠猜。
• 第二步：检查RunnableBranch的分支顺序，确认是否有两个条件同时为真。
• 第三步：检查条件函数本身的语义边界，尤其是关键词路由里的否定语境和同义词场景。
• 第四步：看 default 分支是否有兜底策略，避免“所有条件都没命中”时直接报错。

事故的根源往往不是模型出问题，而是路由策略本身有逻辑漏洞。把路由决策理解成一个独立的“调度服务”，而不是一行简单的 if else，很多问题就能在设计和 review 阶段提前暴露。

我还想提一个容易忽略的细节：LLM 路由的解析失败。如果你用了调度模型来决定走哪个分支，注意它可能输出“1.”而不是“1”，也可能输出“我认为应该是 2”。解析失败时必须走默认分支，不能把错误抛给用户。我在生产环境里会对调度模型输出做一次规范化映射，并用 few-shot 示例强制它只输出数字编号。

至于流式兼容，RunnableBranch的每个分支都是标准 Runnable，天然支持stream方法。你可以在测试环境里分别验证每个分支的流式输出是否正常，这个步骤虽然简单，但能避免上线后才发现某个模型不支持流式导致的体验问题。

这个系列的下一篇，我打算写 LangChain 里如何设计一套可观测的模型调用日志系统。有了路由之后，日志的重要性甚至会超过路由本身。先把这个基础打好，后面做成本分析和质量评估就都有据可查了。