LLM 应用的成本优化策略：从 Token 精简到模型分层的实战路径

LLM 应用的成本优化策略：从 Token 精简到模型分层的实战路径

一、LLM 应用的成本陷阱：为什么模型调用费用会成为最大开支

在 LLM 应用从原型走向生产的过程中，成本往往是最被低估的因素。一个简单的 RAG 系统，每次查询消耗约 2000 Token（1000 输入 + 1000 输出），使用 GPT-4 的单次成本约 0.06 美元。当 QPS 达到 10 时，月成本约 15 万美元——这还不包括向量数据库、Embedding 模型等周边成本。

更隐蔽的是，很多成本浪费隐藏在实现细节中。典型的浪费场景包括：系统提示词（System Prompt）冗长且每次请求重复发送，占用了 30-40% 的输入 Token；检索到的上下文文档未做压缩，大量无关内容被注入 LLM；所有请求都使用最强模型，即使简单问题用小模型就能解决；输出 Token 未做长度限制，模型可能生成冗长的回答。

这些浪费叠加起来，实际有效 Token 占比可能不到 50%。也就是说，一半以上的 API 费用花在了无用信息上。优化 LLM 应用成本，不是简单地换便宜模型，而是从 Token 流转的每个环节寻找压缩空间。

二、LLM 成本优化的四层模型

LLM 应用的成本优化需要从四个层面系统性地考虑，每个层面的优化策略和收益不同。

flowchart TD A[用户请求] --> B{请求路由层} B -->|简单问题| C[小模型 GPT-3.5] B -->|复杂问题| D[大模型 GPT-4] B -->|可缓存| E[语义缓存] C --> F[Token 优化层] D --> F F --> G[系统提示词压缩] F --> H[上下文文档裁剪] F --> I[输出长度限制] G --> J[执行层] H --> J I --> J J --> K[流式输出] J --> L[提前终止] K --> M[监控层] L --> M M --> N[Token 用量追踪] M --> O[成本归因分析] M --> P[异常用量告警]

请求路由层：不是所有请求都需要最强模型。简单问答用小模型即可，只有复杂推理才需要大模型。通过请求路由，可以将 60-70% 的流量分流到便宜模型，成本降低 50% 以上。

Token 优化层：在请求发送给 LLM 之前，压缩输入 Token 数量。包括系统提示词精简、上下文文档裁剪、历史对话摘要等。这一层的优化可以将输入 Token 减少 30-50%。

执行层：在 LLM 生成过程中控制输出成本。包括流式输出（避免生成完整响应后才返回）、提前终止（检测到关键信息后停止生成）、输出长度限制等。

监控层：持续追踪 Token 用量和成本，按业务维度归因分析，及时发现异常用量。

三、生产级成本优化实现

3.1 请求路由：基于复杂度的模型选择

# router.py # 基于请求复杂度的模型路由 from dataclasses import dataclass from enum import Enum from typing import Optional class ModelTier(Enum): SMALL = "gpt-3.5-turbo" # 简单问答，成本约 $0.002/1K token MEDIUM = "gpt-4o-mini" # 中等复杂度，成本约 $0.015/1K token LARGE = "gpt-4o" # 复杂推理，成本约 $0.06/1K token @dataclass class RoutingDecision: model: ModelTier reason: str estimated_tokens: int estimated_cost: float class RequestRouter: def __init__(self): # 模型成本表（每 1K Token 的美元价格） self.cost_per_1k = { ModelTier.SMALL: 0.002, ModelTier.MEDIUM: 0.015, ModelTier.LARGE: 0.06, } def route(self, query: str, context: Optional[str] = None) -> RoutingDecision: """根据查询复杂度选择模型""" complexity = self._assess_complexity(query, context) if complexity <= 0.3: model = ModelTier.SMALL reason = "简单问答，无需复杂推理" elif complexity <= 0.7: model = ModelTier.MEDIUM reason = "中等复杂度，需要上下文理解" else: model = ModelTier.LARGE reason = "复杂推理，需要深度分析" est_tokens = self._estimate_tokens(query, context, model) est_cost = est_tokens / 1000 * self.cost_per_1k[model] return RoutingDecision( model=model, reason=reason, estimated_tokens=est_tokens, estimated_cost=est_cost, ) def _assess_complexity(self, query: str, context: Optional[str]) -> float: """评估查询复杂度，返回 0-1 之间的分数""" score = 0.0 # 长查询通常更复杂 if len(query) > 200: score += 0.2 if len(query) > 500: score += 0.1 # 包含推理关键词的查询更复杂 reasoning_keywords = [ "分析", "比较", "为什么", "如何", "评估", "analyze", "compare", "why", "how", "evaluate", ] for kw in reasoning_keywords: if kw in query.lower(): score += 0.15 break # 需要上下文的查询更复杂 if context and len(context) > 500: score += 0.2 # 多步骤问题更复杂 if any(sep in query for sep in ["并且", "同时", "另外", "and", "also"]): score += 0.15 return min(score, 1.0) def _estimate_tokens(self, query: str, context: Optional[str], model: ModelTier) -> int: """估算 Token 用量""" # 粗略估算：1 个中文字符 ≈ 1.5 Token，1 个英文单词 ≈ 1.3 Token input_chars = len(query) + (len(context) if context else 0) input_tokens = int(input_chars * 1.5) # 输出 Token 根据模型类型估算 output_ratios = { ModelTier.SMALL: 0.5, # 小模型输出较短 ModelTier.MEDIUM: 0.8, ModelTier.LARGE: 1.2, # 大模型输出较长 } output_tokens = int(input_tokens * output_ratios[model]) return input_tokens + output_tokens

