使用 Redis 实现精确缓存

Why Cache LLM Responses?

Exact Cache: Cache Key Design

import hashlib
import json

def make_cache_key(messages: list[dict], model: str, temperature: float) -> str:
    # Create a canonical, order-stable representation
    key_data = {
        'model': model,
        'temperature': temperature,
        'messages': messages,  # list order matters
    }
    # Serialize to JSON with sorted keys for determinism
    serialized = json.dumps(key_data, sort_keys=True, ensure_ascii=False)
    # Hash to a fixed-length key safe for Redis
    return 'llm_cache:' + hashlib.sha256(serialized.encode()).hexdigest()

Connecting to Redis

import redis
import redis.asyncio as aioredis

# Synchronous Redis client
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,  # return str instead of bytes
)

# Async Redis client (for FastAPI)
async_r = aioredis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True,
)

# Test connection
print(r.ping())  # True if Redis is running

Cache-Aside Pattern Implementation

import json
from openai import OpenAI

client = OpenAI()

def cached_completion(
    messages: list[dict],
    model: str = 'gpt-4o-mini',
    temperature: float = 0.7,
    ttl_seconds: int = 3600,
) -> str:
    cache_key = make_cache_key(messages, model, temperature)

    # Cache hit?
    cached = r.get(cache_key)
    if cached is not None:
        print('[CACHE HIT]')
        return json.loads(cached)

    # Cache miss: call API
    print('[CACHE MISS]')
    response = client.chat.completions.create(
        model=model,
        messages=messages,
        temperature=temperature,
    )
    result = response.choices[0].message.content

    # Store in cache with TTL
    r.setex(cache_key, ttl_seconds, json.dumps(result))
    return result

Async Cache-Aside for FastAPI

from openai import AsyncOpenAI
import redis.asyncio as aioredis
import json

async_client = AsyncOpenAI()
async_r = aioredis.Redis(host='localhost', port=6379, decode_responses=True)

async def async_cached_completion(
    messages: list[dict],
    model: str = 'gpt-4o-mini',
    temperature: float = 0.0,
    ttl: int = 86400,
) -> str:
    key = make_cache_key(messages, model, temperature)

    cached = await async_r.get(key)
    if cached:
        return json.loads(cached)

    response = await async_client.chat.completions.create(
        model=model, messages=messages, temperature=temperature
    )
    result = response.choices[0].message.content
    await async_r.setex(key, ttl, json.dumps(result))
    return result

Choosing the Right TTL

# TTL strategy by use case
TTL_STRATEGY = {
    'faq_answering':          86400 * 7,   # 7 days — stable facts
    'code_explanation':       86400,        # 1 day — code rarely changes
    'document_summarization': 3600 * 6,    # 6 hours
    'news_analysis':          300,          # 5 minutes — stale quickly
    'creative_writing':       0,            # 0 = don't cache (non-deterministic)
}

def get_ttl_for_use_case(use_case: str) -> int:
    return TTL_STRATEGY.get(use_case, 3600)  # default 1 hour

Caching Metrics and Monitoring

CACHE_HITS_KEY = 'llm_cache_metrics:hits'
CACHE_MISSES_KEY = 'llm_cache_metrics:misses'

async def async_cached_completion_instrumented(messages, model, temperature=0.0):
    key = make_cache_key(messages, model, temperature)
    cached = await async_r.get(key)

    if cached:
        await async_r.incr(CACHE_HITS_KEY)
        return json.loads(cached)

    await async_r.incr(CACHE_MISSES_KEY)
    response = await async_client.chat.completions.create(
        model=model, messages=messages, temperature=temperature
    )
    result = response.choices[0].message.content
    await async_r.setex(key, 3600, json.dumps(result))
    return result

async def get_cache_stats():
    hits = int(await async_r.get(CACHE_HITS_KEY) or 0)
    misses = int(await async_r.get(CACHE_MISSES_KEY) or 0)
    total = hits + misses
    return {'hit_rate': hits / total if total > 0 else 0, 'total': total}

Cache Invalidation Strategies

async def invalidate_cache_entry(messages, model, temperature):
    key = make_cache_key(messages, model, temperature)
    deleted = await async_r.delete(key)
    print(f'Deleted {deleted} cache entries')

async def invalidate_all_llm_cache():
    # Scan for all keys with prefix 'llm_cache:'
    keys_to_delete = []
    async for key in async_r.scan_iter(match='llm_cache:*'):
        keys_to_delete.append(key)
    if keys_to_delete:
        await async_r.delete(*keys_to_delete)
    print(f'Invalidated {len(keys_to_delete)} cache entries')

Serializing Complex Responses

from openai.types.chat import ChatCompletion

async def cached_completion_full_response(
    messages, model='gpt-4o-mini', temperature=0.0
):
    key = make_cache_key(messages, model, temperature) + ':full'
    cached = await async_r.get(key)

    if cached:
        return ChatCompletion.model_validate_json(cached)  # reconstruct object

    response = await async_client.chat.completions.create(
        model=model, messages=messages, temperature=temperature
    )
    # Serialize Pydantic model to JSON
    await async_r.setex(key, 3600, response.model_dump_json())
    return response

Caching and Non-Determinism

Redis Cluster and Production Setup

# Redis configuration for production LLM caching
# In redis.conf:
# maxmemory 2gb
# maxmemory-policy allkeys-lru

# Connection with retry and connection pool
import redis
from redis.retry import Retry
from redis.backoff import ExponentialBackoff

retry = Retry(ExponentialBackoff(base=0.1), 3)
production_redis = redis.Redis(
    host='your-redis-host.cache.amazonaws.com',
    port=6379,
    ssl=True,
    decode_responses=True,
    max_connections=50,
    retry=retry,
    retry_on_error=[redis.ConnectionError, redis.TimeoutError],
)

Quick Check

Lesson Recap