【高速缓存】RedisVL缓存 LLM 响应实践指南

引言

在现代 AI 应用中，调用大语言模型（LLM）API 不仅会产生可观的费用，还会带来不可忽视的延迟。当用户反复提出相同或相似的问题时，每次都调用 LLM 无疑是一种浪费。语义缓存（Semantic Cache）正是为了解决这个问题而诞生 —— 它利用向量相似度搜索，将用户查询的语义嵌入与缓存中的历史问题进行比较，如果找到语义足够接近的条目，则直接返回缓存的答案，从而避免重复调用 LLM API。

RedisVL 提供了SemanticCache类，基于 Redis 作为向量数据库和缓存存储，实现了高效、可定制的语义缓存。本指南将带你从零开始，逐步掌握其使用方法和核心原理。

前置条件

在开始之前，请确保你已具备以下条件：

• 已安装 RedisVL：pip install redisvl
• 一个正在运行的 Redis 实例（推荐 Redis 8+ 或 Redis Cloud）
• 一个有效的 OpenAI API 密钥（用于演示调用 LLM）

你将学到什么

完成本指南后，你将能够：

• 配置并初始化一个语义缓存实例
• 存储和检索缓存的 LLM 响应
• 理解缓存条目的entry_id和 Rediskey的区别，并利用它们进行精确操作
• 自定义语义相似度阈值，平衡召回率与精确率
• 配置 TTL（生存时间）策略，并理解check()方法对 TTL 的刷新行为
• 通过标签和过滤器实现多用户场景下的访问控制

语义缓存的工作原理

语义缓存的核心是向量化和相似度搜索。当用户提出问题时，系统会：

1. 使用嵌入模型将问题文本转换为高维向量（例如 768 维）。
2. 在 Redis 中执行向量相似度搜索，找出与当前问题向量最接近的已缓存问题。
3. 计算它们之间的余弦距离（取值范围 0~2，0 表示完全相同，2 表示完全相反）。
4. 如果最小距离小于设定的阈值，则认为语义匹配，返回对应的缓存响应。
5. 如果未匹配，则调用 LLM 获取真实响应，并将（问题向量、问题文本、响应、元数据）存入缓存，供后续使用。

下面这张流程图清晰地展示了这一过程：

理解这个流程后，我们开始动手实践。

环境准备

首先，导入必要的库并设置 OpenAI 客户端。

importosimportgetpassimporttimeimportnumpyasnpfromopenaiimportOpenAI# 避免 tokenizers 并行警告os.environ["TOKENIZERS_PARALLELISM"]="False"# 获取 OpenAI API 密钥api_key=os.getenv("OPENAI_API_KEY")orgetpass.getpass("输入你的 OpenAI API 密钥: ")client=OpenAI(api_key=api_key)defask_openai(question:str)->str:"""调用 OpenAI 的补全接口回答问题"""response=client.completions.create(model="gpt-4o-mini",prompt=f"请用简洁的方式回答以下问题：{question}",max_tokens=200)returnresponse.choices[0].text.strip()# 测试一下print(ask_openai("法国的首都是哪里？"))# 输出: 巴黎

初始化语义缓存

SemanticCache在初始化时会自动在 Redis 中创建所需的索引结构（若不存在）。我们使用 Hugging Face 的HFTextVectorizer作为嵌入模型（本例使用redis/langcache-embed-v2，但你可以替换为任何其他模型）。

