从 PilotDeck 源码看 AI Agent 路由架构的设计方法论

当我们谈 AI Agent 时，常常忽略了一个关键问题：请求来了，谁来响应？

一个对话请求应该分配给哪个模型？简单问答用小模型，复杂推理用大模型。一个 Agent 实例故障了，如何自动切换到备用实例？上下文太长了，怎么压缩才能保留关键信息？

这些问题，PilotDeck 给出了一套完整的工程答案。

一、为什么需要路由？

传统的 LLM 调用是"一对一"的：来了请求，用固定的模型处理。但生产环境的 AI Agent 需要：

• 根据任务复杂度选择合适的模型（省钱）
• 多个模型形成降级链，一个失败了自动切换下一个
• 跨 Agent 的上下文压缩和管理
• 子 Agent 的编排和状态维护

这时候，一个简单的 if-else 已经不够用了，需要一个专业的路由层。

二、PilotDeck 路由的整体架构

用户请求
     ↓
RouterRuntime.decide()   ← 决策：用什么模型？
     ↓
RouterRuntime.execute()  ← 执行：调用模型，返回流式事件
     ↓
AgentLoop                ← 循环：工具调用、上下文管理、重试

RouterRuntime 负责"用什么模型"，AgentLoop 负责"模型怎么用"。两者职责分离，各司其职。

三、决策链：优先级层层递进

decide() 方法中，路由决策按优先级层层传递：

Custom Router（插件自定义）→ Scenario（显式指定）
→ TokenSaver Sticky（会话粘性）→ TokenSaver Classification（智能分类）
→ Default（默认兜底）

每一层都是独立的策略，失败时传递到下一层。这种设计叫做责任链模式。

1. Custom Router：插件化的魔力

通过 PilotDeckCustomRouter 接口，可以插入任意的路由策略：

export type PilotDeckCustomRouter = {
  id: string;
  decide(input: CustomRouterDecideInput): Promise<Partial<RouterDecision> | undefined>;
};

外部扩展可以实现自己的路由逻辑，通过 CustomRouterRegistry 注册进来。RouterRuntime 在决策时优先调用自定义路由。

2. TokenSaver：用小模型为任务分类

TokenSaver 是 PilotDeck 最有趣的设计之一。它用一个小模型（Judge）来判断当前任务应该用哪个级别的模型：

用户消息 → Judge 模型 → tier name → 对应模型

四层分类：

Judge 模型只需要返回 tier 名称，计算量极小，却能显著节省成本。

Smart Continuation 优化： 用户只发"继续"、"好的"，这类短确认消息不应该被重新分类，而是继承上一轮的 tier。因为小模型容易把这类消息误判为"simple"。

3. Session Sticky：会话粘性机制

同一个会话的连续请求，如果内容高度相似，每次都调用 Judge 分类是浪费。SessionRouterStore 用内存缓存保存会话状态：

get(sessionId, isSubagent) {
  // 检查 TTL 是否过期
  // LRU 提升最近访问的条目
  return this.map.get(key)?.state;
}

• 支持 TTL（默认 60 分钟）
• 支持 LRU 淘汰（默认容量 500）
• 主/子 Agent 分离存储（key = sessionId 或 sessionId:sub）

四、执行层：容错与恢复

execute() 方法不仅要执行模型调用，还要处理各种故障场景。

1. Fallback 降级链

attempts = [
  主模型,
  ...fallbackPlan.attempts  // 配置的多级降级模型
]

当主模型失败时，按顺序尝试降级模型。但降级不是盲目的，有些错误适合降级，有些不适合：

function isFallbackEligible(error) {
  if (error.code === "invalid_tool_arguments") return true;  // 可自修复
  if (!error.retryable) return false;  // 非重试错误不降级
  if (error.code === "prompt_too_long") return false;  // 长度问题降级也解决不了
  return true;
}

2. Zero-Usage Retry：空响应检测

