5.5 成本控制策略

B --> B1[输入 Token 费用]
B --> B2[输出 Token 费用]
B --> B3[重试与失败消耗]
style B fill:#ff6b6b,color:#fff
​

</div>

上图红色标记的 LLM API 调用是成本优化的**第一优先级**，因为它通常占 RAG 系统总运营成本的 60%–80%。嵌入模型调用和向量数据库成本相对固定且可控，基础设施成本则与部署方式（云端 vs 自建）强相关。

### 成本优化的四层策略

我把 RAG 成本优化归纳为四层，按投入产出比从高到低排列：

| 策略层 | 核心思路 | 预期节省 | 实施难度 |
|--------|----------|----------|----------|
| **L1：避免调用** | 语义缓存，命中直接返回 | 20%–40% | 低 |
| **L2：减少 Token** | Prompt 压缩、上下文裁剪 | 15%–30% | 中 |
| **L3：降低单价** | 多模型路由，简单问题用便宜模型 | 20%–50% | 中 |
| **L4：批量优化** | 批量嵌入、异步处理、本地部署 | 30%–60% | 高 |

**关键原则**：先做 L1 和 L2（低成本高回报），再做 L3 和 L4。很多团队一上来就想本地部署大模型（L4），但忽略了 40% 的查询其实是重复问题，用缓存就能零成本解决。

## 环境准备 / 前置知识

- Python 3.10+，已安装 OpenAI SDK、LangChain、NumPy
- 理解 Token 概念（本教程 5.3 节上下文管理中已有详细说明）
- 了解主流 LLM 的定价模型（OpenAI、Anthropic、通义千问等）
- 已有可运行的 RAG 检索流程（本教程第 4 章内容）

## 分步实战

### 步骤 1：建立成本基线——先算清楚花多少

在优化之前，你必须知道当前的准确成本。很多团队凭感觉说「API 费用太高」，但说不出具体是哪个环节、哪类查询最贵。成本监控是所有优化的前提。

下面是一个实用的成本追踪器实现：

