手搓生产级 AI Agent 系统（13）：别让所有任务都上旗舰模型——路由、预算和降级

文章摘要

前十二篇把生产级 Agent 的执行链路基本补齐了：任务理解、Tool Calling、状态、Memory、Planner-Executor-Reviewer、Checkpoint、人工审批、多 Agent、安全沙箱和发布评测。

系统走到这里，会出现一个非常现实的问题：它能跑了，但可能太贵。

一个 Agent 请求背后可能包含规划、检索、三四次工具调用、子 Agent、Reviewer 和最终生成。如果这些节点全部默认使用最强模型，成本会很快变成产品上线后的主要约束；反过来，如果为了省钱全部切到低价模型，复杂任务成功率和异常恢复能力又会下降。

这一篇不再讨论“哪个模型最好”，而是实现一套模型调度层：根据任务类型、风险、上下文、失败历史和预算选择模型；为每个 Run 预留成本；允许低价模型先跑、失败再升级；高风险任务直接锁定高能力模型；Provider 故障时按照能力而不是名字降级；所有路由结果进入评测闭环，最终用真实 Task Success 和 Cost per Successful Task 调整策略。

先看一个很容易烧钱的 Agent

用户请求：

读取三份合同，比较差异，
查询供应商历史履约数据，
生成采购建议并发给负责人。

一个未经成本治理的流程可能是：

旗舰模型：理解任务
旗舰模型：生成计划
旗舰模型：合同A
旗舰模型：合同B
旗舰模型：合同C
旗舰模型：供应商分析
旗舰模型：Reviewer
旗舰模型：最终邮件

8 次模型调用。

如果中间两个步骤失败重试：

10—12 次

再加长合同 Context，单请求成本非常容易失控。

而实际上，里面很多任务不需要同一档模型。

更合理的拆法：

任务分类：低成本模型
合同结构抽取：低成本/中档模型
复杂差异判断：高能力模型
供应商数据整理：低成本模型
最终风险判断：高能力模型
邮件润色：中档模型

这就是模型调度层存在的意义。

一、不要让业务代码直接写模型名

最糟糕的写法：

chatClient
    .model("gpt-5.6-sol")
    .prompt(...)

然后散落在几十个 Service 里。

模型升级、价格变化、Provider 故障时，全项目改。

业务代码应该请求“能力”：

public record ModelRequirement(
        TaskType taskType,
        CapabilityLevel capability,
        RiskLevel risk,
        int contextTokens,
        boolean toolsRequired,
        boolean visionRequired,
        Duration latencyTarget,
        BigDecimal maximumCost) {
}

路由层再决定具体模型。

二、模型注册表

public record ModelProfile(
        String profileId,
        String provider,
        String model,
        CapabilityLevel capability,
        long contextWindow,
        boolean toolCalling,
        boolean vision,
        BigDecimal inputPricePerMillion,
        BigDecimal outputPricePerMillion,
        Set preferredTasks,
        RiskLevel maximumAutonomousRisk,
        ModelHealth health) {
}

配置示例：

models:
  - profile-id: fast-agent
    provider: google
    model: gemini-3.7-flash
    capability: STANDARD
    input-price-per-million: 0.75
    output-price-per-million: 3.75

  - profile-id: balanced
    provider: openai
    model: gpt-5.6-terra
    capability: ADVANCED
    input-price-per-million: 2.50
    output-price-per-million: 15.00

  - profile-id: frontier-openai
    provider: openai
    model: gpt-5.6-sol
    capability: FRONTIER
    input-price-per-million: 5.00
    output-price-per-million: 30.00

  - profile-id: frontier-anthropic
    provider: anthropic
    model: claude-opus-5
    capability: FRONTIER
    input-price-per-million: 5.00
    output-price-per-million: 25.00

价格必须版本化。Gemini 3.7 Flash 当前促销价格到 2026 年底，不能把这套价格永久写死。

三、先做“硬过滤”，再做评分

Router 第一步不是让 LLM 选模型。

先用确定性条件淘汰不合格候选：

public List eligible(
        ModelRequirement req,
        List profiles) {

    return profiles.stream()
            .filter(p -> p.health().isAvailable())
            .filter(p -> p.contextWindow()
                    >= req.contextTokens())
            .filter(p -> !req.toolsRequired()
                    || p.toolCalling())
            .filter(p -> !req.visionRequired()
                    || p.vision())
            .filter(p -> p.capability()
                    .atLeast(req.capability()))
            .toList();
}

这样可以防止模型自己“推荐”一个根本不支持需求的候选。

四、任务难度不要只让 LLM 自评

任务难度可以来自一组信号：

输入长度
文档数量
工具数量
历史失败率
是否需要多步计划
是否开放式判断
是否高风险
是否需要代码执行
是否存在冲突证据

public record TaskComplexity(
        int score,
        int documentCount,
        int expectedToolCalls,
        boolean multiStep,
        boolean ambiguous,
        boolean conflictingEvidence,
        RiskLevel risk) {
}

