LangChain 多模型调度

在我们做agent时很多情况下要接入不同的模型比如文字有GPT、deepseek、Qwen语音模型 ASR、TTS Paraformer图片模型 Seedream Wan Kolors视频模型Kling、Seedream Wan2.7-Video等。在文本模型中有便宜的也有贵的模型所以在做agent时我们会调很多模型专业模型做专业的事但在做多模型agent时要先问几个问题怎么判断任务的复杂度怎么把任务分到对应的模型分错了怎么办下面我们从最简单的方式开始一步步解决这三个问题。第一步方案一 — 静态复杂度路由最简单最常用静态路由的核心思想是用一个轻量分类器判断任务复杂度然后路由到对应模型。听起来简单但有个关键细节——分类器本身不能贵。很多新手犯的错误是用GPT-4o去判断这个任务该不该用GPT-4o这就是典型的用人来检查人——成本没省下来反而多花了一次。正确的做法是用最便宜的模型做分类。import json from typing import Literal, TypedDict, Annotated from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages # 关键用最便宜的模型做分类——分类器成本必须远小于模型切换带来的节省 classifier_llm ChatOpenAI( modeldeepseek-chat, # 也可以用 gpt-4o-mini temperature0, api_keysk-xxx, base_urlhttps://api.deepseek.com/v1, ) # 这三个模型才是真正干活的 light_model ChatOpenAI( modeldeepseek-chat, # 轻量简单任务 temperature0, api_keysk-xxx, base_urlhttps://api.deepseek.com/v1, ) medium_model ChatOpenAI( modelgpt-4o-mini, # 中等中等任务 temperature0.3, api_keysk-xxx, ) heavy_model ChatOpenAI( modelgpt-4o, # 旗舰复杂任务 temperature0.5, api_keysk-xxx, )关键设计问题分类器怎么判断复杂度两个方案方案A用LLM做分类最灵活推荐CLASSIFIER_PROMPT 分析用户问题的复杂度返回 JSON { complexity: simple | medium | complex, reason: 分类理由一句话, confidence: 0.0-1.0 } 分类标准 - simple: 事实问答、简单翻译、实体提取、关键词生成、天气查询 - medium: 多步推理、代码解释、文档总结、格式转换、简单分析 - complex: 复杂代码生成、多条件规划、数学证明、系统设计、架构方案 用户问题{query} def classify_complexity(query: str) - dict: response classifier_llm.invoke( CLASSIFIER_PROMPT.format(queryquery) ) # 解析 JSON处理可能的格式问题 text response.content.strip() text text.removeprefix(json).removeprefix().strip() text text.removesuffix().strip() return json.loads(text)方案B用规则做分类最快零成本适合边界清晰的场景import re def classify_by_rules(query: str) - dict: 纯规则分类零成本 # 复杂信号 complex_patterns [ r设计.*系统, r架构, r百万并发, r性能优化, r数学证明, r算法.*实现, r代码生成, ] # 简单信号 simple_patterns [ r是.*首都, r天气, r翻译, r提取, r分类, r关键词, r是什么, r多少, ] for p in complex_patterns: if re.search(p, query): return {complexity: complex, reason: 命中复杂规则, confidence: 0.8} for p in simple_patterns: if re.search(p, query): return {complexity: simple, reason: 命中简单规则, confidence: 0.9} return {complexity: medium, reason: 未命中规则默认中等, confidence: 0.6}方案A更灵活方案B更快。生产环境建议两者结合先过规则没过规则的丢给LLM。第二步用LangGraph编排路由图有了分类器接下来就是怎么把模型节点串起来。# 定义状态 class RouterState(TypedDict): query: str complexity: str reason: str confidence: float answer: str model_used: str # 三个模型节点 def simple_agent(state: RouterState) - dict: response light_model.invoke([ {role: system, content: 简洁准确回答直接给出结果即可}, {role: user, content: state[query]}, ]) return {answer: response.content, model_used: deepseek-chat} def medium_agent(state: RouterState) - dict: response medium_model.invoke([ {role: system, content: 详细解答提供步骤说明和必要的背景信息}, {role: user, content: state[query]}, ]) return {answer: response.content, model_used: gpt-4o-mini} def complex_agent(state: RouterState) - dict: response heavy_model.invoke([ {role: system, content: 深入分析给出完整方案、代码示例和设计思路}, {role: user, content: state[query]}, ]) return {answer: response.content, model_used: gpt-4o} # 分类节点 def classifier_node(state: RouterState) - RouterState: result classify_complexity(state[query]) return { complexity: result[complexity], reason: result[reason], confidence: result.get(confidence, 0.5), } # 路由节点——用条件边实现 def router_decision(state: RouterState) - str: 返回下一个节点的名称 # ⚠️ 关键细节置信度低时走medium避免频繁误分类 if state[confidence] 0.6: return medium_agent return state[complexity] _agent # 构建图 builder StateGraph(RouterState) builder.add_node(classifier, classifier_node) builder.add_node(simple_agent, simple_agent) builder.add_node(medium_agent, medium_agent) builder.add_node(complex_agent, complex_agent) # 入口 builder.add_edge(START, classifier) # 条件路由 builder.add_conditional_edges( classifier, router_decision, { simple_agent: simple_agent, medium_agent: medium_agent, complex_agent: complex_agent, } ) # 出口 builder.add_edge(simple_agent, END) builder.add_edge(medium_agent, END) builder.add_edge(complex_agent, END) router_graph builder.compile()运行效果test_queries [ 巴黎是哪个国家的首都, # → simple 用Python实现一个二分查找算法, # → medium 设计一个支持百万并发的高性能消息队列架构, # → complex ] for q in test_queries: result router_graph.invoke({query: q}) print(f[{result[model_used]}] {result[complexity]}: {q[:30]}) print(f 理由: {result[reason]}) print(f 回答: {result[answer][:50]}...) print() # 输出示例 # [deepseek-chat] simple: 巴黎是哪个国家的首都... # 理由: 简单事实问答 # [gpt-4o-mini] medium: 用Python实现一个二分查找算法... # 理由: 需要代码解释和多步分析 # [gpt-4o] complex: 设计一个支持百万并发的高性能消息队列架构... # 理由: 复杂系统设计任务静态路由的核心问题分错了怎么办如果巴黎是哪个国家的首都这种简单问题分类器误判成了complex——虽然最终答案会正确但成本多花了25倍。反过来如果复杂问题被误判成simple——答案质量会严重下降。这就是为什么我们需要方案二。第三步方案二 — 带兜底的静态路由工程化方案一最大的风险是误分类。工程化的做法是加一层兜底。def safe_router(state: RouterState) - str: 带兜底的路由决策 complexity state[complexity] confidence state.get(confidence, 0.5) # 低置信度 → 走中等模型兜底 if confidence 0.6: return medium_agent # 正常路由 return complexity _agent但这还不够。 真正的生产环境还需要记录每次路由决策用于后续优化设置最大调用次数防止死循环设置Fallback节点所有模型都失败时走人工def classification_node_with_audit(state: RouterState) - RouterState: 带审计日志的分类节点 result classify_complexity(state[query]) # 记录路由决策日志 audit_log { query: state[query][:100], classified_as: result[complexity], confidence: result.get(confidence, 0.5), reason: result.get(reason, ), timestamp: __import__(time).time(), } # 这里可以写入数据库或日志文件 # log_to_db(audit_log) return { complexity: result[complexity], reason: result[reason], confidence: result.get(confidence, 0.5), }方案二的优缺点维度优点缺点成本最低只调用一次分类一次模型分类器不准时质量或成本会翻车延迟最低一次路由决策一次模型调用误分类没有纠错机会实现最简单10行代码搞定需要人工定期审核分类结果适用任务边界清晰、分类器准确率高的场景任务类型多样、边界模糊的场景第四步方案三 — 动态并行路由最准但最贵方案二的问题是如果分类错了没有第二次机会。动态并行路由的思路是不分类了同时让多个模型跑选最好的结果。from langgraph.types import Send from typing import Annotated, List import operator class ParallelRouterState(TypedDict): query: str results: Annotated[List[dict], operator.add] # 累加多个模型的结果 best_answer: str model_used: str # 并行派发 def dispatch_to_all(state: ParallelRouterState): 同时分发到三个模型 return [ Send(simple_agent, {query: state[query]}), Send(medium_agent, {query: state[query]}), Send(complex_agent, {query: state[query]}), ] # 模型节点每个返回一个结果 def simple_node(state: dict) - dict: ... # 同上 return {results: [{answer: ..., model: deepseek-chat}]} def medium_node(state: dict) - dict: ... return {results: [{answer: ..., model: gpt-4o-mini}]} def complex_node(state: dict) - dict: ... return {results: [{answer: ..., model: gpt-4o}]} # 选择最佳结果 def select_best(state: ParallelRouterState) - ParallelRouterState: 从所有结果中选择最佳 results state[results] # 方案A用 LLM 打分选择最准 selection_prompt ( 从以下回答中选择最准确完整的一个返回序号:\n\n \n---\n.join( f[{i}] {r[answer][:300]} for i, r in enumerate(results) ) ) selection classifier_llm.invoke(selection_prompt) try: best_idx int(.join(c for c in selection.content if c.isdigit())) except ValueError: best_idx 0 # 解析失败默认选第一个 return { best_answer: results[best_idx][answer], model_used: results[best_idx][model], }动态并行路由的优缺点维度优点缺点准确性最高不会漏掉任何可能的优质回答成本也最高同时调用多个模型延迟受最慢模型限制可能要等GPT-4o出结果不适合对延迟敏感的场景适用准确性要求极高、延迟不敏感的场景成本和延迟要求严格的场景一个更实用的折中方案竞速模式同时发轻量和旗舰两个模型谁先返回可靠结果就用谁的。轻量模型通常快得多如果它先返回且结果质量过关就立刻返回如果质量不过关等旗舰模型的结果。def race_mode(query: str) - str: 竞速模式轻量先跑旗舰兜底 import concurrent.futures with concurrent.futures.ThreadPoolExecutor(max_workers2) as executor: light_future executor.submit(light_model.invoke, query) heavy_future executor.submit(heavy_model.invoke, query) # 先等轻量模型 light_result light_future.result(timeout3) # 快速验证轻量结果 verification classifier_llm.invoke( f这个回答是否完整准确地回答了问题回答{light_result.content[:200]} ) if 是 in verification.content: return light_result.content # 轻量够用不等旗舰了 # 轻量不行等旗舰 return heavy_future.result().content第五步方案四 — 基于历史统计的自适应路由生产级前面三个方案都有一个问题它们不考虑历史表现。分类器说这个任务简单它就去了轻量模型不管这个分类器在过去类似问题上翻过多少次车。真正的生产级方案是用数据驱动路由决策。import sqlite3 import json from datetime import datetime, timedelta class AdaptiveRouter: 基于历史统计的自适应路由引擎 def __init__(self, db_path: str router_stats.db): self.conn sqlite3.connect(db_path) self.conn.execute( CREATE TABLE IF NOT EXISTS routing_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, query_hash TEXT, query_text TEXT, classified_as TEXT, routed_to TEXT, actual_model TEXT, latency_ms REAL, token_cost REAL, user_rating INTEGER, -- 1-5, 用户后续反馈 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) self.conn.execute( CREATE TABLE IF NOT EXISTS model_stats ( task_type TEXT, model TEXT, total_calls INTEGER DEFAULT 0, avg_latency_ms REAL DEFAULT 0, avg_cost REAL DEFAULT 0, avg_rating REAL DEFAULT 0, PRIMARY KEY (task_type, model) ) ) self.conn.commit() def record(self, query: str, classified_as: str, routed_to: str, actual_model: str, latency_ms: float, token_cost: float): 记录一次路由决策 query_hash str(hash(query) % 10000) self.conn.execute( INSERT INTO routing_history (query_hash, query_text, classified_as, routed_to, actual_model, latency_ms, token_cost) VALUES (?, ?, ?, ?, ?, ?, ?) , (query_hash, query[:200], classified_as, routed_to, actual_model, latency_ms, token_cost)) # 更新模型统计 task_type classified_as self.conn.execute( INSERT INTO model_stats (task_type, model, total_calls, avg_latency_ms, avg_cost) VALUES (?, ?, 1, ?, ?) ON CONFLICT(task_type, model) DO UPDATE SET total_calls total_calls 1, avg_latency_ms (avg_latency_ms * total_calls ?) / (total_calls 1), avg_cost (avg_cost * total_calls ?) / (total_calls 1) , (task_type, actual_model, latency_ms, token_cost, latency_ms, token_cost)) self.conn.commit() def get_best_model(self, task_type: str, max_cost: float None) - str: 根据历史数据获取某类任务的最佳模型 query SELECT model, avg_rating, avg_latency_ms, avg_cost FROM model_stats WHERE task_type ? ORDER BY avg_rating DESC, avg_latency_ms ASC rows self.conn.execute(query, (task_type,)).fetchall() if not rows: # 无历史数据用默认策略 default_map { simple: deepseek-chat, medium: gpt-4o-mini, complex: gpt-4o, } return default_map.get(task_type, gpt-4o-mini) # 如果有成本上限排除超预算的模型 if max_cost: rows [r for r in rows if r[3] max_cost] return rows[0][0] if rows else gpt-4o-mini def get_classification_accuracy(self, hours: int 24) - dict: 获取分类器准确率 cutoff datetime.now() - timedelta(hourshours) rows self.conn.execute( SELECT classified_as, actual_model, COUNT(*) as cnt FROM routing_history WHERE created_at ? GROUP BY classified_as, actual_model , (cutoff,)).fetchall() # 简单统计分类和实际调用的模型是否匹配 correct sum(r[2] for r in rows if r[0] r[1]) total sum(r[2] for r in rows) return { accuracy: round(correct / total * 100, 1) if total 0 else 0, total: total, correct: correct, } # 实际使用 router AdaptiveRouter() def data_driven_router_node(state: RouterState) - RouterState: 数据驱动的路由节点 result classify_complexity(state[query]) task_type result[complexity] # 查询历史数据获取此类任务的最佳模型 best_model router.get_best_model(task_type) # 记录路由决策 router.record( querystate[query], classified_astask_type, routed_tobest_model, actual_modelbest_model, # 实际调用时再更新 latency_ms0, token_cost0, ) return { complexity: task_type, reason: f历史数据显示此类任务最优模型为{best_model}, confidence: result.get(confidence, 0.5), }自适应路由的核心优势越用越准——跑得越多历史数据越丰富路由决策越准确自动适应模型变化——如果GPT-4o-mini出了新版本变强了数据会自动反映出来可量化路由质量——每天看分类准确率低于阈值就调第六步完整可运行Demo下面是完整的、可运行的Demo基于DeepSeek GPT-4o-mini GPT-4o三模型路由含静态路由兜底机制运行日志。 Agent多模型路由调度 Demo 功能根据任务复杂度自动路由到最合适的模型 模型DeepSeek-chat简单→ GPT-4o-mini中等→ GPT-4o复杂 依赖pip install langchain-openai langgraph import json import time from typing import TypedDict, Literal from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END # 配置区 DEEPSEEK_API_KEY sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx DEEPSEEK_BASE_URL https://api.deepseek.com/v1 OPENAI_API_KEY sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx # 模型池 # 分类器——用最便宜的模型 classifier_llm ChatOpenAI( modeldeepseek-chat, temperature0, api_keyDEEPSEEK_API_KEY, base_urlDEEPSEEK_BASE_URL, ) # 轻量模型——处理简单任务成本极低 light_model ChatOpenAI( modeldeepseek-chat, temperature0, api_keyDEEPSEEK_API_KEY, base_urlDEEPSEEK_BASE_URL, ) # 中等模型——处理中等任务平衡质量和成本 medium_model ChatOpenAI( modelgpt-4o-mini, temperature0.3, api_keyOPENAI_API_KEY, ) # 旗舰模型——处理复杂任务质量最高 heavy_model ChatOpenAI( modelgpt-4o, temperature0.5, api_keyOPENAI_API_KEY, ) # 状态定义 class RouterState(TypedDict): query: str complexity: str reason: str confidence: float answer: str model_used: str latency_ms: float # 分类器 CLASSIFIER_PROMPT 分析用户问题的复杂度返回 JSON { complexity: simple | medium | complex, reason: 分类理由一句话, confidence: 0.0-1.0 } 分类标准 - simple: 事实问答、简单翻译、实体提取、关键词生成、天气查询、简单计算 - medium: 多步推理、代码解释、文档总结、格式转换、简单分析、方案对比 - complex: 复杂代码生成、多条件规划、数学证明、系统设计、架构方案、算法设计 用户问题{query} def classify_complexity(query: str) - dict: 用最便宜的模型做分类 response classifier_llm.invoke( CLASSIFIER_PROMPT.format(queryquery) ) text response.content.strip() # 清理可能的 markdown 围栏 text text.replace(json, ).replace(, ).strip() try: return json.loads(text) except json.JSONDecodeError: # 解析失败时默认中等 return {complexity: medium, reason: JSON解析失败默认中等, confidence: 0.5} # 模型节点 SIMPLE_SYSTEM 简洁准确回答直接给出结果。不要多余的解释不要背景信息。 MEDIUM_SYSTEM 详细解答提供步骤说明和必要的背景信息。如果涉及代码给出代码示例。 COMPLEX_SYSTEM 深入分析给出完整方案、代码示例和设计思路。考虑边界情况、性能优化和最佳实践。 def simple_agent(state: RouterState) - dict: start time.time() response light_model.invoke([ {role: system, content: SIMPLE_SYSTEM}, {role: user, content: state[query]}, ]) latency (time.time() - start) * 1000 return { answer: response.content, model_used: deepseek-chat, latency_ms: round(latency, 1), } def medium_agent(state: RouterState) - dict: start time.time() response medium_model.invoke([ {role: system, content: MEDIUM_SYSTEM}, {role: user, content: state[query]}, ]) latency (time.time() - start) * 1000 return { answer: response.content, model_used: gpt-4o-mini, latency_ms: round(latency, 1), } def complex_agent(state: RouterState) - dict: start time.time() response heavy_model.invoke([ {role: system, content: COMPLEX_SYSTEM}, {role: user, content: state[query]}, ]) latency (time.time() - start) * 1000 return { answer: response.content, model_used: gpt-4o, latency_ms: round(latency, 1), } # 图编排 def classifier_node(state: RouterState) - RouterState: result classify_complexity(state[query]) print(f分类器判断: {result[complexity]} (置信度: {result.get(confidence, 0.5):.2f})) print(f 理由: {result[reason]}) return { complexity: result[complexity], reason: result[reason], confidence: result.get(confidence, 0.5), } def router_decision(state: RouterState) - str: 带兜底的路由决策 complexity state[complexity] confidence state.get(confidence, 0.5) # 低置信度 → 走中等模型兜底 if confidence 0.6: print(f ⚠️ 置信度低({confidence:.2f})降级至 medium 兜底) return medium_agent # 正常路由 target complexity _agent print(f 路由至: {target}) return target # 构建图 builder StateGraph(RouterState) # 添加节点 builder.add_node(classifier, classifier_node) builder.add_node(simple_agent, simple_agent) builder.add_node(medium_agent, medium_agent) builder.add_node(complex_agent, complex_agent) # 连线 builder.add_edge(START, classifier) builder.add_conditional_edges( classifier, router_decision, { simple_agent: simple_agent, medium_agent: medium_agent, complex_agent: complex_agent, } ) builder.add_edge(simple_agent, END) builder.add_edge(medium_agent, END) builder.add_edge(complex_agent, END) # 编译 router_graph builder.compile() # 可视化 def print_routing_result(query: str, result: dict): print(f\n{*60}) print(f问题: {query}) print(f{*60}) print(f 分类: {result[complexity]} ({result[reason]})) print(f 模型: {result[model_used]}) print(f 耗时: {result.get(latency_ms, 0)}ms) print(f 回答: {result[answer][:100]}...) print(f{*60}\n) # 测试运行 if __name__ __main__: test_queries [ 巴黎是哪个国家的首都, # simple 用Python实现二分查找解释时间复杂度, # medium 设计一个支持百万并发的消息队列系统架构, # complex 今天天气怎么样, # simple 对比RESTful和GraphQL的优缺点给出选型建议, # medium 用Python实现一个简单的神经网络, # medium ] print(Agent 多模型路由引擎启动\n) print(f{测试问题:30} {分类:10} {模型:15} {耗时:10}) print(- * 65) for query in test_queries: try: result router_graph.invoke({query: query}) print(f{query[:28]:30} {result[complexity]:10} f{result[model_used]:15} {result.get(latency_ms, 0):10.0f}ms) except Exception as e: print(f{query[:28]:30} 错误: {str(e)[:30]})运行结果Agent 多模型路由引擎启动 测试问题 分类 模型 耗时 ----------------------------------------------------------------- 巴黎是哪个国家的首都 simple deepseek-chat 1200ms 用Python实现二分查找解释时间复杂 medium gpt-4o-mini 2100ms 设计一个支持百万并发的消息队列系统架 complex gpt-4o 4500ms 今天天气怎么样 simple deepseek-chat 800ms 对比RESTful和GraphQL的优缺点给 medium gpt-4o-mini 2500ms 用Python实现一个简单的神经网络 medium gpt-4o-mini 2800ms第七步四种方案对比维度方案一静态路由方案二带兜底方案三并行路由方案四自适应实现难度最低低高最高成本节省60-80%60-80%20-40%70-85%回答质量依赖分类器较稳定最高逐步提升延迟最低低高等最慢模型低适用场景快速验证、POC生产环境起步对准确性要求极高长期运行的生产系统学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】