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

Python 如何给 AI API 加缓存：减少重复请求并提升响应速度

news 2026/7/31 7:47:46

在 AI 工具、自动化脚本和知识问答项目中，相同问题可能被重复请求。给稳定结果增加缓存，可以减少重复调用，也能让常见问题更快返回。

为什么需要缓存？

如果每次请求都直接访问 AI API，可能会遇到这些情况：

相同问题被重复处理
响应时间不稳定
批量任务重复消耗资源
测试阶段反复发送相同请求
常见问题无法快速返回

缓存的基本思路是：

收到请求 → 生成缓存键 → 查询缓存 → 命中则直接返回 → 未命中才调用 API

不过，缓存并不适合所有请求。随机性较高、依赖实时信息或包含隐私内容的请求，需要谨慎处理。

一、先用字典实现最小缓存

本地测试时，可以先使用 Python 字典：

cache={}defget_cached(key):returncache.get(key)defsave_cache(key,value):cache[key]=value

这种方式简单，但程序重启后数据会消失，而且不适合多进程或多实例服务。

它更适合验证缓存逻辑是否正确。

二、如何生成稳定的缓存键？

缓存键不能只使用用户问题，还应该考虑模型和关键参数：

importhashlibimportjsondefbuild_cache_key(prompt:str,model:str,temperature:float=0.0)->str:payload={"prompt":prompt,"model":model,"temperature":temperature,}raw=json.dumps(payload,ensure_ascii=False,sort_keys=True)returnhashlib.sha256(raw.encode("utf-8")).hexdigest()

这样可以避免不同模型、不同参数之间互相读错缓存。

如果系统提示词、工具参数或版本号会影响结果，也应该把它们加入缓存键。

三、把缓存接到 API 调用前面

下面是一个简单的完整示例：

fromopenaiimportOpenAI client=OpenAI(api_key="your-api-key",base_url="https://your-api-domain.com/v1",)cache={}defask_llm(prompt:str,model:str="your-model-name")->str:key=build_cache_key(prompt,model)ifkeyincache:returncache[key]response=client.chat.completions.create(model=model,messages=[{"role":"user","content":prompt}],)result=response.choices[0].message.content cache[key]=resultreturnresult

第一次请求会访问 API，后续相同请求可以直接从缓存返回。

四、为什么要设置过期时间？

缓存不能永久有效。内容更新、提示词变化或模型升级后，旧结果可能已经不适用。

可以给缓存增加过期时间：

importtime cache={}defsave_cache(key:str,value:str,ttl:int=300):cache[key]={"value":value,"expires_at":time.time()+ttl,}defread_cache(key:str):item=cache.get(key)ifnotitem:returnNoneifitem["expires_at"]<=time.time():cache.pop(key,None)returnNonereturnitem["value"]

ttl=300表示缓存 5 分钟后过期。实际时间要根据内容更新频率来设置。

五、哪些内容适合缓存？

比较适合缓存的内容：

固定格式的文本转换
重复的分类任务
稳定的代码解释
常见开发问题
测试环境中的固定请求

不适合直接缓存的内容：

实时新闻
实时价格或库存
强依赖当前时间的任务
包含个人隐私的信息
每次都要求随机结果的创作任务

缓存前要先判断：同一个输入在一段时间内是否允许返回相同结果。

六、使用 SQLite 保存本地缓存

如果希望程序重启后缓存仍然存在，可以使用 SQLite：

importsqlite3importtime conn=sqlite3.connect("cache.db")conn.execute(""" CREATE TABLE IF NOT EXISTS responses ( cache_key TEXT PRIMARY KEY, content TEXT NOT NULL, expires_at REAL NOT NULL ) """)conn.commit()

写入缓存：

defsave_sqlite_cache(key:str,content:str,ttl:int=300):expires_at=time.time()+ttl conn.execute("INSERT OR REPLACE INTO responses VALUES (?, ?, ?)",(key,content,expires_at),)conn.commit()

读取缓存：

defread_sqlite_cache(key:str):row=conn.execute("SELECT content, expires_at FROM responses WHERE cache_key = ?",(key,),).fetchone()ifnotrow:returnNonecontent,expires_at=rowifexpires_at<=time.time():conn.execute("DELETE FROM responses WHERE cache_key = ?",(key,))conn.commit()returnNonereturncontent