模型可以参与分类，但最终阈值由程序控制。

五、默认模型不应该是最强模型

我更推荐：

先选满足能力要求的最低成本模型

然后允许升级。

示例：

public ModelProfile selectInitial(
        ModelRequirement req,
        List candidates) {

    return candidates.stream()
            .filter(p ->
                    p.capability()
                            .atLeast(req.capability()))
            .min(Comparator.comparing(
                    this::estimatedUnitCost))
            .orElseThrow();
}

但“最低成本”不是单纯按输入价排序，还要结合历史成功率。

六、真正应该优化的是 Cost per Successful Task

定义：

CPS = 总成本 / 成功任务数

假设：

Model A：
单次 $0.02
成功率 60%

Model B：
单次 $0.04
成功率 95%

如果失败要重试：

A 未必便宜。

所以 Registry 里还要维护线上统计：

public record ModelTaskStats(
        String profileId,
        TaskType taskType,
        long sampleCount,
        double taskSuccessRate,
        double retryRate,
        double p95LatencyMs,
        double averageInputTokens,
        double averageOutputTokens,
        BigDecimal costPerSuccessfulTask) {
}

七、升级式路由

第一次使用低成本模型。

出现以下条件升级：

结构化输出连续失败
Reviewer阻断
Tool循环
证据冲突
低置信
任务超出能力
用户要求更深入分析

public EscalationDecision evaluate(
        AgentStepResult result,
        AgentTrace trace) {

    if (trace.retryCount() >= 2) {
        return EscalationDecision.yes(
                "RETRY_LIMIT");
    }

    if (trace.toolCallCount() >= 8) {
        return EscalationDecision.yes(
                "TOOL_LOOP_RISK");
    }

    if (result.hasBlockingConflict()) {
        return EscalationDecision.yes(
                "EVIDENCE_CONFLICT");
    }

    return EscalationDecision.no();
}

八、升级不是重新跑整个 Agent

这是很重要的一点。

错误：

第5步失败
→换旗舰模型
→从第1步重跑

会浪费：

  • 已完成检索；
  • 已生成 Artifact；
  • 已执行 Tool；
  • Token；
  • 时间。

正确：

Checkpoint
＋Artifact
＋Step Ledger

只升级当前失败节点。

九、Run Budget

public record AgentRunBudget(
        BigDecimal maximumCost,
        long maximumInputTokens,
        long maximumOutputTokens,
        int maximumModelCalls,
        int maximumEscalations,
        Instant deadline) {
}

每个步骤执行前先预估并预留。

十、预算预留

public record ModelCallReservation(
        String reservationId,
        String runId,
        String stepId,
        String modelProfile,
        BigDecimal reservedCost,
        long reservedTokens,
        Instant expiresAt) {
}

执行前：

Estimate
→Reserve
→Call
→Settle

避免并行 Sub-Agent 同时把剩余预算花掉。

十一、如何估成本

public BigDecimal estimateCost(
        ModelProfile model,
        long estimatedInput,
        long estimatedOutput) {

    BigDecimal input =
            BigDecimal.valueOf(estimatedInput)
                    .divide(
                            BigDecimal.valueOf(1_000_000))
                    .multiply(
                            model.inputPricePerMillion());

    BigDecimal output =
            BigDecimal.valueOf(estimatedOutput)
                    .divide(
                            BigDecimal.valueOf(1_000_000))
                    .multiply(
                            model.outputPricePerMillion());

    return input.add(output);
}

生产代码要处理精度和舍入，这里只展示思路。

十二、预算不足时怎么降级

顺序我会这样设计：

1. 取消可选 Sub-Agent
2. 减少 Reviewer 轮次
3. 压缩 Context
4. 使用缓存 Artifact
5. 切低成本模型
6. 缩短输出
7. 返回部分结果并说明缺失

不能降级的：

权限
高风险审批
幂等
安全规则
租户隔离

十三、Provider 故障的 Fallback 不应只按名字

不要：

GPT失败 → Claude
Claude失败 → Gemini

应该按 Capability Contract 找替代：

public List fallbacks(
        ModelProfile failed,
        ModelRequirement requirement) {

    return registry.available().stream()
            .filter(p -> !p.profileId()
                    .equals(failed.profileId()))
            .filter(p -> p.capability()
                    .atLeast(requirement.capability()))
            .filter(p -> p.contextWindow()
                    >= requirement.contextTokens())
            .sorted(
                    Comparator.comparing(
                            this::historicalSuccess)
                            .reversed())
            .toList();
}

十四、Fallback 还要考虑行为差异

不同 Provider 的：

  • Tool Schema；
  • Reasoning；
  • Prompt Cache；
  • Structured Output；
  • 多模态；
  • 拒绝策略；

都可能不同。

所以要有 Provider Adapter：

public interface ModelGateway {