```python
# 各模型的定价表（2026年7月参考价，请以官方最新定价为准）
MODEL_PRICING = {
    "gpt-4o": {"input_per_1m": 2.50, "output_per_1m": 10.00},
    "gpt-4o-mini": {"input_per_1m": 0.15, "output_per_1m": 0.60},
    "claude-sonnet-4-20250514": {"input_per_1m": 3.00, "output_per_1m": 15.00},
    "claude-haiku-4-20250414": {"input_per_1m": 0.80, "output_per_1m": 4.00},
    "qwen2.5-72b": {"input_per_1m": 0.40, "output_per_1m": 1.20},
    "qwen2.5-7b": {"input_per_1m": 0.00, "output_per_1m": 0.00},  # 本地部署
}

def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    """根据模型和 Token 数计算费用"""
    pricing = MODEL_PRICING.get(model)
    if not pricing:
        return 0.0
    return (input_tokens / 1_000_000) * pricing["input_per_1m"] + \
           (output_tokens / 1_000_000) * pricing["output_per_1m"]
​

成本追踪器的核心职责有三个：记录每次调用的费用、按天汇总并拆分到不同模型、在接近预算时发出告警。我在设计中使用了 dataclass 来存储每次调用的完整记录（时间、模型、Token 数、延迟、是否命中缓存），这样可以方便地做多维度分析。

实际使用时，你需要在每次 LLM 调用的前后插入埋点：调用前记录开始时间，调用后从响应中提取 usage.prompt_tokens 和 usage.completion_tokens，然后调用 calculate_cost 计算费用并存入记录。建议把成本数据同时写入日志文件和时序数据库（如 Prometheus + Grafana 做可视化），方便后续分析趋势。

一个实用的日报指标集：每日总费用、按模型拆分的费用占比、平均每次调用成本、缓存命中率、P99 单次调用成本。其中 P99 很关键——一次异常调用可能花掉平时 100 次的钱，平均值会被平滑掉。

步骤 2：语义缓存——避免重复调用

企业知识库场景中，大量用户提问是高度相似的。比如「公司年假怎么休」「报销流程是什么」「VPN 怎么连」这些问题可能每天被问几十次。如果每次都走完整的检索 + LLM 调用链路，纯属浪费。

语义缓存的核心思路是：把用户的查询先做嵌入，在缓存向量库中搜索相似查询，如果找到语义相似度超过阈值的历史查询，直接返回缓存的答案。注意，这是「语义相似」而非「字面相同」，所以「年假怎么请」和「休假政策是什么」能命中同一缓存。

class SemanticCache:
    """语义缓存：基于向量相似度匹配历史查询"""
    
    def __init__(self, embedding_model, similarity_threshold: float = 0.92,
                 max_cache_size: int = 10000, ttl_seconds: int = 86400 * 7):
        self.embedding_model = embedding_model
        self.similarity_threshold = similarity_threshold
        self.max_cache_size = max_cache_size
        self.ttl_seconds = ttl_seconds
        self._cache_vectors = []
        self._cache_entries = []
        self._hits = 0
        self._misses = 0
    
    async def get(self, query: str):
        """查询缓存，命中返回答案，未命中返回 None"""
        if not self._cache_entries:
            self._misses += 1
            return None
        query_vector = await self.embedding_model.aembed_query(query)
        if query_vector is None:
            self._misses += 1
            return None
        similarities = cosine_similarity([query_vector], self._cache_vectors)[0]
        best_idx = int(np.argmax(similarities))
        best_score = float(similarities[best_idx])
        if best_score >= self.similarity_threshold:
            entry = self._cache_entries[best_idx]
            age = (datetime.now() - entry["timestamp"]).total_seconds()
            if age < self.ttl_seconds:
                self._hits += 1
                return {"answer": entry["answer"], "similarity": round(best_score, 4),
                        "cache_hit": True}
        self._misses += 1
        return None
    
    async def put(self, query: str, answer: str, metadata=None):
        """将查询-答案对写入缓存"""
        query_vector = await self.embedding_model.aembed_query(query)
        if query_vector is None:
            return
        if len(self._cache_entries) >= self.max_cache_size:
            self._cache_vectors.pop(0)
            self._cache_entries.pop(0)
        self._cache_vectors.append(np.array(query_vector, dtype=np.float32))
        self._cache_entries.append({"query": query, "answer": answer,
                                    "metadata": metadata or {}, "timestamp": datetime.now()})
    
    def get_stats(self):
        total = self._hits + self._misses
        return {"cache_size": len(self._cache_entries), "hits": self._hits,
                "misses": self._misses,
                "hit_rate": round(self._hits / total * 100, 1) if total else 0}
​

相似度阈值怎么选？ 这是最关键的参数。我建议从 0.90 开始，观察一周的命中率误判情况：如果用户投诉「回答的是另一个问题」，说明阈值太低，调到 0.93–0.95；如果命中率低于 10%，说明阈值太高，降到 0.88–0.90。企业内部知识库场景，0.92 通常是个不错的起点，因为用户提问往往比较规范。

生产环境建议：不要把缓存向量放在内存里（如上例），而是存到你的向量数据库中单独建一个 collection。这样重启不丢失，也支持分布式部署。上面的代码是简化演示，生产中把 _cache_vectors 替换为向量数据库查询即可。另外，知识库内容更新后必须主动清除相关缓存，否则用户可能收到过期答案。

步骤 3：多模型智能路由——不同问题用不同价格的模型

不是所有问题都需要 GPT-4o 级别的模型来回答。一个典型的 RAG 知识库中，查询大致分三类：

• 简单查询（约 40%）：事实型问题如「公司地址在哪」「报销额度多少」——用 GPT-4o-mini 甚至本地小模型就能准确回答
• 中等查询（约 45%）：需要一定推理或综合多个文档片段的问题——用 GPT-4o-mini 或 Qwen2.5-72B
• 复杂查询（约 15%）：需要深度推理、多步逻辑链、长文总结——才需要上 GPT-4o 或 Claude Sonnet

class ModelRouter:
    """多模型智能路由器：按查询复杂度路由到不同价位的模型"""
    
    def __init__(self):
        self.route_config = {
            "simple": {"model": "gpt-4o-mini", "max_input": 2000, "max_output": 300,
                        "temperature": 0.0},
            "medium": {"model": "gpt-4o-mini", "max_input": 4000, "max_output": 800,
                        "temperature": 0.3},
            "complex": {"model": "gpt-4o", "max_input": 8000, "max_output": 2000,
                        "temperature": 0.5},
        }
    
    def classify_query(self, query: str, retrieved_docs=None) -> str:
        """分类查询复杂度"""
        q = query.lower()
        # 规则1：短查询且无复杂关键词 → simple
        if len(query) < 20 and not any(kw in q for kw in ["为什么", "如何", "对比", "分析"]):
            return "simple"
        # 规则2：包含复杂关键词 → complex
        complex_kw = ["对比", "区别", "优缺点", "分析", "评估", "总结", "原理", "机制"]
        if any(kw in q for kw in complex_kw):
            return "complex"
        # 规则3：检索结果多 → medium
        if retrieved_docs and len(retrieved_docs) >= 3:
            return "medium"
        return "medium"
    
    def get_model_config(self, query: str, retrieved_docs=None) -> dict:
        complexity = self.classify_query(query, retrieved_docs)
        config = self.route_config[complexity].copy()
        config["complexity"] = complexity
        return config
    
    def estimate_savings(self, daily_queries=5000) -> dict:
        """估算路由策略的成本节省"""
        avg_input, avg_output = 2000, 500
        dist = {"simple": 0.40, "medium": 0.45, "complex": 0.15}
        # 全部走 GPT-4o
        all_premium = daily_queries * calculate_cost("gpt-4o", avg_input, avg_output)
        # 路由后
        routed = sum(
            daily_queries * ratio * calculate_cost(
                self.route_config[level]["model"], avg_input, avg_output)
            for level, ratio in dist.items()
        )
        return {"daily_all_premium": round(all_premium, 2),
                "daily_routed": round(routed, 2),
                "savings_pct": round((1 - routed / all_premium) * 100, 1) if all_premium else 0}
​

路由器的实际效果如何？ 以默认配置为例（40% simple + 45% medium + 15% complex），日查询 5000 次的场景下，全部走 GPT-4o 约 112.5 美元/天，路由后约 9.75 美元/天，节省约 91%。当然，实际节省取决于你的查询分布和模型选择。但即便保守估计（30% simple + 50% medium），节省也在 70% 以上。这是成本优化中投入产出比最高的策略之一。

一个容易犯的错误：很多团队基于词频或正则做路由分类，但效果往往不好，因为中文查询的表达方式太多样。更健壮的做法是用一个小模型（如 GPT-4o-mini）专门做分类——让模型输出 JSON 格式的 {"complexity": "simple/medium/complex"}。这样分类准确率通常能从 70% 提升到 90%+，但每次分类调用本身也有成本（约 0.0001 美元），需要权衡。

步骤 4：Token 压缩——每次调用少花钱

即使确定了用哪个模型，每次调用的 Token 数仍然有优化空间。RAG 系统的输入 Token 主要由三部分组成：系统提示（System Prompt）、检索结果（Retrieved Context）、用户问题。其中检索结果通常占 70%–80% 的输入 Token，是优化的主战场。

策略一：检索结果精简——按相关性截取，只保留最有价值的内容。

class ContextCompressor:
    """上下文压缩器：控制送入 LLM 的 Token 在预算内"""
    
    def __init__(self, max_context_tokens: int = 3000):
        self.max_context_tokens = max_context_tokens
    
    def compress_by_relevance(self, query: str, documents: list, target_tokens=None) -> str:
        """
        按相关性分数从高到低，逐篇加入直到 Token 预算用完。
        贪心策略简单但效果很好，因为高相关性文档往往能回答问题。
        """
        target = target_tokens or self.max_context_tokens
        sorted_docs = sorted(documents, key=lambda d: d.get("score", 0), reverse=True)
        selected, current_tokens = [], 0
        for doc in sorted_docs:
            text = doc["text"]
            doc_tokens = self._estimate_tokens(text)
            if current_tokens + doc_tokens > target:
                remaining = target - current_tokens
                if remaining > 100:
                    selected.append(text[:int(remaining / 1.3)])
                break
            selected.append(text)
            current_tokens += doc_tokens
        return "\n\n---\n\n".join(selected)
    
    def _estimate_tokens(self, text: str) -> int:
        """粗略估算：中文约 1.5 Token/字，英文约 1.3 Token/词"""
        chinese = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
        english = len([w for w in text.split() if w.isascii()])
        return int(chinese * 1.5 + english * 1.3)
​

策略二：Prompt 模板精简——去掉系统提示中的冗余描述。

很多 RAG 系统的系统提示写得像散文：「你是一个专业的知识库助手，请基于以下检索到的文档内容来回答用户的问题。如果文档中没有相关信息，请如实告知。回答时请引用来源……」这段话大约 80 个 Token，每次调用都会重复消耗。精简版：「基于下方文档回答。无相关信息则回答'暂无'。标注来源。」——同样意思约 20 个 Token，每次节省 60 Token。日查询 5000 次就是每天省 30 万 Token。

策略三：关键句提取——从长文档中只提取与查询最相关的 5–8 个句子，而不是把整段文档塞进 Prompt。这对长文档（单篇超过 1000 字）效果尤其明显，通常能减少 40%–60% 的输入 Token，同时保持甚至提升回答质量（因为噪声减少了）。

步骤 5：预算守卫——防止费用失控

你需要一个安全网——当日成本接近预算上限时，自动将所有查询降级到便宜模型，避免预算失控。

class BudgetGuard:
    """预算守卫：当日成本接近上限时自动降级
    
    降级策略（按预算使用比例触发）：
    - < 60%：正常路由
    - 60%-80%：medium 查询也降级到 simple 模型
    - 80%-100%：所有查询都走最便宜的模型
    - > 100%：返回预设回复，不调用 LLM
    """
    
    def __init__(self, daily_budget_usd: float, cost_tracker):
        self.daily_budget = daily_budget_usd
        self.cost_tracker = cost_tracker
    
    def get_allowed_models(self) -> dict:
        """根据当前预算使用比例返回各复杂度是否允许正常模型"""
        summary = self.cost_tracker.get_daily_summary(date.today())
        ratio = summary.total_cost_usd / self.daily_budget if self.daily_budget else 0
        if ratio < 0.6:   return {"simple": True, "medium": True, "complex": True}
        if ratio < 0.8:   return {"simple": True, "medium": False, "complex": True}
        if ratio < 1.0:   return {"simple": True, "medium": False, "complex": False}
        return {"simple": False, "medium": False, "complex": False}
    
    def should_block(self) -> bool:
        """预算是否已完全耗尽"""
        summary = self.cost_tracker.get_daily_summary(date.today())
        return summary.total_cost_usd >= self.daily_budget
​

为什么不直接用 API 提供商的 Rate Limit？ API 提供商的限流是按 QPS 或 TPM（Tokens Per Minute）控制的，而你的预算是按月/日控制的。两者维度不同。比如你设置了 OpenAI 的月度预算 $100，但某天一个爬虫疯狂请求，一天就能花掉 $50。BudgetGuard 在应用层做日级预算控制，是 API 限额的补充而非替代。

步骤 6：完整 Pipeline 串联

把以上所有策略串起来，形成完整的成本优化 RAG Pipeline。调用链路为：语义缓存检查（L1）→ 查询分类 + 模型路由（L3）→ 文档检索 → 上下文压缩（L2）→ LLM 调用 → 结果缓存写入 → 成本记录。

async def cost_optimized_query(user_query, cache, router, compressor, 
                                vector_store, llm_client, cost_tracker):
    """完整的成本优化 RAG 查询流程"""
    start = time.time()
    
    # L1: 语义缓存
    cached = await cache.get(user_query)
    if cached:
        cost_tracker.record("cache", "cache", 0, 0, 
                           (time.time()-start)*1000, cache_hit=True)
        return {**cached, "source": "semantic_cache"}
    
    # L3: 模型路由
    config = router.get_model_config(user_query)
    
    # 文档检索 + L2: 上下文压缩
    docs = await vector_store.search(user_query, top_k=5)
    compressed = compressor.compress_by_relevance(
        user_query, docs, target_tokens=config["max_input"] - 500)
    
    # LLM 调用
    response = await llm_client.chat.completions.create(
        model=config["model"],
        messages=[{"role": "system", "content": "基于下方文档回答。无信息则回答'暂无'。"},
                  {"role": "user", "content": f"文档：\n{compressed}\n\n问题：{user_query}"}],
        max_tokens=config["max_output"], temperature=config["temperature"])
    
    answer = response.choices[0].message.content
    latency = (time.time() - start) * 1000
    
    # 成本记录 + 缓存写入
    cost_tracker.record("openai", config["model"],
                       response.usage.prompt_tokens, response.usage.completion_tokens,
                       latency, query_type=config["complexity"])
    await cache.put(user_query, answer)
    
    return {"answer": answer, "model": config["model"], 
            "complexity": config["complexity"], "cache_hit": False}
​

这个 Pipeline 看起来简单，但每个环节都有工程细节需要注意。比如缓存查询和向量检索可以并行执行（异步并发），模型路由的 classify 方法应该加缓存避免重复计算，上下文压缩失败时应该有 fallback（直接用原始文档）。

完整示例

下面是一个最小可运行的端到端演示，展示路由分类和成本估算：

# 演示路由分类效果
test_queries = [
    ("公司年假有多少天？", "simple"),
    ("Python 列表推导式和 map 函数哪个性能更好？", "medium"),
    ("对比分析 FAISS 和 Milvus 在百万级数据量下的检索延迟", "complex"),
    ("年假怎么请？", "simple"),
]

router = ModelRouter()
for query, expected in test_queries:
    config = router.get_model_config(query)
    match = "✅" if config["complexity"] == expected else "❌"
    print(f"{match} [{config['complexity']:>6}] {query[:40]} → {config['model']}")

# 成本节省估算
savings = router.estimate_savings(daily_queries=5000)
print(f"\n全部 GPT-4o: ${savings['daily_all_premium']}/天")
print(f"路由后: ${savings['daily_routed']}/天")
print(f"节省: {savings['savings_pct']}%")
​

常见问题 FAQ

Q1：RAG 知识库的 LLM API 费用大概占总成本多少？如何估算月度费用？

A：在典型的云部署 RAG 系统中，LLM API 调用占总运营成本的 60%–80%。其余为嵌入模型（约 5%–10%）、向量数据库（约 5%–15%）、应用服务器和网络（约 5%–10%）。月度费用估算公式为：日均查询量 × 30 × (平均输入Token × 输入单价 + 平均输出Token × 输出单价) / 1000000。建议先用成本追踪器跑一周真实数据，再据此做精确预算。

Q2：语义缓存的相似度阈值设多少合适？会不会返回错误答案？

A：企业知识库场景建议从 0.90 开始调试。阈值过低（<0.85）会导致语义不相关的问题命中同一缓存，返回错误答案；阈值过高（>0.95）会导致命中率极低，缓存形同虚设。实际调优方法：每周抽检 50 条缓存命中记录人工判断准确性，据此微调。另外可以在缓存命中时返回标注「此回答来自缓存」的提示，让用户自行判断。

Q3：多模型路由会不会导致简单问题和复杂问题的回答质量不一致？

A：会，但这是合理的工程权衡。90% 的用户分不清 GPT-4o-mini 和 GPT-4o 的回答差异。关键是做好路由分类的准确度——如果复杂问题被错误路由到小模型，质量下降会很明显。建议每周抽查各复杂度分类的准确率，持续优化。对于质量敏感场景（如医疗、法律），建议调高阈值，宁可多用贵模型也不要答错。

Q4：本地部署开源模型真的比 API 便宜吗？什么规模值得本地部署？

A：取决于查询量。以 Qwen2.5-7B 为例，单卡 A10（约 1.2 元/小时）可支撑约 50 QPS。日查询量超过 5 万次通常更便宜；只有几千次则 API 调用更划算。建议：日查询 <1 万次用 API，>3 万次考虑本地部署，中间地带根据实际数据决定。本地部署还有额外的好处：数据不出内网、无网络延迟、不受 API 限流。

Q5：成本优化和回答质量如何平衡？有没有量化评估方法？

A：最佳实践是建立「成本-质量 Pareto 曲线」：横轴是平均每次查询成本，纵轴是回答质量评分（可用本教程 4.5 节介绍的 NDCG 等指标）。每实施一项优化后在曲线上标记一个点。如果成本降 30% 质量只降 2%，就是好策略；如果成本降 10% 质量降 15%，就该回退。目标是找到曲线的「拐点」——再降成本质量就会断崖式下降的位置。

最佳实践与避坑

最佳实践

1. 先建基线再优化：不要凭感觉优化，用成本追踪器跑一周真实数据，知道钱花在哪了
2. 优先做缓存：语义缓存的投入产出比最高，通常是第一个该做的优化
3. 渐进式路由：先只把最明显的简单查询路由到便宜模型，观察一周再扩大范围
4. 设置预算硬限制：用 BudgetGuard 做日级预算控制，防止异常流量导致费用失控
5. 定期审查定价：API 提供商经常调价（通常降价），每季度审查一次模型定价表

常见陷阱

• 缓存过期策略缺失：知识库内容更新后，缓存中可能还是旧答案。必须设置合理 TTL，并在知识库更新时主动清除相关缓存
• 路由规则一刀切：不同业务场景的查询分布不同，不要直接复制别人的路由规则，要根据自己数据调优
• 只看平均成本：平均成本低不意味着没问题。要关注 P99 单次调用成本——一次异常调用可能花掉平时 100 次的钱
• 忽略嵌入成本：很多人只关注 LLM 调用成本，但文档入库时的嵌入调用也可能是一笔不小的费用
• 过度压缩上下文：把检索结果压缩得太狠会导致 LLM 缺少关键信息，回答质量下降。压缩目标是减少 20%–30% 的 Token，而不是 50%+

本节小结

本节围绕 RAG 知识库实战中的成本控制，系统讲解了四层优化策略：语义缓存避免重复调用（L1）、Token 压缩减少每次调用消耗（L2）、多模型路由降低模型单价（L3）、以及预算守卫作为安全网。核心收获有三点：第一，成本优化必须建立在真实数据之上，先用成本追踪器建立基线；第二，缓存和路由是投入产出比最高的两个策略，优先实施；第三，成本和质量不是对立的——通过合理的分层策略，可以在几乎不影响质量的前提下大幅降低成本。

下一节我们将进入本教程的总结与进阶方向，探讨 RAG 系统的未来发展趋势。

延伸阅读

• OpenAI 官方文档中的用量监控与费用管理最佳实践
• LangChain 官方文档 v0.3 版本中的缓存与路由模块说明
• 本教程 4.5 节检索结果评估（质量评估方法论与本节成本-质量平衡直接相关）
• 本教程 5.1 节 LLM 选择与配置（模型定价和选型框架的详细展开）
• 本教程 5.3 节上下文管理（Token 预算和上下文裁剪的基础知识）

关键词：RAG 知识库实战, 成本控制, 语义缓存, 模型路由, Token 压缩, 预算管理, 费用优化
难度：进阶
预计阅读：25 分钟