LLM 多 Provider 路由与降级：把月成本从 ¥15,000 降到 ¥8,000

上个月团队 review 月账单，发现 LLM 调用占了 ¥15,000。做了 2 周优化，月成本降到 ¥8,000，节省 47%，而且可用性还提升了。

这篇文章复盘我们怎么设计多 Provider 路由、自动降级和成本监控。

---

一、背景：为什么需要多 Provider

做 AI 应用只用一家 LLM 看似简单，实际上有 4 个隐患：

1. 成本不可控：单 provider 价格固定，没有谈判空间
2. 可用性风险：单 provider 故障 = 全业务停摆（去年 Claude API 529 故障一晚上）
3. 场景不匹配：不同任务对模型要求不同——简单分类用 GPT-4o 是浪费，复杂推理用 Qwen-Turbo 又不够
4. 数据合规：部分客户要求数据不出境，必须有国内 provider 兜底

于是我们设计了一套统一 LLM 接入层，根据任务类型路由到不同 provider，自动降级到备选，自动核算成本。

---

二、统一 LLM 接入层架构

┌─────────────────────────────────────────────────────────┐
│  业务侧（LangChain.js / Claude Agent SDK / LangGraph）       │
│   - 统一调用 LLMClient.chat(messages, task_type=...)      │
└────────────────────┬────────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│              LLMClient 抽象层                              │
│   - task_type → model_config 路由表                        │
│   - 自动降级 chain（主→备1→备2）                          │
│   - 成本核算（每请求 token 成本记录）                       │
│   - 重试 + 限流 + 监控埋点                                 │
└────────────────────┬────────────────────────────────────┘
                     │
        ┌────────────┼────────────┬─────────────┐
        ▼            ▼            ▼             ▼
   ┌────────┐  ┌────────┐  ┌────────┐  ┌────────┐
   │ Claude │  │  Qwen  │  │ DeepSeek│  │  GPT-4 │
   │Anthropic│  │  阿里云 │  │         │  │ OpenAI │
   └────────┘  └────────┘  └────────┘  └────────┘

2.1 核心代码

from typing import List
import asyncio
from dataclasses import dataclass

@dataclass
class ModelConfig:
    primary: str
    fallback: List[str]
    cost_per_1k_input: float
    cost_per_1k_output: float
    timeout_s: int = 30


MODEL_ROUTING = {
    # task_type: ModelConfig
    "agent_decision": ModelConfig(
        primary="qwen-plus",
        fallback=["claude-sonnet-4-6", "gpt-4o"],
        cost_per_1k_input=0.0008,
        cost_per_1k_output=0.002,
    ),
    "policy_summary": ModelConfig(
        primary="qwen-plus",
        fallback=["claude-sonnet-4-6", "deepseek-chat"],
        cost_per_1k_input=0.0008,
        cost_per_1k_output=0.002,
    ),
    "classification": ModelConfig(
        primary="qwen-turbo",
        fallback=["gpt-4o-mini", "deepseek-chat"],
        cost_per_1k_input=0.0003,
        cost_per_1k_output=0.0006,
    ),
    "complex_reasoning": ModelConfig(
        primary="claude-sonnet-4-6",
        fallback=["qwen-plus", "gpt-4o"],
        cost_per_1k_input=0.003,
        cost_per_1k_output=0.015,
    ),
    "embedding": ModelConfig(
        primary="bge-large-zh-v1.5",
        fallback=["qwen3-embedding", "text-embedding-3-small"],
        cost_per_1k_input=0.0001,
        cost_per_1k_output=0.0,
    ),
}


