上个月我把内部 Agent 服务的模型调用策略从头改了一遍原来所有请求都发给同一个大模型改成了在 Kimi K3 和 Claude 之间做多模型路由。这个改动同时解决了两个长期痛点——长文档处理成本降下来了代码类任务的准确率也提上去了。如果你也在做 AI Agent 开发正在纠结模型选型或者觉得单一模型怎么调都不够用这篇就拿我的真实重构过程当个参考。之所以想到做这件事是因为我发现“选一个最好的模型”本身可能就是个伪命题。Kimi K3 和 Claude 各有明显优势前者在中文长文本、高性价比、结构化输出上非常能打后者在代码推理、复杂工具调用、多步规划上明显更强。把两者塞进同一个 Agent用一层路由来分配请求既省成本又能保质量。下面从场景分析、路由设计、代码实现到踩坑细节完整拆一遍。1. 为什么 Agent 要同时接 Kimi K3 和 Claude先看清各自的本事1.1 我的真实业务场景一个 Agent 管着六类任务我做的是一个带 AI 的知识管理平台用户会上传几十甚至上百页的中文文档也会要求自动生成周报、写技术方案、解读代码仓库、从合同里抽取关键字段、回答日常问题。这些任务如果按模型能力拆开看其实是完全不同的需求文档摘要需要超长上下文和对中文的敏感度代码解读需要扎实的程序语义理解字段抽取需要严格的指令遵循和稳定输出格式普通问答则讲究速度和便宜。一个 Agent 要同时扛住这六类任务在单模型体系下几乎无解。最开始我的实现很粗暴所有请求统一交给 Claude。质量确实不错但用户每次上传几十页 PDFtoken 费用高得离谱。后来我把文档预处理、分块压缩全做了成本还是压不下来。接着试了 Kimi K3长文档处理效果好、价格也低一大截但让它写 SQL 或者改代码时效果不如 Claude 稳定尤其是涉及多文件关联修改的场景。这个问题让我想明白了一个道理模型各有侧重与其寄希望于某一个模型“全能”不如让 Agent 在多个模型之间做路由。注意我这里说的 Kimi K3是月之暗面 Kimi 系列的新版本API 用法和 K2.6 差别不大但指令遵循、工具调用、长文档输出稳定性上都有明显提升。当时测评里我还专门对比过 K3 和 K2.6 写文档的表现K3 在“给定大纲后严格按结构输出”这件事上要听话得多。1.2 我怎么确认这两个模型的能力边界在动手改架构之前我先花了一个星期做了一组小规模 benchmark。选了 30 个真实任务分成三类文档理解类长文摘要、指定章节改写、关键信息定位代码类单文件解释、Bug 定位、SQL 编写、多文件重构建议结构化输出类从合同提取字段、把对话整理成固定 JSON。每个任务跑三遍记录耗时、token 消耗和主要质量问题。结果很能说明问题任务类型Kimi K3 表现Claude 表现备注中文长文档摘要优秀长上下文处理稳定良好但长文档下费用高K3 输入成本优势非常明显代码解释与 Bug 定位中等能应付简单场景优秀多文件逻辑把握更强代码类任务应优先 ClaudeSQL/脚本生成良好标准场景没问题优秀复杂查询更少出错可用回退策略拉平差距结构化字段抽取优秀格式稳定优秀偶尔过度发散看模板复杂度切换普通问答良好速度快良好延迟略高哪个便宜用哪个这张表就是我后面设计路由策略的依据。很多人做多模型路由会陷入一个误区把路由当成不透明黑盒或者只按“模型排行榜”选最大模型。我的理解是Agent 的多模型路由本质是一个运筹问题不是比较问题。你不需要找到那个“唯一的强者”你需要把每个任务分给最合适的模型让总体成本和质量达到平衡。2. 路由层到底在路由什么任务分类、模型画像与决策边界2.1 路由的本质是“分配决策”很多做 AI 开发的朋友一听“多模型路由”第一反应是“不就是 if 分支吗根据任务类型选模型”。早期我也是这么干的但真在 Agent 里跑起来就发现不对。Agent 不是一次调用而是一个循环它要理解用户意图、策划子任务、调用工具、观察结果、再次规划……这个循环里每一步都可能交给不同模型。所以路由层的决策对象不是“这一个问题”而是“下一步动作”。举一个真实例子。用户发来一条指令“帮我看看这个仓库里哪个模块性能最差然后写一份优化方案文档。”这个任务至少有四个决策点第一步“分析仓库结构”需要强的代码理解能力应走 Claude接着“定位性能瓶颈”要结合测试输出分析交给 Claude 更稳然后“输出优化方案文档”这部分是长文写作可以让 Kimi K3 接手如果中间有“总结已有结论、继续下一步”的轻量操作甚至可以走一个更便宜的小模型。每一个决策点都要对上下文长度、工具结果、当前 token 消耗做出判断不是一个静态 if 能解决的。2.2 四个决策维度任务、上下文、成本、可靠性我把路由决策拆成四个维度每次照这个维度打分任务类型代码、文档、抽取、问答用轻量分类器或规则确定。上下文规模如果当前对话历史的 token 超过某个阈值就优先选择上下文窗口更大或成本更低的模型。成本预算记录每个请求累计花费如果本轮预算已用掉 80%后续高成本模型自动降权。可靠性要求比如金融字段抽取或涉及生产的代码修改可靠性要求高即使贵也要用更强的模型普通问答则不必。这四个维度有时会冲突。比如一个代码解释任务需要很强的推理能力但当前上下文已经有 50k token预算也快用完了。这时候就要看阈值怎么定我的做法是优先保证高可靠任务的模型能力预算超了宁可通知调用方也不在代码任务上强行降级。路由决策还要考虑到 Claude 和 Kimi 各自的工具调用生态差异。Claude 对工具的指令遵循更细腻工具调用格式更适合多步推理Kimi K3 在 API 层面同样支持函数调用但两边参数结构不完全一致这需要适配层去转换而不是让上层 Agent 关心模型差异。2.3 从“硬路由”到“软路由”硬路由就是每个任务固定对应一个模型比如“只要写文档就 Kimi只要写代码就 Claude”。这种方案简单直接但存在边界情况有些任务是混合型的比如“解释一段代码并写成培训文档”前半段属于代码理解后半段属于文档生成。用硬路由只能选一方结果总有一半不理想。软路由则是给每个模型算一个置信度再结合约束条件做分配。比如一个任务被分类器判定为“代码文档混合”Kimi 得分 0.4Claude 得分 0.6同时上下文有 60k token预算还有富余最后决定让 Claude 先做代码分析、生成中间结果再由 Kimi 续写文档。这个过程不依赖单一判断而是多次决策的组合。软路由对基建要求也更高你要有统一的任务输入格式、模型无关的中间表达、以及回退机制。我在实际项目里两个阶段都做了第一版用硬路由快速上线跑出基线数据第二版升级成软路由优化混合类任务的效果。建议你也按这个节奏走不要一上来就上复杂架构。3. 从零写一个带路由层的 Agent代码和配置一起给3.1 项目结构设计先看整体目录结构。这不是玩具项目而是我生产项目里简化后的版本可以直接拷贝改造agent-router/ ├── main.py ├── config.yaml ├── router/ │ ├── __init__.py │ ├── route_decision.py │ ├── task_classifier.py │ └── model_router.py ├── executors/ │ ├── __init__.py │ ├── base.py │ ├── kimi_executor.py │ └── claude_executor.py └── agent/ ├── __init__.py └── multi_model_agent.py分层的原则是agent 层只负责对话循环和工具调度不关心底层模型是谁executors 层把 Kimi 和 Claude 的 API 差异隐藏掉router 层负责接收任务特征并给出模型选择建议。这样未来再接新模型只需要新增一个 executorrouter 里加一个候选策略agent 层完全不用动。3.2 路由层任务分类和决策逻辑路由层我倾向于用轻量分类器加一套规则策略。纯正则太脆弱纯大模型分类又贵所以折中方案是先用一个小 prompt 分类把任务打上标签再把标签和上下文信息喂给策略模块。下面给核心代码。任务分类器我用了一次 Kimi 调用来完成因为路由决策本身的成本也要控制。以这个体量每次路由只花几十 token完全可接受。# router/task_classifier.py import json from pydantic import BaseModel class TaskTag(BaseModel): tag: str # code / doc / extract / chat / mixed_code_doc confidence: float TAG_PROMPT 你是任务分类器。判断下面用户请求属于哪类 - code: 代码理解、Bug修复、SQL、脚本编写、代码重构 - doc: 文档写作、摘要、改写、长文生成 - extract: 字段抽取、信息整理、固定JSON输出 - chat: 普通问答、闲聊、概念解释 - mixed_code_doc: 既有代码分析又有文档产出 只返回JSON: {tag: ..., confidence: 0.0-1.0} 用户请求{input} async def classify_with_kimi(user_input: str, kimi_client) - TaskTag: resp await kimi_client.chat.completions.create( modelkimi-k3, messages[ {role: system, content: TAG_PROMPT.replace({input}, user_input)} ], temperature0, response_format{type: json_object}, ) data json.loads(resp.choices[0].message.content) return TaskTag(**data)路由决策的数据结构和决策模块# router/route_decision.py from dataclasses import dataclass dataclass class RouteDecision: model: str # kimi-k3 / claude task_plan: str # 本次循环的下一步动作描述 max_tokens: int 4096 temperature: float 0.3 # router/model_router.py from dataclasses import dataclass dataclass class RequestContext: task_tag: str confidence: float current_tokens: int budget_remaining: float requires_tool: bool class ModelRouter: def __init__(self, cost_table: dict): self.cost_table cost_table def decide(self, ctx: RequestContext) - RouteDecision: if ctx.task_tag in (code, mixed_code_doc): return RouteDecision(modelclaude, task_plandeep_analysis) if ctx.current_tokens 30000 and ctx.budget_remaining 0.3: return RouteDecision(modelkimi-k3, task_plansummary_and_extract) if ctx.task_tag in (doc, extract): return RouteDecision(modelkimi-k3, task_plandraft_output) return RouteDecision(modelkimi-k3, task_planchat_reply)这个decide只是最小实现。实际生产里我会让路由层读取config.yaml的策略表把“哪些 tag 用哪个模型”“什么预算区间切换模型”都配置化而不是硬编码在代码里。硬编码的问题在于每次调策略都要改代码、走发布流程效率太低了。3.3 执行器层屏蔽两个模型的 API 差异为什么一定要有 executor 层因为 Agent 上层如果直接写 Anthropic SDK 和 Kimi SDK代码会散落得到处都是而且一旦后续换模型改动地点非常多。我抽出一个统一的run接口# executors/base.py from abc import ABC, abstractmethod from typing import AsyncIterable, Optional class BaseExecutor(ABC): model_name: str abstractmethod async def run(self, messages, system_prompt, temperature0.3, toolsNone): pass abstractmethod async def stream(self, messages, system_prompt, temperature0.3, toolsNone) - AsyncIterable[str]: passKimi 执行器用 OpenAI 兼容协议# executors/kimi_executor.py from openai import AsyncOpenAI class KimiExecutor(BaseExecutor): def __init__(self, api_key: str, base_url: str, model: str kimi-k3): self.client AsyncOpenAI(api_keyapi_key, base_urlbase_url) self.model_name model async def run(self, messages, system_prompt, temperature0.3, toolsNone): msgs [] if system_prompt: msgs.append({role: system, content: system_prompt}) msgs.extend(messages) kwargs { model: self.model_name, messages: msgs, temperature: temperature, } if tools: kwargs[tools] tools # Kimi 函数调用格式与 OpenAI 兼容 resp await self.client.chat.completions.create(**kwargs) return resp.choices[0].messageClaude 执行器用 Anthropic SDK# executors/claude_executor.py import anthropic class ClaudeExecutor(BaseExecutor): def __init__(self, api_key: str, model: str claude-sonnet-4): self.client anthropic.AsyncAnthropic(api_keyapi_key) self.model_name model async def run(self, messages, system_prompt, temperature0.3, toolsNone): conv_msgs [ {role: m[role], content: m[content]} for m in messages if m[role] ! system ] kwargs { model: self.model_name, system: system_prompt, messages: conv_msgs, temperature: temperature, max_tokens: 4096, } if tools: # Claude 的工具格式与 OpenAI 有差异需要转换 kwargs[tools] convert_tools_to_anthropic(tools) resp await self.client.messages.create(**kwargs) return parse_anthropic_message(resp)注意模型名要按官方文档实际支持的版本填写我这里示例用的claude-sonnet-4不是固定值。环境准备也很简单安装几个依赖、设置环境变量就行pip install openai anthropic pydantic pyyaml export KIMI_API_KEYxxxxxx export KIMI_BASE_URLhttps://api.moonshot.cn/v1 export ANTHROPIC_API_KEYxxxxxx3.4 Agent 主体循环、路由、执行Agent 主体负责把用户请求、历史消息、工具结果拼在一起。每次循环开始前询问路由层再交给 executor 执行# agent/multi_model_agent.py class MultiModelAgent: def __init__(self, router: ModelRouter, executors: dict[str, BaseExecutor]): self.router router self.executors executors self.messages [] async def handle(self, user_input: str): self.messages.append({role: user, content: user_input}) max_steps 5 step 0 while step max_steps: ctx await self._build_context() decision self.router.decide(ctx) executor self.executors[decision.model] result await self._run_with_tool_loop(executor, decision, self.messages) if result.finished: break step 1 return self.messages[-1][content]这个循环是 Agent 的核心每一步都重新路由而不是整段任务固定一个模型。示例里刻意没展开工具调用实现因为代码量会很大后面避坑部分我再补充典型的工具调用处理。4. 生产环境里的路由策略别只看质量成本和延迟也是命4.1 按任务类型路由最基础也是最先要做的最简单的策略就是任务类型路由。我会维护一张策略表任务标签首选模型备选模型说明codeclaudekimi-k3代码理解优先 ClaudeK3 兜底dockimi-k3claude长文档以 K3 为主复杂改写再升级extractkimi-k3claude按模板复杂度动态切换chatkimi-k3无保证响应速度按任务类型路由最大的好处是逻辑透明、容易解释出问题后能快速定位是策略表配置错误还是模型能力问题。缺点是比较死板无法应对混合任务和动态预算。我在第一版就只用了这张表上线后整体成本降低了大约 56%代价是代码类任务里约 3% 的疑难问题被 K3 兜底时质量略降。这个指标可以接受因为后续还有升级回退机制兜底。4.2 预算感知路由把成本约束纳入决策任务标签路由解决“哪个模型更合适”但没解决“钱够不够”。我加了一个预算计数器每个请求进来都预估并记录 token 消耗。具体逻辑每次请求前查询会话级 budget初始值为 1.0每次调用后按消费比例扣减如果剩余预算低于 0.3且当前任务标签不是“高可靠”类型则自动降级到 Kimi K3如果是高可靠类型预算不够时直接拒绝调用返回“需要增加预算”而不是偷偷降级。这里的“高可靠”在代码里就是一个标记从上游业务传入。生产中发现不少业务方并不关心模型效果只关心“别超预算”所以预算感知路由要支持按会话、按用户、按项目多级配额。对每个配额维度我单独维护一个计数器不然一个用户的任务会把整个项目的预算全耗光。4.3 自动回退与升级让故障和效果都有兜底在多模型体系里回退非常重要。任何一个外部 API 都可能出现超时、限流、返回格式异常。我总结回退策略为两种失败回退比如 Claude 请求超时或返回 5xx框架自动改走 Kimi K3并把这次回退记录到日志。质量升级比如 Kimi K3 生成的抽取结果里 JSON 解析失败超过两次自动升级到 Claude 重跑同一任务如果 Claude 也失败就返回明确失败信息给用户而不是堆一个坏结果上去。这里有个细节升级重跑的时候最好带上第一次的失败上下文和模型自己的报错信息这样第二个模型才知道要怎么修正。只把原始任务重新丢过去第二次大概率还是错。5. 实测数据与踩坑记录路由不是加一层转发那么简单5.1 我记录的一组对比数据下面这组数据来自我第二版软路由上线后一周的线上统计样本量大约 2000 个请求。五个指标分别是任务成功率、工具调用成功率、平均延迟、每请求平均成本、人工抽检满意度。方案任务成功率工具调用成功率平均延迟单请求成本抽检满意度只用 Claude92.3%89.1%2.8s0.38 元91%只用 Kimi K388.6%82.4%1.6s0.11 元84%硬路由91.5%87.9%2.1s0.17 元89%软路由93.8%91.2%2.0s0.19 元93%从数据能看出软路由不是各方面都最强但它在成本和满意度的平衡上最理想。延迟上比单用 K3 略高但换来的是工具调用成功率和满意度大幅上升。这组数据也说明评估路由方案不能只看准确率要把成本、延迟、满意度放在一起看。5.2 坑一工具调用格式不兼容导致 Agent 陷入死循环第一版上线没几天我就发现一个诡异问题Agent 在处理“分析代码并提交 Git commit”这类任务时会出现反复调用同一个工具、返回相同错误的情况。日志显示模型一直在请求执行read_file工具却始终拿不到内容。排查半天发现根因是路由切换第一次循环用的 Claude它返回的工具调用是 Anthropic 的tool_use块格式切换到 Kimi 之后executor 没有把历史消息里的tool_use结果转换成 Kimi 认识的格式Kimi 读不懂之前的工具输出就会重新发起同样的工具调用。死循环就是这么来的。解决办法是做一个统一的工具调用消息规范。在 Agent 内部把工具调用的中间态统一成一种内部 schema任何模型真正执行时再由适配器做转换。也就是上面代码里convert_tools_to_anthropic存在的意义。这不是模型的问题而是路由层没做好格式抽象。5.3 坑二上下文状态在两个模型之间“串台”第二个坑更隐蔽。Claude 在处理多轮对话时会生成thinking块或者用专属标记包装部分输出当这个上下文被原封不动传给 Kimi 时Kimi 有时会模仿这种标签格式甚至把内部推理内容当作正常输出返回给用户。这会导致用户看到的答案里夹着一堆不该出现的内部文字。解决方式是在 executor 层增加一个“净化”步骤在模型切换时把上一模型特有的控制标签、工具调用块等内容剥离只保留干净的对话内容。同时我还做了一个白名单只有系统 prompt 和工具输出才允许保留特殊标记普通用户消息里一旦出现模型特有标签就做转义。这个净化步骤要在路由切换前执行不能等两个模型都调用完了再处理。5.4 坑三流式输出和路由切换结合时时序容易乱后来我做流式响应优化让 Claude 生成内容时逐字输出到前端。但 Agent 循环里一旦中途路由切换到 Kimi就得先终止 Claude 的流再启动 Kimi 的流这个切换过程如果没处理好前端会出现一段乱码或半截消息。我踩到的具体问题不是所有 WebSocket 连接都支持中途更换生成来源。我在服务端用一个pending_output缓存区切换时先把已生成但未发送的文本发送掉再清空缓冲区启动新流这样前端看到的就非常连贯。这个坑告诉我多模型路由不是只在后端做一个决策就完事它会影响整个数据管道。流式处理尤其要在架构设计阶段就想好别等上线了再补。5.5 评估没有统一口径调优全靠感觉最后也算一个经验层面的坑。刚上路由那两周我靠人工抽查来判断效果但后来发现人工抽查根本发现不了“混合任务在 K3 上偶尔变差”这类问题。后来我搭了一套离线评测集用统一的评分标准来对比不同路由策略。这五个指标后来也变成了线上监控的核心指标答案正确率、JSON 可解析率、工具调用成功率、平均延迟、单次成本。有了这套评测后面调策略才不是拍脑袋。6. 后续演进与我的最终建议6.1 下一步引入反馈闭环让路由策略自己进化第一版路由策略本质上还是“人肉配置”我这个开发根据 benchmark 结果写死规则。后续我计划加入一个轻量的反馈模块用户可以对 Agent 输出点赞或点踩系统把负反馈样本收集起来定期重新评估路由策略必要时通过检索或小模型分类器自动修改任务标签。也就是说路由策略也要能迭代不能一成不变。成本控制方面可以做得更细。我现在是按会话计费后面会做成按项目、按团队多维度预算并接入一个统一的 Metrics 面板实时展示“每个模型每类任务的调用量、成本、平均质量评分”。到这一步多模型路由才真正变成一个可运营的基础设施而不是一段 if-else。6.2 我的几条实际建议最后聊点实实在在的。如果你也要在自己的 Agent 里做多模型路由我有几个建议先别急着上架构。建议先花一周时间把自己业务的任务类型摸清楚用真实样本测 Kimi K3 和 Claude 的差异数据驱动比感觉靠谱。路由层和模型执行层一定要解耦。哪怕你这一版只有一个模型也要把 executor 抽象出来后面加模型会非常省事。每个路由决策都要记录。我每次请求都会记录模型、任务类型、token 消耗、耗时、是否回退这些日志是后续调优的原材料。做好失败回退但别过度依赖。回退机制保证可用性但如果某个模型频繁触发回退说明路由策略可能有问题要去看是不是分类分错了。质量评测的口径一定要统一否则两个模型之间没法比较。我现在每个任务都有一份固定的评估 prompt所有模型跑同一份评测集。另外这套路由思路并不局限于 Kimi 和 Claude。你后面接入本地模型、专用小模型、或者换成其他大模型 API核心架构都不用怎么动只需要在 executors 里加一个类、在 router 里加几条策略。这也是分层设计最大的回报。我个人做这个项目最大的感受是多模型路由不是让你夹在两个模型之间左右为难而是让每个模型做自己最擅长的事。Kimi K3 负责长文档和成本敏感场景Claude 负责高难推理和代码任务路由层只做一个冷静的调度者。这个框架搭好之后你再加任何新模型都是加分项不是重构。