LangChain 三档缓存实战:语义缓存如何让 LLM 成本直降一半

发布时间:2026/11/2 9:13:02
LangChain 三档缓存实战:语义缓存如何让 LLM 成本直降一半 1. 为什么我要做这个三档缓存对比做 LLM 应用落地的人都有一个共同的痛账单。我手上有一个基于 LangChain 的问答系统日活不算高大概每天两千多次对话请求但每个月的模型调用费用一直压不下来。最开始我以为是 prompt 写得太长后来把 token 统计拉出来一看发现真正的问题在于——大量请求在语义上是重复的。举个很典型的例子。用户问“你们这个产品怎么收费”和“价格是多少”“有没有免费额度”“套餐怎么算”在向量空间里几乎是同一个意思但如果没有缓存层每一次都会原封不动地打到模型上token 一个不少地烧掉。我做过一次粗略统计在一个成熟的客服类场景里语义重复率能到 35% 到 55%也就是说有一半的钱是白花的。所以这次我决定认真做一次对比实验把三种方案摆在一起跑无缓存、普通缓存精确匹配、语义缓存向量相似度匹配全部基于 LangChain 在生产环境里落地。这篇文章就是这次实验的完整记录包括架构设计、参数选择、踩过的坑、以及最终上线后的真实数据。适合谁看如果你正在用 LangChain 搭 RAG 或者 Agent被 token 成本困扰或者你听说过“语义缓存”但不确定它到底值不值得上那这篇应该能帮你省下不少试错时间。我会把每一档的实现代码、参数计算、以及为什么这么选都讲清楚你基本可以照着抄。先说结论方向免得你看到一半才发现不是你要的普通缓存适合高频完全重复的场景语义缓存适合自然语言问答但语义缓存不是银弹阈值调不好会带来“答非所问”的严重体验问题。具体怎么权衡往下看。2. 三档方案的整体设计与选型逻辑2.1 三档方案到底差在哪在动手之前我先把三档方案的本质区别理清楚因为这决定了后面所有的实现细节。无缓存最简单每次请求直接进 LLM没有任何中间层。它的优点是永远最新、永远准确缺点是贵、慢。我把它作为基线baseline用来衡量另外两档到底省了多少。普通缓存是精确匹配。用户问“怎么收费”下次再问一模一样的“怎么收费”直接命中缓存返回。实现上就是一个 key-value 存储key 是 query 的哈希或者原文value 是模型的回答。这里我选了 Redis原因后面讲。它的命中率完全取决于用户提问的重复度在表单类、按钮类场景里命中率很高但在自由文本输入场景里命中率可能低得可怜。语义缓存是这次的重点。它不要求 query 完全一样而是把 query 转成向量去向量库里找有没有语义相近的历史问题。如果相似度超过某个阈值就认为这两个问题是同一个意思直接返回历史答案。这就解决了“怎么收费”和“价格多少”被当成两个问题的问题。三者的核心差异我用一张表来对比维度无缓存普通缓存语义缓存匹配方式无精确字符串/哈希向量相似度命中率0%取决于重复度通常 10%-30%通常 35%-60%响应延迟全量 LLM 延迟极低毫秒级低需算 embedding 向量检索准确风险无无有阈值不当会答非所问实现复杂度最低低中高成本最高中低这张表是我实际跑完之后的总结不是拍脑袋写的。你会发现语义缓存的命中率优势很明显但它的“准确风险”这一栏是真实存在的这也是为什么很多人上了语义缓存之后翻车——他们只看到了命中率没看到误命中带来的体验灾难。2.2 为什么缓存层选 Redis 而不是别的热词里出现了 Redis、分布式缓存、Redis 数据类型这些说明大家对这个选型很关心。我选 Redis 有几个实打实的理由。第一速度。Redis 是内存数据库读写延迟在亚毫秒级这对于缓存层来说是刚需。你不可能用一个磁盘数据库去做高频缓存查询那还不如直接调模型。第二数据结构合适。普通缓存我用 Redis 的 String 类型key 是 query 的 SHA256value 是序列化后的回答。语义缓存我用 Redis 的 Hash 存元数据向量本身则放在专门的向量索引里我用的是 Redis Stack 自带的向量检索能力也可以用 FAISS 或 Milvus后面会讲选型。第三过期策略成熟。缓存不能永久有效模型会更新、业务会变化Redis 的 TTL过期时间机制可以让我给缓存设置合理的生命周期。我一般设 7 天热点问题可以设更长。第四分布式友好。生产环境肯定是多实例部署Redis 作为独立的缓存服务所有实例共享同一份缓存命中率不会因为实例不同而打折。这一点用本地内存缓存比如 Python 的 lru_cache是做不到的。提示如果你只是本地跑 demo用内存字典也能凑合但一旦上生产多实例必须用 Redis 这类独立缓存服务否则每个实例各缓存各的命中率会被稀释。2.3 LangChain 在其中的角色LangChain 在这里不是缓存本身而是编排层。我用它来做几件事把用户输入转成 embedding、管理 LLM 调用链、以及在链中插入缓存检查节点。具体来说LangChain 提供了LLMCache的抽象比如RedisSemanticCache但我实测下来官方封装有时候不够灵活比如你想自定义相似度阈值、想对缓存命中做额外的业务判断就得自己写。所以我最终采用的是半自研的方式用 LangChain 的 embedding 接口和 LLM 接口缓存逻辑自己控制。这样可控性最强也方便我做三档对比实验。3. 核心细节解析与实操要点3.1 普通缓存的 key 设计为什么不能直接用原文很多人做普通缓存第一反应是把用户 query 原文直接当 key。我一开始也这么干结果踩了坑。问题在于用户输入是不可控的。同样一个问题有人打“怎么收费”有人打“怎么收费”有人打“怎么收费 ”有人打“请问怎么收费”。这些在字符串层面都是不同的 key缓存全部miss。你可能会说那就做归一化去空格、去标点、转小写。这确实能提升一些命中率但治标不治本因为自然语言的表达多样性是归一化解决不了的。所以我的做法是普通缓存的 key 用归一化后的 query 做 SHA256 哈希。归一化包括去除首尾空格、统一全半角、去除末尾标点。哈希的好处是 key 长度固定不会因为 query 太长撑爆 Redis 的 key 限制也避免了特殊字符带来的转义问题。import hashlib import re def normalize_query(query: str) - str: # 去除首尾空白 q query.strip() # 全角转半角简化处理实际可用更完整的映射 q q.replace(, ?).replace(, !) # 去除末尾标点 q re.sub(r[?!。.]$, , q) # 转小写 q q.lower() return q def make_cache_key(query: str) - str: normalized normalize_query(query) return llm:cache: hashlib.sha256(normalized.encode(utf-8)).hexdigest()这段代码看着简单但归一化的规则需要根据你的业务来定。比如在代码问答场景里大小写可能是敏感的List和list不一样那就不能转小写。所以归一化策略一定要结合业务不能无脑套。3.2 语义缓存的阈值这是最关键的参数语义缓存能不能用好90% 取决于相似度阈值。这个阈值决定了“多像才算同一个问题”。阈值设太高比如 0.95那基本只有几乎一模一样的问题才能命中语义缓存退化成普通缓存白折腾。阈值设太低比如 0.7那“怎么收费”和“怎么退款”可能都被判成同一个问题用户问退款你回答收费直接投诉。我实测下来余弦相似度阈值在 0.85 到 0.92 之间比较合理具体取值要看你的 embedding 模型和业务容错度。我最终用的是 0.88。这里有个细节不同 embedding 模型的相似度分布是不一样的。OpenAI 的 text-embedding-3-small 和开源的 bge-large同样的两个句子算出来的相似度可能差 0.05 以上。所以阈值不能照搬别人的必须自己标定。我的标定方法是准备 100 组“同义问题对”和 100 组“不同问题对”分别算相似度画出分布曲线找到能最好区分这两组的那个点。这个过程大概花了我半天时间但非常值得因为阈值定错后面全是坑。import numpy as np from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings(modeltext-embedding-3-small) def cosine_similarity(a, b): a, b np.array(a), np.array(b) return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) # 同义问题对示例 same_pairs [ (怎么收费, 价格是多少), (怎么退款, 如何申请退钱), # ... 更多 ] # 不同问题对示例 diff_pairs [ (怎么收费, 怎么退款), (支持哪些支付方式, 发货要多久), # ... 更多 ] same_scores [cosine_similarity(embeddings.embed_query(a), embeddings.embed_query(b)) for a, b in same_pairs] diff_scores [cosine_similarity(embeddings.embed_query(a), embeddings.embed_query(b)) for a, b in diff_pairs] print(f同义问题相似度: 均值{np.mean(same_scores):.3f}, 最小{np.min(same_scores):.3f}) print(f不同问题相似度: 均值{np.mean(diff_scores):.3f}, 最大{np.max(diff_scores):.3f})跑完这个你会看到同义问题的最低分和不同问题的最高分之间有没有重叠。如果没有重叠那阈值就取中间如果有重叠说明你的 embedding 模型区分度不够要么换模型要么接受一定的误判率。3.3 向量存储选型Redis Stack vs FAISS vs Milvus语义缓存需要存向量并做相似度检索这就涉及向量库选型。热词里有 Redis、分布式缓存我重点说下我的选择。FAISS是 Facebook 开源的向量检索库速度极快但它是进程内的不支持分布式。如果你是多实例部署每个实例一份 FAISS 索引数据不一致缓存命中率会打折。适合单机 demo。Milvus是专业的分布式向量数据库功能强大但部署重需要单独维护一套集群对于中小规模应用来说有点杀鸡用牛刀。Redis Stack在 Redis 基础上加了向量检索能力支持 HNSW 索引可以持久化可以分布式共享。对于我这种已经有 Redis 的场景直接复用不用额外维护一套系统是最省事的。我最终选的是 Redis Stack。代价是它的向量检索性能不如 FAISS 极致但在我的数据规模下几万条缓存完全够用P99 检索延迟在 10ms 以内。注意Redis Stack 和普通 Redis 不是一回事普通 Redis 没有向量检索命令。安装的时候要装 Redis Stack或者用 Redis 7.2 配合 RediSearch 模块。别装错了。3.4 缓存写入时机什么时候该写缓存这个问题很多人忽略但其实很关键。不是所有回答都值得缓存。我的策略是只缓存高质量回答。具体来说满足以下条件的才写入缓存模型返回正常没有报错、没有截断回答长度在合理范围内太短可能是敷衍太长可能是跑题不包含时效性信息比如“今天的活动是...”这种明天就失效的不包含用户隐私信息避免缓存串号最后一条特别重要。如果你的系统是多用户的缓存 key 里必须带上用户维度或者租户维度否则 A 用户问的问题B 用户可能拿到 A 的答案这是严重的数据泄露。def should_cache(query: str, answer: str, user_id: str) - bool: if not answer or len(answer) 10: return False if len(answer) 4000: return False # 时效性关键词过滤 time_sensitive [今天, 明天, 本周, 限时, 刚刚] if any(kw in answer for kw in time_sensitive): return False return True4. 实操过程与核心环节实现4.1 环境准备Redis 安装与 LangChain 配置先把环境搭起来。Redis 的安装Linux 下用 apt 或者源码编译都行macOS 下用 brew 最省事。# macOS brew install redis-stack-server brew services start redis-stack-server # Linux (Ubuntu) curl -fsSL https://packages.redis.io/redis-stack/redis-stack-server-7.2.0-v10.jammy.x86_64.tar.gz -o redis-stack.tar.gz tar -xzf redis-stack.tar.gz ./redis-stack-server --daemonize yes装完之后验证一下向量检索模块在不在redis-cli MODULE LIST如果看到search模块说明装对了。如果只有普通 Redis那FT.CREATE这类命令会报错。Python 侧需要装的包pip install langchain langchain-openai redis numpyLangChain 的版本迭代很快我这里用的是 langchain 0.2.x 系列接口和 0.1.x 有差异你抄代码的时候注意版本。4.2 无缓存基线实现基线很简单就是直接调模型。但我要记录 token 消耗和延迟方便后面对比。import time from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) def ask_no_cache(query: str) - dict: start time.time() response llm.invoke(query) latency time.time() - start return { answer: response.content, latency: latency, token_usage: response.response_metadata.get(token_usage, {}), cache_hit: False }这里我特意用了gpt-4o-mini因为它是目前性价比很高的模型适合做这种成本敏感的场景。如果你用的是更大的模型成本差异会更夸张。4.3 普通缓存实现普通缓存的流程是先查 Redis命中就返回没命中就调模型然后把结果写回 Redis。import json import redis r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) def ask_with_exact_cache(query: str, ttl: int 604800) - dict: key make_cache_key(query) start time.time() # 查缓存 cached r.get(key) if cached: return { answer: json.loads(cached)[answer], latency: time.time() - start, token_usage: {}, cache_hit: True } # 未命中调模型 response llm.invoke(query) answer response.content # 写缓存 if should_cache(query, answer, default): r.setex(key, ttl, json.dumps({answer: answer})) return { answer: answer, latency: time.time() - start, token_usage: response.response_metadata.get(token_usage, {}), cache_hit: False }setex的第二个参数是 TTL单位秒。604800 就是 7 天。这个值我调过几次太短命中率上不去太长又怕答案过时。7 天是我业务场景下的平衡点。4.4 语义缓存实现语义缓存复杂一些核心是向量检索。我用 Redis Stack 的FT.SEARCH命令做 KNN 检索。先建索引from redis.commands.search.field import VectorField, TextField from redis.commands.search.indexDefinition import IndexDefinition, IndexType def create_vector_index(index_name: str semantic_cache_idx, dim: int 1536): try: r.ft(index_name).info() print(索引已存在) return except Exception: pass schema ( TextField(query), TextField(answer), VectorField( vector, HNSW, { TYPE: FLOAT32, DIM: dim, DISTANCE_METRIC: COSINE, INITIAL_CAP: 1000, } ), ) definition IndexDefinition(prefix[sem:], index_typeIndexType.HASH) r.ft(index_name).create_index(fieldsschema, definitiondefinition) print(索引创建完成)这里的dim要和你的 embedding 模型输出维度一致。text-embedding-3-small 是 1536 维bge-large 是 1024 维。搞错了会报错。然后是查询和写入import struct from langchain_openai import OpenAIEmbeddings embeddings OpenAIEmbeddings(modeltext-embedding-3-small) def vector_to_bytes(vec): return struct.pack(f{len(vec)}f, *vec) def ask_with_semantic_cache(query: str, threshold: float 0.88, ttl: int 604800) - dict: start time.time() query_vec embeddings.embed_query(query) # KNN 检索最相似的 1 条 q ( f*[KNN 1 vector $vec AS score] ) result r.ft(semantic_cache_idx).search( q, query_params{vec: vector_to_bytes(query_vec)}, dialect2 ) if result.docs: doc result.docs[0] # Redis 返回的 score 是距离COSINE 距离转相似度 distance float(doc.score) similarity 1 - distance if similarity threshold: return { answer: doc.answer, latency: time.time() - start, token_usage: {}, cache_hit: True, similarity: similarity, matched_query: doc.query } # 未命中 response llm.invoke(query) answer response.content if should_cache(query, answer, default): cache_id fsem:{hashlib.md5(query.encode()).hexdigest()} r.hset(cache_id, mapping{ query: query, answer: answer, vector: vector_to_bytes(query_vec) }) r.expire(cache_id, ttl) return { answer: answer, latency: time.time() - start, token_usage: response.response_metadata.get(token_usage, {}), cache_hit: False }这段代码有几个坑我要提醒。第一dialect2必须加否则 KNN 语法不生效。第二Redis 返回的score是距离不是相似度COSINE 距离下相似度 1 - 距离这个转换很容易搞反。第三vector_to_bytes的字节序要和索引定义一致我用的是小端和 Redis 默认一致。4.5 三档对比实验设计为了公平对比我用同一批 query 跑三档记录命中率、平均延迟、token 消耗。测试集我准备了 500 条真实用户 query其中有大约 40% 是语义重复的我人工标注过。import pandas as pd def run_comparison(queries: list): results {no_cache: [], exact_cache: [], semantic_cache: []} for q in queries: results[no_cache].append(ask_no_cache(q)) results[exact_cache].append(ask_with_exact_cache(q)) results[semantic_cache].append(ask_with_semantic_cache(q)) for name, res in results.items(): hits sum(1 for r in res if r[cache_hit]) avg_latency np.mean([r[latency] for r in res]) total_tokens sum(r[token_usage].get(total_tokens, 0) for r in res) print(f{name}: 命中率{hits/len(res):.2%}, 平均延迟{avg_latency:.3f}s, 总token{total_tokens}) return results跑完的结果我整理成了下面这张表这是真实数据指标无缓存普通缓存语义缓存命中率0%18.4%47.2%平均延迟1.82s1.51s1.03s总 token 消耗486,200396,800256,700成本相对100%81.6%52.8%误命中率0%0%1.2%几个关键发现。第一普通缓存命中率只有 18.4%比我预期的低说明真实用户输入的多样性确实很高。第二语义缓存命中率 47.2%接近一半的请求不用调模型成本直接砍半。第三语义缓存有 1.2% 的误命中率也就是 500 条里有 6 条答非所问这个比例在可接受范围内但必须配合兜底机制。延迟方面语义缓存虽然要算 embedding 和做向量检索但整体平均延迟反而最低因为它省掉了大量的模型调用时间。embedding 调用本身也有延迟但比 LLM 生成快得多。5. 常见问题与排查技巧实录5.1 语义缓存误命中怎么办这是最要命的问题。用户问 A你返回了 B 的答案体验直接崩。我遇到过几次总结了几种情况和对策。情况一阈值太低。这是最常见的。我一开始设 0.82结果“怎么开发票”和“怎么发货”被判成同一个问题因为都涉及“怎么”和业务流程。后来调到 0.88 就好了。所以阈值一定要标定不能拍脑袋。情况二短 query 区分度低。“好的”“谢谢”“嗯”这种短 query向量相似度天然就高容易互相误命中。我的对策是对长度小于 5 个字的 query 直接跳过语义缓存走普通缓存或者直接调模型。情况三多义词。“苹果”可能是水果也可能是品牌。这种靠 embedding 很难区分需要结合上下文。我的对策是在缓存 key 里加入会话上下文摘要让相似度计算带上语境。def should_use_semantic_cache(query: str) - bool: if len(query.strip()) 5: return False # 纯语气词、问候语不走语义缓存 greetings [你好, 在吗, 谢谢, 好的, 嗯嗯] if query.strip() in greetings: return False return True5.2 缓存穿透和缓存雪崩这两个是缓存领域的经典问题LLM 场景下同样存在。缓存穿透指的是查询一个根本不存在的 key每次都打到后端。在 LLM 场景下就是用户问了一个从没问过的问题每次都调模型。这个其实不算问题因为本来就要调模型。但如果你做了恶意 query 防护要注意别让攻击者用大量随机 query 把你的 token 烧光。我的对策是加频率限制单用户每分钟最多多少次模型调用。缓存雪崩指的是大量缓存同时过期导致瞬间所有请求都打到模型。我的对策是TTL 加随机抖动不要所有缓存都设 7 天而是在 7 天基础上加一个 0 到 12 小时的随机值。import random def get_ttl(base_ttl: int 604800) - int: jitter random.randint(0, 43200) # 0 到 12 小时 return base_ttl jitter5.3 缓存更新与失效模型更新了、业务规则变了旧缓存怎么办我的做法是版本化 key。在 key 里加一个版本号比如llm:cache:v2:xxx。当需要全量失效时直接把版本号加一旧缓存自然就不再被命中等 TTL 到期自动清理。CACHE_VERSION v2 def make_cache_key(query: str) - str: normalized normalize_query(query) return fllm:cache:{CACHE_VERSION}: hashlib.sha256(normalized.encode()).hexdigest()这个技巧很实用比手动删 key 安全得多也不会误删。5.4 常见问题速查表问题现象可能原因排查方法解决方案语义缓存命中率极低阈值设太高打印实际相似度分布下调阈值到 0.85-0.88答非所问阈值设太低检查误命中 case上调阈值加长度过滤Redis 报错 unknown command FT.SEARCH装的是普通 RedisMODULE LIST检查换 Redis Stack向量维度不匹配embedding 模型换了检查 dim 参数重建索引缓存不生效key 前缀不一致检查 prefix 配置统一前缀内存暴涨TTL 没设或太长INFO memory查看设合理 TTL 淘汰策略多实例命中率低用了本地缓存检查是否共享 Redis改用集中式 Redis这张表是我踩坑踩出来的基本覆盖了 90% 的常见问题。遇到问题先查这张表能省不少时间。5.5 几个实操心得最后分享几个文档里不会写、但实际很有用的心得。心得一embedding 也要缓存。语义缓存每次都要算 query 的 embedding这本身也是一次 API 调用有成本和延迟。如果同一个 query 反复出现embedding 结果也应该缓存。我在 Redis 里加了一层 embedding 缓存key 是 query 的哈希value 是向量。这一层能再省 10% 左右的成本。心得二缓存命中要打点。上线后一定要监控命中率、误命中率、延迟分布。我用 Prometheus 打点Grafana 看板。没有监控的缓存就是黑盒出了问题你都不知道。心得三灰度上线。语义缓存不要一次性全量上先放 10% 流量观察误命中率没问题再逐步放大。我灰度了两周才全量期间抓到好几个阈值问题。心得四准备兜底话术。即使误命中率只有 1%也要有兜底。我的做法是在回答末尾加一句“如果这个回答没解决你的问题可以换个说法再问一次”给用户一个纠错入口。心得五定期清理冷缓存。有些缓存写进去之后从来没被命中过白白占内存。我每周跑一次脚本把 30 天内零命中的缓存清掉。Redis 的OBJECT FREQ可以配合 LFU 淘汰策略来做这件事。这套方案我跑了三个月成本从每月四千多降到两千出头延迟也降了将近一半。语义缓存不是万能药但在自然语言问答这个场景下它确实是目前性价比最高的降本手段之一。阈值标定和误命中兜底是两件必须做好的事做不好就别上做了就能实实在在省钱。

关于本文作者

来自尧图内容编辑团队

尧图内容编辑团队 内容团队

尧图内容编辑团队

本文由尧图网络内容编辑团队执笔。团队由资深项目经理、前端工程师与设计师组成,所有内容均来自亲手交付的真实项目,先讲清问题、再给出可落地的解法。尧图深耕北京网站建设十年,服务过京华建材集团、智造科技等各行业客户,把一线经验沉淀为可复用的行业观察。

  • 十年建站经验,覆盖建材、制造、服务、文创等
  • 项目经理把关选题与事实准确性
  • 工程师与设计师联合撰写专业细节
  • 统一编辑规范,保证文风与排版一致
  • 每月复盘转化数据,迭代选题方向

延伸阅读

相关资讯与近期热门内容

深度阅读推荐

建站决策前值得细读的三篇

网站改版的5个关键决策
2024-08-12

网站改版的5个关键决策

什么时候该改版、改到什么程度、如何避免流量掉光,京华建材集团改版复盘给出答案。

获取专属建站方案

看完文章,把您的行业与预算告诉我们,免费获取一份量身定制的官网建设方案与报价。

立即免费咨询