class LLMClient:
    """统一 LLM 接入层"""

    def __init__(self, providers: dict):
        self.providers = providers  # {"qwen-plus": qwen_client, ...}

    async def chat(
        self,
        messages: list,
        task_type: str,
        temperature: float = 0.0,
    ) -> LLMResponse:
        config = MODEL_ROUTING.get(task_type)
        if not config:
            raise ValueError(f"Unknown task_type: {task_type}")

        # 按 primary → fallback1 → fallback2 顺序尝试
        models = [config.primary] + config.fallback
        last_error = None

        for model in models:
            try:
                response = await asyncio.wait_for(
                    self._call_single_model(model, messages, temperature),
                    timeout=config.timeout_s,
                )
                # 成功：记录 metrics，返回
                self._record_success(model, task_type, response)
                return response
            except (TimeoutError, RateLimitError, ModelError) as e:
                last_error = e
                logger.warning(f"{model} failed: {type(e).__name__}, trying next")
                self._record_failure(model, task_type, e)
                continue

        # 全部失败
        raise AllModelsFailedError(
            f"All models failed for {task_type}: {last_error}"
        )

    async def _call_single_model(self, model: str, messages, temperature):
        """调用单个 provider（带超时）"""
        provider = self.providers[model]
        return await provider.chat(messages, temperature=temperature)

    def _record_success(self, model, task_type, response):
        cost = self._calculate_cost(model, response.usage)
        metrics.increment("llm_request_total", tags={"model": model, "task_type": task_type, "status": "success"})
        metrics.histogram("llm_cost_usd", cost, tags={"model": model, "task_type": task_type})
        metrics.histogram("llm_token_usage", response.usage.total_tokens, tags={"model": model, "task_type": task_type})

    def _calculate_cost(self, model, usage):
        config = next(c for c in MODEL_ROUTING.values() if c.primary == model or model in c.fallback)
        return (usage.input_tokens / 1000) * config.cost_per_1k_input + \
               (usage.output_tokens / 1000) * config.cost_per_1k_output

2.2 路由规则设计原则

我们花了 1 周讨论路由规则，最终定下 4 个原则：

原则 1：按”任务复杂度”分层

任务复杂度	典型场景	主 provider
极简单	分类、提取、关键词	Qwen-Turbo（最便宜）
一般	摘要、改写、翻译	Qwen-Plus（中文最优）
复杂	推理、规划、Agent	Claude Sonnet 4.6（推理最强）

原则 2：同任务类型内”主+备”，跨任务不混

Agent 决策任务的主备是 [Qwen-Plus → Claude → GPT-4o]，但**分类任务的主备是 [Qwen-Turbo → GPT-4o-mini → DeepSeek]**——不交叉。

原则 3：fallback 链不超过 3 个

链太长（如 5 个 fallback）会拖慢 P99 延迟。只保留 3 个最有把握的备选。

原则 4：复杂推理任务用 Claude 模型

不是”用 Qwen 省成本”——复杂推理场景 Qwen 与 Claude 的质量差距是数量级的，省下的钱不够一次客户投诉的损失。

---

三、降级策略：什么时候触发 fallback

不是所有失败都触发降级——要分清”暂时性失败”和”持续性失败”。

3.1 触发降级的条件

class ShouldFallback:
    @staticmethod
    def decide(error: Exception, retry_count: int) -> bool:
        # 超时：触发降级（大概率是 provider 慢）
        if isinstance(error, TimeoutError):
            return True

        # 429 Rate Limit：触发降级
        if isinstance(error, RateLimitError):
            return True

        # 500/502/503：触发降级（provider 服务异常）
        if isinstance(error, ModelError) and error.status_code >= 500:
            return True

        # 400 Bad Request：不降级（请求本身有问题，换 provider 也一样）
        if isinstance(error, ModelError) and error.status_code == 400:
            return False

        # 内容审核拦截：不降级（换 provider 也可能被拦）
        if "content_policy" in str(error):
            return False

        return False

3.2 降级时的用户体验

降级不能让用户感知到。我们的策略：

async def chat_with_user_friendly_fallback(self, messages, task_type):
    try:
        return await self.llm_client.chat(messages, task_type)
    except AllModelsFailedError:
        # 全部失败：返回降级答案
        return LLMResponse(
            content="抱歉，系统暂时繁忙，请稍后再试。",
            model="fallback_static",
            is_degraded=True,
        )