importwarnings warnings.filterwarnings('ignore')fromredisvl.extensions.cache.llmimportSemanticCachefromredisvl.utils.vectorizeimportHFTextVectorizer llmcache=SemanticCache(name="llmcache",# Redis 索引名称redis_url="redis://localhost:6379",# Redis 连接地址distance_threshold=0.1,# 余弦距离阈值（0~2，越小越严格）vectorizer=HFTextVectorizer("redis/langcache-embed-v2")# 嵌入模型)

注意：如果你使用的嵌入模型版本与 RedisVL 预期的不一致，可能会看到警告信息，这通常不影响功能，但建议保持版本匹配。

你可以通过 RedisVL 命令行工具查看索引的详细信息：

rvl index info-illmcache

输出会显示索引的字段结构，其中包括prompt（文本）、response（文本）、inserted_at、updated_at和prompt_vector（向量字段）等。

基本缓存操作

1. 检查缓存（首次为空）

question="法国的首都是哪里？"# 检查缓存是否命中ifresponse:=llmcache.check(prompt=question):print(response)else:print("缓存为空")# 输出: 缓存为空

2. 存储条目

# 存储问题、答案及任意元数据llmcache.store(prompt=question,response="巴黎",metadata={"city":"巴黎","country":"法国"})# 返回完整的 Redis key，例如 'llmcache:115049a...'

3. 再次检查缓存

# 用完全相同的问题查询result=llmcache.check(prompt=question,return_fields=["prompt","response","metadata"])print(result)# 输出包含缓存内容# 用语义相似的问题查询similar_question="法国真正的首都是哪里？"result=llmcache.check(prompt=similar_question)print(result[0]['response'])# 输出: 巴黎

条目 ID 与 Redis Key

每个缓存条目有两个重要的标识符：

• entry_id：由prompt和过滤条件组合后通过 SHA256 哈希生成。相同的 prompt + 相同 filters 会生成相同的entry_id，因此重复存储会覆盖之前的条目。
• key：完整的 Redis 键，格式为{index_name}:{entry_id}。在 Redis 中，键是唯一的，用于直接操作。

# 存储时返回完整的 keykey=llmcache.store(prompt="法国的首都是哪里？",response="巴黎",metadata={"source":"地理"})print(f"完整 Redis key:{key}")# 检查时可以通过 return_fields 获取 entry_id 和 keyresult=llmcache.check(prompt="法国的首都是哪里？",return_fields=["entry_id","prompt","response"])print(f"Entry ID:{result[0]['entry_id']}")print(f"Key:{result[0]['key']}")

获取和删除特定条目

你可以通过entry_id或key来精确获取或删除缓存条目。

fromredisvl.queryimportFilterQueryfromredisvl.query.filterimportFilterExpression# 列出所有缓存条目（使用底层索引查询）query=FilterQuery(filter_expression=FilterExpression("*"),return_fields=["entry_id","prompt","response"])all_entries=llmcache._index.query(query)print(f"缓存中共有{len(all_entries)}条记录:")forentryinall_entries:print(f" - entry_id:{entry['entry_id'][:20]}... prompt:{entry['prompt'][:30]}...")# 根据 entry_id 获取特定记录entry_id=result[0]['entry_id']record=llmcache._index.fetch(entry_id)print(f"获取的记录:{record}")# 删除指定条目（通过 entry_id 列表）llmcache.drop(ids=[entry_id])# 也可以使用 keys 参数删除：llmcache.drop(keys=[key])# 验证删除result=llmcache.check(prompt="法国的首都是哪里？")print(f"删除后:{result}")# 输出空列表

自定义相似度阈值

相似度阈值决定了缓存命中的严格程度。阈值（distance_threshold）采用余弦距离，取值范围为[0, 2]，其中 0 表示完全相同，2 表示完全相反。阈值越小，匹配越严格（高精确率，低召回率）；阈值越大，匹配越宽松（高召回率，低精确率）。

你可以随时调整阈值：

# 将阈值放宽到 0.5（允许更不相似的问题命中）llmcache.set_threshold(0.5)# 重新存储一个条目llmcache.store(prompt="法国的首都是哪里？",response="巴黎")# 测试一个稍远的问题question="尼斯所在国家的首都是哪里？"# 尼斯在法国result=llmcache.check(prompt=question)print(result[0]['response'])# 输出: 巴黎（因为阈值放宽，仍然命中）

如果需要清空整个缓存，使用clear()方法：

llmcache.clear()# 删除所有条目print(llmcache.check(prompt=question))# 空列表

TTL（生存时间）策略

TTL 可以让缓存条目在指定时间后自动过期，避免缓存无限增长。RedisVL 的SemanticCache支持灵活的 TTL 设置。

基本用法

# 设置 TTL 为 5 秒llmcache.set_ttl(5)llmcache.store("这是一个 TTL 测试","这是测试响应")time.sleep(6)# 确认缓存已自动过期result=llmcache.check("这是一个 TTL 测试")print(result)# 输出: []# 重置 TTL 为 None（永久存储）llmcache.set_ttl()# 不传参数即为 None

TTL 刷新行为详解

check()方法在命中缓存时，会自动刷新所有匹配条目的 TTL（滑窗模式）。这意味着频繁访问的条目会一直保持活跃。下表总结了不同场景下的行为：

重要提醒：因为check()会刷新所有匹配的条目（可能有多条），所以即使你原本只想获取一条数据，所有匹配的条目都会延长生存时间。如果你不希望某些条目被意外刷新，可以通过过滤器来精确控制匹配范围。

# 示例：设置 TTL 为 5 分钟llmcache.set_ttl(300)llmcache.store("什么是 Python？","一种编程语言")# 每次查询命中都会重置 TTLresult=llmcache.check("什么是 Python？")# TTL 被刷新为 300 秒# 重置并清空llmcache.set_ttl()llmcache.clear()

简单性能测试

我们来对比一下使用缓存前后的响应时间。

defanswer_question(question:str)->str:"""先查缓存，未命中则调用 LLM"""results=llmcache.check(prompt=question)ifresults:returnresults[0]["response"]else:answer=ask_openai(question)returnanswer# 首次请求（未命中缓存，调用 LLM）question="美国第一任总统是谁？"start=time.time()answer=answer_question(question)end=time.time()print(f"未使用缓存，调用 OpenAI 耗时:{end-start:.4f}秒")# 存储正确答案到缓存llmcache.store(prompt=question,response="乔治·华盛顿")# 后续 10 次请求（命中缓存）times=[]for_inrange(10):cached_start=time.time()cached_answer=answer_question(question)cached_end=time.time()times.append(cached_end-cached_start)avg_time=np.mean(times)print(f"使用缓存平均耗时:{avg_time:.4f}秒")print(f"节省时间百分比:{((end-start)-avg_time)/(end-start)*100:.2f}%")# 通常可节省 95% 以上的时间

查看索引统计信息：

rvl stats-illmcache

最后，删除整个缓存及索引：

llmcache.delete()# 清除所有数据并删除索引

访问控制：标签与过滤器

在多用户或多租户场景下，你需要确保不同用户的数据相互隔离。SemanticCache支持自定义可过滤字段（filterable fields），允许你为每个条目打上标签或数值，并在查询时通过过滤器限定范围。

基本标签过滤

# 创建一个带过滤字段的缓存实例private_cache=SemanticCache(name="private_cache",filterable_fields=[{"name":"user_id","type":"tag"}]# 定义标签字段)# 存储用户 abc 的数据private_cache.store(prompt="我的账户绑定的电话号码是多少？",response="档案中的号码是 123-555-0000",filters={"user_id":"abc"},)# 存储用户 def 的数据private_cache.store(prompt="我的账户绑定的电话号码是多少？",response="档案中的号码是 123-555-1111",filters={"user_id":"def"},)

查询时，使用Tag过滤器：

fromredisvl.query.filterimportTag# 定义过滤器：只查询 user_id 为 abc 的条目user_filter=Tag("user_id")=="abc"response=private_cache.check(prompt="我的账户绑定的电话号码是多少？",filter_expression=user_filter,num_results=2)print(f"找到{len(response)}条记录")print(response[0]['response'])# 输出: 档案中的号码是 123-555-0000

复杂组合过滤

你可以定义多个字段，包括标签（tag）和数值（numeric），并组合成复杂表达式。

complex_cache=SemanticCache(name='account_data',filterable_fields=[{"name":"user_id","type":"tag"},{"name":"account_type","type":"tag"},{"name":"transaction_amount","type":"numeric"},])# 存入多条记录complex_cache.store(prompt="我最近一笔低于100美元的支票账户交易是什么？",response="您最近的交易是75美元",filters={"user_id":"abc","account_type":"checking","transaction_amount":75},)complex_cache.store(prompt="我最近的储蓄账户交易是什么？",response="您最近的存款是300美元",filters={"user_id":"abc","account_type":"savings","transaction_amount":300},)complex_cache.store(prompt="我最近一笔超过200美元的支票账户交易是什么？",response="您最近的交易是350美元",filters={"user_id":"abc","account_type":"checking","transaction_amount":350},)

查询时，组合数值和标签条件：

fromredisvl.query.filterimportNum,Tag value_filter=Num("transaction_amount")>100account_filter=Tag("account_type")=="checking"complex_filter=value_filter&account_filter complex_cache.set_threshold(0.3)# 适当放宽阈值response=complex_cache.check(prompt="我最近的支票账户交易是什么？",filter_expression=complex_filter,num_results=5)print(f"找到{len(response)}条记录")print(response[0]["response"])# 输出: 您最近的交易是350美元

最后，清理资源：

private_cache.delete()complex_cache.delete()