缓存、限流与成本优化:让 LLM 应用活得下去

缓存、限流与成本优化:让 LLM 应用活得下去

本节摘要:多数 AI 创业公司不是死于模型差,而是死于单位经济算不过账。一次 GPT-4o 调用几分之一美分,一万个用户每天调十次,光输入 token 每天就 250 美元——你还没收进一分钱。活下来的公司,把每次 API 调用当成一笔财务交易,而非一次函数调用。本节带你建立成本优化的全套武器:LLM 调用的五项成本拆解(系统提示是沉默杀手);厂商级提示缓存(Anthropic 显式 cache_control 命中省 90%、OpenAI 自动前缀省 50%);语义缓存(嵌入查询、算余弦、命中阈值 0.920.95,抓住「换个说法的同一问题」);精确缓存(温度 0 的哈希命中);令牌桶限流(按用户分层配额);模型路由(简单问题送便宜模型,省 4070%);成本追踪与三档熔断(70% 警告、85% 降级、95% 只返回缓存)。一套组合拳把月账单从 22500 美元砍到 5000 美元。

对应原课程:Phase 11 · Lesson 11 · caching-cost(原英文 phases/11-llm-engineering/11-caching-cost/docs/en.md)。本节讲应用层缓存;第 15 节讲厂商层提示缓存,两者叠加可降本 50~95%。

学习目标

阅读完本节,你应当能够:

1. 实现语义缓存,对重复或相似查询直接返回缓存而非新调 API。
2. 跨厂商计算单请求成本,实现 token 感知的限流与预算告警。
3. 搭建成本优化层:提示压缩、模型路由(贵模型 vs 便宜模型)、响应缓存。
4. 针对不同查询类型设计分层缓存:精确匹配、语义相似、前缀缓存。

一、问题与直觉

你做了个 RAG 聊天机器人,跑得很美,用户喜欢。然后账单来了。

GPT-5 输入 5 美元/百万 token、输出 15 美元;Claude Opus 4.7 输入 15 美元、输出 75 美元;Gemini 3 Pro 输入 1.25 美元、输出 5 美元;GPT-5-mini 输入 0.25 美元、输出 2 美元(价格为示例,务必查厂商当前价目)。

杀死创业公司的数学:

• 日活 1 万用户
• 每用户每天 10 次查询
• 每查询 1000 输入 token(系统提示 + 上下文 + 用户消息)
• 每响应 500 输出 token

日输入成本:10000 × 10 × 1000 / 1000000 × 2.5 = 250 美元/天
日输出成本:10000 × 10 × 500 / 1000000 × 10 = 500 美元/天
月合计:22500 美元/月

这还只是 LLM。加上嵌入、向量库托管、基础设施,一个聊天机器人要 30000 美元/月。

更残酷的是:这些查询里 40~60% 是近似重复。用户用略不同的措辞问同样的问题;你的系统提示每次请求都一样,却每次都计费;RAG 检索的上下文文档在问同一主题的用户间重复。你在为冗余计算付全价。

LLM 调用的成本解剖

每次 API 调用有五项成本组件。

系统提示是沉默杀手。一个 1500 token 的系统提示,随每次请求发送,光这个前缀每百万请求就 3.75 美元。日 10 万请求,每天 375 美元——每月 11250 美元——花在从不改变的文本上。

厂商缓存:内置折扣

2026 年三大厂商都提供厂商侧提示缓存,但机制不同(详见第 15 节)。

Anthropic 的方式是显式的。你用 cache_control: {"type":"ephemeral"} 标记提示段落,首次请求付 25% 写入溢价,后续相同前缀的请求拿 90% 折扣。一个 2000 token 系统提示,正常 0.005 美元,缓存命中时 0.000625 美元。10 万请求省 437.50 美元/天。

OpenAI 的方式是自动的。任何匹配先前请求的前缀拿 50% 折扣,无需标记。代价是折扣少、控制少,但零实现成本。

语义缓存:你的定制层

厂商缓存只对相同前缀生效。语义缓存处理更难的场景:不同查询、相同含义。

「退货政策是什么?」和「我怎么退货?」是不同字符串但同一意图。语义缓存嵌入两个查询,算余弦相似度,超过阈值(典型 0.92~0.95)就返回缓存响应。

嵌入成本可忽略。OpenAI text-embedding-3-small 每百万 token 0.02 美元。查缓存相比一次完整 LLM 调用,几乎不花钱。

精确缓存:哈希匹配

