企业官网建设流程全解析

模型路由机制

摘要

在大模型应用落地的过程中，单一模型往往难以同时满足成本、延迟、推理能力、上下文长度等多维度需求。Harness 作为一套面向生产环境的 AI 编排框架，通过**模型路由机制（Model Routing）**实现了根据任务类型动态选择混元、DeepSeek、Kimi 等不同模型的能力。本文将深入剖析 Harness 模型路由的设计理念、核心架构、路由策略、代码实现以及生产实践中的优化技巧，帮助读者构建一套可扩展、可观测、可回滚的智能模型调度系统。

引言

为什么需要模型路由？

假设你正在构建一个企业级 AI 助手，用户可能提出以下几类请求：

• 闲聊与常识问答：成本敏感，延迟要求高，小模型即可胜任。
• 长文档摘要：需要 128K 甚至更长的上下文窗口，Kimi 是理想选择。
• 复杂代码生成与数学推理：需要强推理能力，DeepSeek-R1 或混元 Pro 更合适。
• 中文创意写作：混元的文风与中文语感更自然。

如果所有请求都打到同一个"最强模型"上，会出现两个问题：

1. 成本爆炸：简单任务用大模型是巨大的浪费。
2. 体验下降：长上下文任务被短上下文模型截断，推理任务被弱模型答非所问。

模型路由机制的核心目标，就是让合适的任务交给合适的模型——在质量、成本、延迟三者之间取得最优平衡。

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 基于规则的特征提取

importrefromdataclassesimportdataclassfromtypingimportList@dataclassclassTaskFeatures:"""从用户请求中提取的任务特征"""prompt:strtoken_count:int# 估算的 token 数has_code:bool# 是否包含代码has_math:bool# 是否包含数学表达式is_long_context:bool# 是否超长上下文language:str# 主要语言intent:str# 意图分类结果classFeatureExtractor:"""规则 + 启发式的特征提取器"""CODE_PATTERN=re.compile(r"```|def |class |function |import |#include")MATH_PATTERN=re.compile(r"\d+\s*[\+\-\*/=]\s*\d+|证明|推导|求解|积分|导数")defextract(self,prompt:str)->TaskFeatures:token_count=self._estimate_tokens(prompt)returnTaskFeatures(prompt=prompt,token_count=token_count,has_code=bool(self.CODE_PATTERN.search(prompt)),has_math=bool(self.MATH_PATTERN.search(prompt)),is_long_context=token_count>32000,language=self._detect_language(prompt),intent=self._classify_intent(prompt),)def_estimate_tokens(self,text:str)->int:# 中文约 1.5 字/token，英文约 4 字符/token，这里做粗略估算chinese_chars=len(re.findall(r"[\u4e00-\u9fff]",text))other_chars=len(text)-chinese_charsreturnint(chinese_chars/1.5+other_chars/4)def_detect_language(self,text:str)->str:chinese_chars=len(re.findall(r"[\u4e00-\u9fff]",text))return"zh"ifchinese_chars/max(len(text),1)>0.3else"en"def_classify_intent(self,text:str)->str:# 生产环境可替换为小型 BERT 分类器ifself.CODE_PATTERN.search(text):return"code"ifself.MATH_PATTERN.search(text):return"reasoning"ifany(kintextforkin["总结","摘要","summarize"]):return"summarize"return"chat"

2.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.client=lite_client# 混元 Lite 等低成本模型defclassify(self,query:str)->Dict:resp=self.client.chat(messages=[{"role":"user","content":CLASSIFY_PROMPT.format(query=query)}],temperature=0.0,max_tokens=64,)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.intent=="code","deepseek"),(lambdaf:f.intent=="reasoning","deepseek"),(lambdaf:f.intent=="creative"andf.language=="zh","hunyuan"),(lambdaf:f.intent=="chat","hunyuan-lite"),]defroute(self,features)->Optional[str]:forcondition,modelinself.rules:ifcondition(features):returnmodelreturnNone# 未命中规则，交给下一层策略

