API 响应缓存策略：减少重复调用，降低成本

调用大模型 API 时，相同的输入往往会得到相同的输出。如果不加缓存，每次都要重新计算，既浪费钱又浪费时间。

这篇文章把 API 响应缓存从简单到生产级讲清楚，帮你节省 30-50% 的 API 成本。

为什么需要缓存

场景	是否适合缓存
相同问题多次查询	适合
系统提示词 + 固定上下文	适合
实时性要求高的对话	不适合
每次输入都不同	不适合

缓存的核心原则：只缓存幂等的、可复用的响应。

基础缓存实现

使用字典缓存

最简单的缓存方式：

from openai import OpenAI
import hashlib
import json

client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="YOUR_BASE_URL",
)

# 全局缓存
cache = {}

def get_cache_key(messages, model):
    """生成缓存键"""
    data = json.dumps({"messages": messages, "model": model}, sort_keys=True)
    return hashlib.md5(data.encode()).hexdigest()

def chat_with_cache(messages, model="YOUR_MODEL"):
    cache_key = get_cache_key(messages, model)
    
    # 检查缓存
    if cache_key in cache:
        print("命中缓存")
        return cache[cache_key]
    
    # 调用 API
    print("调用 API")
    response = client.chat.completions.create(
        model=model,
        messages=messages,
    )
    result = response.choices[0].message.content
    
    # 存入缓存
    cache[cache_key] = result
    
    return result

# 测试
messages = [{"role": "user", "content": "什么是 Python？"}]

# 第一次调用
result1 = chat_with_cache(messages)
print(result1)

# 第二次调用（相同输入）
result2 = chat_with_cache(messages)
print(result2)

第一次调用会请求 API，第二次调用直接返回缓存结果。

使用 diskcache 持久化缓存

字典缓存重启后就丢失了。使用 diskcache 可以把缓存持久化到磁盘：

pip install diskcache

from openai import OpenAI
import hashlib
import json
from diskcache import Cache

client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="YOUR_BASE_URL",
)

# 创建磁盘缓存
cache = Cache("./api_cache")

def get_cache_key(messages, model):
    data = json.dumps({"messages": messages, "model": model}, sort_keys=True)
    return hashlib.md5(data.encode()).hexdigest()

def chat_with_disk_cache(messages, model="YOUR_MODEL"):
    cache_key = get_cache_key(messages, model)
    
    # 检查缓存
    if cache_key in cache:
        print("命中磁盘缓存")
        return cache[cache_key]
    
    # 调用 API
    print("调用 API")
    response = client.chat.completions.create(
        model=model,
        messages=messages,
    )
    result = response.choices[0].message.content
    
    # 存入缓存
    cache.set(cache_key, result)
    
    return result

# 测试
messages = [{"role": "user", "content": "什么是 Python？"}]
result = chat_with_disk_cache(messages)
print(result)

缓存数据会保存在 ./api_cache 目录中，重启程序后仍然有效。

设置缓存过期时间

缓存不应该永久有效，需要设置过期时间：

from diskcache import Cache
import time

cache = Cache("./api_cache")

def chat_with_ttl(messages, model="YOUR_MODEL", ttl=3600):
    """
    ttl: 缓存过期时间（秒），默认 1 小时
    """
    cache_key = get_cache_key(messages, model)
    
    # 检查缓存
    if cache_key in cache:
        print("命中缓存")
        return cache[cache_key]
    
    # 调用 API
    print("调用 API")
    response = client.chat.completions.create(
        model=model,
        messages=messages,
    )
    result = response.choices[0].message.content
    
    # 存入缓存，设置过期时间
    cache.set(cache_key, result, expire=ttl)
    
    return result

# 测试：缓存 1 小时后过期
result = chat_with_ttl(
    [{"role": "user", "content": "什么是 Python？"}],
    ttl=3600
)

缓存统计

需要了解缓存的命中率，评估缓存效果：

from collections import defaultdict

class CachedAPIClient:
    def __init__(self, api_key, base_url):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.cache = Cache("./api_cache")
        self.stats = defaultdict(int)
    
    def get_cache_key(self, messages, model):
        data = json.dumps({"messages": messages, "model": model}, sort_keys=True)
        return hashlib.md5(data.encode()).hexdigest()
    
    def chat(self, messages, model="YOUR_MODEL", ttl=3600):
        cache_key = self.get_cache_key(messages, model)
        
        # 检查缓存
        if cache_key in self.cache:
            self.stats["cache_hit"] += 1
            return self.cache[cache_key]
        
        # 调用 API
        self.stats["cache_miss"] += 1
        response = self.client.chat.completions.create(
            model=model,
            messages=messages,
        )
        result = response.choices[0].message.content
        
        # 存入缓存
        self.cache.set(cache_key, result, expire=ttl)
        
        return result
    
    def get_stats(self):
        total = self.stats["cache_hit"] + self.stats["cache_miss"]
        hit_rate = self.stats["cache_hit"] / total if total > 0 else 0
        
        return {
            "cache_hit": self.stats["cache_hit"],
            "cache_miss": self.stats["cache_miss"],
            "hit_rate": f"{hit_rate:.2%}",
            "total_requests": total,
        }