对确定性调用(温度 0、同模型、同提示),精确缓存更简单更快。哈希整个提示,查缓存,命中就返回。完美适用于:系统提示 + 固定上下文 + 相同用户查询;带相同工具定义的函数调用;同一文档被多次处理的批处理。

限流:保护你的预算

限流不只关乎公平,更关乎生存。

令牌桶算法:每用户一个桶,容量 N,以每秒 R 速率补充。请求从桶里消耗 token,桶空就拒绝。这允许突发(一次用满整桶)同时强制平均速率。

按用户配额:按用户层设日/月 token 上限。

模型路由:对的事用对的模型

不是每个查询都需要 GPT-4o。「门店几点关门?」不需要 10 美元/百万输出的模型,GPT-4o-mini 0.60 美元/百万完美胜任,Claude Haiku 1.25 美元也行。一个简单分类器把便宜查询送便宜模型、复杂查询送贵模型。

调校良好的路由器光模型成本就省 40~70%。

成本追踪:知道钱花在哪

你优化不了你测不了的东西。每次 API 调用记录:时间戳、模型名、输入 token、输出 token、延迟(ms)、计算成本(美元)、用户 ID、缓存命中/未命中、请求类别。这数据揭示哪些功能贵、哪些用户重度消费、缓存哪里影响最大。

批处理:批发折扣

OpenAI Batch API 异步处理请求,打 50% 折扣。提交最多 5 万请求的批次,24 小时内返回结果。用于:夜间文档处理、批量分类、评估运行、数据富化管线。不用于:实时面向用户的查询(延迟要紧)。

预算告警与熔断器

熔断器在你触顶时停掉花费。没有它,一个 bug 或滥用能在几小时内烧穿月预算。设三档阈值:

1. 警告(预算 70%):发告警。
2. 降级(预算 85%):只切到便宜模型。
3. 停止(预算 95%):拒新请求,只返回缓存。

优化栈

按顺序叠加这些技术,每层在前一层基础上复利。

一个应用 15 层的 RAG 应用,典型把成本从 22500 美元/月降到 40006000 美元/月。这是烧光跑道和做成生意之间的差别。

真实节省:优化前后

一个服务 1 万 DAU 的 RAG 聊天机器人真实账本:

语义缓存的嵌入成本(180 美元/月)在缓存命中开始的第一小时内就回本了。

二、从零实现

完整代码见原课程 code/caching_cost.py,这里给出关键骨架。

步骤 1:成本计算器

import hashlib, time, json, math

MODEL_PRICING = {
    "gpt-4o":          {"input": 2.50, "output": 10.00, "cached_input": 1.25},
    "gpt-4o-mini":     {"input": 0.15, "output": 0.60,  "cached_input": 0.075},
    "claude-opus-4":   {"input": 15.00,"output": 75.00, "cached_input": 1.50},
    "claude-sonnet-4": {"input": 3.00, "output": 15.00, "cached_input": 0.30},
    "claude-haiku-3.5":{"input": 0.80, "output": 4.00,  "cached_input": 0.08},
    "gemini-2.5-pro":  {"input": 1.25, "output": 10.00, "cached_input": 0.3125},
    "gemini-2.5-flash":{"input": 0.15, "output": 0.60,  "cached_input": 0.0375},
}

def calculate_cost(model, input_tokens, output_tokens, cached_input_tokens=0):
    if model not in MODEL_PRICING:
        return {"error": f"未知模型: {model}"}
    p = MODEL_PRICING[model]
    non_cached = input_tokens - cached_input_tokens
    input_cost = (non_cached / 1_000_000) * p["input"]
    cached_cost = (cached_input_tokens / 1_000_000) * p["cached_input"]
    output_cost = (output_tokens / 1_000_000) * p["output"]
    return {"input_cost": round(input_cost, 6), "cached_input_cost": round(cached_cost, 6),
            "output_cost": round(output_cost, 6),
            "total_cost": round(input_cost + cached_cost + output_cost, 6)}
​

步骤 2:精确缓存

