Caching Strategies

Redis Cache (production standard)

import redis
import json
import hashlib

class RedisEmbeddingCache:
    """Production caching with Redis"""
    
    def __init__(self, host='localhost', port=6379, ttl=86400):
        self.redis = redis.Redis(host=host, port=port, decode_responses=True)
        self.ttl = ttl  # 24 hours
    
    def _key(self, text, model):
        """Generate the cache key"""
        content = f"{model}:{text}"
        return f"emb:{hashlib.md5(content.encode()).hexdigest()}"
    
    def get(self, text, model):
        """Read from cache"""
        key = self._key(text, model)
        cached = self.redis.get(key)
        
        if cached:
            return json.loads(cached)
        return None
    
    def set(self, text, model, embedding):
        """Save to cache with TTL"""
        key = self._key(text, model)
        self.redis.setex(key, self.ttl, json.dumps(embedding))
    
    def stats(self):
        """Cache statistics"""
        keys = self.redis.keys("emb:*")
        return {
            'total_keys': len(keys),
            'memory_used': self.redis.info('memory')['used_memory_human']
        }

# Usage
cache = RedisEmbeddingCache()

# Get/Set pattern
embedding = cache.get(text, model="text-embedding-3-small")
if not embedding:
    embedding = call_openai_api(text)
    cache.set(text, model, embedding)

---

Disk Cache (simple, no Redis)

import json
import hashlib
from pathlib import Path

class DiskCache:
    """On-disk cache (for development)"""
    
    def __init__(self, cache_dir="./cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
    
    def _key(self, text):
        return hashlib.md5(text.encode()).hexdigest()
    
    def get(self, text):
        cache_file = self.cache_dir / f"{self._key(text)}.json"
        if cache_file.exists():
            with open(cache_file, 'r') as f:
                return json.load(f)
        return None
    
    def set(self, text, embedding):
        cache_file = self.cache_dir / f"{self._key(text)}.json"
        with open(cache_file, 'w') as f:
            json.dump(embedding, f)

---

In-Memory Cache (LRU)

from functools import lru_cache

@lru_cache(maxsize=1000)
def get_embedding_cached(text):
    """LRU cache (1000 most recent embeddings)"""
    return call_openai_api(text)

---

Cache invalidation

# TTL (Time To Live)
# - An embedding never changes → long TTL (7 days)
# - Dynamic data → short TTL (1 hour)

cache.setex(key, ttl=604800, value=embedding)  # 7 days

---

Summary

Strategy	Pros	Cons	Use case
Redis	Shared, persistent, TTL	Requires a server	Production
Disk	Simple, no dependencies	Slow	Development
LRU	Fast, built-in	Not persistent	Single process

Production: Redis mandatory.

---

Module 7 - Capsule 02