# 测试
client = CachedAPIClient(
    api_key="YOUR_API_KEY",
    base_url="YOUR_BASE_URL"
)

# 多次调用
for i in range(10):
    client.chat([{"role": "user", "content": "什么是 Python？"}])

for i in range(5):
    client.chat([{"role": "user", "content": f"问题{i}"}])

# 查看统计
print(client.get_stats())
# 输出: {'cache_hit': 9, 'cache_miss': 6, 'hit_rate': '60.00%', 'total_requests': 15}

使用 Redis 缓存

生产环境通常使用 Redis 作为缓存：

pip install redis

import redis
import json
import hashlib
from openai import OpenAI

class RedisCachedAPIClient:
    def __init__(self, api_key, base_url, redis_host="localhost", redis_port=6379):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True)
    
    def get_cache_key(self, messages, model):
        data = json.dumps({"messages": messages, "model": model}, sort_keys=True)
        return f"api_cache:{hashlib.md5(data.encode()).hexdigest()}"
    
    def chat(self, messages, model="YOUR_MODEL", ttl=3600):
        cache_key = self.get_cache_key(messages, model)
        
        # 检查缓存
        cached = self.redis.get(cache_key)
        if cached:
            print("命中 Redis 缓存")
            return cached
        
        # 调用 API
        print("调用 API")
        response = self.client.chat.completions.create(
            model=model,
            messages=messages,
        )
        result = response.choices[0].message.content
        
        # 存入 Redis，设置过期时间
        self.redis.setex(cache_key, ttl, result)
        
        return result

# 测试
client = RedisCachedAPIClient(
    api_key="YOUR_API_KEY",
    base_url="YOUR_BASE_URL"
)

result = client.chat([{"role": "user", "content": "什么是 Python？"}])
print(result)

缓存预热

对于高频查询，可以提前预热缓存：

def warmup_cache(client, common_questions):
    """预热缓存"""
    print(f"开始预热缓存，共 {len(common_questions)} 个问题")
    
    for i, question in enumerate(common_questions):
        messages = [{"role": "user", "content": question}]
        client.chat(messages)
        
        if (i + 1) % 10 == 0:
            print(f"已预热 {i + 1}/{len(common_questions)}")
    
    print("缓存预热完成")

# 测试
common_questions = [
    "什么是 Python？",
    "什么是机器学习？",
    "什么是深度学习？",
    # ... 更多常见问题
]

warmup_cache(client, common_questions)

缓存失效策略

当输入变化时，需要主动清除缓存：

def invalidate_cache(client, messages):
    """清除指定缓存"""
    cache_key = client.get_cache_key(messages, "YOUR_MODEL")
    
    # 清除 Redis 缓存
    client.redis.delete(cache_key)
    
    # 清除 diskcache
    if cache_key in client.cache:
        del client.cache[cache_key]
    
    print(f"已清除缓存: {cache_key}")

# 测试
messages = [{"role": "user", "content": "什么是 Python？"}]
invalidate_cache(client, messages)

缓存大小限制

缓存不能无限增长，需要限制大小：

from diskcache import Cache

# 限制缓存大小为 1GB
cache = Cache("./api_cache", size_limit=1e9)

# 或者限制条目数
cache = Cache("./api_cache", eviction_policy="least-recently-used")

策略	说明
LRU	最近最少使用，优先淘汰
LFU	最不经常使用，优先淘汰
FIFO	先进先出

成本计算

假设：

• API 价格：$0.01 / 1K tokens
• 每天 1000 次请求
• 缓存命中率：40%

场景	每日请求数	每日成本
无缓存	1000	$10
有缓存（40% 命中）	600	$6
节省	-	$4/天，$120/月

缓存命中率越高，节省越多。

快速排错表

问题	可能原因	解决方法
缓存命中率低	输入变化太大	优化输入，提取公共部分
缓存占用空间大	没有设置过期时间	设置 TTL，限制缓存大小
缓存数据过期	TTL 设置太短	根据业务需求调整 TTL
缓存不一致	没有及时失效	实现主动失效机制

配置检查清单

检查项	建议值
缓存类型	开发用 diskcache，生产用 Redis
TTL	根据业务需求，通常 1-24 小时
缓存大小限制	根据磁盘/内存情况
是否统计命中率	是
是否支持主动失效	是

缓存是降低 API 成本最简单有效的方法。对于重复查询较多的场景，缓存可以节省 30-50% 的成本。选择合适的缓存策略，设置合理的过期时间，监控缓存命中率，就能最大化缓存效果。