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

2301_77724322

497人浏览 · 2026-07-30 17:25:53

2301_77724322 · 2026-07-30 17:25:53 发布

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

为什么需要缓存？

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

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

缓存的基本思路是：

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

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

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

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

cache = {}


def get_cached(key):
    return cache.get(key)


def save_cache(key, value):
    cache[key] = value

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

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

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

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

import hashlib
import json


def build_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)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

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

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

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

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

from openai import OpenAI

client = OpenAI(
    api_key="your-api-key",
    base_url="https://your-api-domain.com/v1",
)

cache = {}


def ask_llm(prompt: str, model: str = "your-model-name") -> str:
    key = build_cache_key(prompt, model)

    if key in cache:
        return cache[key]

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
    )
    result = response.choices[0].message.content
    cache[key] = result
    return result

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

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

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

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

import time

cache = {}


def save_cache(key: str, value: str, ttl: int = 300):
    cache[key] = {
        "value": value,
        "expires_at": time.time() + ttl,
    }


def read_cache(key: str):
    item = cache.get(key)
    if not item:
        return None

    if item["expires_at"] <= time.time():
        cache.pop(key, None)
        return None

    return item["value"]

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

五、哪些内容适合缓存？

比较适合缓存的内容：

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

不适合直接缓存的内容：

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

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

六、使用 SQLite 保存本地缓存

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

import sqlite3
import time

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()

写入缓存：

def save_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()

读取缓存：

def read_sqlite_cache(key: str):
    row = conn.execute(
        "SELECT content, expires_at FROM responses WHERE cache_key = ?",
        (key,),
    ).fetchone()

    if not row:
        return None

    content, expires_at = row
    if expires_at <= time.time():
        conn.execute("DELETE FROM responses WHERE cache_key = ?", (key,))
        conn.commit()
        return None

    return content