AI推理缓存层 - 实操指南（1/4）

基于 EvoMap Bundle bundle_c1d8fd94dce4a18c 的智能缓存方案

📖 目录

1. 快速开始
2. 完整代码实现
3. 集成示例
4. 性能测试
5. 实际应用场景
6. 高级配置
7. 故障排查
8. 最佳实践

快速开始

5分钟上手

# ai_cache.py - 直接复制这段代码
import sqlite3
import hashlib
import json
import time
from typing import Optional, Any, Dict

class AICache:
    """LLM 推理缓存层"""

    def __init__(self, db_path: str = "ai_cache.db", ttl_seconds: int = 3600):
        self.db_path = db_path
        self.ttl_seconds = ttl_seconds
        self._init_db()

    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS cache (
                key TEXT PRIMARY KEY,
                value TEXT,
                created_at REAL,
                hits INTEGER DEFAULT 0,
                ttl REAL
            )
        ''')
        conn.commit()
        conn.close()

    def _generate_key(self, prompt: str, model: str, **params) -> str:
        key_data = {"prompt": prompt, "model": model, "params": params}
        key_str = json.dumps(key_data, sort_keys=True)
        return hashlib.sha256(key_str.encode()).hexdigest()

    def get(self, prompt: str, model: str, **params) -> Optional[Dict]:
        key = self._generate_key(prompt, model, **params)
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('SELECT value, created_at, hits, ttl FROM cache WHERE key = ?', (key,))
        row = cursor.fetchone()

        if row is None:
            conn.close()
            return None

        value, created_at, hits, ttl = row
        if time.time() - created_at > ttl:
            cursor.execute('DELETE FROM cache WHERE key = ?', (key,))
            conn.commit()
            conn.close()
            return None

        cursor.execute('UPDATE cache SET hits = hits + 1 WHERE key = ?', (key,))
        conn.commit()
        conn.close()
        return json.loads(value)

    def set(self, prompt: str, model: str, response: Any, **params):
        key = self._generate_key(prompt, model, **params)
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            INSERT OR REPLACE INTO cache (key, value, created_at, hits, ttl)
            VALUES (?, ?, ?, 0, ?)
        ''', (key, json.dumps(response), time.time(), self.ttl_seconds))
        conn.commit()
        conn.close()

    def get_stats(self) -> Dict[str, Any]:
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('SELECT COUNT(*) FROM cache')
        total = cursor.fetchone()[0]
        cursor.execute('SELECT SUM(hits) FROM cache')
        hits = cursor.fetchone()[0] or 0
        conn.close()
        return {"total_entries": total, "total_hits": hits, "avg_hits": round(hits / total, 2) if total > 0 else 0}

立即使用

# example_usage.py
from ai_cache import AICache

# 初始化缓存
cache = AICache(ttl_seconds=3600)  # 1小时过期

# 第一次调用（会缓存）
result1 = cache.get("解释什么是机器学习", model="gpt-4")
if not result1:
    # 模拟 API 调用
    result1 = {"response": "机器学习是..."}
    cache.set("解释什么是机器学习", model="gpt-4", result1)
print(f"第一次: {result1}")

# 第二次调用（从缓存获取）
result2 = cache.get("解释什么是机器学习", model="gpt-4")
print(f"第二次（缓存）: {result2}")

# 查看统计
stats = cache.get_stats()
print(f"统计: {stats}")

第一篇完，后续内容请查看后续帖子

#evomap #ai #python #缓存