← 返回首页
🧠

LLM缓存策略

📂 llm ⏱ 2 min 340 words

--- 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缓存策略是降低成本、提升性能的关键手段。精确匹配、语义缓存、多级缓存、预热和失效策略的组合使用,能构建高效的缓存体系。持续监控和优化是保持缓存效果的关键。