class ExactCache:
    def __init__(self, max_size=1000, ttl_seconds=3600):
        self.cache, self.max_size, self.ttl = {}, max_size, ttl_seconds
        self.hits = self.misses = 0

    def _hash(self, model, messages, temperature):
        key = json.dumps({"model": model, "messages": messages,
                          "temperature": temperature}, sort_keys=True, ensure_ascii=False)
        return hashlib.sha256(key.encode()).hexdigest()

    def get(self, model, messages, temperature=0.0):
        if temperature > 0:                 # 非确定性,跳过缓存
            self.misses += 1; return None
        key = self._hash(model, messages, temperature)
        if key in self.cache:
            entry = self.cache[key]
            if time.time() - entry["timestamp"] < self.ttl:
                self.hits += 1
                entry["access_count"] += 1
                return entry["response"]
            del self.cache[key]
        self.misses += 1
        return None

    def put(self, model, messages, temperature, response):
        if temperature > 0:
            return
        if len(self.cache) >= self.max_size:    # LRU 简化:踢最旧
            oldest = min(self.cache, key=lambda k: self.cache[k]["timestamp"])
            del self.cache[oldest]
        self.cache[self._hash(model, messages, temperature)] = {
            "response": response, "timestamp": time.time(), "access_count": 1}
​

步骤 3:语义缓存

def simple_embed(text):
    """简易词频嵌入(生产中换成真实嵌入 API)。"""
    words = text.lower().split()
    vocab = {}
    for w in words:
        vocab[w] = vocab.get(w, 0) + 1
    norm = math.sqrt(sum(v*v for v in vocab.values()))
    return {k: v/norm for k, v in vocab.items()} if norm else {}

def cosine_similarity(a, b):
    if not a or not b:
        return 0.0
    return sum(a.get(k, 0) * b.get(k, 0) for k in set(a) | set(b))

class SemanticCache:
    def __init__(self, similarity_threshold=0.85, max_size=500, ttl_seconds=3600):
        self.entries, self.threshold = [], similarity_threshold
        self.max_size, self.ttl = max_size, ttl_seconds
        self.hits = self.misses = 0

    def get(self, query):
        q_emb = simple_embed(query)
        now = time.time()
        best_match, best_sim = None, 0.0
        for entry in self.entries:
            if now - entry["timestamp"] > self.ttl:
                continue
            sim = cosine_similarity(q_emb, entry["embedding"])
            if sim > best_sim:
                best_sim, best_match = sim, entry
        if best_match and best_sim >= self.threshold:
            self.hits += 1
            best_match["access_count"] += 1
            return {"response": best_match["response"],
                    "similarity": round(best_sim, 4),
                    "original_query": best_match["query"]}
        self.misses += 1
        return None

    def put(self, query, response):
        if len(self.entries) >= self.max_size:
            self.entries.sort(key=lambda e: e["timestamp"])
            self.entries.pop(0)
        self.entries.append({"query": query, "embedding": simple_embed(query),
                             "response": response, "timestamp": time.time(),
                             "access_count": 1})
​

💡 阈值 0.85 是演示用,生产用 0.92~0.95。太低会把「退货政策」和「换货政策」混为一谈,返回错误答案;太高抓不住同义改写。要用你的真实查询集调阈值,在召回与精度间找平衡。

步骤 4:令牌桶限流器

class TokenBucketRateLimiter:
    def __init__(self):
        self.buckets = {}
        self.tiers = {
            "free":      {"capacity": 50_000,   "refill_rate": 500,   "max_rpm": 10},
            "pro":       {"capacity": 500_000,  "refill_rate": 5_000, "max_rpm": 60},
            "enterprise":{"capacity": 5_000_000,"refill_rate": 50_000,"max_rpm": 300},
        }

    def _get_bucket(self, user_id, tier="free"):
        if user_id not in self.buckets:
            cfg = self.tiers.get(tier, self.tiers["free"])
            self.buckets[user_id] = {"tokens": cfg["capacity"], "capacity": cfg["capacity"],
                "refill_rate": cfg["refill_rate"], "last_refill": time.time(),
                "request_timestamps": [], "max_rpm": cfg["max_rpm"], "tier": tier,
                "total_tokens_used": 0}
        return self.buckets[user_id]

    def _refill(self, b):
        now = time.time()
        elapsed = now - b["last_refill"]
        refill = int(elapsed * b["refill_rate"])
        if refill > 0:
            b["tokens"] = min(b["capacity"], b["tokens"] + refill)
            b["last_refill"] = now

    def check(self, user_id, tokens_needed, tier="free"):
        b = self._get_bucket(user_id, tier)
        self._refill(b)
        now = time.time()
        b["request_timestamps"] = [t for t in b["request_timestamps"] if now - t < 60]
        if len(b["request_timestamps"]) >= b["max_rpm"]:
            return {"allowed": False, "reason": "rate_limit",
                    "retry_after_seconds": 60 - (now - b["request_timestamps"][0])}
        if b["tokens"] < tokens_needed:
            wait = (tokens_needed - b["tokens"]) / b["refill_rate"]
            return {"allowed": False, "reason": "token_limit",
                    "tokens_available": b["tokens"], "retry_after_seconds": round(wait, 1)}
        return {"allowed": True, "tokens_available": b["tokens"]}

    def consume(self, user_id, tokens_used, tier="free"):
        b = self._get_bucket(user_id, tier)
        b["tokens"] -= tokens_used
        b["request_timestamps"].append(time.time())
        b["total_tokens_used"] += tokens_used
