一、为什么要关注提示词缓存
在AI应用落地时,你会发现一个很常见的现象:很多用户会在同一时间问出差不多一样的问题。比如一个智能客服机器人,同时有一万人点进来问“怎么退订会员”,或者前端因为网络不稳定,同一个请求被自动重发了三次。这些请求到了大模型那边,做的事情几乎一模一样:把同样的文字拆成token,再跑一遍神经网络,算出一堆中间结果,最后生成回答。仔细想想,这一万个请求里,有九千九百九十九次其实都在做重复劳动。
更扎心的是,长文本的输入会让模型在预填充阶段消耗大量的计算资源。如果提示词有5000个token,跑一次就要对5000个token做完整的前向计算。一万个相同请求,就是五千万token的重复计算。这中间有大量算力被白白浪费了。所以,提示词缓存不是锦上添花,而是省钱省力的刚需。
二、缓存的两种形态
2.1 结果缓存
最直白的缓存思路就是“把回答存起来”。用户问过“怎么退订会员”,模型给了一段回答,我们把“这个问题·回答”存到缓存里。下次再有人问一模一样的问题,直接把缓存里的回答扔回去,模型连调用都不用调用。
结果缓存虽然简单,但它有一个致命问题:没法处理相似但不完全相同的内容。比如用户问“怎么退订会员”和“我想取消会员”,意思明明一样,但字面不一样。普通的结果缓存一看key不同,直接就穿透了。另外,模型输出还受温度、top_p这些参数影响,同一个问题在不同参数下可能有不同回答,所以结果缓存的key必须考虑这些参数。
2.2 前缀缓存(KV Cache复用)
大模型生成回答的时候,内部其实是分两个阶段的。第一个阶段叫预填充,模型会把输入的所有token都看一遍,算出每个token对应的Key矩阵和Value矩阵,这些中间结果叫KV Cache。第二个阶段叫解码,模型根据已经算好的KV Cache,一个token一个token地往外蹦答案。
如果两个请求的开头一样,那它们在预填充阶段算出来的KV Cache就是一样的。比如系统提示词是“你是一名专业的金融客服”,这个前缀很多请求都会用。我们完全可以把这一段KV Cache存起来,新请求来了,直接复用这一部分,只算后面不同的部分。这能省掉一大半重复计算。
本文要聊的“提示词缓存架构”,主要就是围绕KV Cache复用展开的。而怎么快速判断两个提示词“是不是同一路货色”,就要靠语义哈希。
三、语义哈希:给提示词找指纹
3.1 普通哈希的尴尬
普通哈希算法,比如MD5、SHA256,大家应该都熟。它们有一个特点:输入只要改一个字符,输出就完全变样。比如“今天天气好”和“今天天气真好”,这两个字符串在肉眼看来差别很小,但MD5值完全是两回事。如果拿这种哈希值当缓存key,那只能处理完全相同的文本,稍微改个词就命中不了。
3.2 语义哈希的原理
语义哈希想解决的就是这个问题。它的思路是:先用模型把文本变成一个向量,然后把这个高维向量压缩、二值化,最终生成一串二进制码。二进制码有点像指纹,但和普通哈希不同的是,语义上相似的文本,它们的指纹也相似。
怎么判断两个指纹相似呢?看汉明距离。汉明距离就是两个二进制串之间不同的位数。比如“10110”和“10100”,只有最后一位不同,汉明距离就是1。汉明距离越小,说明两个文本越接近。
3.3 一个演示用的语义哈希器
在真实项目里,我们通常会用sentence-transformers这类模型来生成句子向量。但为了让你快速理解核心逻辑,我这里写了一个简化版,只用Python标准库就能跑。它把文本切成字符n-gram,然后给每个n-gram分配一个随机向量,累加后取正负号,最终得到一个二进制的语义指纹。
# 技术栈:Python 3.9+(仅使用标准库)
import re
import random
class SimpleSemanticHasher:
"""
一个极简的语义哈希器。
核心思路:字符n-gram作为特征,每个特征分配一个随机向量,
文本向量累加后取符号,得到二进制指纹。
生产环境请换成训练好的句子编码器,这里只是演示原理。
"""
def __init__(self, bits=64, ngram=2, seed=42):
self.bits = bits # 指纹位数
self.ngram = ngram # n-gram长度
self._rand = random.Random(seed) # 固定随机种子,保证可复现
self._feature_vectors = {} # 特征 -> 随机向量
def _get_feature_vector(self, feature):
# 同一个特征永远使用同一个随机向量
if feature not in self._feature_vectors:
self._feature_vectors[feature] = [
self._rand.uniform(-1, 1) for _ in range(self.bits)
]
return self._feature_vectors[feature]
def _features(self, text):
# 清洗输入:只保留中文、字母和数字,然后切成n-gram
text = re.sub(r'[^a-zA-Z0-9\u4e00-\u9fff]', '', text.lower())
if len(text) < self.ngram:
return [text]
return [text[i:i + self.ngram] for i in range(len(text) - self.ngram + 1)]
def hash(self, text):
"""
返回十六进制字符串形式的语义指纹。
例如:'3f9a7b...'
"""
sums = [0.0] * self.bits
for feature in self._features(text):
vec = self._get_feature_vector(feature)
for i in range(self.bits):
sums[i] += vec[i]
bit_string = ''.join('1' if v > 0 else '0' for v in sums)
# 把二进制串转成十六进制,便于阅读和存储
return format(int(bit_string, 2), '0{}x'.format(self.bits // 4))
@staticmethod
def hamming_distance(a, b):
"""计算两个十六进制语义指纹的汉明距离,越小越相似。"""
x = int(a, 16)
y = int(b, 16)
return bin(x ^ y).count('1')
# 使用示例
if __name__ == '__main__':
hasher = SimpleSemanticHasher(bits=64, ngram=2)
text1 = "今天天气真好,适合出去玩"
text2 = "今天天气不错,去公园散步吧"
text3 = "量子力学是一种很难懂的理论"
h1 = hasher.hash(text1)
h2 = hasher.hash(text2)
h3 = hasher.hash(text3)
print("文本1指纹:", h1)
print("文本2指纹:", h2)
print("文本3指纹:", h3)
print("文本1与文本2的汉明距离:", SimpleSemanticHasher.hamming_distance(h1, h2))
print("文本1与文本3的汉明距离:", SimpleSemanticHasher.hamming_distance(h1, h3))
从结果里你能看到,文本1和文本2的汉明距离明显小于文本1和文本3的距离,因为它们共享了“今天”“天气”这些特征。这就是语义哈希的直觉:越相似的东西,指纹越接近。
四、缓存架构整体设计
4.1 请求来了怎么办
有了一套能生成语义指纹的工具,我们就能设计缓存流程了。假设现在来了一个请求,完整的处理步骤是这样的:
- 拿到用户提示词,以及模型名、温度、top_p等参数。
- 用语义哈希器给提示词算一个语义指纹。
- 把模型参数序列化以后,用普通哈希生成一个参数哈希。
- 用“语义指纹 + 参数哈希”组合成缓存key。
- 拿这个key去缓存里查。
- 如果命中了,还不能直接返回,要做一步相似度校验,确认缓存里的原始文本和当前提示词真的够接近,防止语义哈希碰撞造成错误回答。
- 校验通过后,要么直接返回缓存好的回答,要么把缓存的KV Cache接着往下用。
- 如果没命中,就走正常的模型推理,算完之后把结果和中间状态写回缓存。
4.2 Key怎么设计
缓存key是架构里牵一发动全身的地方。如果只用语义指纹当key,有两个隐患。第一,不同文本可能因为哈希碰撞得到同一个指纹,虽然概率很低,但在高并发下依然会有脏数据。第二,语义接近但不完全相同的请求,直接返回同一个回答,有时候是没问题的,有时候会闹笑话。
所以key一定要是组合形式,类似下面这样:
sem_hash = hasher.hash(prompt)
param_hash = hashlib.md5(f"{model}:{temperature}:{top_p}".encode()).hexdigest()[:16]
key = f"{sem_hash}:{param_hash}"
这样做的好处是,既允许语义相似的提示词互相碰撞,又用参数哈希把那些“语义一样但参数不同”的请求分开。
4.3 缓存的存储结构
缓存不能无限涨,否则内存会爆炸。我们需要两个机制:一个是过期时间TTL,放太久的数据自动扔掉;另一个是LRU淘汰,容量满了以后,优先淘汰最久没被访问的数据。
下面这段代码是一个带TTL和LRU的本地缓存实现。生产环境里你可以换成Redis,但核心逻辑是一样的。
# 技术栈:Python 3.9+(仅使用标准库)
import time
from collections import OrderedDict
class PromptCache:
"""
一个简单的本地缓存,支持:
- 过期时间(TTL)
- LRU淘汰
key为字符串,value为任意对象(比如缓存的响应文本或KV Cache引用)
"""
def __init__(self, capacity=1024, ttl=300):
self.capacity = capacity
self.ttl = ttl
self._store = OrderedDict()
def get(self, key):
if key not in self._store:
return None
value, expire_at = self._store[key]
if time.time() > expire_at:
# 已过期,删掉并返回None
del self._store[key]
return None
# 移动到末尾表示最近被访问
self._store.move_to_end(key)
return value
def put(self, key, value):
if key in self._store:
del self._store[key]
# 如果容量不够,弹出最久未使用的那一项
while len(self._store) >= self.capacity:
self._store.popitem(last=False)
self._store[key] = (value, time.time() + self.ttl)
self._store.move_to_end(key)
五、高并发下的请求合并
5.1 缓存穿透问题
缓存虽然好用,但有一个经典的高并发难题叫缓存穿透。假设缓存里还没有数据,这时候突然来了一千个完全相同的请求。如果没有保护措施,这一千个请求会全部认为“缓存没中”,然后一起冲到模型层。模型会被打得很惨。所以我们还需要一个请求合并的机制。
所谓请求合并,也叫Single-Flight,简单说就是:同一时刻,同一个key的请求,只允许一个真正去执行模型调用,其他请求在旁边等着,等第一个计算完了,大家一起用结果。
5.2 用asyncio实现请求合并
Python里的asyncio可以优雅地实现这个机制。核心是用一个字典保存“正在执行的任务”,key就是缓存key,value是一个Future对象。当第二个相同请求到达时,发现已经在执行了,就直接等着那个Future完成。
# 技术栈:Python 3.9+
import asyncio
class RequestCoalescer:
"""
请求合并器:同一时刻同一个key的请求只会触发一次真正的计算。
别的请求都会等待这个计算结果。
"""
def __init__(self):
self._inflight = {} # key -> Future
async def execute(self, key, coro_func):
"""
key: 缓存key,通常就是语义哈希+参数哈希
coro_func: 一个异步函数,调用它代表执行真正的模型请求
"""
# 如果已经有请求在处理中,直接复用
if key in self._inflight:
future = self._inflight[key]
return await future
# 创建新的Future,保存到字典里
loop = asyncio.get_running_loop()
future = loop.create_future()
self._inflight[key] = future
try:
# 执行真正的模型调用
result = await coro_func()
future.set_result(result)
except Exception as e:
# 把异常也传递给等待方
future.set_exception(e)
finally:
# 清理掉这个key,避免占用内存
self._inflight.pop(key, None)
return await future
这样一来,即使缓存里没有数据,同一瞬间并发一万个相同请求,模型也只会收到一次调用。剩下的请求只是挂在那里等待,不会有额外的算力消耗。
5.3 组合使用
实际项目中,我们会把PromptCache和RequestCoalescer放在同一个服务里。先查缓存,缓存命中了就直接返回。缓存没中的话,用请求合并器把并发请求聚合成一个,然后统一去模型那边拿结果,拿到以后再回填缓存。这套组合拳能避免大部分重复计算。
六、实际应用场景详解
6.1 场景一:Prompt模板复用
很多产品会用固定的模板拼接用户输入。比如“你是{行业}领域的专家,请回答:{问题}”。不同用户只是行业和问题不一样。如果把模板的固定前缀,比如“你是”“领域的专家,请回答”这些部分对应的KV Cache缓存起来,再结合语义哈希对动态部分做相似匹配,就能省掉大量前缀计算。比如“汽车行业”和“汽车领域”语义很接近,可以复用同一段前缀。
6.2 场景二:用户反复重试
用户在聊天窗口里点“重新生成”按钮,前端经常一瞬间发出好几个一模一样的请求。以前模型要傻乎乎地算好几遍,现在有了缓存和请求合并,第一个请求算完后,后面几个请求直接拿结果。用户感受到的只有丝滑,后端压力也小很多。
6.3 场景三:Agent/多轮对话循环
Agent在思考过程中会反复调用大模型,而且系统提示词往往是不变的。比如每一轮推理都用“你是一个擅长拆解任务的AI助手”这句话开头。语义哈希可以快速识别出这个前缀,把对应的KV Cache复用起来,Agent的执行成本会大幅下降。
七、技术优缺点分析
7.1 优点
- 能显著减少重复的预填充计算,省下真金白银的算力成本。
- 相同或相似请求的响应速度更快,用户体验更好。
- 语义哈希让缓存不仅仅是“完全匹配”,还能覆盖“意思差不多”的请求,命中率更高。
- 配合请求合并,可以有效缓解高并发场景下的缓存穿透和击穿问题。
7.2 缺点
- 语义哈希是近似算法,存在误判风险。如果把两个其实不相关的文本判成相似,就会返回错误回答。
- 生成语义指纹本身需要跑一次文本编码模型,这个开销如果太大,缓存省下的算力可能还不够填这个坑。
- KV Cache复用会增加缓存系统的复杂度,特别是分布式环境下的数据一致性很难保证。
- 模型版本升级后,旧缓存可能失效,甚至产生不兼容的结果,版本管理很麻烦。
八、注意事项
8.1 阈值要谨慎设定
汉明距离到底小于多少才算“相似”?这个阈值直接影响命中率和准确率。阈值设太小,命中率太低;设太大,误判率太高。建议离线准备一批样本,统计相似请求和不相似请求的汉明距离分布,选一个中间值。实在拿不准的时候,宁可不命中,也不要错。
8.2 缓存key必须带上模型参数
温度、top_p、max_tokens这些参数会直接影响输出。如果缓存key里只放提示词指纹,不同参数的用户可能会拿到同一个回答,那肯定不行。所以参数哈希是key里必不可少的组成部分。
8.3 复用之前做二次校验
语义指纹只能作为粗筛。真正要复用KV Cache或者返回缓存结果之前,最好把缓存里存的原始提示词和当前提示词进行一次更严格的相似度计算,比如用向量算一下余弦相似度,或者做一次token级的前缀匹配。这样就算语义哈希发生了碰撞,也有最后一道防线。
8.4 敏感信息不要缓存
如果提示词里带着用户名、手机号、住址之类的隐私数据,缓存得越久,泄露风险越大。需要设计脱敏策略,或者直接跳过缓存。这个问题在客服系统里特别常见,一定要重视。
8.5 容量和淘汰策略
KV Cache本身是很重的数据,不像一段文本那么简单。缓存容量需要严格限制,最好能监控显存和内存占用。LRU和TTL是基本配置,有时候还需要按请求量做动态调整。
8.6 模型版本管理
模型升级以后,同样一段提示词内部算出的向量会变,语义指纹也会变。如果不把模型版本写进key,旧缓存的数据可能会被新模型错误地复用。最简单的办法就是每个请求在构建缓存key时,把模型版本号一起拼接进去。
九、总结
提示词缓存并不是一个万能开关,它更像是一把双刃剑。用好了,能让高并发的大模型服务又快又省钱;用不好,轻则命中率低下,重则返回荒谬的错误答案。语义哈希是整套架构的灵魂,它让缓存从“一字不差才能命中”升级成“意思相近也能复用”。再配上KV Cache前缀缓存、请求合并、LRU淘汰和TTL过期,一个完整的提示词缓存架构就搭建起来了。关键是记住一个原则:大胆缓存,小心验证,在算力成本和回答质量之间找到最合适的平衡点。
Comments