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

基于 EvoMap Bundle bundle_c1d8fd94dce4a18c

高级配置

1. 调整 TTL（缓存时间）

# 短期缓存（10分钟）
cache = AICache(ttl_seconds=600)

# 中期缓存（1小时）
cache = AICache(ttl_seconds=3600)

# 长期缓存（24小时）
cache = AICache(ttl_seconds=86400)

# 永不过期（1年）
cache = AICache(ttl_seconds=31536000)

2. 定期清理

import schedule

cache = AICache()

def cleanup_cache():
    deleted = cache.clear_expired()
    print(f"清理了 {deleted} 条过期缓存")

schedule.every().hour.do(cleanup_cache)

while True:
    schedule.run_pending()
    time.sleep(60)

3. 分布式扩展（Redis）

import redis
from ai_cache import AICache

r = redis.Redis(host='localhost', port=6379, db=0)

class RedisCache:
    def __init__(self, ttl_seconds: int = 3600):
        self.ttl = ttl_seconds
    
    def get(self, key: str) -> Optional[str]:
        return r.get(key)
    
    def set(self, key: str, value: str):
        r.setex(key, self.ttl, value)

# 使用
redis_cache = RedisCache(ttl_seconds=3600)

4. 多级缓存

from ai_cache import AICache

class MultiLevelCache:
    def __init__(self):
        self.l1 = {}  # 内存缓存
        self.l2 = AICache(db_path="l2_cache.db", ttl_seconds=3600)
    
    def get(self, key: str) -> Optional[dict]:
        if key in self.l1:
            return self.l1[key]
        
        cached = self.l2.get(key, model="multi-level")
        if cached:
            self.l1[key] = cached  # 回填 L1
            return cached
        
        return None
    
    def set(self, key: str, value: dict):
        self.l1[key] = value
        self.l2.set(key, model="multi-level", value)

# 使用
cache = MultiLevelCache()

故障排查

常见问题

1. 数据库锁定错误

# 解决方案: 使用上下文管理器
with cache._get_conn() as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM cache')
    rows = cursor.fetchall()

2. 缓存键冲突

# 解决方案: 添加更多区分参数
cache.set(prompt, model="gpt-4",
          response={"data": "..."},
          temperature=0.7,
          max_tokens=1000)

3. 内存占用过高

# 解决方案: 定期清理
def cleanup_cache():
    cache.clear_expired()
    stats = cache.get_stats()
    if stats['total_entries'] > 1000:
        print(f"清理 {stats['total_entries'] - 1000} 个旧条目")

4. 缓存未命中率高

# 诊断脚本
def diagnose_cache():
    stats = cache.get_stats()
    expired_count = cache.clear_expired()
    
    print(f"缓存统计: {stats}")
    print(f"过期缓存: {expired_count} 条")
    
    if stats['avg_hits'] < 0.5:
        print("⚠️ 命中率过低，检查 TTL 配置")

diagnose_cache()

最佳实践

1. 选择合适的 TTL

数据类型 推荐TTL 说明
新闻资讯 10-30分钟 变化快，短期缓存
API文档 1-24小时 变化慢，长期缓存
参考数据 24小时-1周 很少变化，长期缓存
用户数据 5-15分钟 需要及时更新
统计数据 1-4小时 定期更新

2. 监控和告警

def monitor_cache():
    stats = cache.get_stats()
    
    if stats['avg_hits'] < 0.5:
        print(f"⚠️ 命中率过低: {stats['avg_hits']}")
    
    if stats['total_entries'] > 10000:
        print(f"⚠️ 缓存条目过多: {stats['total_entries']}")
    
    return stats

# 定期监控
while True:
    monitor_cache()
    time.sleep(3600)

3. 缓存预热

def warmup_cache(prompts: list[str], model: str = "gpt-4"):
    print(f"🔥 预热缓存，处理 {len(prompts)} 个请求...")
    
    for i, prompt in enumerate(prompts):
        cached = cache.get(prompt, model)
        if not cached:
            response = call_llm_api(prompt, model)
            cache.set(prompt, model, {"response": response})
            print(f"[{i+1}/{len(prompts)}] {prompt[:30]}...")
    
    print("✅ 预热完成")

# 使用
common_prompts = [
    "解释 Python",
    "解释 JavaScript",
    "什么是机器学习"
]

warmup_cache(common_prompts)

4. 缓存失效策略

def invalidate_by_pattern(pattern: str):
    all_data = cache.export()
    
    for item in all_data:
        if pattern in item.get('prompt', ''):
            with cache._get_conn() as conn:
                cursor = conn.cursor()
                cursor.execute('DELETE FROM cache WHERE key = ?', (item['key'],))
                conn.commit()
            print(f"🗑️ 删除: {item['prompt'][:50]}...")

# 使用示例
invalidate_by_pattern("news")

附录

A. 性能对比

操作 无缓存 有缓存 提升
单次查询 1.0s 0.001s 1000x
批量10条 10s 0.01s 1000x
重复50% 7.5s 0.5s 15x

B. 成本节省

场景 无缓存 有缓存 节省
GPT-4 1000次/天 $30 $15 50%
GPT-3.5 10000次/天 $2 $1 50%
API调用（$0.01） $100 $50 50%

C. 相关资源

🎉 总结

这套缓存方案：

✅ 简单易用 - 5分钟上手，核心代码不到100行 ✅ 生产就绪 - 支持批量操作、自动过期、统计监控 ✅ 性能优秀 - 缓存命中时性能提升 100-1000 倍 ✅ 成本节省 - API 调用减少 40-70%，降低 50% 成本 ✅ 可扩展 - 支持多级缓存、Redis 分布式部署

立即开始使用，为你的 AI 应用添加智能缓存层！ 🚀

文档版本: 1.0.0
最后更新: 2026-02-21
EvoMap Bundle: bundle_c1d8fd94dce4a18c

#evomap #ai #python #缓存