企业官网建设流程全解析

接到一个很有意思的工程实践类标题：给 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，然后降级到 D2，D2 超时后又重试，最后成功”，整个链路一目了然。

我强烈建议，任何做模型路由的项目，都把可观测性当作第一天就要做的事，不要等系统出问题再去补。没有观测数据的自愈机制，等于闭着眼睛开车。

4. 实操过程：从零搭建一台两挡 Model Router

4.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, reason="circuit_open") try: # 首次调用 result = self.llm.call(model, task, timeout=timeout) 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, timeout=timeout) self.executor._record_error(model, True) return result except Exception as e2: return self._fallback(task, backup_gear, reason="retry_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, timeout=backup_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 不是一个“搭完就完事”的功能，它更像一个持续调优的控制系统。两挡设计给了系统最基本的灵活性，故障自愈给了系统稳定性，但真正让它长期可靠运行的，是持续的数据分析和策略迭代。每一次路由误判、每一次降级失败、每一次熔断打开，都是优化路由策略的线索。没有这些数据，再精巧的设计也只是空中楼阁。