3.2 语义缓存：避免重复计算

# semantic_cache.py # 基于向量相似度的语义缓存 import hashlib import time from typing import Optional class SemanticCache: def __init__(self, vector_store, similarity_threshold: float = 0.95, ttl: int = 3600): self.vector_store = vector_store self.similarity_threshold = similarity_threshold self.ttl = ttl # 缓存有效期（秒） self.stats = {"hits": 0, "misses": 0} def get(self, query: str) -> Optional[str]: """查询语义缓存""" # 向量化查询 query_embedding = self.vector_store.embed(query) # 搜索最相似的缓存条目 results = self.vector_store.search( query_embedding, top_k=1, ) if not results: self.stats["misses"] += 1 return None # 检查相似度是否超过阈值 best_match = results[0] if best_match.score < self.similarity_threshold: self.stats["misses"] += 1 return None # 检查缓存是否过期 if time.time() - best_match.metadata["cached_at"] > self.ttl: self.vector_store.delete(best_match.id) self.stats["misses"] += 1 return None self.stats["hits"] += 1 return best_match.metadata["response"] def set(self, query: str, response: str): """将查询和响应存入语义缓存""" query_embedding = self.vector_store.embed(query) self.vector_store.insert( embedding=query_embedding, metadata={ "query": query, "response": response, "cached_at": time.time(), "cache_key": hashlib.md5(query.encode()).hexdigest(), }, ) def hit_rate(self) -> float: total = self.stats["hits"] + self.stats["misses"] if total == 0: return 0.0 return self.stats["hits"] / total

3.3 上下文压缩

# context_compressor.py # RAG 检索结果的上下文压缩 class ContextCompressor: def __init__(self, llm_client, max_context_tokens: int = 2000): self.llm = llm_client self.max_tokens = max_context_tokens def compress(self, query: str, documents: list[str]) -> str: """压缩检索到的文档，只保留与查询相关的部分""" # 第一步：按相关性排序文档 ranked_docs = self._rank_by_relevance(query, documents) # 第二步：逐条添加文档，直到达到 Token 上限 compressed_parts = [] current_tokens = 0 for doc in ranked_docs: doc_tokens = self._count_tokens(doc) if current_tokens + doc_tokens > self.max_tokens: # 超出预算，对最后一条文档做摘要压缩 remaining_budget = self.max_tokens - current_tokens if remaining_budget > 200: summary = self._summarize(doc, query, remaining_budget) compressed_parts.append(summary) break compressed_parts.append(doc) current_tokens += doc_tokens return "\n\n".join(compressed_parts) def _summarize(self, doc: str, query: str, max_tokens: int) -> str: """使用小模型对文档做针对性摘要""" prompt = f"""请从以下文档中提取与问题相关的关键信息，不超过 {max_tokens} 个 Token。 问题：{query} 文档：{doc} 关键信息：""" response = self.llm.chat( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}], max_tokens=max_tokens, ) return response.content def _count_tokens(self, text: str) -> int: return int(len(text) * 1.5) def _rank_by_relevance(self, query: str, docs: list[str]) -> list[str]: # 简单实现：按文档长度排序，短文档优先（信息密度通常更高） return sorted(docs, key=len)

四、架构权衡与适用边界

缓存命中率与语义阈值的矛盾。语义缓存的相似度阈值越低，命中率越高，但误缓存的风险也越大。阈值 0.95 时命中率约 15-20%，阈值 0.90 时命中率可达 30-40%，但可能出现语义相近但答案不同的误命中。建议对答案唯一性强的场景（如事实查询）使用 0.90 阈值，对答案多样性的场景（如创意生成）使用 0.95 或不使用缓存。

模型路由的准确率与延迟。路由决策本身需要计算，如果路由逻辑过于复杂（如调用 Embedding 模型做相似度计算），路由延迟可能抵消模型切换节省的成本。建议使用基于规则的轻量路由（关键词匹配 + 长度判断），将路由延迟控制在 1ms 以内。

上下文压缩的信息损失。压缩检索文档时，可能丢失对 LLM 推理有用的细节信息。实测表明，将上下文从 4000 Token 压缩到 2000 Token，回答质量下降约 5-10%。需要在成本节省和回答质量之间找到平衡点。

适用边界：成本优化策略适用于月 API 费用超过 1000 美元的 LLM 应用。对于月费用在 100 美元以内的简单应用，优化收益有限，不值得投入开发资源。对于对回答质量要求极高的场景（如医疗诊断），不建议使用模型路由和上下文压缩。

五、总结

LLM 应用的成本优化需要从四个层面系统推进：请求路由层将简单问题分流到便宜模型，Token 优化层压缩输入中的冗余信息，执行层控制输出长度和提前终止，监控层追踪成本归因和异常。其中请求路由的收益最大（可降低 50% 以上成本），语义缓存对重复查询场景效果显著（命中率 15-30%），上下文压缩可减少 30-50% 的输入 Token。工程落地时需要重点权衡缓存命中率与误缓存风险、路由准确率与路由延迟、压缩率与回答质量损失。对于月费用低于 100 美元的应用，优先保证功能完整性，成本优化可以延后。