3.2 打分路由（Score-based Routing）

规则无法覆盖所有情况，此时可以用加权打分。每个模型在多个维度上有评分，根据任务特征动态加权：

fromdataclassesimportdataclassfromtypingimportDict,List@dataclassclassModelProfile: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:# 根据任务特征动态调整权重weights=self._compute_weights(features)best_model,best_score=None,float("-inf")forname,pinself.profiles.items():iffeatures.token_count>p.context_window:continue# 上下文不够，直接淘汰score=self._score(p,weights,features)ifscore>best_score:best_model,best_score=name,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.intent=="code":base.update({"code":0.5,"reasoning":0.3,"cost":0.1,"latency":0.1})eliffeatures.intent=="reasoning":base.update({"reasoning":0.6,"cost":0.2,"latency":0.2})eliffeatures.intent=="chat":base.update({"cost":0.5,"latency":0.4,"chinese":0.1})eliffeatures.intent=="creative":base.update({"chinese":0.6,"cost":0.2,"latency":0.2})returnbasedef_score(self,p:ModelProfile,w:Dict[str,float],features)->float:# 成本、延迟需要归一化（越小越好，取负向）cost_score=1.0/(1.0+p.cost_per_1k)latency_score=1.0/(1.0+p.latency_ms/1000)return(w["cost"]*cost_score+w["latency"]*latency_score+w["reasoning"]*p.reasoning_score+w["chinese"]*p.chinese_score+w["code"]*p.code_score)

3.3 模型画像示例

MODEL_PROFILES=[ModelProfile("hunyuan-lite",cost_per_1k=0.001,latency_ms=300,reasoning_score=0.5,chinese_score=0.85,context_window=32000,code_score=0.5),ModelProfile("hunyuan-pro",cost_per_1k=0.02,latency_ms=900,reasoning_score=0.8,chinese_score=0.95,context_window=128000,code_score=0.8),ModelProfile("deepseek-v3",cost_per_1k=0.008,latency_ms=800,reasoning_score=0.85,chinese_score=0.85,context_window=64000,code_score=0.9),ModelProfile("deepseek-r1",cost_per_1k=0.015,latency_ms=1500,reasoning_score=0.95,chinese_score=0.85,context_window=64000,code_score=0.9),ModelProfile("kimi",cost_per_1k=0.012,latency_ms=1000,reasoning_score=0.8,chinese_score=0.9,context_window=200000,code_score=0.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_key=api_key self.endpoint=endpointdefchat(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:str="https://api.deepseek.com"):self.api_key=api_key self.base_url=base_urldefchat(self,messages,**kwargs)->str:# DeepSeek 兼容 OpenAI 协议payload={"model":kwargs.get("model","deepseek-chat"),"messages":messages,"temperature":kwargs.get("temperature",0.7),}resp=self._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 delta=json.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:str="https://api.moonshot.cn"):self.api_key=api_key self.base_url=base_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 logger=logging.getLogger(__name__)classModelRouter:"""Harness 模型路由引擎"""def__init__(self,extractor:FeatureExtractor,rule_router:RuleRouter,score_router:ScoreRouter,adapters:Dict[str,BaseAdapter],fallback_model:str="hunyuan-pro",):self.extractor=extractor self.rule_router=rule_router self.score_router=score_router self.adapters=adapters self.fallback_model=fallback_model self.metrics={"route_count":0,"fallback_count":0}defroute(self,prompt:str)->str:"""决定使用哪个模型"""features=self.extractor.extract(prompt)# 1. 规则优先model=self.rule_router.route(features)ifmodelandmodelinself.adapters:logger.info(f"[route] rule hit ->{model}")returnmodel# 2. 打分兜底model=self.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"]+=1model=self.route(prompt)try:adapter=self.adapters[model]returnadapter.chat(messages=[{"role":"user","content":prompt}],**kwargs,)exceptExceptionase:logger.error(f