接到一个很有意思的工程实践类标题给 Pi Agent 装上自动挡Model Router 的两挡设计与故障自愈。我基于自己在 LLM 应用工程和 Coding Agent 落地过程中积累的经验把整个方案的思路、设计细节、实现步骤和踩坑记录完整展开。给 Pi Agent 装上自动挡Model Router 的两挡设计与故障自愈最近在调教 Pi coding agent 时我发现一个特别典型的痛点单一模型根本没法同时满足“响应快”和“能力强”两个诉求。轻量模型跑得飞起但代码质量时常拉胯重量模型输出漂亮但每次等待都像开盲盒而且一旦上游接口抖动整个 Agent 的任务链直接卡死。后来我索性给 Pi Agent 的模型调用层加了一个 Model Router用“两挡设计”解决了这个问题——轻量挡负责日常高频简单操作重载挡负责复杂重构和深层推理再配上一套故障自愈机制让 Agent 在模型异常时自己知道换路绕行。这篇文章把这套路由器的完整设计思路、实现细节和排障经验写出来适合正在做 Agent 应用、或者想优化模型调用链路的工程师参考。1. 为什么需要给 Pi Coding Agent 装“自动挡”1.1 一个真实的崩盘现场单一模型的短板先说我遇到的实际场景。当时的 Pi coding agent 跑的是一个固定模型日常任务包括变量重命名、小函数重构、补注释、生成单元测试之类。这类任务有个共同特点上下文不长、目标明确、模型只需要做局部修改但数量特别大一天能触发上千次调用。用轻量模型跑这些任务单次响应大概 1~2 秒体感很快。但问题出在那种“看起来简单、实际需要全局理解”的任务上比如跨文件重构、提取公共模块、排查一个诡异的状态更新 bug。轻量模型经常给你改一个文件就完事完全没有意识到另一个文件里还有依赖关系结果代码能跑通但架构越来越烂。后来我换上重量模型统一处理所有请求质量确实上来了但两个新问题非常致命。一是单次调用延迟翻了 4~5 倍原本 2 秒能完成的重命名变成了 8 秒起步用户侧明显感觉到 Agent “变笨变慢了”二是成本直接飙升因为简单任务的 token 消耗比复杂任务少了一个量级但重量模型的定价是轻量模型的十几倍甚至几十倍这笔账一算下来项目还没上线预算先扛不住了。我一度在“质量”和“效率”之间反复横跳直到想明白一个事不是模型不行是我没有给任务分诊。不同复杂度的任务本来就该交给不同能力的模型去处理就像开车一样——城市通勤用经济挡高速超车才挂运动挡。这才有了给 Pi Agent 加 Model Router 的想法。1.2 两挡设计的核心思路从“大而全”到“够用就好”Model Router 的本质是在 Agent 和底层模型之间插入一个路由决策层。它接收任务的上下文、难度、类型等信息决定这次调用应该走哪条模型链路。“两挡设计”是我刻意保持的简化策略不是不能做三挡四挡而是两挡在工程上最清晰D1 挡轻量挡对应轻量模型主要承担高频简单任务。目标是把 P50 响应时间压在 3 秒以内单次调用成本尽量低。D2 挡重载挡对应重量模型主要承担复杂推理、大规模重构、跨文件分析等任务。目标是质量优先延迟和成本可以适当放宽。这个挡位切换逻辑很像变速箱的换挡逻辑任务简单就挂高档位速度快任务复杂就降挡动力足。当然这里的“挡位”和车速是反过来的我用“两挡”只是为了让团队沟通时有个好记的名字。这个设计最核心的价值不在于“省了多少钱”而在于让系统有了可控的兜底策略。一旦重量模型不可用路由器可以自动切到轻量挡服务不会全挂轻量挡质量不够时又可以升挡尝试重量模型。两者互为备份这是单一模型方案永远做不到的。2. 两挡路由器的关键设计挡位判定与切换逻辑2.1 挡位划分轻量挡与重载挡的适用场景在设计阶段我先把 Pi coding agent 的典型任务做了分类这个分类直接决定路由策略怎么定。D1 轻量挡适用场景单文件的变量重命名、函数签名调整基于已有代码补注释、生成 docstring单测用例批量生成不涉及跨模块 mock 分析格式化、lint 修复、小范围错误修正代码解释、片段总结、commit message 生成这些任务的特点是输入输出都在一个小范围内模型不需要全局推理上下文窗口小即使模型能力弱一些结果也基本可控。D2 重载挡适用场景跨文件重构、公共模块抽取分析运行时错误、排查数据流异常复杂算法实现与优化基于多文件上下文的设计方案生成长对话历史下的需求变更理解这些任务的特点需要模型同时理解多个文件的依赖关系或者需要较强的推理链轻量模型往往“有心无力”生成的方案看起来对实际执行时漏洞百出。这里有一个重要的经验不要让任务类型单一决定挡位。比如“修改函数”听起来很简单但如果这个函数被十几个文件引用那修改方案就涉及影响面分析必须升级到重载挡。所以挡位判定需要综合考虑“任务类型 上下文规模 影响范围”三个维度。2.2 路由判定的三种策略规则、语义与混合确定挡位划分后就要设计路由判定逻辑。我经过三轮迭代最终确定用“混合规则”方式而不是单纯依赖某一种信号。第一版纯规则关键字匹配。我在提示词里埋了任务标签比如[simple_task]、[complex_task]由 Agent 自己标注。这个方案简单直接但问题很大——Agent 对任务难度的“自我认知”不靠谱经常把简单任务标成复杂任务路由准确率只有 60% 多。第二版基于上下文维度的启发式规则。我统计了任务的输入 token 数、涉及文件数、调用深度、是否包含“重构/优化/分析”等关键词建立了一个加权打分公式complexity_score ( token_count / 2000 * 0.4 file_count * 2 * 0.3 keyword_score * 0.2 history_turns * 0.1 )实测下来这个打分比纯关键字准确不少但问题出在 keyword_score 的判定上——有些任务描述里带着“分析”但实际是简单解读有些任务没说“重构”但改动牵连很广。规则终究是规则的局限。第三版最终规则打分 语义分类双通道。我加了一个轻量分类模型做语义路由专门对任务描述做二分类“简单 / 复杂”再用规则打分做校准。两者取“任一判为复杂则走重载挡”的策略虽然会多一些保守路由但误伤率把复杂任务路由到轻量模型基本降到了 5% 以下。这个双通道设计最务实的点在于语义通道负责理解“任务意图”规则通道负责捕获“任务规模”。两者互为补充不会因为模型分类漂移而导致路由完全失效。2.3 切换时机与防抖策略挡位切换不是越快越好里面有个非常容易踩的坑频繁切换会让系统抖得像筛子。设想一个场景任务调用了重载挡结果重量模型响应超时路由器立刻降挡到轻量模型下一次任务又因为规则打分偏高升到重载挡再次超时再次降挡。系统就在两个挡位之间反复横跳用户看到的就是时而快时而慢体验反而更差。我的解决方案是引入“防抖窗口”和“最小停留时间”两个机制。防抖窗口的意思是说在一次任务结束后记录本次调用的挡位情况如果系统判定需要切换挡位先等一个冷却时间比如 30 秒在这个时间内即使新的任务也满足切挡条件也不会马上切而是累计到一个“切换意向”计数器里达到阈值再执行。最小停留时间更简单粗暴每个挡位至少连续跑满 N 个任务才能切换。比如 D2 挡至少停留 10 个任务D1 挡至少停留 5 个任务。这样做的逻辑是挡位切换本身有成本模型上下文预热、路由决策开销频繁切换省下的延迟远不足以弥补额外开销。我现在的实现里还加了一个“会话级状态”同一个用户的同一个会话挡位变化尽量控制在相邻挡位之间避免从一个极端跳到另一个极端。比如正在跑一个大重构任务中间插入几个小任务系统仍然保持在 D2 挡直到所有任务都结束后才重新分诊。这个设计非常关键它避免了同一代码库上下文被反复切换模型带来的“记忆断层”。3. 故障自愈机制不是“不犯错”而是“快恢复”3.1 自愈链路的三层防御超时、重试、熔断Model Router 解决了“选对模型”的问题但还没有解决“模型挂了怎么办”的问题。真实生产环境中模型 API 的故障率比你想象的高得多尤其是重量模型高峰期超时率 10% 都是常态。这时候就需要故障自愈机制兜底。我的自愈链路分三层层层递进每一层处理不同粒度的异常第一层超时控制。每个模型调用都设了独立超时阈值D1 挡 8 秒D2 挡 30 秒。超过阈值直接判定本次调用失败进入重试逻辑。这里的经验是超时阈值应该基于模型的历史 P95 延迟来定不是拍脑袋。我统计过 D2 挡正常场景 P95 是 12 秒P99 是 20 秒所以把超时设在 30 秒既不会频繁误杀也不会让用户等太久。第二层有限重试。重试不是无限重试而是“1 次快速重试 1 次切换重试”。第一次超时后先重试同一模型一次如果仍然失败就执行“切换重试”——把这次请求改走另一个挡位的模型。比如 D2 挡失败就降级到 D1 挡重跑D1 挡失败就升级到 D2 挡重跑。这样既能处理瞬时抖动同一模型重试成功又能处理模型级故障切换后绕行成功。第三层熔断器。如果某个模型在单位时间窗口内错误率超过阈值熔断器直接打开短时间内不再把任何请求路由到这个模型。我的熔断参数是在 60 秒滑动窗口内请求数超过 20 次且错误率超过 50%熔断 120 秒。熔断期间所有该档位请求自动切换到另一个挡位。这三层防御合在一起的效果是单个模型的小抖动在重试层消化模型整体故障在熔断层消化系统不会因为任何一个上游故障而整体不可用。3.2 降级与回退策略挡位不是单向的很多同学设计自愈时有个误区以为降级就是“从贵的模型降到便宜的模型”单向操作。实际工程里降级和回退必须是双向的而且回退比降级更难做对。降级场景指的是D2 挡失败后请求落到 D1 挡。这里要注意D1 挡虽然能跑但输出质量可能不达标所以我在降级路径上额外加了一个“质量校验”步骤——把 D1 挡的输出和原始任务的目标做一次比对如果明显不满足比如测试用例生成失败、代码有语法错误标记为“降级不成功”并尝试再次切回 D2 挡。回退场景指的是D1 挡连续失败后系统判定这个任务可能比预想的复杂自动升级到 D2 挡重跑。这个逻辑看起来简单但容易踩一个坑——升级触发的条件必须足够保守否则简单任务都会因为一次偶然超时而被升级到重载挡两挡的成本优势就没了。我在实现里用了一个非常保守的回退触发规则同一任务在 D1 挡连续失败 2 次或者返回结果经质量校验不通过 2 次才允许升级到 D2 挡。这样既保证了兜底能力又不会太灵敏。另一个容易被忽视的点降级和回退必须记录审计日志。每次降级、回退、熔断都写入日志后面做模型运维时这些日志是判断模型服务质量和路由策略是否合理的第一手数据。3.3 可观测性日志、指标与迭代基础故障自愈不是一锤子买卖它需要持续迭代而迭代的基础就是可观测性。我在 Model Router 里埋了三类观测数据结构化日志每次路由决策、每次模型调用、每次降级/回退都输出一条 JSON 日志字段包括任务 ID、路由挡位、模型名称、延迟、token 数、错误码、重试次数、最终状态。这些日志用于事后问题追踪比如“某个任务明明该走 D2 挡为什么走了 D1”。聚合指标我把关键指标暴露成 Prometheus Metrics包括路由分布D1/D2 占比、模型调用成功率、P50/P95/P99 延迟、降级次数、回退次数、熔断次数、平均 token 消耗、预估成本。这些指标直接上 Grafana 面板每天看一眼就能快速判断路由器策略是否健康。调用链追踪因为 Pi Coding Agent 一个任务通常包含多次模型调用我把每次路由决策和模型调用串成一条调用链配上 trace_id这样定位问题时就能看到“这个任务先走了 D1然后降级到 D2D2 超时后又重试最后成功”整个链路一目了然。我强烈建议任何做模型路由的项目都把可观测性当作第一天就要做的事不要等系统出问题再去补。没有观测数据的自愈机制等于闭着眼睛开车。4. 实操过程从零搭建一台两挡 Model Router4.1 核心配置与路由判定模块下面是我在 Pi Coding Agent 项目中实际用的一版 Model Router 核心代码精简掉业务细节后如下。首先是配置模块所有的挡位、模型、阈值都在这里管理方便后续调优# router_config.py from dataclasses import dataclass, field dataclass class ModelRouteConfig: # D1 轻量挡 light_model: str deepseek-v3-lite light_timeout: float 8.0 # 秒 light_max_retries: int 1 light_min_stay: int 5 # 最小停留任务数 # D2 重载挡 heavy_model: str deepseek-r1 heavy_timeout: float 30.0 heavy_max_retries: int 1 heavy_min_stay: int 10 # 熔断参数 circuit_breaker_window: int 60 # 秒 circuit_breaker_threshold: float 0.5 # 错误率 circuit_breaker_open_seconds: int 120 # 防抖窗口 debounce_seconds: int 15 switch_intent_threshold: int 3这里我给每个挡位配置了独立的超时和最小停留时间核心逻辑是挡位的能力差异决定了它们对延迟的容忍度不同D2 挡任务更复杂超时阈值理应更高。然后是路由判定模块采用“规则打分 语义分类”双通道模式# router.py class TaskComplexityEstimator: def __init__(self, config: ModelRouteConfig): self.config config def _rule_score(self, task: dict) - float: 基于 token 数、文件数、关键词的规则打分。 token_count task.get(input_tokens, 500) file_count task.get(related_files, 1) keywords task.get(keywords, []) kw_score min(len(keywords) * 0.5, 1.0) token_score min(token_count / 2000, 1.0) * 0.4 file_score min(file_count / 5, 1.0) * 0.3 keyword_score kw_score * 0.2 history_score min(task.get(history_turns, 0) / 10, 1.0) * 0.1 return token_score file_score keyword_score history_score def _semantic_score(self, task_description: str) - float: 轻量语义分类返回 0~1 的复杂度得分这里简化成基于关键词的启发式判断。 complex_indicators [重构, 分析, 排查, 优化, 跨文件, 设计, 架构] score 0.0 for word in complex_indicators: if word in task_description: score 0.2 return min(score, 1.0) def decide(self, task: dict) - str: rule_s self._rule_score(task) semantic_s self._semantic_score(task.get(description, )) # 双通道任一判定复杂则走 D2保守优先 if semantic_s 0.6 or rule_s 0.65: return D2 return D1关于_semantic_score的简化真实生产中这个位置应该替换为一个二分类模型或向量检索匹配核心逻辑不变——语义通道和规则通道互为校验避免单一信号误判。我在项目里是先跑了一个小的 bert 分类模型随后为了部署简单又换成了聚类模板匹配准确率差不了太多。4.2 路由执行引擎与切换控制路由判定只解决了“往哪走”的问题还要有一个执行引擎来管理“怎么走”和“什么时候切换”。我把执行引擎设计成支持“同步阻塞调用”和“异步任务队列”两种模式。同步模式适合需要立即拿到模型结果的场景比如用户正在对话中请求代码补全。异步模式适合后台批量任务比如批量生成单测。核心执行逻辑如下# executor.py import time import threading from collections import deque class RouterExecutor: def __init__(self, config: ModelRouteConfig, llm_client): self.config config self.llm llm_client self.current_gear D1 self.gear_since time.time() self.task_count_in_gear 0 self.switch_intent 0 self.lock threading.Lock() self.error_window deque() def _record_error(self, model_name: str, success: bool): 在滑动窗口内记录模型错误率。 self.error_window.append({ ts: time.time(), model: model_name, success: success, }) while self.error_window and time.time() - self.error_window[0][ts] self.config.circuit_breaker_window: self.error_window.popleft() def _is_circuit_open(self, model_name: str) - bool: 熔断判定窗口内请求数够且错误率超阈值则熔断。 window [e for e in self.error_window if e[model] model_name] if len(window) 20: return False error_rate sum(1 for e in window if not e[success]) / len(window) return error_rate self.config.circuit_breaker_threshold def _can_switch_gear(self) - bool: 防抖 最小停留时间联合判定。 min_stay self.config.light_min_stay if self.current_gear D1 else self.config.heavy_min_stay if self.task_count_in_gear min_stay: return False if time.time() - self.gear_since self.config.debounce_seconds: return False return True这里我特别说明一个容易忽略的点切换挡位不是在单次请求过程中切换而是在两次请求之间切换。也就是说一个任务一旦开始执行就用固定的挡位跑完不会因为中途超时而临时改道。临时改道是“重试”和“降级”的职责与“挡位切换”是两回事。把这两件事分开逻辑才清晰。4.3 自愈与故障恢复模块的实现最后是故障恢复模块它封装了超时、重试、降级和回退的完整流程# recovery.py class RecoveryHandler: def __init__(self, config: ModelRouteConfig, executor: RouterExecutor, llm_client): self.config config self.executor executor self.llm llm_client def invoke_with_recovery(self, task: dict, gear: str): 统一入口按挡位执行带自愈逻辑。 if gear D1: model, timeout self.config.light_model, self.config.light_timeout backup_gear D2 else: model, timeout self.config.heavy_model, self.config.heavy_timeout backup_gear D1 if self.executor._is_circuit_open(model): # 熔断打开直接切备用挡 return self._fallback(task, backup_gear, reasoncircuit_open) try: # 首次调用 result self.llm.call(model, task, timeouttimeout) self.executor._record_error(model, True) return result except Exception as e: self.executor._record_error(model, False) # 第一次重试同模型 try: result self.llm.call(model, task, timeouttimeout) self.executor._record_error(model, True) return result except Exception as e2: return self._fallback(task, backup_gear, reasonretry_failed) def _fallback(self, task, backup_gear: str, reason: str): 降级/回退到备用挡位。 backup_model ( self.config.light_model if backup_gear D1 else self.config.heavy_model ) backup_timeout ( self.config.light_timeout if backup_gear D1 else self.config.heavy_timeout ) # 记录审计日志 log_event(model_router_fallback, { task_id: task.get(task_id), reason: reason, fallback_gear: backup_gear, }) return self.llm.call(backup_model, task, timeoutbackup_timeout)这里的核心设计决策是重试只做一轮第二次失败直接降级不做三轮重试的“有耐心”策略。原因在于一次模型调用如果连续两次超时大概率是模型服务本身的问题而不是偶发抖动继续重试只会让任务失败得更慢并且侵蚀用户体验。4.4 接入 Pi Coding Agent 的完整链路与测试验证把上述模块串起来后接入流程是这样的用户提交任务 →TaskComplexityEstimator.decide()给出建议挡位 → 检查当前挡位与建议挡位是否一致若不一致且满足切换条件先切换 →RecoveryHandler.invoke_with_recovery()执行模型调用并处理自愈 → 返回结果给 Agent 编排层。整个链路我在本地用 mock LLM 服务做了测试验证覆盖了这几个场景正常 D1 任务路由到轻量模型单次成功延迟 1.5 秒。正常 D2 任务路由到重量模型单次成功延迟 9 秒。D2 首次超时、重试成功总延迟约 40 秒用户侧看到任务变慢但最终成功。D2 两次超时、降级到 D1 成功D1 返回结果任务在限定时间内完成。D1 连续失败 2 次、自动升级 D2 成功系统判定任务复杂度被低估回退逻辑生效。模型 A 熔断打开、所有请求走模型 B链路无感知自愈生效。这里我建议所有做类似系统的人都做一个故障注入测试故意让某个模型 100% 失败观察系统是否能在 1 分钟内切换到备用链路并恢复服务。这个测试越早做越好能提前暴露很多只在“故障状态”下才会出现的问题。5. 常见问题与排查技巧实录5.1 路由误判导致频繁切换上线第一周我就遇到了一个典型的误判场景用户让 Pi Agent “帮我优化这段查询逻辑”任务描述里出现了“优化”这个关键词语义通道直接打高分路由到了 D2 重载挡。但实际代码只有十行D1 挡足以处理。结果就是大量简单任务被“过度路由”到 D2 挡成本飙升延迟变长但输出质量和 D1 挡几乎没差。这个问题的根源在于我把“优化”这类词不加区分地当成了复杂度强信号。解决办法是给复杂度打分加条件约束如果任务描述包含强动作词但代码文件数和上下文 token 都很少则降权处理。同时我把“优化”拆成两种局部优化单文件、单函数走 D1整体优化多文件、架构级走 D2。具体实现里我在_semantic_score中加入了否定削弱逻辑if 优化 in task.get(description, ): if file_count 2 and input_tokens 1500: semantic_s - 0.3 # 削弱误判用这类规则校准后D2 挡的路由占比从 47% 降到了 22%整体成本降了一大截而任务成功率没有变化。吃一堑长一智我现在对所有关键词信号都抱有怀疑态度语义通道必须和规模信号做联合判定不能只盯着关键词。5.2 模型失败后重试反而更慢注意退避与重试计数有一次我在压测时发现一个非常蹊跷的现象某些任务在 D2 挡超时后进入重试但重试的响应时间比初始调用还要长很多最终任务总耗时接近 60 秒用户已经等得不耐烦。排查后发现问题出在重试时没有重置超时计时器。我最初的代码是把整个invoke_with_recovery调用包在一个总超时里第一次调用用了 28 秒重试又等 28 秒两个加起来远超用户可接受的范围。这个问题的解决方案其实很简单每次模型调用单独使用独立的超时计数器并且在重试前加入一个 500ms~1s 的随机退避防止所有请求同时重试撞上同一个故障窗口。另一个和重试相关的经验是重试请求应该携带和初始请求完全不同的 request id否则部分模型 API 的网关会识别为重复请求拒绝重放反而导致重试必失败。我在测试中就遇到过某家模型 API 对相同 request id 的重复请求直接返回限流错误的坑。5.3 观测数据揭示的“隐形降级”问题接入 Grafana 面板后我发现一个之前完全没意识到的现象D1 挡的失败率比 D2 挡还低但 D1 挡的超时次数占比却非常高。一开始以为是路由策略有问题导致 D1 挡压力过大后来才发现是轻量模型的服务端存在“长时间空闲连接被切断”的问题。具体表现是如果某个轻量模型实例超过 2 分钟没有收到请求下一次调用就需要重新建连导致延迟从 1.5 秒直接飙到 10 秒以上大量请求踩着超时阈值边缘失败。这个问题在单请求场景下几乎不可见但 Pi Coding Agent 是高频调用场景每天几千次请求这个问题被无限放大。最后我通过两个手段解决了一是给模型调用客户端加连接池二是加定时心跳保活。这个问题也验证了自愈机制的一个好处——即使连接层有问题系统也会通过重试和降级兜住不会直接挂掉但如果没有观测数据这类隐形瓶颈可能上线几个月都发现不了。结合这段时间的实践我的总体感觉是Model Router 不是一个“搭完就完事”的功能它更像一个持续调优的控制系统。两挡设计给了系统最基本的灵活性故障自愈给了系统稳定性但真正让它长期可靠运行的是持续的数据分析和策略迭代。每一次路由误判、每一次降级失败、每一次熔断打开都是优化路由策略的线索。没有这些数据再精巧的设计也只是空中楼阁。