绝不让用户看到 “Anthropic 529 Error” 这种技术错误。

---

四、成本优化：从 ¥15,000 到 ¥8,000 的 5 个具体动作

动作 1：任务分类 + 路由（节省 30%）

问题：之前所有任务都用 Claude Sonnet，单价 ¥0.003/1K（input）。

优化：按任务复杂度路由到不同模型：

任务类型	优化前	优化后	月成本变化
简单分类	Claude Sonnet	Qwen-Turbo	¥4500 → ¥600
中文摘要	Claude Sonnet	Qwen-Plus	¥6000 → ¥1500
复杂推理	Claude Sonnet	Claude Sonnet（保留）	¥3000 → ¥3000
Embedding	OpenAI	BGE-large-zh	¥1500 → ¥200

节省：¥4500 + ¥4500 + ¥1300 = ¥10,300/月（68% 节省）

动作 2：启用 Prompt Caching（节省 25%）

Chatomni 的 prompt 结构是：系统提示（500 tokens）+ 政策原文（5000 tokens）+ 用户问题（50 tokens）。80% 的请求里”政策原文”不变。

启用 Anthropic Prompt Caching 后，命中部分按 10% 价格计费：

# 第一次请求：创建缓存
response = client.messages.create(
    model="claude-sonnet-4-6",
    system=[
        {"type": "text", "text": system_prompt, "cache_control": {"type": "ephemeral"}},
    ],
    messages=[{"role": "user", "content": [
        {"type": "text", "text": policy_doc, "cache_control": {"type": "ephemeral"}},
        {"type": "text", "text": user_query},
    ]}]
)
# cache_creation_input_tokens: 5500, cache_read_input_tokens: 0

# 第二次请求（同一文档）：命中缓存
# cache_creation_input_tokens: 0, cache_read_input_tokens: 5500
# 计费：cache_read 价格 = input 价格 × 0.1

节省：policy_summary 任务成本降低 70%（因为 cache read 占 90%）

动作 3：限流保护（防止意外烧钱）

最初我们没有限流保护。一次代码 bug 导致单次循环里调用了 5000 次 Qwen-Plus，一晚上烧了 ¥500。

加限流后：

class RateLimiter:
    def __init__(self, max_requests_per_minute=60, max_cost_per_hour_usd=50):
        self.request_limiter = TokenBucket(rate=max_requests_per_minute / 60, capacity=max_requests_per_minute)
        self.hourly_cost = 0
        self.max_cost_per_hour = max_cost_per_hour_usd

    async def acquire(self, estimated_cost_usd=0.01):
        await self.request_limiter.acquire()
        if self.hourly_cost + estimated_cost_usd > self.max_cost_per_hour:
            raise RateLimitExceeded(f"Hourly cost limit {self.max_cost_per_hour} reached")
        self.hourly_cost += estimated_cost_usd

效果：再没出现过”单晚烧 ¥500”的事故。

动作 4：批量请求合并（节省 15%）

有些场景（如异步评估、批量处理）可以合并请求：

# ❌ 优化前：1000 条独立调用
for item in items:
    response = await llm.chat(f"评估这条数据：{item}")

# ✅ 优化后：1 次 batch 调用
prompt = "\n".join([f"{i}. {item}" for i, item in enumerate(items)])
response = await llm.chat(f"请评估以下数据：\n{prompt}")

效果：减少 90% 的请求数，但输入 token 增加有限。

动作 5：成本监控 + 告警

# Prometheus 告警规则
- alert: LLMHighCost
  expr: increase(llm_cost_usd[1h]) > 50
  for: 10m
  annotations:
    summary: "LLM 1h 成本超 ¥{{ $value }}"

- alert: LLMCostPerTask
  expr: avg by (task_type) (rate(llm_cost_usd[1h]) > 20)
  for: 30m
  annotations:
    summary: "{{ $labels.task_type }} 任务小时成本飙高"

