LLM调用时重复生成、缺乏缓存，如何用装饰器优化？

Question

在LLM服务调用中，相同prompt反复请求导致冗余计算、响应延迟升高、Token浪费及API成本激增——根本原因在于未对确定性输入（如固定system/user prompt + 参数）做结果缓存。传统手动缓存逻辑侵入业务代码，易出错且难以复用；而LLM输出虽具随机性（如`temperature>0`），但对`temperature=0`或确定性模式（如函数调用、JSON Schema约束）完全可缓存。如何在不修改模型调用函数内部、不耦合缓存实现的前提下，自动识别“可缓存调用”，支持LRU过期、序列化键归一化（忽略空格/换行）、多级缓存（内存+Redis）并兼容异步？这正是装饰器模式的理想场景：通过`@llm_cache(ttl=300)`透明拦截输入、生成稳定哈希键、代理执行与缓存读写，兼顾性能、可维护性与可观测性。

Answer 1

```html

一、现象层：LLM调用冗余的可观测表征

• 相同 prompt 在 5 分钟内重复触发 ≥12 次（监控平台采样数据）
• 平均端到端延迟从 850ms 升至 2.3s（含模型排队+推理+网络）
• OpenAI API token 消耗中，37% 来自完全重复的 system+user 输入组合（Trace ID 聚类分析）
• 成本审计显示：temperature=0 的确定性请求占总调用量 64%，却未启用任何缓存

二、归因层：为什么传统缓存方案失效？

方案类型	耦合度	键稳定性	异步支持	可观测性
手写 if cache.get(...) / cache.set(...)	高（侵入业务函数）	差（JSON.dumps 未 normalize whitespace）	需手动 await，易漏	无埋点、无命中率统计
中间件级代理（如 Envoy + Redis）	低（但无法识别 temperature=0 语义）	中（HTTP body 哈希，忽略参数语义）	不透明（gRPC/HTTP/SDK 多协议难统一）	仅网络层指标，无 LLM 语义标签

三、设计层：@llm_cache 装饰器的核心契约

该装饰器需满足以下正交能力：

1. 可缓存性自动判定：基于 temperature==0、response_format.type=="json_object"、tool_choice!="auto" 等规则动态启用缓存
2. 键归一化引擎：对 prompt 字符串执行 re.sub(r'\s+', ' ', s).strip() + json.dumps(sorted_dict, separators=(',', ':'))
3. 多级缓存协同：内存 LRU（maxsize=1000, ttl=300）→ Redis（fallback + 长期存储）
4. 全异步原生支持：同步函数与 async def 均自动适配（通过 inspect.iscoroutinefunction 分支）

四、实现层：关键代码骨架（Python 3.11+）

from functools import wraps, lru_cache
import hashlib
import json
import asyncio
from typing import Any, Callable, Dict, Optional, Union
import redis.asyncio as redis

def llm_cache(ttl: int = 300, maxsize: int = 1000):
    memory_cache = lru_cache(maxsize=maxsize)
    
    async def get_redis_client():
        return redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

    def normalize_key(args, kwargs) -> str:
        # 提取确定性字段：model, messages, temperature, response_format, tools...
        clean_kwargs = {k: v for k, v in kwargs.items() 
                       if k in ['model', 'messages', 'temperature', 'response_format', 'tools', 'tool_choice']}
        # 归一化 messages 中 content 的空白符
        if 'messages' in clean_kwargs:
            for m in clean_kwargs['messages']:
                if 'content' in m and isinstance(m['content'], str):
                    m['content'] = ' '.join(m['content'].split())
        key_str = json.dumps(clean_kwargs, sort_keys=True, separators=(',', ':'))
        return hashlib.sha256(key_str.encode()).hexdigest()

    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def sync_wrapper(*args, **kwargs):
            if not _is_cacheable(kwargs): 
                return func(*args, **kwargs)
            key = normalize_key(args, kwargs)
            # 先查内存
            cached = memory_cache(key)
            if cached is not None:
                return cached
            # 再查 Redis
            r = asyncio.run(get_redis_client())
            cached_val = asyncio.run(r.get(f'llm:{key}'))
            if cached_val:
                result = json.loads(cached_val)
                memory_cache.cache_clear()  # 简化示例，实际应带值缓存
                memory_cache(key)
                return result
            # 执行并双写
            result = func(*args, **kwargs)
            asyncio.run(r.setex(f'llm:{key}', ttl, json.dumps(result)))
            memory_cache(key)
            return result

        @wraps(func)
        async def async_wrapper(*args, **kwargs):
            if not _is_cacheable(kwargs):
                return await func(*args, **kwargs)
            key = normalize_key(args, kwargs)
            cached = memory_cache(key)
            if cached is not None:
                return cached
            r = await get_redis_client()
            cached_val = await r.get(f'llm:{key}')
            if cached_val:
                result = json.loads(cached_val)
                memory_cache(key)
                return result
            result = await func(*args, **kwargs)
            await r.setex(f'llm:{key}', ttl, json.dumps(result))
            memory_cache(key)
            return result

        return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper
    return decorator

五、验证层：端到端效果对比（A/B 测试）

graph LR A[原始调用] -->|QPS=42
Avg Latency=2140ms
Cache Hit=0%| B[启用 @llm_cache] B --> C{可缓存判定} C -->|temperature=0 ✓| D[归一化键 → SHA256] C -->|tool_choice=required ✓| D D --> E[LRU 内存查询] E -->|Miss| F[Redis 查询] F -->|Hit| G[返回 JSON 解析结果] F -->|Miss| H[调用 OpenAI SDK] H --> I[双写内存+Redis] I --> G

六、演进层：生产就绪增强项

• 可观测性注入：自动注入 OpenTelemetry Span，打标 llm.cache.hit_ratio、llm.cache.level（memory/redis/miss）
• 缓存穿透防护：对高频 miss key 启用布隆过滤器（RedisBloom）预检
• 语义去重扩展：接入 sentence-transformers 计算 prompt embedding 余弦相似度（>0.98 视为等价）
• 灰度发布能力：支持按 model 或 user_id 百分比开启缓存，避免冷热 key 不均

```