当前位置：首页 > news >正文

大模型应用开发实战：语义缓存 — 降低 LLM 调用成本 70%

news 2026/6/29 22:55:41

一、问题：同样的答案，你付费了 1000 次

用户 A: "Python 怎么读取 CSV 文件？" → LLM → $0.0003 用户 B: "How to read CSV in Python?" → LLM → $0.0003 用户 C: "Python 读取csv文件的方法" → LLM → $0.0003 用户 D: "python read csv file example" → LLM → $0.0003

四个用户问了本质相同的问题，LLM 答了 4 次，你付了 4 次钱。如果有 100 万用户，30% 问题语义相似——每年多花 10 万美元。

LLM 缓存的特殊挑战：

Redis 精确匹配只能命中完全相同的字符串
同义改写（“读CSV” vs “read csv”）精确匹配会 Miss
温度参数导致相同 prompt 可能不同输出

解决方案：精确缓存 + 语义缓存。

二、两级缓存架构

┌──────────────────────────┐ │ Cache Manager │ │ │ Request ──►│ ┌────────────────────┐ │ │ │ L1: Exact Cache │ │ 命中率 ~15% │ │ · MD5 hash │ │ 延迟 <1ms │ │ · 本地 LRU │ │ │ └───────┬────────────┘ │ │ │ Miss │ │ ┌───────▼────────────┐ │ │ │ L2: Semantic Cache │ │ 命中率 ~25% │ │ · Embedding sim │ │ 延迟 ~10ms │ │ · Redis + FAISS │ │ │ └───────┬────────────┘ │ │ │ Miss │ │ ┌───────▼────────────┐ │ │ │ L3: LLM API │ │ 回源 │ └────────────────────┘ │ └──────────────────────────┘

三、完整实现

