AI 应用的成本失控往往不是因为「模型太贵」，而是因为设计上没做任何节约：每次都全文重发、不用缓存、不控制输出长度、不看 trace。这篇给一套可落地的降本方法，按投入产出排。
一、先搞清楚钱花在哪
# 每次调用的成本构成
1. 输入 token × 输入单价 + 输出 token × 输出单价
1. 注意：输出单价通常是输入的 2-4 倍
1. 实例：某次 RAG 问答
1. 系统提示词 300 + 检索片段 1500 + 用户问题 20 = 1820 输入
1. 生成回答 400 输出
1. 假设每百万 token 输入 ¥5、输出 ¥15
1. 成本 = 1820/1e6 * 5 + 400/1e6 * 15 ≈ ¥0.015 / 次
1. 1 万次/天 → ¥150/天 → ¥4500/月第一个关键认知：输入是主要成本，不是输出。RAG 场景里检索片段常占输入的 80% 以上。
二、降本手段排序（按投入产出）
| 手段 | 降本幅度 | 实现难度 | 优先级 |
|---|---|---|---|
| 缓存相同请求 | 视重复率，10-50% | 低 | 最高 |
| 压缩检索片段数量 | 20-40% | 低 | 高 |
| 限制输出长度 | 10-30% | 低 | 高 |
| 精简系统提示词 | 5-15% | 低 | 高 |
| 按任务路由到便宜模型 | 50-80% | 中 | 高 |
| 批量合并请求 | 10-20% | 中 | 中 |
| 本地模型承接简单任务 | 视分流比例 | 高 | 中 |
三、缓存：最大的一块免费收益import hashlib, json
import redis.asyncio as redis
rdb = redis.from_url("redis://localhost:6379")
async def cached_chat(messages: list, model: str, **kw) -> dict:
"""只对「确定性」请求做缓存（temperature=0 或很低）。"""
key = "llm:" + hashlib.sha256(
json.dumps({"m": messages, "model": model, "kw": kw},
sort_keys=True, ensure_ascii=False).encode()
).hexdigest()[:32]
if (hit := await rdb.get(key)):
return json.loads(hit)
resp = await llm.chat(messages=messages, model=model, **kw)
await rdb.setex(key, 3600, json.dumps(resp)) # 缓存 1 小时
return resp
- 只缓存 temperature≈0 的请求：创意类任务缓存会让用户拿到重复内容
- 缓存键要包含完整上下文：model、messages、参数全都要，否则会串味
- RAG 场景分两层缓存：检索结果单独缓存（更稳定），生成结果按「问题+片段哈希」缓存
- TTL 要短：1-6 小时足够，太长会导致知识过期
四、压缩检索片段：少给但给准
# 之前：取 top 5，每块 800 字 → 4000 字进提示词
1. 之后：重排后取 top 3，且每块只保留「与问题相关的句子」
def compress_chunk(chunk: str, query: str, max_sentences: int = 3):
"""从块里挑与问题最相关的几句。"""
sents = re.split(r'(?<=[。！？])', chunk)
scored = sorted(sents, key=lambda s: lexical_overlap(s, query), reverse=True)
keep = [s for s in scored[:max_sentences] if s.strip()]
return ''.join(keep)
1. 效果：4000 字 → 约 1200 字，输入成本降 70%
1. 风险：压缩可能切掉上下文，要配合重排一起用
import hashlib, json
import redis.asyncio as redis
rdb = redis.from_url("redis://localhost:6379")
async def cached_chat(messages: list, model: str, **kw) -> dict:
"""只对「确定性」请求做缓存（temperature=0 或很低）。"""
key = "llm:" + hashlib.sha256(
json.dumps({"m": messages, "model": model, "kw": kw},
sort_keys=True, ensure_ascii=False).encode()
).hexdigest()[:32]
if (hit := await rdb.get(key)):
return json.loads(hit)
resp = await llm.chat(messages=messages, model=model, **kw)
await rdb.setex(key, 3600, json.dumps(resp)) # 缓存 1 小时
return resp# 之前：取 top 5，每块 800 字 → 4000 字进提示词
1. 之后：重排后取 top 3，且每块只保留「与问题相关的句子」
def compress_chunk(chunk: str, query: str, max_sentences: int = 3):
"""从块里挑与问题最相关的几句。"""
sents = re.split(r'(?<=[。！？])', chunk)
scored = sorted(sents, key=lambda s: lexical_overlap(s, query), reverse=True)
keep = [s for s in scored[:max_sentences] if s.strip()]
return ''.join(keep)
1. 效果：4000 字 → 约 1200 字，输入成本降 70%
1. 风险：压缩可能切掉上下文，要配合重排一起用