Prompt缓存策略与成本优化Semantic Cache在企业Agent中的应用一、深度引言企业Agent的运营成本中LLM API调用费用往往占据60%以上的比例。一个中等规模的企业知识问答Agent每天处理5000次查询如果每次查询都完整发送上下文和Prompt仅GPT-4级别的API调用费用就在每日数百美元级别——按年计算是一笔数十万的支出。但深入分析这些查询的分布会发现一个优化空间大约30%-40%的查询在语义上是高度相似的。公司的年假政策是什么和年假有多少天本质上是同一个问题的不同表述但标准实现会发送两次完整的API请求。这两次请求消耗的Token总量、响应时间几乎相同但其中有一次是冗余的。Semantic Cache语义缓存的思路是将相同语义而非相同字符串的查询结果缓存起来在发送API请求之前先用语义相似度匹配检查缓存。本文将拆解Semantic Cache在企业Agent场景下的完整实现方案。flowchart TD A[用户查询] -- B{精确缓存匹配} B --|命中| C[直接返回] B --|未命中| D{语义相似度匹配} D --|相似度 阈值| E[返回缓存 标注] D --|相似度 阈值| F[发送LLM请求] F -- G[LLM响应] G -- H[生成Embedding] H -- I[存入向量缓存] I -- J[返回给用户] subgraph 缓存策略 K[LRU淘汰策略] L[TTL过期策略] M[相似度阈值动态调整] end style D fill:#4A90D9,color:#fff style F fill:#F39C12,color:#fff style K fill:#27AE60,color:#fff二、原理剖析语义相似度的计算和阈值选择Semantic Cache的核心是将用户查询映射到向量空间通过余弦相似度判断是否命中缓存。但这引入了两个关键问题阈值设多少合适向量相似度高是否意味着回答可以复用对于阈值经验和实验数据指向0.92-0.95的余弦相似度区间。低于0.92语义差异开始出现——退款流程和退货流程在向量空间中可能相似度0.90但回答内容有本质区别。高于0.95命中率显著下降缓存的成本节约效果有限。但仅靠相似度阈值是不够的。需要在缓存的值中额外存储问题的意图标签——由LLM在首次回答时同时输出。当候选缓存的相似度在0.92-0.95的灰色区间时通过意图标签做二次校验如果意图标签相同即使相似度稍低也接受缓存命中如果意图标签不同即使相似度高也拒绝。缓存生命周期管理语义缓存的生命周期管理比传统KV缓存复杂。传统缓存直接设置TTL 5分钟就行但语义缓存的TTL需要考虑两个维度时效性——某些问题的答案随时间变化如当前股价需要短TTL稳定性——某些问题的答案长期不变如公司地址可以长TTL甚至永不过期。实践中采用分级TTL策略知识类问题公司政策、产品文档TTL设置为24小时实时类问题股票、天气、新闻TTL设置为5分钟通用类问题概念解释、技术问答TTL设置为1小时。TTL分级信息需要从查询的意图分类中获得。淘汰策略方面LRULeast Recently Used是基础方案但语义缓存有一个独特的优化维度如果一个向量在特征空间中密度很高——周围有大量低相似度的查询——这个向量的潜在命中概率更高可以考虑增加其驻留权重。这是热点保护策略与LRU仅看访问频次不同它还看该缓存在向量空间中的可复用性。Embedding模型的选择语义缓存的精度取决于Embedding模型的语义区分能力。在成本敏感的场景下可以分层使用不同精度的Embedding模型粗筛阶段使用轻量模型如text2vec-base-chinese128MB筛选到Top-K候选后用高精度模型做精细匹配。这种两级流水线可以在保证精度的同时降低计算成本。三、生产级代码以下是一个完整的Semantic Cache实现覆盖向量缓存、相似度匹配、TTL管理和分层淘汰策略。import hashlib import json import logging import threading import time from collections import OrderedDict from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum from typing import Any, Callable, Dict, List, Optional, Tuple import numpy as np logger logging.getLogger(semantic_cache) # 向量工具 class VectorUtils: 向量计算工具 —— 处理Embedding的各种向量操作 staticmethod def cosine_similarity(vec1: List[float], vec2: List[float]) - float: 计算余弦相似度 为什么不用np.dot直接算 需要先做向量范数检查零向量会导致除零错误 a np.array(vec1, dtypenp.float32) b np.array(vec2, dtypenp.float32) norm_a np.linalg.norm(a) norm_b np.linalg.norm(b) if norm_a 0.0 or norm_b 0.0: return 0.0 return float(np.dot(a, b) / (norm_a * norm_b)) staticmethod def normalize(vec: List[float]) - List[float]: 向量归一化 —— 预处理后的向量相似度计算更快 a np.array(vec, dtypenp.float32) norm np.linalg.norm(a) if norm 0: return vec return (a / norm).tolist() # TTL分级策略 class ContentType(Enum): 内容类型 —— 决定缓存TTL KNOWLEDGE knowledge # 知识类公司政策、文档 → 24h REALTIME realtime # 实时类股票、天气 → 5min GENERAL general # 通用类概念解释 → 1h # TTL映射秒 TTL_MAP { ContentType.KNOWLEDGE: 86400, # 24小时 ContentType.REALTIME: 300, # 5分钟 ContentType.GENERAL: 3600, # 1小时 } dataclass class CacheEntry: 缓存条目 query: str query_embedding: List[float] # 查询的向量表示 answer: str intent_label: str # 意图标签用于相似度灰色区域的二次校验 content_type: ContentType ContentType.GENERAL ttl_seconds: int 3600 created_at: float field(default_factorytime.time) hit_count: int 0 # 命中次数影响淘汰策略 last_hit_at: Optional[float] None def is_expired(self) - bool: 检查是否过期 return (time.time() - self.created_at) self.ttl_seconds def heat_score(self) - float: 计算热度分数 —— 用于淘汰决策 为什么综合命中次数和时间衰减 纯LRU会淘汰很久之前热门但现在无人访问的条目 加入时间衰减让淘汰更准确 age_hours (time.time() - self.created_at) / 3600 time_decay max(0.1, 1.0 - age_hours / 48) # 48小时线性衰减 return self.hit_count * time_decay # Semantic Cache 引擎 class SemanticCache: 语义缓存引擎 —— 核心实现 def __init__( self, similarity_threshold: float 0.92, grey_zone_min: float 0.88, max_entries: int 10000, embedding_dim: int 768, ): self.similarity_threshold similarity_threshold self.grey_zone_min grey_zone_min self.max_entries max_entries self.embedding_dim embedding_dim # 数据存储 self._entries: OrderedDict[str, CacheEntry] OrderedDict() # 精确缓存文本 → 缓存键处理完全相同的查询 self._exact_cache: Dict[str, str] {} # 向量索引简化版内存索引生产环境用FAISS/Milvus self._embeddings: List[np.ndarray] [] self._entry_ids: List[str] [] self._lock threading.RLock() # 统计指标 self._total_queries 0 self._exact_hits 0 self._semantic_hits 0 self._misses 0 def get( self, query: str, query_embedding: List[float], embedder: Optional[Callable] None, ) - Tuple[Optional[str], Dict]: 查询缓存 —— 返回(缓存的回答, 元信息) 查询流程精确匹配 → 语义匹配 → 缓存未命中 with self._lock: self._total_queries 1 # 第一层精确匹配速度快成本零 query_hash hashlib.md5(query.encode()).hexdigest() if query_hash in self._exact_cache: entry_id self._exact_cache[query_hash] entry self._entries.get(entry_id) if entry and not entry.is_expired(): entry.hit_count 1 entry.last_hit_at time.time() self._exact_hits 1 # LRU: 将命中的条目移到末尾 self._entries.move_to_end(entry_id) logger.debug(f精确命中: {query[:30]}...) return entry.answer, { source: exact_cache, hits: entry.hit_count, } else: # 过期清理 if entry_id in self._exact_cache: del self._exact_cache[query_hash] if entry_id in self._entries: del self._entries[entry_id] # 第二层语义匹配速度中等成本取决于向量搜索实现 normalized VectorUtils.normalize(query_embedding) best_entry, best_similarity self._semantic_search(normalized) if best_entry and best_similarity self.similarity_threshold: # 高相似度直接命中 best_entry.hit_count 1 best_entry.last_hit_at time.time() self._semantic_hits 1 self._entries.move_to_end(best_entry.query) logger.debug( f语义命中: {query[:30]}... f- {best_entry.query[:30]}... f(相似度: {best_similarity:.3f}) ) return best_entry.answer, { source: semantic_cache, similarity: round(best_similarity, 3), matched_query: best_entry.query, } if best_entry and best_similarity self.grey_zone_min: # 灰色区域意图标签二次校验 if self._intent_match(query, best_entry.intent_label): best_entry.hit_count 1 best_entry.last_hit_at time.time() self._semantic_hits 1 self._entries.move_to_end(best_entry.query) logger.info( f灰色区域命中(意图匹配): f{query[:30]}... (相似度: {best_similarity:.3f}) ) return best_entry.answer, { source: semantic_cache_grey, similarity: round(best_similarity, 3), matched_query: best_entry.query, } # 缓存未命中 self._misses 1 return None, {source: miss} def set( self, query: str, query_embedding: List[float], answer: str, intent_label: str general, content_type: ContentType ContentType.GENERAL, ): 写入缓存 with self._lock: # 淘汰检查 if len(self._entries) self.max_entries: self._evict() normalized VectorUtils.normalize(query_embedding) ttl TTL_MAP.get(content_type, 3600) entry CacheEntry( queryquery, query_embeddingnormalized, answeranswer, intent_labelintent_label, content_typecontent_type, ttl_secondsttl, ) self._entries[query] entry query_hash hashlib.md5(query.encode()).hexdigest() self._exact_cache[query_hash] query # 更新向量索引 self._embeddings.append(np.array(normalized, dtypenp.float32)) self._entry_ids.append(query) logger.debug( f缓存写入: {query[:30]}... f(TTL: {ttl}s, 类型: {content_type.value}) ) def _semantic_search( self, query_vec: List[float] ) - Tuple[Optional[CacheEntry], float]: 向量相似度搜索 —— 找最相似的缓存条目 为什么用暴力搜索而非近似搜索 缓存条目通常在万级以下暴力搜索的延迟在毫秒级。 超过10万条时应升级为FAISS的IVF索引 if not self._embeddings: return None, 0.0 query_arr np.array(query_vec, dtypenp.float32) # 批量计算余弦相似度 embeddings_matrix np.stack(self._embeddings) # 为什么用dot product而非cosine_similarity函数 # 向量已归一化dot product等价于cosine similarity # 且numpy的批量dot运算比循环调用cosine_similarity快10倍 similarities np.dot(embeddings_matrix, query_arr) max_idx int(np.argmax(similarities)) max_sim float(similarities[max_idx]) if max_sim self.grey_zone_min: return None, max_sim entry_id self._entry_ids[max_idx] entry self._entries.get(entry_id) if entry and not entry.is_expired(): return entry, max_sim return None, max_sim def _intent_match(self, query: str, cached_intent: str) - bool: 意图标签匹配 —— 灰色区域的二次校验 为什么用意图标签而非让LLM再判断一次 LLM判断的成本接近一次完整请求失去了缓存的意义。 意图标签是回答时附带生成的成本极低 # 生产环境应使用轻量意图分类器 # 这里简化实现检查关键词共现 intent_keywords { policy_query: [政策, 规定, 流程, 要求, 条件], definition: [什么是, 定义, 解释, 含义, 意思], troubleshooting: [报错, 失败, 不行, 问题, 解决], comparison: [区别, 对比, 哪个好, 选择], } keywords intent_keywords.get(cached_intent, []) return any(kw in query for kw in keywords) def _evict(self): 淘汰策略 —— 综合热度和访问时间的混合淘汰 为什么不用纯LRU LRU在热点数据频繁变化时表现好但语义缓存的 特点是热点数据高度集中——少数问题的变体能命中大量查询。 综合热度分数可以更好地保护这些枢纽缓存 if not self._entries: return # 计算所有条目的热度分数 scored [ (entry_id, entry.heat_score()) for entry_id, entry in self._entries.items() ] scored.sort(keylambda x: x[1]) # 按热度升序最低的最先淘汰 # 淘汰最低热度的条目 evict_count max(1, len(self._entries) // 10) # 至少淘汰1个最多10% for i in range(evict_count): if i len(scored): break entry_id scored[i][0] entry self._entries[entry_id] # 从所有索引中移除 del self._entries[entry_id] query_hash hashlib.md5(entry.query.encode()).hexdigest() if query_hash in self._exact_cache: del self._exact_cache[query_hash] # 移除向量索引 try: idx self._entry_ids.index(entry_id) self._entry_ids.pop(idx) self._embeddings.pop(idx) except ValueError: pass logger.info( f缓存淘汰: 移除了 {evict_count} 个条目 f当前缓存大小: {len(self._entries)} ) def cleanup_expired(self): 清理过期条目 —— 建议在低负载时定时调用 with self._lock: expired_ids [ eid for eid, entry in self._entries.items() if entry.is_expired() ] for eid in expired_ids: entry self._entries[eid] del self._entries[eid] query_hash hashlib.md5(entry.query.encode()).hexdigest() if query_hash in self._exact_cache: del self._exact_cache[query_hash] try: idx self._entry_ids.index(eid) self._entry_ids.pop(idx) self._embeddings.pop(idx) except ValueError: pass if expired_ids: logger.info(f清理了 {len(expired_ids)} 个过期缓存条目) def get_stats(self) - Dict: 获取缓存统计 with self._lock: total max(self._total_queries, 1) return { total_queries: self._total_queries, exact_hits: self._exact_hits, semantic_hits: self._semantic_hits, misses: self._misses, hit_rate: round( (self._exact_hits self._semantic_hits) / total, 3 ), cache_size: len(self._entries), max_entries: self.max_entries, estimated_savings_tokens: ( (self._exact_hits self._semantic_hits) * 500 # 估算每次命中节省500 tokens ), } # 使用示例 # 模拟Embedding函数生产环境替换为真实的Embedding模型API def mock_embed(text: str) - List[float]: 模拟Embedding —— 实际使用OpenAI/本地模型 import random random.seed(hash(text) % (2**32)) return [random.random() for _ in range(768)] if __name__ __main__: cache SemanticCache( similarity_threshold0.92, max_entries1000, ) # 场景1写入基础查询 base_query 公司的年假政策是什么 base_embedding mock_embed(base_query) cache.set( querybase_query, query_embeddingbase_embedding, answer公司年假政策入职满1年享有5天带薪年假满3年享有10天..., intent_labelpolicy_query, content_typeContentType.KNOWLEDGE, ) # 场景2语义相似查询应命中缓存 similar_query 年假有多少天 similar_embedding mock_embed(similar_query) answer, meta cache.get(similar_query, similar_embedding) if answer: print(f缓存命中! 来源: {meta[source]}) print(f原始查询: {base_query}) print(f当前查询: {similar_query}) print(f回答: {answer[:50]}...) else: print(缓存未命中需要调用LLM) # 模拟LLM调用省略 # llm_answer call_llm(similar_query) # cache.set(similar_query, similar_embedding, llm_answer, ...) # 打印统计 stats cache.get_stats() print(f\n缓存统计: {json.dumps(stats, ensure_asciiFalse, indent2)})四、边界权衡召回率vs命中率是Semantic Cache最核心的权衡。相似度阈值越高返回的缓存结果越可靠但命中率越低节约的API成本越少。相似度阈值越低命中率越高但准确度可能下降错误缓存返回可能导致用户不信任系统。解决方案不是找最优阈值而是设计多级判断——在高相似度区间直接返回在灰色区间用意图标签或轻量模型做二次校验在低相似度区间放弃缓存。这比调节单一阈值能同时提升召回率和命中率。缓存预热的策略设计新部署的系统缓存是空的前几小时命中率极低。可以在上线前对FAQ、产品文档等高频知识做批量预Embedding写入缓存。但预热量过大又会影响启动速度。建议预热控制在总缓存容量20%以内优先填充访问频次可预判的知识。向量索引的性能拐点当缓存条目超过10万时暴力搜索的延迟从毫秒级上升到十毫秒级。此时应升级为FAISS的IVFPQ索引在损失不到1%召回率的代价下将搜索延迟稳定在毫秒级。但如果缓存量还未达到这个量级引入FAISS的运维复杂度是过度的。五、总结Semantic Cache不是传统缓存的简单升级而是一套完整的查询优化管道精确匹配做最快的过滤语义匹配识别相同意图的不同表述灰色区域用意图标签做安全的二次判断。三层设计让缓存命中率在大部分场景下达到30%-40%对应LLM API成本节省在同等比例。核心设计原则是保守的缓存策略优于激进的缓存策略——返回一个不确定是否正确的缓存结果比重新调用LLM的成本更高因为错误答案损害的是用户信任。在实践中先追求精确匹配的高准确率再逐步调优语义匹配的召回率是一条风险可控的优化路径。
Prompt缓存策略与成本优化：Semantic Cache在企业Agent中的应用
发布时间：2026/7/11 14:54:41
Prompt缓存策略与成本优化Semantic Cache在企业Agent中的应用一、深度引言企业Agent的运营成本中LLM API调用费用往往占据60%以上的比例。一个中等规模的企业知识问答Agent每天处理5000次查询如果每次查询都完整发送上下文和Prompt仅GPT-4级别的API调用费用就在每日数百美元级别——按年计算是一笔数十万的支出。但深入分析这些查询的分布会发现一个优化空间大约30%-40%的查询在语义上是高度相似的。公司的年假政策是什么和年假有多少天本质上是同一个问题的不同表述但标准实现会发送两次完整的API请求。这两次请求消耗的Token总量、响应时间几乎相同但其中有一次是冗余的。Semantic Cache语义缓存的思路是将相同语义而非相同字符串的查询结果缓存起来在发送API请求之前先用语义相似度匹配检查缓存。本文将拆解Semantic Cache在企业Agent场景下的完整实现方案。flowchart TD A[用户查询] -- B{精确缓存匹配} B --|命中| C[直接返回] B --|未命中| D{语义相似度匹配} D --|相似度 阈值| E[返回缓存 标注] D --|相似度 阈值| F[发送LLM请求] F -- G[LLM响应] G -- H[生成Embedding] H -- I[存入向量缓存] I -- J[返回给用户] subgraph 缓存策略 K[LRU淘汰策略] L[TTL过期策略] M[相似度阈值动态调整] end style D fill:#4A90D9,color:#fff style F fill:#F39C12,color:#fff style K fill:#27AE60,color:#fff二、原理剖析语义相似度的计算和阈值选择Semantic Cache的核心是将用户查询映射到向量空间通过余弦相似度判断是否命中缓存。但这引入了两个关键问题阈值设多少合适向量相似度高是否意味着回答可以复用对于阈值经验和实验数据指向0.92-0.95的余弦相似度区间。低于0.92语义差异开始出现——退款流程和退货流程在向量空间中可能相似度0.90但回答内容有本质区别。高于0.95命中率显著下降缓存的成本节约效果有限。但仅靠相似度阈值是不够的。需要在缓存的值中额外存储问题的意图标签——由LLM在首次回答时同时输出。当候选缓存的相似度在0.92-0.95的灰色区间时通过意图标签做二次校验如果意图标签相同即使相似度稍低也接受缓存命中如果意图标签不同即使相似度高也拒绝。缓存生命周期管理语义缓存的生命周期管理比传统KV缓存复杂。传统缓存直接设置TTL 5分钟就行但语义缓存的TTL需要考虑两个维度时效性——某些问题的答案随时间变化如当前股价需要短TTL稳定性——某些问题的答案长期不变如公司地址可以长TTL甚至永不过期。实践中采用分级TTL策略知识类问题公司政策、产品文档TTL设置为24小时实时类问题股票、天气、新闻TTL设置为5分钟通用类问题概念解释、技术问答TTL设置为1小时。TTL分级信息需要从查询的意图分类中获得。淘汰策略方面LRULeast Recently Used是基础方案但语义缓存有一个独特的优化维度如果一个向量在特征空间中密度很高——周围有大量低相似度的查询——这个向量的潜在命中概率更高可以考虑增加其驻留权重。这是热点保护策略与LRU仅看访问频次不同它还看该缓存在向量空间中的可复用性。Embedding模型的选择语义缓存的精度取决于Embedding模型的语义区分能力。在成本敏感的场景下可以分层使用不同精度的Embedding模型粗筛阶段使用轻量模型如text2vec-base-chinese128MB筛选到Top-K候选后用高精度模型做精细匹配。这种两级流水线可以在保证精度的同时降低计算成本。三、生产级代码以下是一个完整的Semantic Cache实现覆盖向量缓存、相似度匹配、TTL管理和分层淘汰策略。import hashlib import json import logging import threading import time from collections import OrderedDict from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum from typing import Any, Callable, Dict, List, Optional, Tuple import numpy as np logger logging.getLogger(semantic_cache) # 向量工具 class VectorUtils: 向量计算工具 —— 处理Embedding的各种向量操作 staticmethod def cosine_similarity(vec1: List[float], vec2: List[float]) - float: 计算余弦相似度 为什么不用np.dot直接算 需要先做向量范数检查零向量会导致除零错误 a np.array(vec1, dtypenp.float32) b np.array(vec2, dtypenp.float32) norm_a np.linalg.norm(a) norm_b np.linalg.norm(b) if norm_a 0.0 or norm_b 0.0: return 0.0 return float(np.dot(a, b) / (norm_a * norm_b)) staticmethod def normalize(vec: List[float]) - List[float]: 向量归一化 —— 预处理后的向量相似度计算更快 a np.array(vec, dtypenp.float32) norm np.linalg.norm(a) if norm 0: return vec return (a / norm).tolist() # TTL分级策略 class ContentType(Enum): 内容类型 —— 决定缓存TTL KNOWLEDGE knowledge # 知识类公司政策、文档 → 24h REALTIME realtime # 实时类股票、天气 → 5min GENERAL general # 通用类概念解释 → 1h # TTL映射秒 TTL_MAP { ContentType.KNOWLEDGE: 86400, # 24小时 ContentType.REALTIME: 300, # 5分钟 ContentType.GENERAL: 3600, # 1小时 } dataclass class CacheEntry: 缓存条目 query: str query_embedding: List[float] # 查询的向量表示 answer: str intent_label: str # 意图标签用于相似度灰色区域的二次校验 content_type: ContentType ContentType.GENERAL ttl_seconds: int 3600 created_at: float field(default_factorytime.time) hit_count: int 0 # 命中次数影响淘汰策略 last_hit_at: Optional[float] None def is_expired(self) - bool: 检查是否过期 return (time.time() - self.created_at) self.ttl_seconds def heat_score(self) - float: 计算热度分数 —— 用于淘汰决策 为什么综合命中次数和时间衰减 纯LRU会淘汰很久之前热门但现在无人访问的条目 加入时间衰减让淘汰更准确 age_hours (time.time() - self.created_at) / 3600 time_decay max(0.1, 1.0 - age_hours / 48) # 48小时线性衰减 return self.hit_count * time_decay # Semantic Cache 引擎 class SemanticCache: 语义缓存引擎 —— 核心实现 def __init__( self, similarity_threshold: float 0.92, grey_zone_min: float 0.88, max_entries: int 10000, embedding_dim: int 768, ): self.similarity_threshold similarity_threshold self.grey_zone_min grey_zone_min self.max_entries max_entries self.embedding_dim embedding_dim # 数据存储 self._entries: OrderedDict[str, CacheEntry] OrderedDict() # 精确缓存文本 → 缓存键处理完全相同的查询 self._exact_cache: Dict[str, str] {} # 向量索引简化版内存索引生产环境用FAISS/Milvus self._embeddings: List[np.ndarray] [] self._entry_ids: List[str] [] self._lock threading.RLock() # 统计指标 self._total_queries 0 self._exact_hits 0 self._semantic_hits 0 self._misses 0 def get( self, query: str, query_embedding: List[float], embedder: Optional[Callable] None, ) - Tuple[Optional[str], Dict]: 查询缓存 —— 返回(缓存的回答, 元信息) 查询流程精确匹配 → 语义匹配 → 缓存未命中 with self._lock: self._total_queries 1 # 第一层精确匹配速度快成本零 query_hash hashlib.md5(query.encode()).hexdigest() if query_hash in self._exact_cache: entry_id self._exact_cache[query_hash] entry self._entries.get(entry_id) if entry and not entry.is_expired(): entry.hit_count 1 entry.last_hit_at time.time() self._exact_hits 1 # LRU: 将命中的条目移到末尾 self._entries.move_to_end(entry_id) logger.debug(f精确命中: {query[:30]}...) return entry.answer, { source: exact_cache, hits: entry.hit_count, } else: # 过期清理 if entry_id in self._exact_cache: del self._exact_cache[query_hash] if entry_id in self._entries: del self._entries[entry_id] # 第二层语义匹配速度中等成本取决于向量搜索实现 normalized VectorUtils.normalize(query_embedding) best_entry, best_similarity self._semantic_search(normalized) if best_entry and best_similarity self.similarity_threshold: # 高相似度直接命中 best_entry.hit_count 1 best_entry.last_hit_at time.time() self._semantic_hits 1 self._entries.move_to_end(best_entry.query) logger.debug( f语义命中: {query[:30]}... f- {best_entry.query[:30]}... f(相似度: {best_similarity:.3f}) ) return best_entry.answer, { source: semantic_cache, similarity: round(best_similarity, 3), matched_query: best_entry.query, } if best_entry and best_similarity self.grey_zone_min: # 灰色区域意图标签二次校验 if self._intent_match(query, best_entry.intent_label): best_entry.hit_count 1 best_entry.last_hit_at time.time() self._semantic_hits 1 self._entries.move_to_end(best_entry.query) logger.info( f灰色区域命中(意图匹配): f{query[:30]}... (相似度: {best_similarity:.3f}) ) return best_entry.answer, { source: semantic_cache_grey, similarity: round(best_similarity, 3), matched_query: best_entry.query, } # 缓存未命中 self._misses 1 return None, {source: miss} def set( self, query: str, query_embedding: List[float], answer: str, intent_label: str general, content_type: ContentType ContentType.GENERAL, ): 写入缓存 with self._lock: # 淘汰检查 if len(self._entries) self.max_entries: self._evict() normalized VectorUtils.normalize(query_embedding) ttl TTL_MAP.get(content_type, 3600) entry CacheEntry( queryquery, query_embeddingnormalized, answeranswer, intent_labelintent_label, content_typecontent_type, ttl_secondsttl, ) self._entries[query] entry query_hash hashlib.md5(query.encode()).hexdigest() self._exact_cache[query_hash] query # 更新向量索引 self._embeddings.append(np.array(normalized, dtypenp.float32)) self._entry_ids.append(query) logger.debug( f缓存写入: {query[:30]}... f(TTL: {ttl}s, 类型: {content_type.value}) ) def _semantic_search( self, query_vec: List[float] ) - Tuple[Optional[CacheEntry], float]: 向量相似度搜索 —— 找最相似的缓存条目 为什么用暴力搜索而非近似搜索 缓存条目通常在万级以下暴力搜索的延迟在毫秒级。 超过10万条时应升级为FAISS的IVF索引 if not self._embeddings: return None, 0.0 query_arr np.array(query_vec, dtypenp.float32) # 批量计算余弦相似度 embeddings_matrix np.stack(self._embeddings) # 为什么用dot product而非cosine_similarity函数 # 向量已归一化dot product等价于cosine similarity # 且numpy的批量dot运算比循环调用cosine_similarity快10倍 similarities np.dot(embeddings_matrix, query_arr) max_idx int(np.argmax(similarities)) max_sim float(similarities[max_idx]) if max_sim self.grey_zone_min: return None, max_sim entry_id self._entry_ids[max_idx] entry self._entries.get(entry_id) if entry and not entry.is_expired(): return entry, max_sim return None, max_sim def _intent_match(self, query: str, cached_intent: str) - bool: 意图标签匹配 —— 灰色区域的二次校验 为什么用意图标签而非让LLM再判断一次 LLM判断的成本接近一次完整请求失去了缓存的意义。 意图标签是回答时附带生成的成本极低 # 生产环境应使用轻量意图分类器 # 这里简化实现检查关键词共现 intent_keywords { policy_query: [政策, 规定, 流程, 要求, 条件], definition: [什么是, 定义, 解释, 含义, 意思], troubleshooting: [报错, 失败, 不行, 问题, 解决], comparison: [区别, 对比, 哪个好, 选择], } keywords intent_keywords.get(cached_intent, []) return any(kw in query for kw in keywords) def _evict(self): 淘汰策略 —— 综合热度和访问时间的混合淘汰 为什么不用纯LRU LRU在热点数据频繁变化时表现好但语义缓存的 特点是热点数据高度集中——少数问题的变体能命中大量查询。 综合热度分数可以更好地保护这些枢纽缓存 if not self._entries: return # 计算所有条目的热度分数 scored [ (entry_id, entry.heat_score()) for entry_id, entry in self._entries.items() ] scored.sort(keylambda x: x[1]) # 按热度升序最低的最先淘汰 # 淘汰最低热度的条目 evict_count max(1, len(self._entries) // 10) # 至少淘汰1个最多10% for i in range(evict_count): if i len(scored): break entry_id scored[i][0] entry self._entries[entry_id] # 从所有索引中移除 del self._entries[entry_id] query_hash hashlib.md5(entry.query.encode()).hexdigest() if query_hash in self._exact_cache: del self._exact_cache[query_hash] # 移除向量索引 try: idx self._entry_ids.index(entry_id) self._entry_ids.pop(idx) self._embeddings.pop(idx) except ValueError: pass logger.info( f缓存淘汰: 移除了 {evict_count} 个条目 f当前缓存大小: {len(self._entries)} ) def cleanup_expired(self): 清理过期条目 —— 建议在低负载时定时调用 with self._lock: expired_ids [ eid for eid, entry in self._entries.items() if entry.is_expired() ] for eid in expired_ids: entry self._entries[eid] del self._entries[eid] query_hash hashlib.md5(entry.query.encode()).hexdigest() if query_hash in self._exact_cache: del self._exact_cache[query_hash] try: idx self._entry_ids.index(eid) self._entry_ids.pop(idx) self._embeddings.pop(idx) except ValueError: pass if expired_ids: logger.info(f清理了 {len(expired_ids)} 个过期缓存条目) def get_stats(self) - Dict: 获取缓存统计 with self._lock: total max(self._total_queries, 1) return { total_queries: self._total_queries, exact_hits: self._exact_hits, semantic_hits: self._semantic_hits, misses: self._misses, hit_rate: round( (self._exact_hits self._semantic_hits) / total, 3 ), cache_size: len(self._entries), max_entries: self.max_entries, estimated_savings_tokens: ( (self._exact_hits self._semantic_hits) * 500 # 估算每次命中节省500 tokens ), } # 使用示例 # 模拟Embedding函数生产环境替换为真实的Embedding模型API def mock_embed(text: str) - List[float]: 模拟Embedding —— 实际使用OpenAI/本地模型 import random random.seed(hash(text) % (2**32)) return [random.random() for _ in range(768)] if __name__ __main__: cache SemanticCache( similarity_threshold0.92, max_entries1000, ) # 场景1写入基础查询 base_query 公司的年假政策是什么 base_embedding mock_embed(base_query) cache.set( querybase_query, query_embeddingbase_embedding, answer公司年假政策入职满1年享有5天带薪年假满3年享有10天..., intent_labelpolicy_query, content_typeContentType.KNOWLEDGE, ) # 场景2语义相似查询应命中缓存 similar_query 年假有多少天 similar_embedding mock_embed(similar_query) answer, meta cache.get(similar_query, similar_embedding) if answer: print(f缓存命中! 来源: {meta[source]}) print(f原始查询: {base_query}) print(f当前查询: {similar_query}) print(f回答: {answer[:50]}...) else: print(缓存未命中需要调用LLM) # 模拟LLM调用省略 # llm_answer call_llm(similar_query) # cache.set(similar_query, similar_embedding, llm_answer, ...) # 打印统计 stats cache.get_stats() print(f\n缓存统计: {json.dumps(stats, ensure_asciiFalse, indent2)})四、边界权衡召回率vs命中率是Semantic Cache最核心的权衡。相似度阈值越高返回的缓存结果越可靠但命中率越低节约的API成本越少。相似度阈值越低命中率越高但准确度可能下降错误缓存返回可能导致用户不信任系统。解决方案不是找最优阈值而是设计多级判断——在高相似度区间直接返回在灰色区间用意图标签或轻量模型做二次校验在低相似度区间放弃缓存。这比调节单一阈值能同时提升召回率和命中率。缓存预热的策略设计新部署的系统缓存是空的前几小时命中率极低。可以在上线前对FAQ、产品文档等高频知识做批量预Embedding写入缓存。但预热量过大又会影响启动速度。建议预热控制在总缓存容量20%以内优先填充访问频次可预判的知识。向量索引的性能拐点当缓存条目超过10万时暴力搜索的延迟从毫秒级上升到十毫秒级。此时应升级为FAISS的IVFPQ索引在损失不到1%召回率的代价下将搜索延迟稳定在毫秒级。但如果缓存量还未达到这个量级引入FAISS的运维复杂度是过度的。五、总结Semantic Cache不是传统缓存的简单升级而是一套完整的查询优化管道精确匹配做最快的过滤语义匹配识别相同意图的不同表述灰色区域用意图标签做安全的二次判断。三层设计让缓存命中率在大部分场景下达到30%-40%对应LLM API成本节省在同等比例。核心设计原则是保守的缓存策略优于激进的缓存策略——返回一个不确定是否正确的缓存结果比重新调用LLM的成本更高因为错误答案损害的是用户信任。在实践中先追求精确匹配的高准确率再逐步调优语义匹配的召回率是一条风险可控的优化路径。