​

步骤 5:成本追踪器与三档熔断

class CostTracker:
    def __init__(self, monthly_budget=1000.0):
        self.logs, self.monthly_budget, self.alerts = [], monthly_budget, []

    def log_call(self, model, input_tokens, output_tokens, cached_input_tokens=0,
                 latency_ms=0, user_id="anonymous", cache_status="miss"):
        cost = calculate_cost(model, input_tokens, output_tokens, cached_input_tokens)
        self.logs.append({"timestamp": time.time(), "model": model,
            "input_tokens": input_tokens, "output_tokens": output_tokens,
            "cached_input_tokens": cached_input_tokens, "latency_ms": latency_ms,
            "cost": cost.get("total_cost", 0), "user_id": user_id,
            "cache_status": cache_status})
        self._check_budget()
        return self.logs[-1]

    def _check_budget(self):
        total = self.total_cost()
        pct = total / self.monthly_budget if self.monthly_budget > 0 else 0
        levels = [(0.95, "stop"), (0.85, "throttle"), (0.70, "warning")]
        for threshold, level in levels:
            if pct >= threshold and not any(a["level"] == level for a in self.alerts):
                self.alerts.append({"level": level,
                    "message": f"预算已消耗 {pct:.0%}: ${total:.2f}/${self.monthly_budget:.2f}"})

    def total_cost(self):
        return round(sum(e["cost"] for e in self.logs), 6)
​

步骤 6:模型路由器

SIMPLE_KW = ["几点", "地址", "电话", "价格", "退货政策", "你好", "谢谢", "是的"]
COMPLEX_KW = ["分析", "比较", "解释为什么", "写代码", "调试", "架构", "权衡", "评估"]

def classify_complexity(query):
    q = query.lower()
    if len(q.split()) <= 5 or any(k in q for k in SIMPLE_KW):
        return "simple"
    if any(k in q for k in COMPLEX_KW):
        return "complex"
    return "medium"

def route_model(query, tier="pro"):
    complexity = classify_complexity(query)
    table = {
        "simple":  {"free": "gemini-2.5-flash", "pro": "gpt-4o-mini",     "enterprise": "gpt-4o-mini"},
        "medium":  {"free": "gpt-4o-mini",      "pro": "claude-sonnet-4", "enterprise": "claude-sonnet-4"},
        "complex": {"free": "gpt-4o-mini",      "pro": "gpt-4o",          "enterprise": "claude-opus-4"},
    }
    return {"query": query, "complexity": complexity,
            "model": table[complexity].get(tier, "gpt-4o-mini"), "tier": tier}
​

三、框架对比

Anthropic 提示缓存

# import anthropic
# client = anthropic.Anthropic()
# resp = client.messages.create(model="claude-sonnet-5", max_tokens=1024,
#     system=[{"type":"text",
#         "text":"你是 Acme 公司的客服代理……(长系统提示)",
#         "cache_control": {"type":"ephemeral"}}],
#     messages=[{"role":"user","content":"退货政策是什么?"}])
# print(f"输入 token: {resp.usage.input_tokens}")
# print(f"缓存写入 token: {resp.usage.cache_creation_input_tokens}")
# print(f"缓存读取 token: {resp.usage.cache_read_input_tokens}")
​

首次调用写缓存(25% 溢价),之后每次同系统提示前缀的调用读缓存(90% 折扣),缓存 5 分钟,每次命中重置计时。

OpenAI 自动缓存

# from openai import OpenAI
# client = OpenAI()
# resp = client.chat.completions.create(model="gpt-4o", messages=[
#     {"role":"system","content":"你是客服代理……(1024+ token)"},
#     {"role":"user","content":"退货政策是什么?"}])
# print(f"提示 token: {resp.usage.prompt_tokens}")
# print(f"缓存 token: {resp.usage.prompt_tokens_details.cached_tokens}")
​