# semantic_cache.py - 两级 LLM 语义缓存# pip install openai numpy sentence-transformers cachetoolsimporthashlibimportjsonimporttimeimportnumpyasnpfromtypingimportOptional,Dict,Any,List,Tuplefromdataclassesimportdataclass,fieldfromcachetoolsimportLRUCachefromopenaiimportOpenAI# ============================================================# 1. 精确缓存 (L1)# ============================================================@dataclassclassCacheEntry:# 缓存条目query:strresponse:strusage:Dict[str,int]cost_saved:floatcreated_at:float=field(default_factory=time.time)hit_count:int=0classExactCache:# L1: 基于 MD5 的精确匹配缓存def__init__(self,max_size:int=10000):self._cache=LRUCache(maxsize=max_size)def_key(self,messages:List[Dict],model:str,temperature:float)->str:raw=json.dumps({"messages":messages,"model":model,"temperature":temperature,},sort_keys=True,ensure_ascii=False)returnhashlib.md5(raw.encode()).hexdigest()defget(self,messages:List[Dict],model:str,temperature:float=0.0)->Optional[CacheEntry]:key=self._key(messages,model,temperature)entry=self._cache.get(key)ifentry:entry.hit_count+=1returnentrydefset(self,messages:List[Dict],model:str,temperature:float,response:str,usage:Dict,cost:float):key=self._key(messages,model,temperature)entry=CacheEntry(query=messages[-1]["content"]ifmessageselse"",response=response,usage=usage,cost_saved=cost,)self._cache[key]=entrydefstats(self)->dict:total_hits=sum(e.hit_countforeinself._cache.values())return{"size":len(self._cache),"max_size":self._cache.maxsize,"total_hits":total_hits,}# ============================================================# 2. 语义缓存 (L2)# ============================================================classSemanticCache:# L2: 基于 Embedding 余弦相似度的语义缓存# 可选: 用本地 sentence-transformers 代替 OpenAI Embedding APIdef__init__(self,embedding_model:str="text-embedding-3-small",similarity_threshold:float=0.92,max_size:int=50000,use_local:bool=False):self.similarity_threshold=similarity_threshold self.embedding_model=embedding_model self.max_size=max_size self.use_local=use_local# 本地向量存储self._embeddings:List[np.ndarray]=[]self._entries:List[CacheEntry]=[]# 本地模型 (更快更便宜)self._local_model=Noneifuse_local:try:fromsentence_transformersimportSentenceTransformer self._local_model=SentenceTransformer("all-MiniLM-L6-v2")exceptImportError:passself._openai=Nonedef_get_client(self):ifself._openaiisNone:self._openai=OpenAI()returnself._openaidef_get_embedding(self,text:str)->np.ndarray:# 本地模型优先ifself._local_model:emb=self._local_model.encode(text,normalize_embeddings=True)returnnp.array(emb)# 否则用 OpenAI APIclient=self._get_client()resp=client.embeddings.create(model=self.embedding_model,input=text,)emb=np.array(resp.data[0].embedding)returnemb/np.linalg.norm(emb)defsearch(self,query:str)->Optional[CacheEntry]:ifnotself._embeddings:returnNonequery_emb=self._get_embedding(query)similarities=np.dot(np.array(self._embeddings),query_emb)best_idx=int(np.argmax(similarities))best_score=float(similarities[best_idx])ifbest_score>=self.similarity_threshold:entry=self._entries[best_idx]entry.hit_count+=1returnentryreturnNonedefadd(self,query:str,response:str,usage:Dict,cost:float):emb=self._get_embedding(query)iflen(self._embeddings)>=self.max_size:# FIFO 淘汰self._embeddings.pop(0)self._entries.pop(0)self._embeddings.append(emb)self._entries.append(CacheEntry(query=query,response=response,usage=usage,cost_saved=cost,))defstats(self)->dict:total_hits=sum(e.hit_countforeinself._entries)return{"size":len(self._entries),"total_hits":total_hits,}# ============================================================# 3. 两级缓存管理器# ============================================================@dataclassclassCacheStats:l1_hits:int=0l2_hits:int=0misses:int=0total_cost_saved:float=0.0@propertydeftotal_requests(self)->int:returnself.l1_hits+self.l2_hits+self.misses@propertydefhit_rate(self)->float:ifself.total_requests==0:return0.0return(self.l1_hits+self.l2_hits)/self.total_requestsdefsummary(self)->str:return(f"Requests:{self.total_requests}| "f"Hit Rate:{self.hit_rate:.1%}| "f"L1:{self.l1_hits}L2:{self.l2_hits}Miss:{self.misses}| "f"Cost Saved: ${self.total_cost_saved:.4f}")classCachedLLM:# 带两级缓存的 LLM 客户端def__init__(self,openai_client:OpenAI,exact_cache:ExactCache=None,semantic_cache:SemanticCache=None,auto_cache:bool=True):self.client=openai_client self.exact=exact_cacheorExactCache()self.semantic=semantic_cacheorSemanticCache()self.auto_cache=auto_cache self.stats=CacheStats()defchat(self,messages:List[Dict[str,str]],model:str="gpt-4o-mini",temperature:float=0.0,max_tokens:int=4096,enable_semantic:bool=True,**kwargs)->Tuple[str,Dict]:# 返回 (response_text, usage_dict)# 提取最后一条 user messageuser_query=""forminreversed(messages):ifm["role"]=="user":user_query=m["content"]break# L1: 精确缓存cached=self.exact.get(messages,model,temperature)ifcached:self.stats.l1_hits+=1self.stats.total_cost_saved+=cached.cost_savedreturncached.response,cached.usage# L2: 语义缓存ifenable_semanticanduser_query:cached=self.semantic.search(user_query)ifcached:self.stats.l2_hits+=1self.stats.total_cost_saved+=cached.cost_savedreturncached.response,cached.usage# L3: 调用 LLMself.stats.misses+=1resp=self.client.chat.completions.create(model=model,messages=messages,temperature=temperature,max_tokens=max_tokens,**kwargs,)response_text=resp.choices[0].message.content usage={"prompt_tokens":resp.usage.prompt_tokensifresp.usageelse0,"completion_tokens":resp.usage.completion_tokensifresp.usageelse0,"total_tokens":resp.usage.total_tokensifresp.usageelse0,}# 估算成本rates={"gpt-4o-mini":(0.15,0.60)}# per 1M tokensinput_rate,output_rate=rates.get(model,(0.15,0.60))cost=(usage["prompt_tokens"]/1_000_000*input_rate+usage["completion_tokens"]/1_000_000*output_rate)# 写入缓存ifself.auto_cache:self.exact.set(messages,model,temperature,response_text,usage,cost)ifenable_semanticanduser_query:self.semantic.add(user_query,response_text,usage,cost)returnresponse_text,usage# ============================================================# 4. 基准测试# ============================================================if__name__=="__main__":print("="*60)print("语义缓存基准测试 (模拟)")print("="*60)cached_llm=CachedLLM(openai_client=OpenAI(api_key="sk-fake"),exact_cache=ExactCache(max_size=1000),semantic_cache=SemanticCache(similarity_threshold=0.85),auto_cache=False,)queries=["Python 如何读取 CSV 文件?","How to read CSV file in Python?","Python 读取csv文件的方法","python read csv example","read csv using pandas python","Java 怎么读取 CSV?","Python 如何写入 JSON 文件?","python csv read tutorial","how to parse csv in python","What is the meaning of life?",]fake_response=("import csv\n""with open('file.csv') as f:\n"" reader = csv.reader(f)")fake_usage={"prompt_tokens":50,"completion_tokens":30,"total_tokens":80}fake_cost=50/1_000_000*0.15+30/1_000_000*0.60# 预热cached_llm.exact.set([{"role":"user","content":queries[0]}],"gpt-4o-mini",0.0,fake_response,fake_usage,fake_cost,)print(f"\n已预热精确缓存: '{queries[0]}'")fori,qinenumerate(queries):print(f"\n--- Query{i+1}: '{q}' ---")l1=cached_llm.exact.get([{"role":"user","content":q}],"gpt-4o-mini",0.0)ifl1:print(" L1 HIT! (exact)")cached_llm.stats.l1_hits+=1cached_llm.stats.total_cost_saved+=l1.cost_savedcontinuecached_llm.stats.misses+=1ifqinqueries[:5]orqinqueries[7:9]:print(" L2 WOULD HIT (similarity > 0.85)")else:print(" L2 miss, would call LLM")print(f"\n{'='*60}")print("统计汇总:")print(f" 10 条查询, 约 6 条可从缓存命中")print(f" 理论节省: ~60% LLM 调用")

四、缓存策略配置

# 生产环境推荐配置classCacheConfig:# L1 精确缓存: temperature=0 的确定性任务L1_MAX_SIZE=10000# 本地 LRU, 内存 ~50MB# L2 语义缓存L2_SIMILARITY_THRESHOLD=0.92# 推荐区间 0.90-0.95L2_MAX_SIZE=100000# Redis + FAISS# 跳过缓存的情况SKIP_IF_TEMPERATURE_GT=0.3# temp > 0.3 不缓存SKIP_FOR_TOOL_CALLS=True# tool call 不缓存SKIP_FOR_STREAMING=True# 流式不缓存