最近我们组里在做Agent选型的时候,遇到一个很现实的场景:不同任务、不同时段、不同预算之下,同一个模型根本不可能“一把梭”。有的查询只是做个关键词抽取,拿旗舰模型跑纯属浪费;有的业务却需要长上下文、复杂推理,轻量模型一上来就露怯。后来我们在LangChain里把多模型路由和智能切换做成了标准能力,效果立竿见影。这篇就结合我们踩过的坑,把“怎么让Agent自己决定调用哪个LLM”这件事讲透。
1. 为什么要做多模型路由:问题背景与核心思路
1.1 单模型策略的三大硬伤
先说说为什么不能只挂一个模型。很多入门项目甚至生产Demo,都喜欢在Agent里写死一个模型:构建Prompt → 调用LLM → 拿到结果。这套链路在小流量、单任务的场景下完全没问题,但一旦Agent要服务多种类型的请求,就会暴露三个硬伤。
第一是成本失配。不同模型的价格能差几十倍。如果Agent每天处理十万次请求,其中八成是“查天气”“抽关键词”这种简单任务,却全部走旗舰模型,月底账单会非常难看。第二是质量失配。轻量模型在简单任务上表现并不差,但在数学推理、长文档理解、代码生成这类任务上,它跟旗舰模型的差距是肉眼可见的。用一个统一模型,要么简单任务费钱,要么复杂任务拉胯,两头不讨好。第三是稳定性失配。单一模型出现限流、超时、接口故障时,Agent整个就瘫了,没有降级方案。
1.2 路由的本质:给请求分流
多模型路由策略的核心思路,说白了就一句话:在模型调用之前,加一道“分流器”,根据请求的特征,决定把这条请求交给哪个模型处理。
这道分流器不神秘,也不用在大模型层面做什么改造。它就是一个普通的函数、一条分支逻辑,或者一个专门用来分类的轻量模型。关键在设计分流规则时想清楚:什么特征的请求该走哪条路。
我习惯把分流维度归纳为四类:
- 任务类型维度:代码生成、数学推理、摘要、翻译、问答,分别对应不同模型能力侧重的模型。
- 复杂度维度:短文本分类用轻量模型,长文档分析或复杂推理用旗舰模型。
- 预算维度:同一个任务类型下,根据用户等级或业务优先级选择高配或低配模型。
- 可用性维度:主模型故障或限流时,自动切换备选模型。
1.3 LangChain在这个场景里的角色
有人可能会问:路由逻辑不难写,为什么要用LangChain?我的体会是,LangChain提供了一套标准化的模型调用抽象,让“路由”这件事不用你重复造轮子。
在LangChain里,所有模型都被包装成统一接口的Runnable对象,不管是哪家的模型,调用方式都是invoke、stream、batch这一套。这样一来,路由层完全不关心背后是哪个模型,只负责“把请求交给某个Runnable”。而且LangChain本身提供了RunnableBranch这类专门做分支路由的组件,配合自定义路由函数,可以实现很优雅的分流逻辑。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. LangChain中的模型路由基础:从静态配置到动态决策
2.1 先把Runnable的底子打好
LangChain从0.1.x开始,核心抽象从Chain转向了Runnable。你可以把Runnable理解为一个“可执行单元”:输入一个东西,输出一个东西。模型是Runnable,Prompt模板套上模型也是一个Runnable,连你自己写的一个普通函数,只要包上@tool或者直接传入,也能当Runnable用。
这个抽象对路由策略特别重要。因为路由本质上就是“把不同的输入导向不同的Runnable”,如果你的模型不是Runnable,你就得自己写一堆if-else去调用不同的SDK,代码会很散。而用Runnable统一包装之后,路由逻辑就变成:
python复制llm_a = ChatModelFactory.create("high-end")
llm_b = ChatModelFactory.create("lightweight")
def route_fn(query: str) -> Runnable:
if is_complex_task(query):
return llm_a
return llm_b
2.2 路由的两种基本形态
LangChain社区里用到的路由方案,大体可以分两种形态。
静态路由。就是提前把规则写死,比如“凡是带代码片段的请求走代码模型,凡是纯文本问答走通用模型”。这种方案简单直接,适合规则清晰、不会频繁变化的场景。缺点是规则写多了之后容易变成一堆if-else,维护起来费劲。
动态路由。请求进去之后,先由一个“分流判断器”实时判断该走哪条路。这个判断器可以是一个小模型、一个关键词规则引擎,也可以是一个嵌入向量相似度匹配器。动态路由的优点是可以处理无规则可循的复杂请求,缺点是增加了一次额外的判断开销,而且判断器本身也可能判断错。
实际项目中,这两者往往是混合使用的:先用规则做一层粗筛(比如带代码块标记的直接走代码模型),再用分类模型做细筛(比如识别请求属于“复杂推理”还是“简单抽取”)。
2.3 路由决策的常用输入信号
做路由判断时,能利用的信号比很多人想象中要多。我整理一下常用的几类:
- Prompt本身:任务内容、语言、长度、是否包含代码、是否包含公式。
- 任务元数据:用户等级、业务线、API来源、请求类型标签。
- 模型实时状态:各模型的错误率、平均响应时间、剩余配额。
- 历史反馈:同一类任务过去在哪个模型上表现更好。
把这些信号组合起来,路由策略就不再是“拍脑袋分路”,而是有依据的决策。比如同样的代码生成请求,普通用户走轻量模型,VIP用户走旗舰模型,这就是“任务类型 + 用户等级”的组合路由。
3. 核心实现:路由策略的落地代码
3.1 用RunnableBranch搭简易路由
LangChain内置了一个组件叫RunnableBranch,它就是专门为这种分流场景设计的。它的用法类似编程语言里的switch-case:一组条件和对应的Runnable,从上往下匹配,命中哪个分支就走哪个Runnable。
下面是一段典型的路由配置代码:
python复制from langchain_core.runnables import RunnableBranch, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
high_end_llm = ChatOpenAI(model="high-end-model", temperature=0.2)
light_llm = ChatOpenAI(model="light-model", temperature=0.3)
code_llm = ChatOpenAI(model="code-model", temperature=0.1)
prompt_template = ChatPromptTemplate.from_template("{query}")
def is_code_request(query: str) -> bool:
code_markers = ["```", "def ", "class ", "function", "import ", "SELECT ", "curl "]
return any(marker in query for marker in code_markers)
def is_complex_request(query: str) -> bool:
# 这里可以换成你需要的判断逻辑
return len(query) > 500 or "推理" in query or "计算" in query
branch = RunnableBranch(
(is_code_request, code_llm | prompt_template),
(is_complex_request, high_end_llm | prompt_template),
(lambda _: True, light_llm | prompt_template) # 默认分支
)
注意这段代码我只是示意,实际跑的时候需要根据你的LangChain版本调整模型导入路径和参数。RunnableBranch的匹配顺序是从上到下,第一个命中就执行,后面不会再匹配。所以默认分支一定要放在最后,用一个恒为True的条件兜底。
3.2 用轻量模型做路由判断器
上面这种基于规则的分支,适合特征明显的请求。但真实场景里很多请求是模糊的,比如“帮我看看这段需求文档有什么坑”,你说它是复杂还是简单?这时候更适合用一个轻量模型来做路由判断。
做法是这样的:先让轻量模型把用户请求分类,输出一个类别标签,然后根据类别标签决定走哪个模型。
python复制from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
router_prompt = ChatPromptTemplate.from_template(
"""请判断以下用户请求属于哪个类别,只输出类别名,不要输出其他内容。
类别:["code", "math", "summary", "translation", "simple_qa", "other"]
用户请求:{query}"""
)
router_chain = router_prompt | light_llm | StrOutputParser()
def route_by_class(query: str) -> Runnable:
category = router_chain.invoke({"query": query}).strip()
if category == "code":
return code_llm
elif category == "math":
return high_end_llm
elif category == "summary":
return medium_llm
elif category == "translation":
return light_llm
else:
return high_end_llm
这里有一个容易被忽略的细节:路由判断用的模型,要跟任务处理用的模型分开。别用同一个旗舰模型既做分类又做生成,那等于多花了一次旗舰模型的钱。轻量模型做分类任务能力足够,而且速度快、成本低。
3.3 组装成完整的路由链路
有了路由判断,下一步就是把整个链路串起来。在LangChain里,我们可以用RunnableLambda把路由函数和分支逻辑组装成一个完整的Agent执行链路:
python复制from langchain_core.runnables import RunnableLambda
def smart_route(query: str) -> str:
selected_model = route_by_class(query)
response = selected_model.invoke(query)
return response.content
agent_chain = RunnableLambda(smart_route)
这么做有个好处:对上层调用方来说,它们只看到agent_chain这一个入口,完全不需要关心内部有几个模型、路由规则是什么。以后要调整路由策略,只改路由函数内部逻辑,不影响外部接口。
3.4 带上下文记忆的Agent路由
如果路由是嵌在Agent流程里的,而不是纯粹的“一问一答”,情况会稍微复杂一点。因为Agent往往有历史对话、工具调用记录、临时状态,这些都要参与路由判断。
我的做法是把历史摘要作为路由信号之一。比如先把最近的对话压缩成一个摘要,判断当前请求是否依赖历史上下文;如果依赖,就优先选上下文窗口更大的模型;如果不依赖,可以根据当前请求单独分流。
python复制def route_with_history(query: str, history_summary: str) -> Runnable:
if len(history_summary) > 1500:
return high_context_llm
if query_needs_tools(query):
return tool_calling_llm
return route_by_class(query)
3.5 模型切换的平滑处理
还有个实操问题:路由到新模型之后,对话历史还能不能用?不同模型的上下文格式、system prompt要求可能不一样,直接拿上一个模型的对话历史怼给下一个模型,经常出乱子。
我的建议是,在切换模型时做一个历史格式的转换:统一的对话历史内部表示,在传给具体模型之前,按目标模型的要求格式化。LangChain里各家模型的消息类格式不完全一样,但大部分都支持HumanMessage、AIMessage这种通用类型,转换成本不大。
4. 智能切换LLM的进阶玩法:场景驱动的路由规则
4.1 成本优先型路由:先把账单降下来
多模型路由最直接的价值就是降成本。我们有个线上服务,之前全部请求走旗舰模型,月成本高得离谱。后来做了一个简单的分级:短问题(少于200字符)走轻量模型,中等长度问题走中端模型,只有长文档和复杂推理走旗舰模型。结果成本直接降了六成,而整体回答质量几乎没有变化,因为简单问题用轻量模型本身也回答得不错。
实现这种成本路由的关键,是定好分级的阈值。阈值定太低了,复杂请求被分到轻量模型,质量崩;阈值定太高了,降本效果不明显。我一般会根据线上请求的长度分布来定,画一个直方图,看自然分界点。
4.2 质量优先型路由:复杂任务绝不妥协
有些业务对质量极其敏感,比如医疗咨询、法律问答、金融分析。这时候路由策略的重心不是省钱,而是确保复杂任务一定落在能力最强的模型上。
这类场景我建议用“双重确认”机制:先用规则或轻量模型做初判,如果判断是复杂任务,再让一个强模型复核一遍分类,确认无误后才走旗舰模型。这样一来,即使轻量模型分类出错,也能在第二道关卡被拦下来。当然,代价是多一次模型调用,但质量敏感的业务值得这个成本。
4.3 可用性优先型路由:故障自动切换
模型接口不是永远稳定的。限流、超时、5xx错误,我们在线上都遇到过。可用性路由就是在主模型出问题时,自动把请求切到备用模型。
LangChain里可以给模型调用包一层重试逻辑,或者用with_fallbacks来实现:
python复制from langchain_core.runnables import Runnable
fallback_chain = high_end_llm.with_fallbacks(
[light_llm, medium_llm],
exceptions_to_handle=(TimeoutError, ConnectionError)
)
这段代码的意思是:调用high_end_llm如果抛出超时或连接错误,自动依次尝试light_llm和medium_llm。注意备用模型的顺序也有讲究,我习惯按“能力从强到弱”排列,这样优先用相对更好的模型顶上,实在不行才用最轻量的。
4.4 混合路由:规则+分类+权重多管齐下
到了生产级别,很少只用单一的路由策略。我常用的混合路由设计是这样的:
- 第一层:硬性规则过滤。比如请求里包含用户明确的模型偏好,或者业务线指定了某种模型,直接走规则。
- 第二层:轻量模型分类。对没有命中硬规则的请求,用轻量模型分成任务类型。
- 第三层:动态加权选路。根据实时监控数据(错误率、响应延迟、成本余量),在同类模型中动态选择权重更高的那个。
这种三层结构的好处是:规则快、分类准、动态调节能应对突发状况。缺点是链路变长,排查问题要逐层看日志。所以每一层我都要记录日志,判断是“规则命中”“分类结果”还是“动态加权最终选择了哪个模型”,这样出了问题能快速定位。
4.5 基于Embedding的语义路由
还有一种方式这两年很流行:用Embedding把用户请求转成向量,再跟预定义的任务模板向量做相似度匹配,相似度最高的就是该走的路由分支。
python复制from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
embeddings = OpenAIEmbeddings(model="embedding-model")
store = InMemoryVectorStore(embeddings)
# 预定义任务模板
route_templates = [
{"text": "帮我写一段业务代码,包括函数定义和异常处理", "route": "code"},
{"text": "总结这篇文章的主要观点,300字以内", "route": "summary"},
{"text": "翻译这句话成英文", "route": "translation"},
]
store.add_texts([item["text"] for item in route_templates])
查询时把用户请求也做Embedding,取最相似的模板对应的路由。这个方案适用于那些“不好用规则描述,但是语义上能区分”的场景。实际部署时注意几点:Embedding模型如果跟任务模型不是同一家的,维度可能不一致,需要统一;模板库要定期维护,不能一套模板用到底。
5. 常见问题与排查技巧实录
5.1 路由判断错了怎么办
多模型路由最常见的坑,就是分类模型把请求分错了。比如本该走旗舰模型的复杂推理,被轻量模型判定成“simple_qa”,导致回答质量拉胯。
我的排查方法分四步:
- 还原路由结果:日志里记录了分类器输出的类别、各层命中的规则,先确认是哪一层出了问题。
- 抽样看分类器输出:把一段时间的分类结果和人工标注对比,计算分类准确率。如果低于九成,说明分类Prompt写得不够清楚,或者类别定义太宽泛。
- 优化分类Prompt:在分类Prompt里补上“当请求涉及多步骤推理或多学科知识时,判定为complex类别”这类边界说明。
- 加兜底机制:如果分类结果置信度低(可以让分类模型顺便输出一个置信度),就默认走旗舰模型。宁可多花点钱,也别让质量崩。
5.2 不同模型返回格式不统一
路由到不同模型之后,你会发现它们的返回风格差异很大:有的喜欢列条,有的喜欢长段落,有的会带markdown。同一个Agent给用户返回的内容,一会儿是列表一会儿是段落,体验很割裂。
解决方案是在每个模型分支后面再接一个统一的格式化Runnable,把模型原始输出转换成统一的展示格式。LangChain里可以用StrOutputParser拿到纯文本,再做一次后置处理,比如强制要求“不要使用markdown列表,用编号段落表达”。
5.3 上下文窗口差异导致切换失败
我之前遇到过一个问题:Agent先用小上下文模型聊了几轮,积累了很长的历史,然后突然路由切换到大上下文模型,倒是没问题;但反过来,先用大上下文模型聊了很久,切换回小模型时,历史超长,直接报错。
这里的关键是:切换前检查历史长度是否超出目标模型的上下文窗口。如果超了,先做历史裁剪或摘要压缩,再切换。LangChain里可以用trim_messages处理,或者自己写一个按token数裁剪的函数。裁剪时注意保留最近的对话,因为最新的上下文通常最关键。
5.4 限流和并发引起的雪崩
路由到模型之后,如果多个路由分支都指向同一个模型,突发流量会把这个模型打限流。更糟糕的是,限流后fallback切到另一个模型,那个模型也被打爆,形成雪崩。
我的经验是给每个模型做独立的并发控制,在路由层加一个简单的“配额计数器”。比如每秒钟最多往某个模型送N个请求,超过的部分排队或走备选。这个N要根据模型的限流阈值和业务峰值来定,留出20%到30%的余量。
5.5 日志和监控的设计
最后说一个容易被忽视的点:多模型路由让系统变复杂了,如果不做好日志,出了问题根本没法查。
我要求在路由链路的每个关键节点都埋点:
- 请求到达时间、原始内容摘要
- 路由判断结果、各层命中情况
- 最终选择的模型、调用耗时、Token消耗
- 模型返回的状态码、错误信息
这些日志聚合起来,可以灰度分析“不同请求类型在不同模型上的表现差异”。我见过一个团队,就是靠这些数据发现某个模型在英文代码生成上表现极差,果断调低了它的路由权重,整体代码质量立刻回升。
写在最后的实操体会
多模型路由这个东西,做成“能跑”不难,做成“跑得稳”才是真功夫。我个人在实际操作中最深的体会是:路由策略不是一次定死的设计,而是一个持续根据线上数据调优的过程。不要指望第一版规则就能涵盖所有场景,上线之后要多看路由分布、多看错误日志、多对比不同模型在同一类任务上的效果。还有一个小技巧:在路由判断结果里加一个trace字段,把“为什么选了A模型”记录下来,等出问题复盘的时候会省很多力气。这个内容后续还可以往“推理成本预算控制”“多模型并发投票”方向扩展,不过先把基础的路由做到位,收益就已经很明显了。
