★ FEATURED ARTICLE 1. 从手动挡到自动挡Pi Agent 为什么需要一个 Model Router用 Pi Agent 跑过一段时间任务的人大概率都经历过这种场景一个本来跑得好好的流程突然因为某个模型接口超时、限流或者返回格式异常整个 Agent 就卡死在那里要么无限重试烧钱要么直接抛错退出。你盯着日志发现它其实只是需要换一个模型继续跑但 Agent 本身没有这个判断能力——它只会认死一个配置好的模型撞了南墙也不回头。这就是典型的手动挡困境。所谓手动挡指的是模型选择这件事完全由人在配置里写死你指定用哪个模型Agent 就从头到尾用哪个模型中途不管发生什么都不换。这种模式在实验阶段没问题一旦进入长时间、多任务的真实运行环境脆弱性就暴露无遗。而 Model Router 要做的就是给 Pi Agent 装上自动挡——让 Agent 自己根据当前情况决定用哪个模型并且在模型出问题时能自动切换、自动恢复。我最初接触这个概念的时候也以为无非就是加个 try-catch 换个模型重试。但真正动手之后才发现这里面涉及的问题远比想象中复杂什么时候该切换切换到哪个切换之后原来的上下文怎么处理如果所有模型都挂了怎么办这些问题如果没有一套清晰的设计所谓的自动只会变成自动乱套。这篇文章想聊的就是我在给 Pi Agent 设计 Model Router 时踩过的坑和最终沉淀下来的方案。核心是两挡设计——一个偏保守的手动挡兜底一个偏激进的自动挡调度两者配合以及故障自愈——当模型链路出现异常时Router 如何一步步把任务救回来。如果你也在做 Agent 相关的工程化落地尤其是涉及多模型调度的场景这些经验应该能帮你少走不少弯路。需要先说明的是本文讨论的 Model Router 是一个位于 Agent 与底层模型调用之间的调度层它不改变 Agent 的推理逻辑只负责这一轮请求该发给谁。理解这个定位很重要因为很多设计上的取舍都源于此——Router 是基础设施不是智能体本身。2. 两挡设计的核心思路手动挡兜底自动挡调度2.1 为什么不是全自动就完事很多人第一反应是既然要自动那就全自动好了让 Router 完全接管模型选择。我一开始也是这么想的结果很快就遇到了问题。全自动模式下Router 的决策逻辑一旦有 bug或者某个模型的健康探测出现误判整个 Agent 的行为就变得不可预测。你根本不知道它这一轮为什么选了这个模型下一轮又为什么换了另一个排查起来极其痛苦。更麻烦的是成本。自动挡倾向于哪个能用用哪个但不同模型的单价可能差几十倍。如果 Router 没有成本意识它可能在一个简单任务上反复调用高成本模型账单直接起飞。而手动挡的价值就在于它给你一个明确的、可控的基准线让你知道至少这条路径是稳的、成本是可预期的。所以两挡设计的第一个原则是自动挡负责效率和韧性手动挡负责可控和兜底。两者不是替代关系而是互补关系。2.2 手动挡显式指定行为可预测手动挡的本质是配置即真相。你在配置里写死模型 A那么除非 A 彻底不可用否则 Router 不会主动换。这种模式适合几类场景调试和复现阶段你需要固定变量确认问题到底出在模型还是出在 Agent 逻辑上。成本敏感的生产任务你明确知道这个任务用便宜模型就够不希望 Router自作聪明升级。合规或质量要求固定的场景某些输出必须由特定模型产生不能随意替换。手动挡的实现其实很简单核心就是一条优先级链。下面是一个简化的配置结构router: mode: manual primary: model-a fallback: [model-b, model-c] retry: max_attempts: 2 backoff_ms: 500这里的关键是fallback列表——即使是手动挡也不应该是单点死磕。当 primary 连续失败达到阈值Router 按顺序尝试 fallback。注意手动挡的 fallback 是有序的、确定的不涉及任何动态评分这样行为才可预测。提示手动挡下 fallback 的触发条件要设得保守一些比如连续 2 次超时或返回明确错误码才切换。如果一有风吹草动就切手动挡就失去了稳定基准的意义。2.3 自动挡动态评分按需调度自动挡才是 Model Router 真正有意思的部分。它的核心是一个动态评分函数每一轮请求前Router 根据当前各模型的状态算出一个分数选最高的那个。评分维度通常包括维度说明权重示例健康度近期成功率、错误率0.4延迟近期 P95 响应时间0.2成本单位 token 价格0.2负载当前并发占用0.1任务匹配模型能力与任务类型的匹配度0.1权重不是拍脑袋定的而是根据你的实际痛点调整。比如你被账单教育过就把成本权重调高如果延迟是用户体验的命门就加大延迟权重。我自己的经验是健康度权重不要低于 0.3否则 Router 容易在模型已经半死不活的时候还硬往上撞。自动挡的另一个关键是任务匹配。不是所有任务都适合同一个模型。比如代码生成任务和长文本摘要任务对模型能力的要求完全不同。Router 可以维护一张任务类型到模型偏好的映射表在评分时给匹配的模型加分。这部分不需要很复杂一张静态表加少量规则就够用过度设计反而难维护。2.4 两挡之间怎么切换两挡不是互斥的而是可以运行时切换的。我的做法是提供一个显式的切换开关同时允许自动降级当自动挡连续多次无法找到可用模型时Router 自动退回到手动挡的 fallback 链保证任务至少能跑完。class ModelRouter: def __init__(self, config): self.mode config.mode # manual or auto self.manual_chain config.manual_chain self.scorer DynamicScorer(config.weights) def select(self, task_context): if self.mode manual: return self._select_manual() try: return self._select_auto(task_context) except NoAvailableModelError: # 自动挡找不到可用模型降级到手动挡兜底 self.mode manual return self._select_manual()这个降级逻辑看起来简单但它解决了一个很实际的问题自动挡的评分依赖健康数据而健康数据在系统刚启动或者数据被清空时可能是空的。这时候自动挡会瞎选降级到手动挡反而更稳。3. 故障自愈当模型链路开始出问题时Router 该做什么3.1 故障的三种典型形态在讲自愈之前得先搞清楚故障到底长什么样。我在实际运行中遇到的模型故障大致可以归为三类每类的处理策略完全不同第一类是硬故障。模型接口直接返回 5xx或者连接超时、DNS 解析失败。这类故障特征明显Router 能立刻感知处理方式也直接——标记该模型不可用切换到下一个。第二类是软故障。接口返回 200但内容有问题比如返回了空响应、JSON 格式错误、或者输出明显被截断。这类故障最阴险因为从 HTTP 层面看一切正常但 Agent 拿到的东西没法用。Router 需要做响应校验把这类假成功识别出来。第三类是性能退化。模型还能用但延迟从 800ms 涨到了 8s或者成功率从 99% 掉到了 70%。这类故障不会立刻让任务失败但会拖垮整体体验。Router 需要通过滑动窗口统计来捕捉这种趋势。注意软故障的识别是自愈能力的核心难点。我的经验是至少要做三层校验——响应非空、结构可解析、关键字段存在。三层都过才认为这次调用成功。3.2 健康度是怎么算出来的自愈的前提是知道谁病了。Router 需要为每个模型维护一个健康度指标这个指标不能是简单的成功/失败布尔值而应该是一个随时间衰减的连续值。我用的是指数加权移动平均EWMA公式大致是这样health_new alpha * result (1 - alpha) * health_old其中result是本次调用的结果成功为 1失败为 0alpha是衰减系数通常取 0.2 到 0.3。这个公式的好处是近期的结果权重高历史结果逐渐淡出。一个模型如果连续成功健康度会快速回升如果连续失败健康度会快速下降。但光有 EWMA 还不够因为不同故障的严重程度不一样。超时和返回格式错误对健康度的影响应该不同。所以我给不同类型的失败设了不同的惩罚系数故障类型惩罚系数说明连接超时0.0完全失败5xx 错误0.0完全失败限流 4290.3部分惩罚可能只是暂时响应为空0.1严重接近完全失败格式错误0.2严重延迟超标0.6轻度惩罚这样算出来的健康度更能反映模型的真实状态。一个偶尔被限流的模型健康度不会掉得太狠而一个频繁返回空响应的模型健康度会迅速跌到谷底。3.3 熔断、半开与恢复探测健康度低到一定程度Router 应该把这个模型熔断——暂时不再向它发请求。但熔断不能是永久的否则模型恢复了 Router 也不知道。所以需要半开状态熔断一段时间后放少量探测请求过去如果成功就恢复失败就继续熔断。这套机制借鉴的是经典的断路器模式但在 Model Router 场景下有几个特殊考量探测请求要选低风险任务。不要拿一个重要的长任务去探测万一探测失败损失太大。我通常用短小的 ping 类请求做探测。熔断时长要动态。第一次熔断可能只停 30 秒如果恢复后很快又失败下次熔断就停 5 分钟再失败停 30 分钟。这种退避策略能避免在模型持续故障时反复无效探测。半开状态要限流。同一时间只允许一个探测请求避免探测流量本身把刚恢复的模型又打挂。class CircuitBreaker: def __init__(self, failure_threshold5, recovery_timeout30): self.state closed # closed / open / half_open self.failure_count 0 self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.last_failure_time None def allow_request(self): if self.state closed: return True if self.state open: if time.time() - self.last_failure_time self.recovery_timeout: self.state half_open return True return False # half_open: 只放一个请求 return self._try_acquire_probe()3.4 自愈的完整链路把上面这些拼起来一次完整的故障自愈链路大概是这样感知调用失败或响应校验不通过记录故障类型。降权更新该模型的健康度按惩罚系数扣分。切换当前请求立即切换到评分最高的备用模型重试。熔断判断如果健康度跌破阈值触发熔断进入 open 状态。探测恢复熔断超时后进入 half_open放探测请求。恢复或继续熔断探测成功则恢复 closed失败则回到 open 并延长熔断时长。这条链路的关键在于每一步都要有明确的触发条件和退出条件不能含糊。我见过一些实现熔断和恢复的边界很模糊结果模型在半死不活的状态下反复横跳反而制造了更多问题。4. 上下文与状态切换模型时最容易翻车的地方4.1 对话历史能不能直接搬这是我在设计 Router 时踩的最大的一个坑。最初我以为切换模型无非就是把同样的对话历史发给另一个模型能有多难结果发现不同模型对消息格式的要求、对 system prompt 的处理、甚至对角色定义的理解都不一样。最典型的问题是system prompt 的兼容性。有些模型对 system 消息支持很好有些则要求把 system 内容合并到第一条 user 消息里。如果你直接把为模型 A 构造的请求原样发给模型 B轻则效果打折重则直接报错。我的解决方案是在 Router 里加一层请求适配器。每个模型注册时附带一个适配函数负责把统一的内部请求格式转换成该模型需要的格式class ModelAdapter: def adapt(self, internal_request): # 默认实现直接透传 return internal_request class ModelBAdapter(ModelAdapter): def adapt(self, internal_request): # 模型 B 不支持 system 角色合并到首条 user 消息 messages internal_request[messages] system_content new_messages [] for msg in messages: if msg[role] system: system_content msg[content] \n else: new_messages.append(msg) if system_content and new_messages: new_messages[0][content] ( system_content new_messages[0][content] ) internal_request[messages] new_messages return internal_request这层适配看起来是额外开销但它把模型差异这个复杂度收敛到了一个地方而不是散落在 Agent 的各个角落。4.2 切换后的上下文一致性还有一个更隐蔽的问题上下文一致性。假设 Agent 在模型 A 上跑了 5 轮模型 A 在第 5 轮输出了一个工具调用Agent 执行了工具并拿到结果。这时候模型 A 挂了Router 切到模型 B。模型 B 看到的是一段它自己没参与过的对话历史它对这个工具调用的理解可能和模型 A 完全不同。这个问题没有完美的解法但有几个缓解手段切换时注入状态摘要。在切换点插入一条系统消息简要说明此前由其他模型处理当前任务状态是 XXX。这能帮新模型快速对齐。优先在任务边界切换。如果可能尽量在一个完整任务结束后再切换模型而不是在任务中途。这需要 Router 和 Agent 有更深的协作。保持工具定义一致。工具调用的 schema 在不同模型间要尽量统一减少理解偏差。提示如果你的 Agent 涉及复杂的多轮工具调用切换模型的风险会显著上升。这种情况下我建议把自动挡的切换阈值调高宁可多等一会儿原模型恢复也不要轻易中途换人。4.3 状态持久化与恢复Router 自身的状态也需要持久化否则进程重启后健康度、熔断状态全丢了自愈能力等于从零开始。我把 Router 的状态分成两类易失状态当前请求的临时上下文不需要持久化。持久状态各模型的健康度、熔断状态、统计数据需要定期落盘。持久化的频率不用太高我一般 30 秒或每 100 次调用落一次盘。太频繁会影响性能太少则重启后状态偏差大。落盘格式用简单的 JSON 就够没必要上数据库。def persist_state(self): state { models: { name: { health: m.health, circuit_state: m.circuit.state, stats: m.stats.snapshot() } for name, m in self.models.items() }, timestamp: time.time() } with open(self.state_file, w) as f: json.dump(state, f)5. 实测中的几个反直觉发现5.1 自动挡不一定比手动挡快这是最让我意外的结论。直觉上自动挡能动态选最优模型应该更快才对。但实测下来在很多场景下自动挡的端到端延迟反而更高。原因有几个评分本身有开销。每次请求前算一遍分数虽然单次很快但高频调用下累积起来不可忽略。切换有成本。切换模型意味着新的连接建立、可能的冷启动、上下文适配这些都会增加延迟。自动挡倾向于抖动。如果两个模型评分接近Router 可能在它们之间反复横跳每次切换都带来额外开销。所以我的建议是自动挡适合长任务和容错要求高的场景手动挡适合短平快和高频调用的场景。不要迷信自动就是好。5.2 健康度阈值设太低等于没设我一开始把熔断阈值设得很宽松想着多给模型一些机会。结果发现一个健康度已经掉到 0.3 的模型实际上已经基本不可用了但因为它还没跌破阈值Router 还在往上发请求导致大量请求失败后才切换用户体验很差。后来我把阈值调高到 0.6 左右效果明显改善。宁可早熔断也不要让用户等一个注定失败的请求。熔断太早的代价只是少用一个模型熔断太晚的代价是用户直接感受到失败。5.3 探测请求也会污染健康度半开状态的探测请求如果也算进健康度统计会带来一个问题一个刚恢复的模型因为探测请求样本少健康度波动很大可能导致 Router 在恢复和熔断之间反复横跳。我的处理是探测请求单独统计不混入主健康度。只有探测连续成功 N 次后才把模型正式标记为恢复并把健康度重置到一个中性值比如 0.7让它重新开始积累正常样本。5.4 成本控制要靠预算而不是评分前面提到评分里有成本维度但实测下来光靠评分权重来控制成本效果有限。因为成本权重再高也只是在选哪个上起作用无法阻止 Router 在预算耗尽后继续调用。真正有效的做法是加一层预算控制给每个任务或每个时间窗口设一个成本上限Router 在选模型前先检查预算超了就强制降级到最便宜的可用模型或者直接拒绝。这比调权重靠谱得多。def select_with_budget(self, task_context, budget): if budget.remaining 0: return self.cheapest_available() candidates self.scorer.rank(task_context) for model in candidates: if model.estimated_cost budget.remaining: return model return self.cheapest_available()6. 把 Router 接进 Pi Agent 的工程细节6.1 接入点选在哪里Router 的接入点很关键。接得太靠上Router 拿不到足够的上下文评分不准接得太靠下又要改很多底层代码。我的经验是接在Agent 的模型调用抽象层——也就是 Agent 里所有发请求给模型的地方都收敛到这一个函数Router 就挂在这个函数上。这样做的好处是Agent 的业务逻辑完全不用改它还是按原来的方式调用模型只是背后多了个 Router 在调度。坏处是这个抽象层必须设计得足够干净否则 Router 会被各种特殊调用绕过。6.2 超时和重试的配合Router 的重试和 Agent 自身的重试容易打架。如果 Agent 层有重试Router 层也有重试一次失败可能触发 N×M 次调用放大故障。我的做法是重试只在 Router 层做Agent 层不重试。Router 负责把一次逻辑调用映射到多次物理调用Agent 只看到最终结果。这样重试策略集中在一处好控制也好观测。超时设置也要分层单次模型调用的超时要短比如 30 秒Router 的整体超时要长比如 90 秒允许切换 2-3 次。这样单次超时不会拖垮整体整体超时又能兜住最坏情况。6.3 可观测性没有日志的自愈等于黑盒自愈逻辑如果不可观测出了问题你根本不知道它为什么做了某个决策。所以 Router 必须输出足够详细的日志和指标每次选择选了哪个模型评分是多少为什么选它。每次切换从哪个切到哪个触发原因是什么。每次熔断/恢复状态变化的时间点和依据。聚合指标各模型的成功率、延迟分布、切换次数。这些数据不仅能用于排查还能反过来优化评分权重。我每个月会看一次这些指标根据实际情况调整权重效果比一次性调好就不管要好得多。7. 一些关于自动挡的边界思考写到这里我想聊一个更本质的问题自动挡的边界在哪里。Model Router 能解决的是模型层面的故障和调度但它解决不了任务层面的失败。如果一个任务本身设计得有问题或者 Agent 的推理逻辑有缺陷换多少个模型都没用。我见过一些团队把 Agent 跑不通的问题全部归咎于模型然后寄希望于 Router 自动切换来救回来结果只是把失败延迟了而已。所以我的观点是Router 是韧性工具不是万能药。它能让系统在模型波动时保持稳定但不能让一个设计糟糕的 Agent 变得聪明。在引入 Router 之前先确认你的 Agent 逻辑本身是站得住的否则就是在给一个漏水的桶加更复杂的盖子。另外自动挡的自动程度也需要克制。全自动听起来很美但实际运行中人在关键节点的介入往往比全自动更可靠。我的做法是日常运行走自动挡但保留一个人工接管的开关遇到重要任务或者异常情况可以一键切回手动挡由人来决定用哪个模型。这种自动为主、人工兜底的模式比纯自动或纯手动都更实用。最后分享一个我自己的小习惯每次 Router 的配置有改动我都会先在一个影子环境里跑一段时间对比新旧配置的指标差异确认没有退化再上生产。Router 这种基础设施改动的风险往往比看起来大谨慎一点总没错。
阅读完成 · 觉得有帮助？