有时候模型返回了，但内容是空的（特别是流式响应中途出错）。Zero-Usage Retry 检测这种场景：

function shouldRetryZeroUsage(state) {
  if (state.observedFinish &&       // 收到 message_end
      !state.observedAnyText &&     // 但没有任何内容
      totalTokens === 0) {          // 且 Token 数为 0
    return true;  // 触发重试
  }
}

3. Streaming 防重复机制

这是最体现工程精细度的地方。Fallback 切换模型时，如果已经向用户输出了部分内容，再切换会导致重复输出。PilotDeck 用 hasYieldedContent 标记来解决：

let hasYieldedContent = false;
let pending: CanonicalModelEvent[] = [];

for await (const event of streamAttempt(...)) {
  if (!hasYieldedContent && isContentEvent(event)) {
    // 先 flush 之前 buffer 的 framing events
    for (const queued of pending) yield queued;
    pending = [];
    yield event;
    hasYieldedContent = true;  // 标记：已经有输出了
    continue;
  }
  if (hasYieldedContent) {
    yield event;  // 直接输出
    continue;
  }
  pending.push(event);  // 还没内容，先 buffer
}

五、AgentLoop：循环中的艺术

AgentLoop 是整个系统的核心循环，它的管理非常精细。

1. 两阶段压缩

压缩（Compaction）在路由决策前后各执行一次：

第一阶段：路由决策前（用主 Agent 的默认上下文窗口）
     ↓
第二阶段：路由决策后（用目标模型的上下文窗口）

为什么？因为不同模型的上下文窗口不同。如果主 Agent 配置了 20k token 窗口，但路由决定用 4k 窗口的模型，就需要二次压缩。

2. Circuit Breaker：防止模型卡死

如果连续 3 轮所有工具调用都是 invalid_tool_input 错误，说明模型陷入了某种死循环（如反复生成空参数），这时候应该熔断：

const MAX_CONSECUTIVE_ALL_INVALID_TURNS = 3;
if (consecutiveAllInvalidTurns >= MAX_CONSECUTIVE_ALL_INVALID_TURNS) {
  throw new Error("模型陷入工具调用错误循环，终止执行");
}

3. JSON Self-Correct

模型生成的 JSON 参数有时会格式错误（如缺少引号、尾随逗号）。PilotDeck 会自动检测并让模型重试：

if (error.code === "invalid_tool_arguments" && jsonSelfCorrectCount < 3) {
  messages.push({
    role: "user",
    content: "你上一个工具调用的参数包含无效 JSON，请用有效 JSON 重试。"
  });
  continue;  // 重试
}

最多重试 3 次。

六、CompactionEngine：上下文压缩

当对话历史太长时，CompactionEngine 用一次额外的模型调用来总结历史：

保留最近 35% 的消息 → 用模型总结前面的内容 → 插入边界标记

总结后的结构：

boundaryMarker → summary → keep → attachments → hookResults

工具对的完整性检查也很重要：如果被总结的消息中有一个工具调用，但它的结果在保留部分，就会产生悬垂引用。CompactionEngine 会剥离这些不成对的工具调用和结果。

七、设计模式总结

八、写在最后

看 PilotDeck 的代码，最大的感受是工程精细度。

很多框架设计一个功能，画个架构图就完了。PilotDeck 的每个功能都有完整的边界情况处理：空响应怎么处理、Token 预算超了怎么处理、连续错误怎么熔断、流式输出怎么防止重复……

这些不是过度设计，而是生产级系统的必备能力。当你的系统每天处理成千上万的请求时，每一个边界情况的处理质量，决定了系统的稳定性和成本。

这也是 AI Agent 框架从"玩具"走向"产品"的关键一步。

相关阅读：

• PilotDeck GitHub：https://github.com/OpenBMB/PilotDeck^[1]
• 本文对应的代码解析：router/ 目录下的模块化设计

如果你对 AI Agent 的路由架构、上下文管理、容错机制有更多想法，欢迎交流。

引用链接