Caching Strategies

Redis Cache (production standard)

import redis
import json
import hashlib

class RedisEmbeddingCache:
    """Production caching con 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 horas
    
    def _key(self, text, model):
        """Generate cache key"""
        content = f"{model}:{text}"
        return f"emb:{hashlib.md5(content.encode()).hexdigest()}"
    
    def get(self, text, model):
        """Get 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 con 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']
        }

# Uso
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, sin Redis)

import json
import hashlib
from pathlib import Path

class DiskCache:
    """Cache en disco (para 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 embeddings más recientes)"""
    return call_openai_api(text)

---

Cache invalidation

# TTL (Time To Live)
# - Embedding nunca cambia → TTL largo (7 días)
# - Datos dinámicos → TTL corto (1 hora)

cache.setex(key, ttl=604800, value=embedding)  # 7 días

---

Resumen

Strategy	Pros	Cons	Use case
Redis	Shared, persistent, TTL	Requiere servidor	Production
Disk	Simple, no dependencies	Lento	Development
LRU	Rápido, built-in	No persistent	Single process

Production: Redis mandatory.

---

Módulo 7 - Cápsula 02