- alert: LLMFallbackSpike
  expr: rate(llm_fallback_total[10m]) > 0.5
  for: 5m
  annotations:
    summary: "Fallback 频率异常，可能主 provider 出问题"

---

五、成本对比

优化项	优化前	优化后	节省
任务路由	¥15,000	¥10,500	¥4,500（30%）
Prompt Caching	¥10,500	¥7,875	¥2,625（25%）
批量请求	¥7,875	¥6,694	¥1,181（15%）
限流保护	¥6,694	¥6,694	¥0（避免事故）
月度成本	¥15,000	¥8,000	¥7,000（47%）

---

六、可用性提升：主 provider 故障时的故事

上个月某天凌晨 2 点，Anthropic API 全区域 529 故障（我们事后才知道）。

如果用单 provider，整个 Chatomni 对话功能就挂了。但我们的多 provider 架构：

[Agent 决策] 主 Qwen-Plus 失败 → 自动降级到 Claude Sonnet 4.6 ✅
[分类任务] 主 Qwen-Turbo 成功 ✅
[政策摘要] 主 Qwen-Plus 失败 → 自动降级到 DeepSeek ✅

用户体验：极少数请求感知到 1-2 秒延迟，没有用户投诉。

第二天 oncall review 时，没有任何业务侧感知到这个故障——这就是多 provider 的价值。

---

七、踩过的坑

坑 1：fallback 不一定能救

某次 Claude API 返回内容审核错误（status 400），我们以为是 provider 故障，触发 fallback。结果每个 provider 都返回内容审核错误——因为请求内容本身有问题，换 provider 也救不了。

教训：fallback 不是万能的，要区分错误类型（400 内容问题 不降级，500 服务问题 才降级）。

坑 2：成本监控要分任务类型

最初我们只有全局成本监控（llm_cost_total）。结果某次 classification 任务因为误用 Claude 模型，月成本从 ¥200 飙到 ¥2000——我们 3 周后才发现。

改成 by (task_type) 分桶后，第 2 天就报警了。

坑 3：不要过度优化

我们曾考虑给每条请求都做”先用便宜模型试，不行再升级”的策略。结果发现：

• 便宜模型试错的成本 > 直接用合适模型
• 延迟翻倍（两次调用）
• 用户体验下降

结论：路由规则要静态稳定，别搞动态试错。

坑 4：Prompt Caching 的 TTL 坑

Anthropic Prompt Caching 默认 TTL 5 分钟。我们有些场景用户间隔 > 5 分钟，缓存命中失败。

最后用 extended_ttl 模式（延长到 1 小时），但成本也增加 25%——按场景权衡。

---

八、给类似场景的建议

如果你也要做多 Provider 路由：

1. 从第 1 天就设计抽象层——别等业务跑起来再重构
2. 任务路由表要简单——别超过 10 个 task_type，越多越难维护
3. 降级要分层——超时/限流降级，内容错误不降级
4. 成本监控按任务类型——全局监控会让你错过问题
5. 限流保护是必需的——一次 bug 可能烧掉一个月预算
6. Prompt Caching 是 ROI 最高的优化——代码改动 5 行，效果立竿见影

---

九、总结

多 Provider 路由不是”为了用多家而用多家”——是为了降本 + 提可用性 + 灵活应对场景。

核心原则：

1. 统一抽象层——业务侧只关心 task_type，不关心 provider
2. 按复杂度路由——简单任务用便宜模型，复杂任务用强模型
3. 自动降级链——超时/限流降级，内容错误不降级
4. 成本分桶监控——按 task_type 分桶，及时发现异常
5. Prompt Caching 必开——重复 prompt 的场景节省 70%+

最后一句话：别只盯着”能不能调通 LLM”，要设计”调得稳 + 调得省 + 调得快”的完整体系。

---

十、参考资料

• Anthropic Prompt Caching
• Qwen 模型计费
• OpenAI Automatic Caching
• LiteLLM 路由

---

作者：魏远标，贝斯平 AI 架构师。技术博客：javai.tech

你的 LLM 月账单是多少？省了多少？留言聊聊～