模型路由机制摘要在大模型应用落地的过程中单一模型往往难以同时满足成本、延迟、推理能力、上下文长度等多维度需求。Harness 作为一套面向生产环境的 AI 编排框架通过**模型路由机制Model Routing**实现了根据任务类型动态选择混元、DeepSeek、Kimi 等不同模型的能力。本文将深入剖析 Harness 模型路由的设计理念、核心架构、路由策略、代码实现以及生产实践中的优化技巧帮助读者构建一套可扩展、可观测、可回滚的智能模型调度系统。引言为什么需要模型路由假设你正在构建一个企业级 AI 助手用户可能提出以下几类请求闲聊与常识问答成本敏感延迟要求高小模型即可胜任。长文档摘要需要 128K 甚至更长的上下文窗口Kimi 是理想选择。复杂代码生成与数学推理需要强推理能力DeepSeek-R1 或混元 Pro 更合适。中文创意写作混元的文风与中文语感更自然。如果所有请求都打到同一个最强模型上会出现两个问题成本爆炸简单任务用大模型是巨大的浪费。体验下降长上下文任务被短上下文模型截断推理任务被弱模型答非所问。模型路由机制的核心目标就是让合适的任务交给合适的模型——在质量、成本、延迟三者之间取得最优平衡。Harness 的定位Harness 并不是一个模型本身而是一个模型编排层。它向上对业务暴露统一接口向下对接多个模型供应商混元、DeepSeek、Kimi、OpenAI 等中间通过路由、缓存、降级、观测等模块完成智能调度。正文一、模型路由的核心架构Harness 的模型路由整体分为四层┌─────────────────────────────────────────────┐ │ 业务调用层 (Unified API) │ ├─────────────────────────────────────────────┤ │ 路由决策层 (Router) │ │ ├─ 任务分类器 (Task Classifier) │ │ ├─ 规则引擎 (Rule Engine) │ │ └─ 策略评分器 (Scorer) │ ├─────────────────────────────────────────────┤ │ 模型适配层 (Adapters) │ │ ├─ HunyuanAdapter │ │ ├─ DeepSeekAdapter │ │ └─ KimiAdapter │ ├─────────────────────────────────────────────┤ │ 基础设施层 (Cache / Retry / Metrics) │ └─────────────────────────────────────────────┘业务调用层对外暴露统一的chat()接口屏蔽底层模型差异。路由决策层根据任务特征选择目标模型是本文的核心。模型适配层把不同厂商的 API 差异鉴权、参数、流式协议统一封装。基础设施层提供缓存、重试、限流、监控等横切能力。二、任务分类路由的第一步路由的前提是知道当前任务是什么类型。Harness 采用轻量分类器 规则兜底的混合策略。2.1 基于规则的特征提取importrefromdataclassesimportdataclassfromtypingimportListdataclassclassTaskFeatures:从用户请求中提取的任务特征prompt:strtoken_count:int# 估算的 token 数has_code:bool# 是否包含代码has_math:bool# 是否包含数学表达式is_long_context:bool# 是否超长上下文language:str# 主要语言intent:str# 意图分类结果classFeatureExtractor:规则 启发式的特征提取器CODE_PATTERNre.compile(r|def |class |function |import |#include)MATH_PATTERNre.compile(r\d\s*[\\-\*/]\s*\d|证明|推导|求解|积分|导数)defextract(self,prompt:str)-TaskFeatures:token_countself._estimate_tokens(prompt)returnTaskFeatures(promptprompt,token_counttoken_count,has_codebool(self.CODE_PATTERN.search(prompt)),has_mathbool(self.MATH_PATTERN.search(prompt)),is_long_contexttoken_count32000,languageself._detect_language(prompt),intentself._classify_intent(prompt),)def_estimate_tokens(self,text:str)-int:# 中文约 1.5 字/token英文约 4 字符/token这里做粗略估算chinese_charslen(re.findall(r[\u4e00-\u9fff],text))other_charslen(text)-chinese_charsreturnint(chinese_chars/1.5other_chars/4)def_detect_language(self,text:str)-str:chinese_charslen(re.findall(r[\u4e00-\u9fff],text))returnzhifchinese_chars/max(len(text),1)0.3elseendef_classify_intent(self,text:str)-str:# 生产环境可替换为小型 BERT 分类器ifself.CODE_PATTERN.search(text):returncodeifself.MATH_PATTERN.search(text):returnreasoningifany(kintextforkin[总结,摘要,summarize]):returnsummarizereturnchat2.2 使用小模型做意图分类对于更精细的意图识别可以调用一个轻量模型如混元 Lite做零样本分类importjsonfromtypingimportDict CLASSIFY_PROMPT你是一个任务分类器。请判断用户请求属于以下哪一类 - chat: 闲聊、常识问答 - code: 代码生成、调试、解释 - reasoning: 数学、逻辑推理 - summarize: 长文档摘要 - creative: 创意写作、文案 只返回 JSON{intent: 类别, confidence: 0.0-1.0} 用户请求{query} classLLMIntentClassifier:def__init__(self,lite_client):self.clientlite_client# 混元 Lite 等低成本模型defclassify(self,query:str)-Dict:respself.client.chat(messages[{role:user,content:CLASSIFY_PROMPT.format(queryquery)}],temperature0.0,max_tokens64,)try:returnjson.loads(resp)exceptjson.JSONDecodeError:return{intent:chat,confidence:0.5}三、路由策略从规则到打分有了任务特征下一步是决定派给哪个模型。Harness 支持三种策略可组合使用。3.1 规则路由Rule-based Routing最简单也最可控的方式适合明确边界的场景fromtypingimportOptionalclassRuleRouter:基于规则的硬路由def__init__(self):self.rules[# (条件函数, 目标模型)(lambdaf:f.is_long_context,kimi),(lambdaf:f.intentcode,deepseek),(lambdaf:f.intentreasoning,deepseek),(lambdaf:f.intentcreativeandf.languagezh,hunyuan),(lambdaf:f.intentchat,hunyuan-lite),]defroute(self,features)-Optional[str]:forcondition,modelinself.rules:ifcondition(features):returnmodelreturnNone# 未命中规则交给下一层策略3.2 打分路由Score-based Routing规则无法覆盖所有情况此时可以用加权打分。每个模型在多个维度上有评分根据任务特征动态加权fromdataclassesimportdataclassfromtypingimportDict,ListdataclassclassModelProfile:name:strcost_per_1k:float# 每千 token 成本元latency_ms:int# 平均首 token 延迟reasoning_score:float# 推理能力 0-1chinese_score:float# 中文能力 0-1context_window:int# 上下文窗口code_score:floatclassScoreRouter:def__init__(self,profiles:List[ModelProfile]):self.profiles{p.name:pforpinprofiles}defroute(self,features)-str:# 根据任务特征动态调整权重weightsself._compute_weights(features)best_model,best_scoreNone,float(-inf)forname,pinself.profiles.items():iffeatures.token_countp.context_window:continue# 上下文不够直接淘汰scoreself._score(p,weights,features)ifscorebest_score:best_model,best_scorename,scorereturnbest_modeldef_compute_weights(self,features)-Dict[str,float]:根据任务类型调整各维度权重base{cost:0.2,latency:0.2,reasoning:0.2,chinese:0.2,code:0.2,}iffeatures.intentcode:base.update({code:0.5,reasoning:0.3,cost:0.1,latency:0.1})eliffeatures.intentreasoning:base.update({reasoning:0.6,cost:0.2,latency:0.2})eliffeatures.intentchat:base.update({cost:0.5,latency:0.4,chinese:0.1})eliffeatures.intentcreative:base.update({chinese:0.6,cost:0.2,latency:0.2})returnbasedef_score(self,p:ModelProfile,w:Dict[str,float],features)-float:# 成本、延迟需要归一化越小越好取负向cost_score1.0/(1.0p.cost_per_1k)latency_score1.0/(1.0p.latency_ms/1000)return(w[cost]*cost_scorew[latency]*latency_scorew[reasoning]*p.reasoning_scorew[chinese]*p.chinese_scorew[code]*p.code_score)3.3 模型画像示例MODEL_PROFILES[ModelProfile(hunyuan-lite,cost_per_1k0.001,latency_ms300,reasoning_score0.5,chinese_score0.85,context_window32000,code_score0.5),ModelProfile(hunyuan-pro,cost_per_1k0.02,latency_ms900,reasoning_score0.8,chinese_score0.95,context_window128000,code_score0.8),ModelProfile(deepseek-v3,cost_per_1k0.008,latency_ms800,reasoning_score0.85,chinese_score0.85,context_window64000,code_score0.9),ModelProfile(deepseek-r1,cost_per_1k0.015,latency_ms1500,reasoning_score0.95,chinese_score0.85,context_window64000,code_score0.9),ModelProfile(kimi,cost_per_1k0.012,latency_ms1000,reasoning_score0.8,chinese_score0.9,context_window200000,code_score0.75),]四、统一适配层屏蔽模型差异路由选出了模型名但各家的 API 参数、流式协议、错误码都不一样。适配层的作用就是让上层无感。fromabcimportABC,abstractmethodfromtypingimportIterator,List,DictclassBaseAdapter(ABC):所有模型适配器的基类abstractmethoddefchat(self,messages:List[Dict],**kwargs)-str:...abstractmethoddefstream_chat(self,messages:List[Dict],**kwargs)-Iterator[str]:...classHunyuanAdapter(BaseAdapter):def__init__(self,api_key:str,endpoint:str):self.api_keyapi_key self.endpointendpointdefchat(self,messages,**kwargs)-str:# 混元特有的字段映射payload{Model:kwargs.get(model,hunyuan-pro),Messages:messages,Temperature:kwargs.get(temperature,0.7),Stream:False,}# 实际调用省略返回文本returnself._post(payload)[Choices][0][Message][Content]defstream_chat(self,messages,**kwargs):payload{**kwargs,Messages:messages,Stream:True}forchunkinself._post_stream(payload):yieldchunk[Choices][0][Delta].get(Content,)def_post(self,payload):...def_post_stream(self,payload):...classDeepSeekAdapter(BaseAdapter):def__init__(self,api_key:str,base_url:strhttps://api.deepseek.com):self.api_keyapi_key self.base_urlbase_urldefchat(self,messages,**kwargs)-str:# DeepSeek 兼容 OpenAI 协议payload{model:kwargs.get(model,deepseek-chat),messages:messages,temperature:kwargs.get(temperature,0.7),}respself._post(/v1/chat/completions,payload)returnresp[choices][0][message][content]defstream_chat(self,messages,**kwargs):payload{**kwargs,messages:messages,stream:True}forlineinself._post_stream(/v1/chat/completions,payload):ifline.startswith(data: )andline!data: [DONE]:importjson deltajson.loads(line[6:])[choices][0][delta]yielddelta.get(content,)def_post(self,path,payload):...def_post_stream(self,path,payload):...classKimiAdapter(BaseAdapter):def__init__(self,api_key:str,base_url:strhttps://api.moonshot.cn):self.api_keyapi_key self.base_urlbase_urldefchat(self,messages,**kwargs)-str:payload{model:kwargs.get(model,moonshot-v1-128k),messages:messages,temperature:kwargs.get(temperature,0.7),}returnself._post(/v1/chat/completions,payload)[choices][0][message][content]defstream_chat(self,messages,**kwargs):payload{**kwargs,messages:messages,stream:True}forchunkinself._post_stream(/v1/chat/completions,payload):yieldchunkdef_post(self,path,payload):...def_post_stream(self,path,payload):...五、路由引擎的完整实现把前面的模块串起来形成 Harness 的核心路由引擎importloggingfromtypingimportDict,List,Optional loggerlogging.getLogger(__name__)classModelRouter:Harness 模型路由引擎def__init__(self,extractor:FeatureExtractor,rule_router:RuleRouter,score_router:ScoreRouter,adapters:Dict[str,BaseAdapter],fallback_model:strhunyuan-pro,):self.extractorextractor self.rule_routerrule_router self.score_routerscore_router self.adaptersadapters self.fallback_modelfallback_model self.metrics{route_count:0,fallback_count:0}defroute(self,prompt:str)-str:决定使用哪个模型featuresself.extractor.extract(prompt)# 1. 规则优先modelself.rule_router.route(features)ifmodelandmodelinself.adapters:logger.info(f[route] rule hit -{model})returnmodel# 2. 打分兜底modelself.score_router.route(features)ifmodelandmodelinself.adapters:logger.info(f[route] score hit -{model})returnmodel# 3. 最终兜底self.metrics[fallback_count]1logger.warning([route] fallback triggered)returnself.fallback_modeldefchat(self,prompt:str,**kwargs)-str:对外统一入口路由 调用 降级self.metrics[route_count]1modelself.route(prompt)try:adapterself.adapters[model]returnadapter.chat(messages[{role:user,content:prompt}],**kwargs,)exceptExceptionase:logger.error(f
