最近在打磨一个面向一线业务的智能助理项目(内部代号叫 Pi Agent)时,我有一半的时间都在跟模型调用较劲。这个 Agent 要处理的问题五花八门,有的请求就三五个字,比如“查一下订单状态”“明天什么天气”,有的请求却是一大段需求描述,比如“把这段聊天记录里的所有待办事项整理出来,按优先级排一下,顺便生成一个提醒列表”。最开始我偷懒,把所有请求都塞给同一个“最强模型”,效果倒是稳,但月底看账单的时候心在滴血,响应速度也一直被人吐槽;后来我又试图全都切到便宜的轻量模型,结果遇到稍微绕一点的逻辑问题就开始胡说八道。折腾了几轮之后,我意识到问题根本不在“选哪个模型”,而在于“有没有根据请求动态选模型的能力”。于是花了一周时间,在 Pi Agent 和模型服务之间加了一层 Model Router,一个两挡的模型路由器,分别对应经济挡和性能挡,同时还做了故障自愈,把整个模型调用从手动挡改成了自动挡。
1. 项目背景与“自动挡”思路
1.1 单模型调用的三个痛:成本、延迟、不稳定
任何做过 Agent 的人都绕不开一个问题:请求到底交给哪个模型?最省心的答案是“全部交给能力最强的那个”,这也是绝大多数原型项目的默认做法。但这个做法在真实业务里撑不了多久。
首先是成本。我在内部做过一次粗略统计:如果所有请求都走旗舰模型,单次调用成本大概是走轻量模型的十倍到二十倍。当时 Pi Agent 的日均请求量在几万级别,一大半是“订单状态”“营业时间”“退换货政策”这类简单查询,直接用大模型回答,等于每天把大量预算烧在了重复劳动上。其次是延迟。旗舰模型平均首字响应时间在 1.2 秒左右,轻量模型能压到 0.4 秒。对用户来说,1 秒和 2 秒的体验差距非常明显,尤其是客服场景,用户等得越久越容易暴躁。最后是不稳定。越强的模型往往承担着越多的并发压力,超时、限流、5xx 错误都是家常便饭,而轻量模型在复杂任务上的胡说八道率又高得离谱。单模型方案意味着这些痛只能暴露在用户面前,没有任何中间缓冲。
后来我做过一张对照表,把单模型方案和路由方案放在一起看,差别非常直观。单模型等于把所有鸡蛋放一个篮子,路由方案则是根据请求类型分拣,再根据健康状态动态切换。
| 维度 | 单一大模型 | 两挡路由 |
|---|---|---|
| 简单请求响应 | 慢,平均 1.2s+ | 快,平均 0.4s 左右 |
| 简单请求成本 | 高 | 低 |
| 复杂请求表现 | 好 | 好(走性能挡) |
| 故障影响范围 | 单一模型故障=全局不可用 | 单挡故障可自动切换 |
| 运维复杂度 | 低 | 中 |
1.2 从“手动挡”到“自动挡”:路由层到底解决什么问题
手动挡开车时,什么时候换挡完全靠司机自己判断,新手很容易把转速拉到红区才想起换挡,老手虽然开得顺,但注意力始终得放在路况和转速上。自动挡则是把换挡逻辑集成到了系统内部,踩油门爬坡时自动降挡,匀速巡航时自动升挡,司机只需要控制方向。
Pi Agent 原本的模型调用方式就是个典型的手动挡:开发者手动配置一个模型,Agent 不管来什么请求都按同一个挡位跑。Model Router 要做的事,就是把“换挡”这个动作从开发者手里收走,变成一个独立的决策模块。它做的事可以拆成四层:请求分流、挡位内模型选择、健康状态感知、故障转移。请求分流指根据输入内容判断当前请求适合哪个挡位;挡位内模型选择是指某个挡位下可能配置了多个模型,根据实时状态选最合适的那个;健康状态感知是持续记录每个模型的成功率、延迟、限流情况;故障转移则是在感知到异常后,自动把流量切到备用通道。
这四层合在一起,就构成了 Agent 和模型服务之间的缓冲带。有了这个缓冲带,上层业务代码不需要关心模型服务今天抽不抽风,只需要调用路由器,拿回一个结果。
1.3 为什么是“两挡”而不是三挡四挡
一开始我也想过要不要做三挡:高速挡处理超复杂任务,性能挡处理一般任务,经济挡处理简单任务。但真拆解了一遍业务后发现,Pi Agent 的请求基本可以按复杂度划分成两大类,一类是信息查询、单轮问答、意图判断,另一类是多步推理、代码生成、长文本加工。所谓“超复杂任务”,其实在第二类里只占很小比例,把它们也塞进性能挡,浪费不了多少成本,反而避免了规则模型的复杂度爆炸。
选择“两挡”还有一个可运维性考虑。挡位越多,路由判定逻辑就越难写。挡位之间的边界模糊,误判率会快速上升,而且一旦误判,排查成本很高。两挡的好处是决策空间小,规则模型很容易调到一个稳定的状态,遇到拿不准的请求就统一归入性能挡,误判带来的损失也最小。任何方案如果让我自己都搞不清楚边界,上线后必然变成黑盒,所以我最后坚持两挡设计,没有继续往上加。
2. 两挡设计:经济挡与性能挡怎么分工
2.1 经济挡:给简单请求用的低成本通道
经济挡的核心定位是“快速、便宜、够用”。它的服务对象是那些不需要深层推理的请求:天气查询、订单状态、地址确认、商品属性问答、常见政策解释、简单的情感安慰,甚至包括对话里的“嗯嗯”“好的”“继续”这类陪聊式回复。
选择模型时,我优先看三个指标:首字延迟、单 token 价格、并发上限。延迟最好在 300 毫秒以内,价格尽量是性能挡的十分之一以下,并发上限别太低,否则高峰期会变成瓶颈。参数配置上,经济挡的 temperature 我固定得很低,基本在 0.1 到 0.3 之间,max_tokens 控制在 300 左右,超时时间给到 5 秒。这么做是因为简单任务需要的是稳定和一致,而不是天马行空的发挥,把创造力参数压低能明显减少胡编乱造。
经济挡还需要单独的 prompt 模板。如果直接复用性能挡的提示词,轻量模型容易被长指令搞糊涂,输出质量和格式都不稳定。我给 Pi Agent 的经济挡单独写了一套极简提示词:先设定角色,再给三到五条输出规则,最后用 JSON 固定返回字段。实测下来,格式错乱率比直接复用复杂提示词降低了六成以上。
2.2 性能挡:给复杂任务用的“大马力”通道
性能挡服务的是需要“认真思考”的请求:多轮对话中的逻辑推演、代码生成与调试、长文档的摘要和结构化抽取、工具调用的规划、需要结合上下文综合判断的任务。这些请求如果交给经济挡,轻则答非所问,重则生成完全不可用的内容。
性能挡的模型选择走的是另一个逻辑,我优先看推理能力和上下文窗口长度。这里的“推理能力”不是靠跑分看的,而是拿 Pi Agent 真实场景里的几十个困难样本去测:比如“用户连续修改了三次需求,最后还要兼容第一次的约束条件”这种场景,模型能不能正确理解。上下文窗口决定了它能不能吞下较长的会话历史,窗口太小的模型会在多轮场景里频繁丢失信息。
性能挡的配置跟经济挡是两个方向:temperature 在 0.6 到 0.8 之间,保留一定的创造性,max_tokens 给到 4096,超时时间放宽到 30 秒。代价是成本高、响应慢,所以它必须被路由判定限制住,只处理真正需要它的请求,而不是让用户随随便便就把流量带进来。
2.3 路由判定:谁来决定挂哪个挡
我把路由判定做了三个层次,优先级从高到低:硬性关键词信号、上下文信号、复杂度评分。
硬性关键词信号比较好理解,比如用户输入里出现“帮我写”“怎么实现”“写一段代码”“分析一下”“规划”等,基本可以确定是复杂任务,直接进性能挡。这里有个技巧:关键词不能列太长,列太长反而容易误伤,重点抓住“写、分析、规划、为什么”这几类高频动作词就够了。
上下文信号用来处理多轮对话里的隐性需求。单看当前这句话可能很简单,比如用户说“那再来一个”,但结合前面几轮聊天记录,他可能是在要求持续生成代码。所以我会检查会话轮数、累计文本长度和最近一条消息是不是在追问复杂主题,只要命中,就升挡。
复杂度评分是最后一个托底策略。我把输入长度、问句模式、是否包含多个意图等特征换算成一个 0 到 1 之间的分数,超过阈值就进性能挡。这个评分可以做得非常简单:
def decide_gear(user_input, context): # 第一层:硬性关键词信号 power_keywords = ["帮我写", "怎么实现", "写一段", "分析", "规划", "为什么", "生成代码"] text = user_input.strip() for kw in power_keywords: if kw in text: return "power" # 第二层:上下文信号 history = context.get("messages", []) if len(history) >= 4: return "power" total_chars = sum(len(m.get("content", "")) for m in history) if total_chars + len(text) > 800: return "power" # 第三层:复杂度评分 score = 0.0 if len(text) > 200: score += 0.4 if any(ch in text for ch in ["为什么", "如何", "怎么"]): score += 0.3 if "并且" in text or "同时" in text or "," in text: score += 0.2 return "power" if score >= 0.6 else "fast"判定逻辑整体思路是“宁可高估,不要低估”。拿不准的请求进性能挡,最坏的结果是多花一点钱;但如果低估了,把复杂请求送进了经济挡,用户看到的就是一堆空话和错话,体验损失远大于成本损失。
3. 故障自愈机制:不能让一个坏挡拖垮整个 Agent
3.1 常见故障类型与影响面分析
在真实服务里,模型 API 的故障形态比想象中多。我整理了一张内部排查时用的故障表:
| 故障现象 | 常见原因 | 对业务的影响 | 能否重试 |
|---|---|---|---|
| 请求超时 | 模型服务负载高、网络抖动 | 用户等待过久,响应失败 | 可重试 |
| 429 限流 | 并发触发配额限制 | 请求被拒,流量丢失 | 可重试,但需退避 |
| 5xx 服务错误 | 模型服务异常或过载 | 返回错误信息 | 视情况重试 |
| 空响应/格式错乱 | 模型产生空内容或非预期格式 | 下游解析失败 | 换模型重试 |
这里容易忽略的一点是,故障不一定是“完全不可用”。很多时候模型服务是部分劣化:比如成功率从 99% 掉到 85%,或者延迟从 1 秒涨到 5 秒。纯看报错数量发现不了这种劣化,只有持续统计成功率、平均延迟、P95 延迟这些指标,才能把“亚健康”状态识别出来。
3.2 健康检查与滑动窗口统计
我给每个挡位都配了一个滑动窗口,记录最近 100 次调用的成功率。每次调用结束后,结果会被塞进队列,同时把超过窗口时长的旧记录踢出去。这样既能反映最近一段时间的真实状态,又不会因为几分钟前的突发故障而长时间误伤一个已经恢复正常的模型。
判断逻辑是:窗口内调用次数达到 20 次以上、失败率超过 20% 时,标记该挡位不健康;连续失败达到阈值时,直接打开熔断开关。健康检查和路由判定是并行执行的,路由判定负责选挡,健康检查负责确认这个挡位当前能不能用。如果健康检查发现性能挡不健康,即使复杂度评分判定要进性能挡,也会自动降级到经济挡或直接走兜底文案。
3.3 三级自愈策略:重试、降级、熔断
自愈机制我分成了三级,每一级解决不同层次的问题。
第一级是同挡内重试。当调用返回超时或者 5xx 错误时,如果系统当前负载不高,我会在同一个挡位内重试一次。这里要注意,重试不能盲目做,如果一个请求已经消耗了 15 秒才超时,再重试一次就是又搭进去 15 秒,用户早就走了。所以同挡重试只适用于快速失败场景,比如 429、快速返回的 5xx,而不适用于已经耗时过半的超时错误。
第二级是跨挡降级。性能挡挂了,就把请求降级到经济挡;经济挡挂了,就降级到默认兜底模型或者直接返回缓存结果。降级不是简单地换一个模型重新调用,它还需要换一套 prompt,因为同一个 prompt 在小模型上可能产生完全不同的效果。我在实现时给每个挡位都配了一套独立的 prompt 模板,降级时把整个执行上下文一起切换,而不是只换一个 base_url。
第三级是熔断。如果一个挡位连续失败超过 10 次,我直接打开熔断开关,接下来 60 秒内任何请求都不再尝试调用这个挡位,全部走降级通道。60 秒后熔断器进入半开状态,放一个探测请求进去,成功就恢复,失败就继续熔断。
下面是简化版的自愈核心代码:
class FailoverEngine: def __init__(self, router): self.router = router self.stats = {} self.circuit_open_until = {} def call_with_failover(self, gear_name, messages): gear = self.router.get_gear(gear_name) key = gear_name # 熔断打开:直接降级 if time.time() < self.circuit_open_until.get(key, 0): return self.router.fallback(gear_name, messages), "fallback" try: result, latency = gear.call(messages) self._record_success(key) return result, gear_name except TimeoutError: self._record_failure(key) if self._should_open_circuit(key): self.circuit_open_until[key] = time.time() + 60 return self.router.fallback(gear_name, messages), "fallback" except RateLimitError: self._record_failure(key) return self.router.fallback(gear_name, messages), "fallback"3.4 自愈恢复:不要永远踩死一个挡位
熔断时间到期后,系统并不会自动把所有流量切回原挡位,而是先进入“半开”状态。半开状态下,我会放 1% 到 5% 的探测流量去调原挡位,如果这批探针请求的成功率恢复正常,就逐步把流量切回去;如果继续失败,熔断器重新打开,进入下一轮冷却。
这里最忌讳的是“熔断结束后直接一把梭”。我踩过这个坑:某个模型服务故障恢复后,瞬间几百个请求同时打进去,又把刚恢复的服务打崩了。后来我加了流量恢复梯度,每 30 秒恢复 20% 的流量,四到五个周期才完全恢复,服务整体稳定了很多。自愈的真正含义是让系统在无人干预的情况下平稳渡过故障期,而不是在恢复的那一瞬间制造第二次故障。
4. 实操落地:把 Model Router 挂到 Pi Agent 上
4.1 路由模块的接口设计与目录结构
我在 Pi Agent 项目里新增了一个独立的 router 模块,尽量不让路由逻辑污染 Agent 的主体代码。目录结构非常简单:
pi_agent/ ├── agent.py # Agent 主流程 ├── config/ │ └── router_config.py # 挡位、模型、超时等配置 ├── router/ │ ├── gear.py # 挡位定义与调用封装 │ ├── router.py # 路由判定与挡位选择 │ └── failover.py # 故障自愈与熔断逻辑 └── utils/ └── logger.py # 结构化日志接口设计上,我尽量让外部只暴露两个方法:route()负责根据输入和上下文选挡,call_with_failover()负责带故障自愈的调用。Agent 主流程不需要关心底层模型变化,只要拿到结果就行。
4.2 核心 Router 类的实现细节
Router 类的核心职责是把“路由判定”和“模型调用”解耦。我在初始化时从配置文件里读取各挡位的模型参数,然后在运行时动态决定走哪个挡位。
class ModelRouter: def __init__(self, config): self.config = config self.gears = {} for name, conf in config.items(): self.gears[name] = Gear(name, conf) def route(self, user_input, context): decision = decide_gear(user_input, context) gear = self.gears[decision] if not gear.is_healthy(): return "fallback" return decision def fallback(self, gear_name, messages): # 降级策略:如果性能挡失败,尝试经济挡;如果经济挡失败,返回固定文案 if gear_name == "power": return self.gears["fast"].call(messages) return {"content": "当前服务繁忙,请稍后再试。", "fallback": True}这里有一个设计细节:路由判定和健康检查分开做,判定走规则,健康检查走统计数据。两者结合起来,才能避免“规则说进性能挡,但性能挡已经挂了”的尴尬局面。
4.3 与 Agent 请求链路的整合方式
Pi Agent 的主流程其实就是一个循环:接收用户输入、组合上下文、调用模型、把结果回传。接入 Model Router 后,主流程几乎没变,只是把“直接调模型”换成了“通过 Router 调模型”。
def agent_handler(user_input, session): context = session.get_context() router = session.router # 1. 路由判定 target_gear = router.route(user_input, context) # 2. 带故障自愈的模型调用 response, used_gear = router.call_with_failover(target_gear, context) # 3. 审计日志 log(f"route={target_gear}, actual={used_gear}, user={session.user_id}") # 4. 返回结果 return response接入之后,最大的变化是日志里多了一个actual字段。以前出问题我只能看到“这个请求失败了”,现在可以看到“这个请求原本打算走性能挡,但因为超时降级到了经济挡”。这个字段在后来的问题排查里帮了大忙。
4.4 监控指标与告警阈值
模型路由本身不产生价值,只有监控能证明它的价值。我在上线后重点盯四个指标:各挡位调用量、失败率、平均延迟、降级率。很简单,直接通过结构化日志上报到日志平台,再做简单的聚合。
告警阈值我调过好几轮。第一版是“失败率超过 1% 就告警”,结果每天醒来都被一堆告警轰炸。后来改成“失败率超过 10% 或连续五分钟降级率超过 30% 才告警”,噪声一下子小了很多。做自愈系统的核心目标不是“永不故障”,而是“故障时尽量不打扰用户、不打扰开发”,所以告警阈值一定要跟故障影响真正挂钩,而不是跟技术指标的绝对值挂钩。
5. 踩坑记录与性能优化建议
5.1 坑一:路由规则过分依赖关键词,误判率高得离谱
第一版路由规则里我写了十几个关键词,结果发现“分析”“怎么”这类词在客服场景里出现频率极高,很多用户随口说一句“请问怎么退货”就被送进了性能挡,成本立刻涨了回去。后来我把规则拆成两层:动作类关键词和场景类关键词,只有动作类关键词才直接触发性能挡,场景类关键词只用来给复杂度评分加分。改动之后,性能挡调用比例从 70% 降到了 35% 左右,成本明显改善。
5.2 坑二:故障自愈造成雪崩,备用模型也被打爆
有次性能挡模型出现故障,自愈机制把所有请求都切到了经济挡,结果经济挡模型原本只有单并发限制,瞬时流量涌入后也出现大量超时。这个问题的本质是“降级路径没有做限流”。后来我在降级调用上套了一层简单的并发限制器,最多允许 50 个降级请求同时处理,超出部分直接返回排队提示。降级是兜底手段,不是替代方案,必须让它“慢一点但可用”,而不是“快但全崩”。
5.3 坑三:故障转移后上下文丢失,回复牛头不对马嘴
有一次性能挡超时降级到经济挡,结果经济挡拿到的是性能挡的完整对话历史,好几千字的上下文把它的脑子彻底撑糊涂了,回复了一堆完全无关的内容。问题出在降级时没有做上下文裁剪。后来我在降级逻辑里加了一个动作:如果目标挡位是经济挡,就先把对话历史压缩成最近两轮用户消息和最后一条系统回复,再传给模型。这个处理虽然损失了一部分上下文,但至少保证了输出是靠谱的。
5.4 压测与效果验证:这样调整才算数
自愈机制不能用“看起来没报错”来验证,必须用真实流量的回放测试。我把 Pi Agent 某天的真实请求做了脱敏处理,然后分别用“单一大模型”和“两挡路由”跑了一遍,对比三个指标:平均首字延迟、总成本、错误率。
| 指标 | 单一大模型 | 两挡路由 | 变化 |
|---|---|---|---|
| 平均首字延迟 | 1.28s | 0.52s | 降低 59% |
| 千次请求成本 | 较高 | 约前者的 1/4 | 降低 75% 左右 |
| 错误率 | 2.1% | 0.8% | 降低 60%+ |
回放数据验证了路由方案的收益,但我也提醒自己:指标变好不等于系统可靠。我连续观察了两周,每天看一下降级日志和熔断日志,确认自愈机制在真实压力下真的生效,才敢把这套方案完全交给业务。
实际体会与一个百试不爽的小技巧
这套 Model Router 在我这边跑了大半年,最大的感触是:所谓“故障自愈”,最重要的不是技术方案有多完美,而是它对业务的影响足够小。用户不会关心你的模型服务今天怎么了,他们只会在意自己的问题有没有被解决。路由器存在的意义,就是把底层服务的各种变化挡在外面,给用户一个稳定的回答。
最后分享一个后续扩展时很有用的小技巧:如果有一天你想从两挡升级成多挡,不要直接改判定逻辑,先在路由日志里加一个expected_gear字段,把真实人工标注的结果和路由判定结果做对比。连续记录两周,你会很清楚地看到哪些请求被误判、被低估,这些数据比任何理论分析都更能指导挡位设计。与其反复调整规则本身,不如先把判定数据沉淀下来,很多优化目标自然就浮出水面了。