AI 推理结果缓存策略相同问题别反复调用模型一、怎么又问了同样的问题然后 GPT 又收了一次钱Agent 被反复问到相似问题是常态。用户可能会换一种问法但意图不变——我的订单到哪了和查一下我的物流是同一个意思。如果每次都用 LLM 重新推理不仅浪费钱还增加了延迟。实际数据一个客服 Agent 日均处理 5000 次对话其中约 30% 的问题语义重复不同用户问同一个 FAQ。每次 LLM 调用成本约 $0.02日浪费 $30。加上延迟——每次调用 1-3 秒重复问题的总等待时间约 1500 秒/天。语义缓存可以把这 30% 的调用短路节省 $30/天和 1500 秒延迟。更隐蔽的浪费场景同一个用户在对话中反复追问同一个问题——先问什么是微服务Agent 回答后用户又问微服务有什么好处。第二次问题的答案中包含什么是微服务的部分内容——如果缓存了第一次的答案第二次可以复用部分内容而非重新生成完整回答。这种渐进追问模式在客服场景中非常常见约占 15% 的对话。语义缓存Semantic Cache是解决方案不是精确匹配问题文本而是对问题做语义向量化匹配相似的历史答案。相似度超过阈值直接返回缓存否则才调用 LLM。flowchart TD A[用户问题] -- B[问题向量化] B -- C{语义缓存查询} C -- D[向量相似度搜索] D -- E{最高相似度 0.95?} E --|是| F[返回缓存答案] E --|否| G[调用 LLM] G -- H[得到回答] H -- I[写入缓存br/问题向量 答案 TTL] I -- J[返回答案] F -- K[更新访问时间] K -- J二、语义缓存 vs 精确缓存精确缓存Exact Match Cache问题必须逐字相同。订单在哪里和我的订单在哪被当做两个不同请求。命中率极低——实际数据约 5-10%只有用户重新问完全相同的问题才命中。但不会返回错误答案——因为没有相似度模糊匹配命中就意味着完全一致。适用于法律合同审查、代码生成等对精确性要求极高的场景——帮我写一个排序函数和帮我写一个搜索函数在向量空间很近但答案完全不同精确缓存不会错误命中。语义缓存Semantic Cache用 embedding 向量衡量问题相似度。两个语义相同但措辞不同的问题向量距离很近——余弦相似度通常在 0.92-0.98 之间。命中率高——实测约 25-35%FAQ 类问题可达 50%。但存在错误命中风险——两个问题语义相似但确实需要不同答案。例如如何重置密码和如何重置 API Key在向量空间很近都涉及重置动作但答案完全不同。混合策略语义相似度 0.98直接返回缓存极高置信。这个阈值下命中率约 15-20%但几乎不会错误命中——因为 0.98 的相似度意味着两个问题几乎等价。例如Python 怎么排序列表和如何对 Python 的 list 排序相似度约 0.97在这个阈值附近。0.90 相似度 0.98返回缓存但标注基于相似问题回答。用户看到这个标注后可以判断答案是否适用如果不适用可以重新提问触发 LLM 调用。这个标注的核心价值是让用户做最终判断——而非让系统强行决定缓存答案是否适用。相似度 0.90调用 LLM。这个阈值下的问题语义差异足够大缓存命中的错误风险太高。分层策略的收益0.98 阈值部分节省约 15% 调用0.90 阈值部分额外节省约 10% 调用。总节省约 25%但准确率保持在 95%。三、生产级语义缓存实现import asyncio import hashlib import json import time from dataclasses import dataclass, field from typing import Optional import numpy as np import redis.asyncio as redis from openai import AsyncOpenAI dataclass class CacheEntry: 缓存条目。 question: str answer: str embedding: list[float] created_at: float field(default_factorytime.time) last_hit_at: float field(default_factorytime.time) hit_count: int 0 ttl_seconds: int 3600 class SemanticCache: 语义缓存层。 设计决策 - 读取路径先做向量搜索再做 LLM 调用——缓存命中短路 - 写入路径异步执行不阻塞用户响应 - 相似度阈值可配平衡命中率和回答准确性 - TTL 按问题类型分级——事实性问题短 TTL说明性问题长 TTL def __init__( self, redis_url: str, similarity_threshold: float 0.95, high_confidence_threshold: float 0.98, ): self._client AsyncOpenAI() self._redis redis.from_url(redis_url) self._threshold similarity_threshold self._high_conf high_confidence_threshold # embedding 缓存——减少重复向量化 self._embed_cache: dict[str, list[float]] {} async def get_or_compute( self, question: str, compute_fn, ttl: int 3600, ) - tuple[str, bool]: 查找缓存或调用 LLM。 Returns: (answer, is_cached) # Step 1: 向量化问题 q_embedding await self._embed(question) # Step 2: 搜索缓存 cached await self._search_cache(q_embedding) if cached is not None: similarity, entry cached # 超高置信度直接返回 if similarity self._high_conf: await self._update_hit(entry, question) return (entry.answer \n\n[来自缓存], True) # 中等置信度返回但带提示 if similarity self._threshold: await self._update_hit(entry, question) return ( entry.answer \n\n[基于相似问题的缓存在回答——如不准确请重新提问], True, ) # Step 3: 缓存未命中——调用 LLM answer await compute_fn() # Step 4: 异步写入缓存 asyncio.create_task( self._write_cache(question, q_embedding, answer, ttl) ) return (answer, False) async def _embed(self, text: str) - list[float]: 文本向量化——带本地缓存。 cache_key hashlib.md5(text.encode()).hexdigest() if cache_key in self._embed_cache: return self._embed_cache[cache_key] resp await self._client.embeddings.create( modeltext-embedding-3-small, inputtext ) embedding resp.data[0].embedding self._embed_cache[cache_key] embedding return embedding async def _search_cache( self, query_embedding: list[float] ) - Optional[tuple[float, CacheEntry]]: 在 Redis 中搜索最相似的缓存条目。 暴力遍历——小规模 1000 条可行。 大规模应使用 Redis StackRediSearch Vector Similarity。 暴力遍历的性能边界 1000 条缓存 × 1536 维向量 × 余弦相似度计算 ≈ 10ms 10000 条时约 100ms——开始影响用户体验 超过 10000 条必须切换到 ANN 索引 best_score 0.0 best_entry: Optional[CacheEntry] None cursor 0 query_np np.array(query_embedding) while True: cursor, keys await self._redis.scan( cursor, matchsemcache:*:entry, count100 ) for key in keys: data await self._redis.get(key) if not data: continue entry_data json.loads(data) entry_emb np.array(entry_data[embedding]) similarity np.dot(query_np, entry_emb) / ( np.linalg.norm(query_np) * np.linalg.norm(entry_emb) 1e-10 ) if similarity best_score: best_score similarity best_entry CacheEntry( questionentry_data[question], answerentry_data[answer], embeddingentry_data[embedding], created_atentry_data[created_at], last_hit_atentry_data[last_hit_at], hit_countentry_data[hit_count], ttl_secondsentry_data[ttl_seconds], ) if cursor 0: break if best_entry and best_score self._threshold: return (best_score, best_entry) return None async def _write_cache( self, question: str, embedding: list[float], answer: str, ttl: int, ) - None: 写入缓存。 entry CacheEntry( questionquestion, answeranswer, embeddingembedding, ttl_secondsttl, ) key fsemcache:{hashlib.md5(question.encode()).hexdigest()}:entry data json.dumps({ question: entry.question, answer: entry.answer, embedding: entry.embedding, created_at: entry.created_at, last_hit_at: entry.last_hit_at, hit_count: entry.hit_count, ttl_seconds: entry.ttl_seconds, }) await self._redis.setex(key, ttl, data) async def _update_hit(self, entry: CacheEntry, question: str) - None: 更新缓存命中的访问信息。 key fsemcache:{hashlib.md5(entry.question.encode()).hexdigest()}:entry data_raw await self._redis.get(key) if data_raw: data json.loads(data_raw) data[hit_count] 1 data[last_hit_at] time.time() await self._redis.setex(key, entry.ttl_seconds, json.dumps(data))四、缓存的局限与代价错误命中语义相似有时是陷阱。如何重置密码和如何重置 API Key在向量空间很近但答案完全不同。这是语义缓存的固有问题——高维向量空间中语义接近不等于问答等价。更隐蔽的错误命中跨领域的语义碰撞。Python 的 list 怎么排序和Python 的 dict 怎么排序相似度可能 0.95都涉及 Python 排序但答案完全不同list.sort() vs sorted(dict.items())。分层阈值可以部分缓解——0.98 阈值下这类碰撞更少但无法完全消除。错误命中的终极防御对于涉及金额、法律、安全的回答永远不走缓存——即使相似度 0.99 也调用 LLM。在缓存配置中标记这类不可缓存的领域确保零风险。时效性缓存的答案可能过时。现在几点永远不该缓存Python 最新版本缓存在有新版本发布时失效。TTL 设置是艺术太短命中率低FAQ 类问题 TTL 24h 可能太短太长答案过时实时数据类问题 TTL 5min 就可能太长。按问题类型分级的 TTL 策略事实性问题Python 3.12 有什么新特性TTL 7 天——事实不会频繁变化状态性问题我的订单到哪了TTL 5 分钟——状态随时可能变化概念性问题什么是微服务TTL 30 天——概念解释几乎不变存储增长每条缓存约 1-3KB文本 1536 维浮点向量。10 万条 100-300MB。Redis 能 hold 住但要注意内存上限。向量存储的开销是主要瓶颈。1536 维 × 4 bytes/dimension 6KB/条 embedding。10 万条 600MB 纯向量数据。如果用 text-embedding-3-small 的 512 维版本存储降至 200MB——但相似度精度也降低约 3-5%。这是一个存储 vs 精度的 trade-off。Trade-off 讨论阈值高低 vs 命中率 vs 准确率阈值 0.98 命中率 15% 准确率 99%阈值 0.90 命中率 35% 准确率 85%。推荐分层阈值而非单一阈值——不同场景用不同阈值。embedding 维度 vs 存储 vs 精度1536 维精度最高但存储最大512 维存储减半但精度降约 3-5%。存储敏感场景用 512 维精度敏感场景用 1536 维。本地缓存 vs Redis 缓存本地缓存快0 嫩 ms但不跨实例共享Redis 缓存慢2-5ms但共享。多实例部署必须用 Redis——否则同一问题在不同实例上被重复计算。五、总结语义缓存是用向量相似度换取 API 调用节省的技术。权衡点在于命中率 vs 准确率——阈值设高0.98命中率低但准确设低0.90命中率高但有错误命中风险。推荐分层阈值超高置信直接返回、中等置信带免责提示、低置信走 LLM。这比简单的命中/未命中二元决策更符合工程直觉。另外TTL 按问题类型分级是必须的——把现在几点缓存 24 小时和把什么是微服务缓存 5 分钟一样荒谬。对于涉及金额、法律、安全的回答永远不走缓存——即使相似度极高也调用 LLM确保零风险。
AI 推理结果缓存策略：相同问题别反复调用模型
AI 推理结果缓存策略相同问题别反复调用模型一、怎么又问了同样的问题然后 GPT 又收了一次钱Agent 被反复问到相似问题是常态。用户可能会换一种问法但意图不变——我的订单到哪了和查一下我的物流是同一个意思。如果每次都用 LLM 重新推理不仅浪费钱还增加了延迟。实际数据一个客服 Agent 日均处理 5000 次对话其中约 30% 的问题语义重复不同用户问同一个 FAQ。每次 LLM 调用成本约 $0.02日浪费 $30。加上延迟——每次调用 1-3 秒重复问题的总等待时间约 1500 秒/天。语义缓存可以把这 30% 的调用短路节省 $30/天和 1500 秒延迟。更隐蔽的浪费场景同一个用户在对话中反复追问同一个问题——先问什么是微服务Agent 回答后用户又问微服务有什么好处。第二次问题的答案中包含什么是微服务的部分内容——如果缓存了第一次的答案第二次可以复用部分内容而非重新生成完整回答。这种渐进追问模式在客服场景中非常常见约占 15% 的对话。语义缓存Semantic Cache是解决方案不是精确匹配问题文本而是对问题做语义向量化匹配相似的历史答案。相似度超过阈值直接返回缓存否则才调用 LLM。flowchart TD A[用户问题] -- B[问题向量化] B -- C{语义缓存查询} C -- D[向量相似度搜索] D -- E{最高相似度 0.95?} E --|是| F[返回缓存答案] E --|否| G[调用 LLM] G -- H[得到回答] H -- I[写入缓存br/问题向量 答案 TTL] I -- J[返回答案] F -- K[更新访问时间] K -- J二、语义缓存 vs 精确缓存精确缓存Exact Match Cache问题必须逐字相同。订单在哪里和我的订单在哪被当做两个不同请求。命中率极低——实际数据约 5-10%只有用户重新问完全相同的问题才命中。但不会返回错误答案——因为没有相似度模糊匹配命中就意味着完全一致。适用于法律合同审查、代码生成等对精确性要求极高的场景——帮我写一个排序函数和帮我写一个搜索函数在向量空间很近但答案完全不同精确缓存不会错误命中。语义缓存Semantic Cache用 embedding 向量衡量问题相似度。两个语义相同但措辞不同的问题向量距离很近——余弦相似度通常在 0.92-0.98 之间。命中率高——实测约 25-35%FAQ 类问题可达 50%。但存在错误命中风险——两个问题语义相似但确实需要不同答案。例如如何重置密码和如何重置 API Key在向量空间很近都涉及重置动作但答案完全不同。混合策略语义相似度 0.98直接返回缓存极高置信。这个阈值下命中率约 15-20%但几乎不会错误命中——因为 0.98 的相似度意味着两个问题几乎等价。例如Python 怎么排序列表和如何对 Python 的 list 排序相似度约 0.97在这个阈值附近。0.90 相似度 0.98返回缓存但标注基于相似问题回答。用户看到这个标注后可以判断答案是否适用如果不适用可以重新提问触发 LLM 调用。这个标注的核心价值是让用户做最终判断——而非让系统强行决定缓存答案是否适用。相似度 0.90调用 LLM。这个阈值下的问题语义差异足够大缓存命中的错误风险太高。分层策略的收益0.98 阈值部分节省约 15% 调用0.90 阈值部分额外节省约 10% 调用。总节省约 25%但准确率保持在 95%。三、生产级语义缓存实现import asyncio import hashlib import json import time from dataclasses import dataclass, field from typing import Optional import numpy as np import redis.asyncio as redis from openai import AsyncOpenAI dataclass class CacheEntry: 缓存条目。 question: str answer: str embedding: list[float] created_at: float field(default_factorytime.time) last_hit_at: float field(default_factorytime.time) hit_count: int 0 ttl_seconds: int 3600 class SemanticCache: 语义缓存层。 设计决策 - 读取路径先做向量搜索再做 LLM 调用——缓存命中短路 - 写入路径异步执行不阻塞用户响应 - 相似度阈值可配平衡命中率和回答准确性 - TTL 按问题类型分级——事实性问题短 TTL说明性问题长 TTL def __init__( self, redis_url: str, similarity_threshold: float 0.95, high_confidence_threshold: float 0.98, ): self._client AsyncOpenAI() self._redis redis.from_url(redis_url) self._threshold similarity_threshold self._high_conf high_confidence_threshold # embedding 缓存——减少重复向量化 self._embed_cache: dict[str, list[float]] {} async def get_or_compute( self, question: str, compute_fn, ttl: int 3600, ) - tuple[str, bool]: 查找缓存或调用 LLM。 Returns: (answer, is_cached) # Step 1: 向量化问题 q_embedding await self._embed(question) # Step 2: 搜索缓存 cached await self._search_cache(q_embedding) if cached is not None: similarity, entry cached # 超高置信度直接返回 if similarity self._high_conf: await self._update_hit(entry, question) return (entry.answer \n\n[来自缓存], True) # 中等置信度返回但带提示 if similarity self._threshold: await self._update_hit(entry, question) return ( entry.answer \n\n[基于相似问题的缓存在回答——如不准确请重新提问], True, ) # Step 3: 缓存未命中——调用 LLM answer await compute_fn() # Step 4: 异步写入缓存 asyncio.create_task( self._write_cache(question, q_embedding, answer, ttl) ) return (answer, False) async def _embed(self, text: str) - list[float]: 文本向量化——带本地缓存。 cache_key hashlib.md5(text.encode()).hexdigest() if cache_key in self._embed_cache: return self._embed_cache[cache_key] resp await self._client.embeddings.create( modeltext-embedding-3-small, inputtext ) embedding resp.data[0].embedding self._embed_cache[cache_key] embedding return embedding async def _search_cache( self, query_embedding: list[float] ) - Optional[tuple[float, CacheEntry]]: 在 Redis 中搜索最相似的缓存条目。 暴力遍历——小规模 1000 条可行。 大规模应使用 Redis StackRediSearch Vector Similarity。 暴力遍历的性能边界 1000 条缓存 × 1536 维向量 × 余弦相似度计算 ≈ 10ms 10000 条时约 100ms——开始影响用户体验 超过 10000 条必须切换到 ANN 索引 best_score 0.0 best_entry: Optional[CacheEntry] None cursor 0 query_np np.array(query_embedding) while True: cursor, keys await self._redis.scan( cursor, matchsemcache:*:entry, count100 ) for key in keys: data await self._redis.get(key) if not data: continue entry_data json.loads(data) entry_emb np.array(entry_data[embedding]) similarity np.dot(query_np, entry_emb) / ( np.linalg.norm(query_np) * np.linalg.norm(entry_emb) 1e-10 ) if similarity best_score: best_score similarity best_entry CacheEntry( questionentry_data[question], answerentry_data[answer], embeddingentry_data[embedding], created_atentry_data[created_at], last_hit_atentry_data[last_hit_at], hit_countentry_data[hit_count], ttl_secondsentry_data[ttl_seconds], ) if cursor 0: break if best_entry and best_score self._threshold: return (best_score, best_entry) return None async def _write_cache( self, question: str, embedding: list[float], answer: str, ttl: int, ) - None: 写入缓存。 entry CacheEntry( questionquestion, answeranswer, embeddingembedding, ttl_secondsttl, ) key fsemcache:{hashlib.md5(question.encode()).hexdigest()}:entry data json.dumps({ question: entry.question, answer: entry.answer, embedding: entry.embedding, created_at: entry.created_at, last_hit_at: entry.last_hit_at, hit_count: entry.hit_count, ttl_seconds: entry.ttl_seconds, }) await self._redis.setex(key, ttl, data) async def _update_hit(self, entry: CacheEntry, question: str) - None: 更新缓存命中的访问信息。 key fsemcache:{hashlib.md5(entry.question.encode()).hexdigest()}:entry data_raw await self._redis.get(key) if data_raw: data json.loads(data_raw) data[hit_count] 1 data[last_hit_at] time.time() await self._redis.setex(key, entry.ttl_seconds, json.dumps(data))四、缓存的局限与代价错误命中语义相似有时是陷阱。如何重置密码和如何重置 API Key在向量空间很近但答案完全不同。这是语义缓存的固有问题——高维向量空间中语义接近不等于问答等价。更隐蔽的错误命中跨领域的语义碰撞。Python 的 list 怎么排序和Python 的 dict 怎么排序相似度可能 0.95都涉及 Python 排序但答案完全不同list.sort() vs sorted(dict.items())。分层阈值可以部分缓解——0.98 阈值下这类碰撞更少但无法完全消除。错误命中的终极防御对于涉及金额、法律、安全的回答永远不走缓存——即使相似度 0.99 也调用 LLM。在缓存配置中标记这类不可缓存的领域确保零风险。时效性缓存的答案可能过时。现在几点永远不该缓存Python 最新版本缓存在有新版本发布时失效。TTL 设置是艺术太短命中率低FAQ 类问题 TTL 24h 可能太短太长答案过时实时数据类问题 TTL 5min 就可能太长。按问题类型分级的 TTL 策略事实性问题Python 3.12 有什么新特性TTL 7 天——事实不会频繁变化状态性问题我的订单到哪了TTL 5 分钟——状态随时可能变化概念性问题什么是微服务TTL 30 天——概念解释几乎不变存储增长每条缓存约 1-3KB文本 1536 维浮点向量。10 万条 100-300MB。Redis 能 hold 住但要注意内存上限。向量存储的开销是主要瓶颈。1536 维 × 4 bytes/dimension 6KB/条 embedding。10 万条 600MB 纯向量数据。如果用 text-embedding-3-small 的 512 维版本存储降至 200MB——但相似度精度也降低约 3-5%。这是一个存储 vs 精度的 trade-off。Trade-off 讨论阈值高低 vs 命中率 vs 准确率阈值 0.98 命中率 15% 准确率 99%阈值 0.90 命中率 35% 准确率 85%。推荐分层阈值而非单一阈值——不同场景用不同阈值。embedding 维度 vs 存储 vs 精度1536 维精度最高但存储最大512 维存储减半但精度降约 3-5%。存储敏感场景用 512 维精度敏感场景用 1536 维。本地缓存 vs Redis 缓存本地缓存快0 嫩 ms但不跨实例共享Redis 缓存慢2-5ms但共享。多实例部署必须用 Redis——否则同一问题在不同实例上被重复计算。五、总结语义缓存是用向量相似度换取 API 调用节省的技术。权衡点在于命中率 vs 准确率——阈值设高0.98命中率低但准确设低0.90命中率高但有错误命中风险。推荐分层阈值超高置信直接返回、中等置信带免责提示、低置信走 LLM。这比简单的命中/未命中二元决策更符合工程直觉。另外TTL 按问题类型分级是必须的——把现在几点缓存 24 小时和把什么是微服务缓存 5 分钟一样荒谬。对于涉及金额、法律、安全的回答永远不走缓存——即使相似度极高也调用 LLM确保零风险。