    ModelCallResult call(
            ModelProfile model,
            NormalizedModelRequest request);
}

业务层只看到统一请求。

十五、模型切换时，Prompt 也可能需要 Profile

不要假设一份 Prompt 在所有模型上表现相同。

public record PromptProfile(
        String promptId,
        String baseVersion,
        Map modelOverrides) {
}

但 Override 不要无限分叉，否则无法维护。

优先让 Prompt 保持通用，只对明显差异做少量适配。

十六、健康检查

public enum ModelHealth {
    HEALTHY,
    DEGRADED,
    RATE_LIMITED,
    UNAVAILABLE
}

健康度来自：

错误率
429
P95
超时
Provider状态
连续失败

处于 DEGRADED 的模型可以降低权重，不一定完全摘除。

十七、熔断

短时间连续超时
→打开熔断
→新请求路由其他模型
→半开探测
→恢复

不要让每个 Agent Step 自己独立疯狂重试。

十八、重试和 Fallback 是两回事

Retry：同一个模型再试
Fallback：换另一个模型

错误分类：

可重试

网络抖动
临时 5xx
少量 429

应 Fallback

模型不可用
持续限流
Context不支持
能力不匹配

不应自动重试

权限失败
策略拒绝
错误业务参数
真实副作用结果未知

十九、路由决策必须记录

public record ModelRoutingDecision(
        String decisionId,
        String runId,
        String stepId,
        String selectedProfile,
        List candidateProfiles,
        String reasonCode,
        BigDecimal estimatedCost,
        String policyVersion,
        Instant createdAt) {
}

否则一个月后成本上涨，你不知道为什么全走了旗舰模型。

二十、核心指标

agent_model_call_total{
  profile,
  task_type,
  result
}

agent_model_escalation_total{
  from,
  to,
  reason
}

agent_model_fallback_total{
  reason
}

agent_task_success_rate{
  profile,
  task_type
}

agent_cost_per_success{
  profile,
  task_type
}

agent_budget_exceeded_total

二十一、最容易被忽略的指标：Escalation Rate

如果某类任务：

80% 都从 Flash 升级到 Sol

说明默认路由就错了。

如果：

只有 5% 升级
且最终成功率接近旗舰模型

这才是高性价比路由。

二十二、用离线评测初始化路由

上一篇的评测平台可以直接给 Router 提供先验：

Task Type × Model
→ Success Rate
→ Cost
→ P95

新模型上线时先影子测试，不直接替换默认模型。

二十三、再用线上数据持续校准

路由策略每周或每月更新：

离线 Benchmark
＋线上 Task Success
＋成本
＋人工修改
＋失败分类

不要实时让算法自动改路由权重，尤其高风险系统。

先生成候选策略，再经过 Canary。

二十四、多 Agent 的预算要从父 Run 分配

Run Budget $2
├─Planner $0.15
├─Research $0.50
├─Finance $0.40
├─Legal $0.40
├─Reviewer $0.30
└─Reserve $0.25

子 Agent 不能认为自己有完整 $2。

否则并发一开，总预算马上穿透。

二十五、一定要留安全预算

我会提前保留：

Reviewer
异常恢复
最终输出

的预算。

不要前面研究 Agent 把钱花光，最后连 Review 都跑不起。

二十六、一个实用的初始策略

如果没有任何历史数据，可以先用：

低风险高频：低成本模型
普通企业任务：中档模型
高风险 / 模糊 / 长任务：旗舰模型

再通过实际数据优化。

别一开始就做一个复杂的 ML Router。

规则路由更容易解释和调试。

二十七、上线检查

□ 业务代码不直接写死模型名
□ Model Registry包含能力、价格、健康度
□ 路由先做硬能力过滤
□ 默认不是最高价模型
□ 失败支持当前Step升级
□ 升级不重跑已完成副作用
□ Run有总预算
□ 并行调用前原子预留预算
□ Fallback按能力契约选择
□ Retry与Fallback错误分类明确
□ Provider差异由Adapter处理
□ 路由决策可审计
□ 统计Cost per Successful Task
□ 监控Escalation Rate
□ 新模型先评测和影子，再进主路由

写在最后

2026 年的模型竞争越来越激烈：更强的模型不断出现，Flash 类模型又不断把价格往下压，开放权重模型同时在扩大部署选择。

这意味着企业架构不应该继续问：

“我们到底选哪一个模型？”

更好的问题是：

“哪一类任务，在什么风险和预算下，应该由哪个模型处理？”

模型调度层的价值，就是把这个问题从人工争论变成数据和策略。

一个成熟的 Agent 系统，不会把最贵模型当默认答案，也不会为了省钱把所有任务都塞给便宜模型。它会知道什么时候该省，什么时候该升级，什么时候该降级，什么时候应该直接停下来交给人。

这才是生产环境里的“模型智能”。

更多企业级 AI 应用、Agent、RAG 与模型工程化内容，我会继续整理在 智元界：

https://www.zyentor.com/