★ FEATURED ARTICLE 不少人做 Agent 项目时一开始都会直接怼一个能力最强的模型上去请求全走大模型省事是省事但到了账单出来的时候心态往往就不那么稳了。我在做 Pi Agent 这个项目的时候前期也是这么干的直到有一次线上跑的批量任务因为模型服务抖动整批任务全部失败重来才痛下决心要搞一套自己的模型路由机制。最终落地的东西就是标题里说的“自动挡”——Model Router一套带两挡变速、自带故障自愈的模型调度模块。这篇文章就把这套设计从头到尾拆开讲包括为什么需要两挡、路由判定的依据、故障自愈的状态机怎么设计以及一个最小可用的 Python 实现。不管你是正在做 Agent 应用还是只想给内部系统省点模型调用成本这波实操都值得参考。1. 为什么非要装“自动挡” —— 单模型调用的三个老大难问题1.1 成本与体验的跷跷板Agent 应用里最明显的矛盾就是“效果”和“成本”打架。一个高质量大模型的 API 价格可能是轻量模型的几倍甚至十几倍而 Agent 的一次完整任务往往要串多个步骤每一步都调大模型的话单个用户的单次会话成本会快速累积。更尴尬的是很多步骤根本不需要那么强的理解能力比如简单的意图识别、格式化输出、关键词抽取用一个轻量模型完全能扛住。你要是统一用顶配模型等于每天开着跑车去买菜油耗全花在低速路段上。我最初就是吃了这个亏。Pi Agent 刚跑起来的时候平均一次完整会话要调 8 到 12 次模型接口全部走最强模型用户量上来之后成本直接失控。当时第一反应是“那就所有请求都换便宜模型”结果效果又崩了复杂指令的下游任务开始频繁出错用户反馈明显变差。这说明什么说明做 Agent 的人必须接受一个现实不同档位的模型适用于不同复杂度的任务真实系统需要的是按需分配。1.2 故障时只能干瞪眼除了钱稳定性更是大问题。模型服务跑在别人家第三方 API 的可用性再高也不是百分之百。在实际运行中我遇到的故障类型远不止“服务挂了”这一种有的是响应超时服务本身还活着但速度慢到不可接受有的是限流短时间请求量太大接口直接返回 429有的是内容质量骤降模型不报错但回复明显跑偏。最头疼的是最后一种。因为不报错监控系统完全无感知用户却在真实地流失。用单一模型的架构遇到这些故障基本只能干瞪眼——没有备用通道没有自动降级只能靠手动切换配置然后重启服务中间要损失大量时间和用户体验。更不要说深夜出了问题你人还在睡觉服务却在悄悄失败。1.3 需求分类大部分请求其实用不到大模型后来我统计了 Pi Agent 里的所有请求发现一个很扎心的事实大约 60% 到 70% 的调用都属于低复杂度任务。它们的特点是目标明确、上下文短、输出格式固定比如“把这段话翻译成英文”“从这段文本里提取所有日期”“判断用户这句话是不是在骂人”。这些活儿一个大参数模型和小参数模型做出来的效果几乎没有差别但价格差了好几倍。这就给了 Model Router 发挥空间把请求分成不同挡位简单任务走便宜快速通道复杂任务才动用高性能模型。整套系统就像汽车的自动变速箱路况好就提前升挡省油需要超车时再降挡给足动力。所以 Pi Agent 的 Model Router 核心逻辑就两个字——分流把合适的活儿安排到合适的模型上。2. 两挡设计的核心思路快挡与慢挡2.1 快挡低成本挡承载什么任务我给快挡定的调子是“能用就行但必须快和稳”。快挡用到的是轻量级模型特点是响应速度快、单价低、并发上限高。Pi Agent 里的命名、意图识别、实体抽取、简单改写、格式整理都走这一挡。因为这类任务的判定标准是“有没有做对”而不是“做得多惊艳”轻量模型已经具备足够强的指令跟随能力完全不需要反复多轮 prompt 调优。快挡的另一个优势是容错空间大。因为单价低就算某个请求需要重试一两次总成本也比较可控。这带来一个潜在好处快挡可以采用更激进的重试策略比如超时后立刻重试换备用实例而不需要像慢挡一样心疼调用次数。在实现的时候我把快挡的超时阈值设得很短一般 5 秒以内无法在预期时间内返回的结果直接判负交给下游的降级逻辑处理。2.2 慢挡高性能挡什么时候才出手慢挡留给了真正难啃的骨头复杂推理、多步规划、长文本总结、代码生成、歧义消除。这些任务的特征是一旦做错整个 Agent 链路的下游就全偏了必须拿能力最强的模型来保证上限。Pi Agent 里的典型场景是用户提出一个跨步骤的需求比如“帮我安排一个三天两夜的行程预算三千以内重点是美食和人文景点”。这种任务牵扯到约束条件、偏好推理、行程编排轻量模型经常顾此失彼。慢挡就是我最后的兜底手段宁可多花点钱也要把一个复杂任务一次做对避免因为返工而浪费更多上下文 token 和用户时间。慢挡在路由策略里的权重很高但触发条件必须严格。我在最初版本里吃过亏只要输入长度超过某个值就走慢挡结果大量简单但冗长的文本被送进了高成本模型费用虚高。后来改成“长度 复杂度特征 指令结构”的联合判定误判率才降下来。2.3 挡位切换的决策依据“什么任务该用哪一挡”听起来是模型能力问题本质上其实是分类问题。我在实现 Model Router 的时候综合了三个维度的信号来做最终决策。第一层是硬规则层命中关键词、正则表达式、系统指令类型都能直接锁定挡位。比如识别到用户只要求“翻译成英文”直接进快挡不用再考虑复杂度判断。这层规则最简单可靠所有规则我都是从历史日志里统计出来的高频模式准确率最高。第二层是动态启发式层比如输入长度、函数调用次数、上下文规模。通常我会把“上下文长度超过 6000 token”作为候选慢挡信号之一但不直接触发慢挡还要结合后面的结构化特征一起看。第三层是历史反馈层这也是初期最容易被忽略的部分。我记录了每个用户请求实际走哪个模型、任务是否成功、用户是否追问或投诉然后定期回填反馈数据。如果发现某一类请求在快挡上频繁失败就自动把这类请求升级到慢挡相反如果某类请求在慢挡上表现过于轻松就尝试降回快挡验证效果。这样整个路由规则就能不断自动迭代而不是靠人肉从日志里翻原因。3. Model Router 的路由决策与故障自愈实现3.1 路由决策的完整链路路由决策不是一个简单的 if-else 就能搞定的至少要经过四个环节特征提取、挡位判定、调用执行、结果校验。在 Pi Agent 的架构中我把这四个环节做成了一个独立的中间层上游应用只负责拼 prompt真正调谁、怎么调、失败了怎么办全部由 Model Router 负责。特征提取阶段Router 会计算当前请求的复杂度向量包括输入 token 数、指令中出现的操作符数量、是否包含多步推理关键词、上下文是否存在歧义标记、是不是首次与用户的交互等。这些特征拼接成一个特征数组输入到判定模块。判定模块是核心。它内部先跑硬规则再跑启发式打分。打分公式不复杂我用的是加权求和为每个特征分配权重超过阈值就判定为慢挡否则走快挡。权重是我用一批历史标注数据调出来的后续也一直通过反馈回填在自动微调。调用执行阶段要考虑的不只是“调模型接口”还包括超时设置、重试次数、并发控制、优先级排队。我在这个阶段最看重的是“优雅降级”快挡调用失败不是立刻报错给用户而是先尝试换另一个同档模型实例如果连续失败再升级到慢挡慢挡也失败才进入最后的兜底逻辑。最后的结果校验环节很多人会忽略。我强烈建议加上因为大模型接口返回 200 不代表内容是对的。检查项至少有JSON 是否可解析、是否包含要求的必填字段、输出长度是否异常、是否有明显的重复内容或空回复。校验不通过同样触发降级或重试。3.2 故障自愈从超时到熔断再到恢复自愈这个词听起来玄乎落到实现上其实就是一套状态机正常 → 熔断 → 探测 → 恢复。我照着熔断器的思想给每个模型通道单独维护一个健康状态。正常状态下请求照常分发如果某个通道连续出现超时、错误码或校验失败就会触发计数器累加。当“连续失败次数”达到阈值我通常设 5 次通道进入熔断状态后续请求短时间内不再打到该通道而是直接走备用通道或降级挡位。熔断不是永久性的否则万一模型服务恢复了我们还在傻乎乎地绕路。我为熔断状态设置了一个冷却窗口比如 30 秒。冷却结束通道进入半开状态放少量探针流量进去如果探针请求成功就把状态切回正常整个通道恢复使用如果探针也失败重新进入熔断并拉长冷却窗口。这套机制看起来不复杂但收益很大。Pi Agent 有一次深夜遇到上游模型 429 限流如果没有熔断机制所有请求都会挤在同一时间点反复抢同一个坏掉的通道造成雪崩效应。有了熔断之后请求自动被切到备用通道线上几乎无感知。3.3 可观测性与审计故障自愈做得再完善也不能完全替代人对系统的掌控所以我特别强调可观测性。Model Router 的每一次路由决策和每一次自愈动作都要留痕。Pi Agent 里的日志格式很简单一条日志记录请求 ID、任务类型、命中挡位、实际模型、判定打分、调用时长、是否熔断、最终成功还是降级。这些日志聚合起来就是调参和优化规则的核心依据。这里有一个很实用的经验一定要能从日志里还原“如果当时走了另一条通道会是什么结果”。我常做的一个动作是把同一个请求用快挡和慢挡各跑一遍做离线对比用来判断当下某个规则的误判率是不是在升高。这种 A/B 对照的数据比拍脑袋调阈值靠谱得多。4. 实操用 Python 写一个最小可用的 Model Router4.1 前置准备与配置结构理论说了不少是时候上点能直接抄作业的东西了。我用一个简化版 Python 实现来演示 Model Router 的核心骨架重点放在路由判定、超时控制、重试和熔断上。先定义配置结构每个模型通道有独立的超时时间、重试次数、单价权重和熔断参数。我用 YAML 文件来管理这样调参不用改代码。# model_router_config.yaml channels: fast: provider: light_model_provider base_url: https://api.example.com/v1/chat/completions timeout_seconds: 5 max_retries: 2 fail_threshold: 5 cooldown_seconds: 30 slow: provider: heavy_model_provider base_url: https://api.example.com/v1/chat/completions timeout_seconds: 15 max_retries: 1 fail_threshold: 3 cooldown_seconds: 45 router: slow_score_threshold: 10配置里的 fail_threshold 就是熔断器连续失败触发阈值每个通道单独维护避免快挡出问题把慢挡也拖下水。慢挡阈值比快挡小因为慢挡单价高连续失败三次就该警觉了。4.2 路由判定与调用实现接下来写路由判定的核心函数。先定义一个特征提取方法把一个请求变为特征字典再按规则打分。# router_engine.py import re import time import json from typing import Dict, List, Tuple class RouteFeatureExtractor: def __init__(self): self.hard_rules { translate: fast, extract: fast, summarize_long: slow, plan: slow, coding: slow, } def extract(self, user_input: str, system_prompt: str, context_tokens: int) - Dict: features {} features[input_length] len(user_input) features[context_tokens] context_tokens features[has_multi_step_keyword] any( word in user_input for word in [然后, 步骤, 规划, 依次, 最后] ) features[has_code_keyword] any( word in user_input for word in [代码, 函数, debug, 报错, python] ) features[has_long_summary] context_tokens 6000 features[hard_hit] None for keyword, channel in self.hard_rules.items(): if keyword in user_input.lower() or keyword in system_prompt.lower(): features[hard_hit] channel break return features def score(self, features: Dict) - int: if features[hard_hit]: return -1 # -1 means hard rule hit score 0 if features[input_length] 800: score 3 if features[context_tokens] 6000: score 4 if features[has_multi_step_keyword]: score 3 if features[has_code_keyword]: score 2 return score打分规则解释一下硬规则命中直接返回 -1表示无需打分按照规定挡位走。软特征打分采用加法分数越高越倾向慢挡。阈值我设定为 10也就是至少要同时命中“长上下文 多步推理 长输入”才会把请求判给慢挡这样的判定比较保守避免过度消耗高成本资源。路由决策函数和通道调用函数合在一起写# model_router.py class ModelChannel: def __init__(self, name: str, config: Dict): self.name name self.timeout config[timeout_seconds] self.max_retries config[max_retries] self.fail_threshold config[fail_threshold] self.cooldown config[cooldown_seconds] self.fail_count 0 self.state closed # closed, open, half-open self.last_half_open_ts 0 def is_open(self) - bool: return self.state open def record_failure(self): self.fail_count 1 if self.fail_count self.fail_threshold: self.state open self.fail_count 0 def record_success(self): self.fail_count 0 self.state closed def set_half_open(self): self.state half-open self.last_half_open_ts time.time() def can_probe(self) - bool: if self.state half-open and time.time() - self.last_half_open_ts self.cooldown: return True if self.state closed: return True return FalseModelChannel 是熔断器的最小实现。状态机三态closed 正常open 熔断half-open 探测。这里有个容易踩坑的点实际业务里通道恢复探测不能太频繁所以我为 half-open 状态加了冷却时间判断只有距离上次探测足够久才允许再次放探针流量。调用主逻辑包含超时、重试、降级# model_router.py (continued) import requests def call_with_retry(channel: ModelChannel, payload: Dict) - Dict: if channel.is_open(): raise RuntimeError(fchannel {channel.name} is open) if not channel.can_probe(): raise RuntimeError(fchannel {channel.name} cooling) last_error None for attempt in range(channel.max_retries 1): try: resp requests.post( channel.base_url, jsonpayload, timeoutchannel.timeout, ) if resp.status_code 200: result validate_response(resp.json()) if result[valid]: channel.record_success() return result[data] else: last_error finvalid response: {result[reason]} else: last_error fhttp {resp.status_code} except requests.Timeout: last_error timeout except Exception as e: last_error str(e) # brief sleep before retry time.sleep(0.5 * (attempt 1)) channel.record_failure() raise RuntimeError(fchannel {channel.name} failed: {last_error})这段代码里有几个细节值得强调。第一每次重试前用短暂退避避免在上游抖动时重复并发冲击。第二response 即使返回 200也要经过 validate_response 的内容校验这个校验就是前面说的结果检查环节。第三状态记录是在整个调用结束后统一处理的不是每次 200 都刷新避免偶发成功把熔断器“洗白”。4.3 自愈与降级的状态流转有了通道对象之后还得写一个控制器把主通道、备用通道和降级逻辑串起来。我以一个两挡场景为例请求先走快挡快挡失败后走慢挡慢挡也失败则返回可读的错误提示。def route_and_call(request: Dict, router_config: Dict) - Tuple[Dict, str]: fast_channel ModelChannel(fast, router_config[channels][fast]) slow_channel ModelChannel(slow, router_config[channels][slow]) features extractor.extract( request[user_input], request.get(system_prompt, ), request[context_tokens], ) if features[hard_hit] fast: target fast_channel backup slow_channel elif features[hard_hit] slow: target slow_channel backup fast_channel else: score scorer.score(features) if score router_config[router][slow_score_threshold]: target slow_channel backup fast_channel else: target fast_channel backup slow_channel for channel, is_backup in [(target, False), (backup, True)]: if channel.is_open(): continue try: result call_with_retry(channel, build_payload(request, channel.name)) return result, f{backup- if is_backup else }{channel.name} except RuntimeError: continue return {error: all channels unavailable}, none这段逻辑最微妙的地方在于主通道失败后代码是用 continue 跳出并且立刻尝试备用通道而不是把异常抛给上层。这样用户侧看到的最坏结果只是多等待了一个超时周期而非直接得到一个错误页。Pi Agent 的线上实践中这种降级路径承担了大约 4% 的流量成功把故障率压到了极低。4.4 参数调优与成本收益验证代码能跑只是第一步参数怎么设才真正考验功力和业务理解。我的经验是先按“安全保守”原则初始再按“误差容忍”逐步收紧。保守的意思是初期宁可多花成本也不让快挡的误判影响效果所以阈值设得高一些跑一段时间之后从日志里统计快挡失败率如果稳定在很低水平再逐步下调阈值。拿 slow_score_threshold 举例。我最初定的是 12跑了两周后统计判定为慢挡的请求里约 10% 最终实际结果是快挡也能完成的简单任务。于是我把阈值降到 10省了一部分成本又过了两周再次统计误判升到 6%效果仍在可接受范围。调参的过程就是这样一步步逼近最优。成本收益验证很容易做直接用两条数据对比接口调用费用汇总和任务成功率。模型路由稳定运行后的 Pi Agent在保持任务成功率持平的前提下模型调用成本下降了约 55%同时快挡的平均响应时间比慢挡短了 70%。这组数字比任何理论推导都更能说明问题。5. 常见问题与排查技巧实录5.1 高频故障速查表先整理一张问题速查表都是 Pi Agent 运行期间真实踩过的坑排查思路比较通用。问题现象可能原因排查路径解决办法快挡模型经常超时轻量模型并发能力不足或网络抖动看通道日志算 P95 响应时间快挡加一个同等级备用实例或调低单请求超时尽早触发降级慢挡调用量异常增长路由打分阈值过低简单任务被误判抽样对比同请求的快慢挡输出质量上调 slow_score_threshold或往硬规则里加更多的“快速”关键词熔断后大量请求失败备用通道容量不够或探测窗口太短看熔断恢复日志确认备用通道是否也超载加长 half-open 冷却时间或把备用通道扩容降级响应时间反而更长主通道重试次数过多熔断前消耗大量时间记录 call_with_retry 的每次耗时减少主通道最大重试次数首个请求失败后立刻切换备用内容校验频繁损坏校验规则过严或模型输出格式不稳定看 validate_response 的失败原因占比放宽非关键字段校验或改用 schema 解析器做容错匹配5.2 三个容易踩的坑第一个坑是重试和熔断的参数打架。我在初期设置的是“每个通道重试 3 次熔断阈值 5 次”结果发现一个通道在熔断前就能消耗掉大量时间。原因很简单一次请求重试 3 次连续两个请求就把失败计数累到 6 了第三个请求进来时通道直接熔断但前两个请求的用户已经在等待中流失。最好的做法是让单批请求的消耗时间小于熔断窗口的一半比如重试 2 次 超时 5 秒单请求最多消耗 15 秒这样熔断触发前的用户体验还在可控范围。第二个坑是半开状态探针完全复用真实流量。刚刚实现完熔断器时我以为半开状态下直接把真实请求放进去就是探针省事又好用。结果发现这会导致一个问题真实流量里凑巧都是简单请求探针全部成功通道被快速恢复但通道实际情况仍然不稳定恢复后马上又有复杂请求进来再次触发熔断形成来回抖动的“熔断震荡”。后来我在半开状态下的探针流量里混合了预设的探针用例确保探针能覆盖一定比例的复杂请求抖动情况才基本消失。第三个坑是日志字段不够导致无法复盘。最初 Model Router 的日志只记录了“最终走了哪个通道”没有记录“最初路由判定是什么”。等到想分析误判率时才发现根本统计不出来——因为我把判定和最终调用的通道混为一谈完全没意识到降级链路会改变实际调用通道。后来改了日志结构增加“route_decision”和“actual_channel”两个独立字段才把两类数据分清楚调参逻辑也终于有了基石。还有一个值得分享的小技巧跟稳定运维相关给 Model Router 加一个“手动干预开关”。线上出了紧急状况时你肯定不希望在降温池里慢慢等状态机恢复而是能一键把某条通道强制置为 closed 或 open。这个开关平时不用但一旦用上能省掉你半夜爬起来改配置再发布的大把时间。写到这里Pi Agent 的自动挡方案就算完整讲透了。这套 Model Router 说白了就是在成本和稳定性之间找平衡快挡省钱保响应慢挡保效果兜底故障自愈机制则是整个系统的安全网。从我个人实践来看路由规则的迭代一定要基于日志反馈的数据不要拍脑袋调阈值熔断和重试的参数要反复测算互相之间的影响可观测性是所有优化的大前提。如果大家也在做类似的 Agent 项目完全可以直接照这套思路去实现一个最小版本跑起来之后再看数据逐步调优。后续我大概率还会在 Pi Agent 里加上基于语义嵌入的自动流量分类让路由判定更智能一些到时候有空再写一篇文章跟大家分享。
阅读完成 · 觉得有帮助？
