LLM缓存策略
--- title: "LLM缓存策略" description: "详解LLM系统中的各种缓存策略,提升响应速度并降低API调用成本" tags: ["缓存策略", "性能优化", "成本控制"] category: "llm" icon: "🧠"
LLM缓存策略
LLM API调用成本高且延迟大,合理的缓存策略能显著降低成本并提升用户体验。缓存策略需要在命中率、新鲜度和存储成本之间找到平衡点。
为什么需要缓存
LLM API调用的平均延迟在1-5秒,每次调用成本从几美分到几美元不等。对于重复性查询,缓存可以将响应时间降低到毫秒级,同时完全避免API调用成本。
常见适合缓存的场景包括:FAQ类问答、模板化内容生成、代码生成常见模式、翻译常用语句等。
精确匹配缓存
最基本的缓存策略是对输入进行精确匹配:
import hashlib
import json
class ExactCache:
def __init__(self, ttl=3600):
self.cache = {}
self.ttl = ttl
def _make_key(self, prompt, **kwargs):
content = json.dumps({"prompt": prompt, **kwargs}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
def get(self, prompt, **kwargs):
key = self._make_key(prompt, **kwargs)
if key in self.cache:
entry = self.cache[key]
if time.time() - entry["time"] < self.ttl:
return entry["response"]
return None
def set(self, prompt, response, **kwargs):
key = self._make_key(prompt, **kwargs)
self.cache[key] = {"response": response, "time": time.time()}
精确匹配缓存简单高效,但命中率依赖于输入的规范性。
语义缓存
基于语义相似度的缓存,能够匹配含义相近但表述不同的查询:
from sentence_transformers import SentenceTransformer
import numpy as np
class SemanticCache:
def __init__(self, similarity_threshold=0.92):
self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
self.threshold = similarity_threshold
self.cache = []
def get(self, query):
query_embedding = self.encoder.encode(query)
for entry in self.cache:
similarity = np.dot(query_embedding, entry["embedding"])
if similarity >= self.threshold:
return entry["response"]
return None
def set(self, query, response):
embedding = self.encoder.encode(query)
self.cache.append({"query": query, "embedding": embedding, "response": response})
语义缓存能显著提升命中率,但需要维护向量索引。
多级缓存
结合不同层级的缓存提升命中率:
class MultiLevelCache:
def __init__(self):
self.l1_cache = {} # 内存缓存
self.l2_cache = RedisCache() # 分布式缓存
self.l3_cache = DiskCache() # 磁盘缓存
async def get(self, key):
# L1: 内存
if key in self.l1_cache:
return self.l1_cache[key]
# L2: Redis
result = await self.l2_cache.get(key)
if result:
self.l1_cache[key] = result
return result
# L3: 磁盘
result = await self.l3_cache.get(key)
if result:
self.l1_cache[key] = result
await self.l2_cache.set(key, result)
return result
return None
多级缓存逐层查找,命中速度递减但容量递增。
缓存预热
提前填充热门查询的缓存:
class CacheWarmer:
def __init__(self, cache, llm_client):
self.cache = cache
self.client = llm_client
async def warm_popular_queries(self, queries):
tasks = []
for query in queries:
if not self.cache.get(query):
tasks.append(self.warm_single(query))
await asyncio.gather(*tasks)
async def warm_single(self, query):
response = await self.client.generate(query)
self.cache.set(query, response)
缓存预热能在服务启动时就准备好热门数据,避免冷启动问题。
缓存失效策略
合理的失效策略确保缓存数据的新鲜度:
class CacheInvalidation:
def __init__(self):
self.strategies = {
"ttl": self.ttl_invalidation,
"lru": self.lru_invalidation,
"manual": self.manual_invalidation,
}
def ttl_invalidation(self, entry, max_age=3600):
return time.time() - entry["time"] > max_age
def lru_invalidation(self, entry, max_size=1000):
return self.get_access_count(entry["key"]) == 0
def manual_invalidation(self, entry, version):
return entry.get("version", 0) < version
TTL适合时效性要求高的场景,LRU适合容量受限场景。
缓存统计与监控
监控缓存性能帮助优化策略:
class CacheMetrics:
def __init__(self):
self.hits = 0
self.misses = 0
def record_hit(self):
self.hits += 1
def record_miss(self):
self.misses += 1
@property
def hit_rate(self):
total = self.hits + self.misses
return self.hits / total if total > 0 else 0
持续监控缓存命中率,低于阈值时需要调整缓存策略。
总结
LLM缓存策略是降低成本、提升性能的关键手段。精确匹配、语义缓存、多级缓存、预热和失效策略的组合使用,能构建高效的缓存体系。持续监控和优化是保持缓存效果的关键。