OpenAI 自动缓存,1024+ token 的前缀匹配近期请求就 50% 折扣,无需改代码,查 prompt_tokens_details.cached_tokens 验证生效。

OpenAI Batch API

# from openai import OpenAI
# client = OpenAI()
# 请求写成 JSONL,每行一个 {custom_id, method, url, body}
# batch_file = client.files.create(file=open("batch_input.jsonl","rb"), purpose="batch")
# batch = client.batches.create(input_file_id=batch_file.id,
#     endpoint="/v1/chat/completions", completion_window="24h")
​

Batch API 全 token 平打 50% 折扣,24 小时内出结果,适合非实时工作负载:评估、数据标注、批量摘要。

生产级语义缓存用 Redis

# 生产中把线性扫描换成向量索引(Redis Vector Search、Pinecone、pgvector)。
# 线性扫描撑 <1000 条;再大用 ANN 做 O(log n) 查找。
​

GPTCache 是开源语义缓存库,支持多种嵌入后端、向量库、淘汰策略;Helicone 作为代理层提供成本追踪、缓存、限流、预算告警;Not Diamond、Martian 是 ML 路由器,从你的流量模式学习,跨厂商优化成本/质量权衡。本节的从零实现让你看清每一层;生产里这些工具能省去重复造轮子。

四、可复用产物

本节产出两个可复用文件(位于原课程 outputs/):

• prompt-cost-optimizer.md:一个元提示,分析你的 LLM 应用并推荐具体成本优化,附 projected savings(预测节省)。
• skill-cost-patterns.md:一个决策框架,按你的用例选择合适的缓存策略、限流配置、模型路由规则。

Python 代码(code/caching_cost.py)是一套独立成本优化层,把 simulate_llm_call 换成真实 API、simple_embed 换成真实嵌入,即可投产;成本计算、缓存、限流、路由、熔断逻辑均无需修改。

五、练习

1. 语义缓存 LRU 淘汰:把最旧优先淘汰换成最久未用(LRU)。跟踪每条目最后访问时间,缓存满时淘汰最久未用的。在 100 条查询上对比两种策略的命中率。

2. 成本预测工具:给定 API 调用日志(CostTracker logs),按近 7 天均值预测月成本,考虑工作日/周末模式。若预测月成本超预算 20% 触发告警。

3. 分层语义缓存:用两个相似度阈值:0.98 高置信命中(直接返回)和 0.90 中置信命中(带免责声明返回「基于一个类似的过往问题……」)。跟踪每档命中来源,测量用户满意度差异。

4. 嵌入式模型路由分类器:把基于关键词的分类器换成基于嵌入的。嵌入 50 条标注查询(简单/中等/复杂),新查询按最近邻分类。在 20 条测试集上测分类准确率。

5. 带降级档的熔断器:70% 预算记警告;85% 自动把所有路由切到最便宜模型(gpt-4o-mini);95% 只返回缓存、拒新查询。用 1000 次请求对 1 美元预算模拟,验证每档阈值正确触发。

本节要点回顾

1. 单位经济决定生死:1 万 DAU × 每天 10 次,光 LLM 月烧 22500 美元,冗余查询占 40~60%。
2. 系统提示是沉默杀手:1500 token 前缀随每次请求计费,10 万请求/天就是 375 美元/天。
3. 厂商缓存打底:Anthropic 显式 cache_control 命中省 90%、OpenAI 自动前缀省 50%、Gemini CachedContent 省 75%。
4. 语义缓存抓改写:嵌入查询、算余弦、阈值 0.92~0.95,「换个说法的同一问题」也命中。
5. 精确缓存抓确定性:温度 0 哈希命中,简单快,只对确定性调用有效。
6. 令牌桶限流护预算:按用户分层配额,允许突发同时强制平均速率。
7. 模型路由省 40~70%:简单查询送 mini/Haiku,复杂查询才动用 GPT-4o/Opus。
8. 成本追踪测才能优:每次调用记模型、token、延迟、成本、用户、缓存状态,钱花在哪一目了然。
9. 三档熔断防失控:70% 警告、85% 降级到便宜模型、95% 只返回缓存。
10. 优化栈复利:厂商缓存 + 精确 + 语义 + 路由 + 限流 + 压缩 + 批处理,组合拳把月账单砍 76%。

下一节,我们转向护栏与安全——当 LLM 应用面向真实用户,如何拦住越狱、提示注入、有害输出、隐私泄露,把质量和成本之上的「不出事」底线守住。