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

引言在现代 AI 应用中调用大语言模型LLMAPI 不仅会产生可观的费用还会带来不可忽视的延迟。当用户反复提出相同或相似的问题时每次都调用 LLM 无疑是一种浪费。语义缓存Semantic Cache正是为了解决这个问题而诞生 —— 它利用向量相似度搜索将用户查询的语义嵌入与缓存中的历史问题进行比较如果找到语义足够接近的条目则直接返回缓存的答案从而避免重复调用 LLM API。RedisVL 提供了SemanticCache类基于 Redis 作为向量数据库和缓存存储实现了高效、可定制的语义缓存。本指南将带你从零开始逐步掌握其使用方法和核心原理。前置条件在开始之前请确保你已具备以下条件已安装 RedisVLpip install redisvl一个正在运行的 Redis 实例推荐 Redis 8 或 Redis Cloud一个有效的 OpenAI API 密钥用于演示调用 LLM你将学到什么完成本指南后你将能够配置并初始化一个语义缓存实例存储和检索缓存的 LLM 响应理解缓存条目的entry_id和 Rediskey的区别并利用它们进行精确操作自定义语义相似度阈值平衡召回率与精确率配置 TTL生存时间策略并理解check()方法对 TTL 的刷新行为通过标签和过滤器实现多用户场景下的访问控制语义缓存的工作原理语义缓存的核心是向量化和相似度搜索。当用户提出问题时系统会使用嵌入模型将问题文本转换为高维向量例如 768 维。在 Redis 中执行向量相似度搜索找出与当前问题向量最接近的已缓存问题。计算它们之间的余弦距离取值范围 0~20 表示完全相同2 表示完全相反。如果最小距离小于设定的阈值则认为语义匹配返回对应的缓存响应。如果未匹配则调用 LLM 获取真实响应并将问题向量、问题文本、响应、元数据存入缓存供后续使用。下面这张流程图清晰地展示了这一过程是否用户输入问题将问题文本向量化在 Redis 中执行向量相似度搜索是否存在距离 阈值的缓存条目返回缓存的响应调用 LLM API 生成响应将新问题、响应、向量和元数据存入缓存返回新生成的响应结束理解这个流程后我们开始动手实践。环境准备首先导入必要的库并设置 OpenAI 客户端。importosimportgetpassimporttimeimportnumpyasnpfromopenaiimportOpenAI# 避免 tokenizers 并行警告os.environ[TOKENIZERS_PARALLELISM]False# 获取 OpenAI API 密钥api_keyos.getenv(OPENAI_API_KEY)orgetpass.getpass(输入你的 OpenAI API 密钥: )clientOpenAI(api_keyapi_key)defask_openai(question:str)-str:调用 OpenAI 的补全接口回答问题responseclient.completions.create(modelgpt-4o-mini,promptf请用简洁的方式回答以下问题{question},max_tokens200)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 llmcacheSemanticCache(namellmcache,# Redis 索引名称redis_urlredis://localhost:6379,# Redis 连接地址distance_threshold0.1,# 余弦距离阈值0~2越小越严格vectorizerHFTextVectorizer(redis/langcache-embed-v2)# 嵌入模型)注意如果你使用的嵌入模型版本与 RedisVL 预期的不一致可能会看到警告信息这通常不影响功能但建议保持版本匹配。你可以通过 RedisVL 命令行工具查看索引的详细信息rvl index info-illmcache输出会显示索引的字段结构其中包括prompt文本、response文本、inserted_at、updated_at和prompt_vector向量字段等。基本缓存操作1. 检查缓存首次为空question法国的首都是哪里# 检查缓存是否命中ifresponse:llmcache.check(promptquestion):print(response)else:print(缓存为空)# 输出: 缓存为空2. 存储条目# 存储问题、答案及任意元数据llmcache.store(promptquestion,response巴黎,metadata{city:巴黎,country:法国})# 返回完整的 Redis key例如 llmcache:115049a...3. 再次检查缓存# 用完全相同的问题查询resultllmcache.check(promptquestion,return_fields[prompt,response,metadata])print(result)# 输出包含缓存内容# 用语义相似的问题查询similar_question法国真正的首都是哪里resultllmcache.check(promptsimilar_question)print(result[0][response])# 输出: 巴黎条目 ID 与 Redis Key每个缓存条目有两个重要的标识符entry_id由prompt和过滤条件组合后通过 SHA256 哈希生成。相同的 prompt 相同 filters 会生成相同的entry_id因此重复存储会覆盖之前的条目。key完整的 Redis 键格式为{index_name}:{entry_id}。在 Redis 中键是唯一的用于直接操作。# 存储时返回完整的 keykeyllmcache.store(prompt法国的首都是哪里,response巴黎,metadata{source:地理})print(f完整 Redis key:{key})# 检查时可以通过 return_fields 获取 entry_id 和 keyresultllmcache.check(prompt法国的首都是哪里,return_fields[entry_id,prompt,response])print(fEntry ID:{result[0][entry_id]})print(fKey:{result[0][key]})获取和删除特定条目你可以通过entry_id或key来精确获取或删除缓存条目。fromredisvl.queryimportFilterQueryfromredisvl.query.filterimportFilterExpression# 列出所有缓存条目使用底层索引查询queryFilterQuery(filter_expressionFilterExpression(*),return_fields[entry_id,prompt,response])all_entriesllmcache._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_idresult[0][entry_id]recordllmcache._index.fetch(entry_id)print(f获取的记录:{record})# 删除指定条目通过 entry_id 列表llmcache.drop(ids[entry_id])# 也可以使用 keys 参数删除llmcache.drop(keys[key])# 验证删除resultllmcache.check(prompt法国的首都是哪里)print(f删除后:{result})# 输出空列表自定义相似度阈值相似度阈值决定了缓存命中的严格程度。阈值distance_threshold采用余弦距离取值范围为[0, 2]其中 0 表示完全相同2 表示完全相反。阈值越小匹配越严格高精确率低召回率阈值越大匹配越宽松高召回率低精确率。你可以随时调整阈值# 将阈值放宽到 0.5允许更不相似的问题命中llmcache.set_threshold(0.5)# 重新存储一个条目llmcache.store(prompt法国的首都是哪里,response巴黎)# 测试一个稍远的问题question尼斯所在国家的首都是哪里# 尼斯在法国resultllmcache.check(promptquestion)print(result[0][response])# 输出: 巴黎因为阈值放宽仍然命中如果需要清空整个缓存使用clear()方法llmcache.clear()# 删除所有条目print(llmcache.check(promptquestion))# 空列表TTL生存时间策略TTL 可以让缓存条目在指定时间后自动过期避免缓存无限增长。RedisVL 的SemanticCache支持灵活的 TTL 设置。基本用法# 设置 TTL 为 5 秒llmcache.set_ttl(5)llmcache.store(这是一个 TTL 测试,这是测试响应)time.sleep(6)# 确认缓存已自动过期resultllmcache.check(这是一个 TTL 测试)print(result)# 输出: []# 重置 TTL 为 None永久存储llmcache.set_ttl()# 不传参数即为 NoneTTL 刷新行为详解check()方法在命中缓存时会自动刷新所有匹配条目的 TTL滑窗模式。这意味着频繁访问的条目会一直保持活跃。下表总结了不同场景下的行为场景行为ttlNone默认条目永久存储。check()不会影响 TTL。初始化时设置ttl3600存储时条目获得 3600 秒 TTL。每次check()命中时TTL 会被重置为 3600 秒。之后通过set_ttl(3600)设置已有条目不会自动获得 TTL但后续的check()命中会为匹配条目添加 TTL并刷新。通过set_ttl(None)移除 TTL已有条目保持原有的 TTL到期后仍会删除但check()命中时不再刷新 TTL。重要提醒因为check()会刷新所有匹配的条目可能有多条所以即使你原本只想获取一条数据所有匹配的条目都会延长生存时间。如果你不希望某些条目被意外刷新可以通过过滤器来精确控制匹配范围。# 示例设置 TTL 为 5 分钟llmcache.set_ttl(300)llmcache.store(什么是 Python,一种编程语言)# 每次查询命中都会重置 TTLresultllmcache.check(什么是 Python)# TTL 被刷新为 300 秒# 重置并清空llmcache.set_ttl()llmcache.clear()简单性能测试我们来对比一下使用缓存前后的响应时间。defanswer_question(question:str)-str:先查缓存未命中则调用 LLMresultsllmcache.check(promptquestion)ifresults:returnresults[0][response]else:answerask_openai(question)returnanswer# 首次请求未命中缓存调用 LLMquestion美国第一任总统是谁starttime.time()answeranswer_question(question)endtime.time()print(f未使用缓存调用 OpenAI 耗时:{end-start:.4f}秒)# 存储正确答案到缓存llmcache.store(promptquestion,response乔治·华盛顿)# 后续 10 次请求命中缓存times[]for_inrange(10):cached_starttime.time()cached_answeranswer_question(question)cached_endtime.time()times.append(cached_end-cached_start)avg_timenp.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_cacheSemanticCache(nameprivate_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_filterTag(user_id)abcresponseprivate_cache.check(prompt我的账户绑定的电话号码是多少,filter_expressionuser_filter,num_results2)print(f找到{len(response)}条记录)print(response[0][response])# 输出: 档案中的号码是 123-555-0000复杂组合过滤你可以定义多个字段包括标签tag和数值numeric并组合成复杂表达式。complex_cacheSemanticCache(nameaccount_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_filterNum(transaction_amount)100account_filterTag(account_type)checkingcomplex_filtervalue_filteraccount_filter complex_cache.set_threshold(0.3)# 适当放宽阈值responsecomplex_cache.check(prompt我最近的支票账户交易是什么,filter_expressioncomplex_filter,num_results5)print(f找到{len(response)}条记录)print(response[0][response])# 输出: 您最近的交易是350美元最后清理资源private_cache.delete()complex_cache.delete()