4. Caching Strategies

Descripción

Reducir llamadas redundantes a la API con estrategias de caching: exact match (hash de prompt), semantic caching (embedding similarity), TTL strategies según volatilidad, y cache invalidation. Implementación con Redis, in-memory, y GPTCache.

Por Qué Cachear Respuestas LLM

Las respuestas LLM tienen características que las hacen buenas candidatas para caching:

Características de requests LLM:
1. Alta repetición: "¿Cuál es el precio?" se hace 1000 veces/día
2. Determinismo (temperature=0): Mismo prompt → misma respuesta
3. Costo alto por request: Vale la pena la complejidad del cache
4. Latencia variable: Caché hace p50 consistente

Cuándo NO cachear:
- Queries que requieren información en tiempo real (precios de bolsa, clima)
- Conversaciones donde cada mensaje depende del historial específico
- Outputs creativos donde la variedad es el objetivo

Los 3 Tipos de Caching

Tipo 1: EXACT MATCH CACHE
  "¿Cuál es el horario de atención?" (exactamente igual)
  → Hash del prompt → respuesta en Redis
  → Hit rate: 30-40% típico

Tipo 2: SEMANTIC CACHE
  "¿A qué hora abren?" ~ "¿Cuál es el horario de atención?"
  → Embeddings + cosine similarity → respuesta de query similar
  → Hit rate adicional: 10-20% más que exact match

Tipo 3: PROMPT CACHING (API-level)
  System prompt reutilizado
  → Descuento en tokens (50-90%) en la API
  → No evita el request, pero reduce el costo

Tipo 1: Exact Match Cache con Redis

import redis
import hashlib
import json
import time
from openai import OpenAI
from typing import Optional

client = OpenAI()

class ExactMatchCache:
    """
    Cache de exact match para respuestas LLM.
    Usa Redis como backend con TTL configurable.
    """
    
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        ttl_default: int = 3600,  # 1 hora default
        prefix: str = "llm_cache:"
    ):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.ttl_default = ttl_default
        self.prefix = prefix
        
        # Stats de cache
        self._hits = 0
        self._misses = 0
    
    def _cache_key(
        self,
        prompt: str,
        model: str = "gpt-4o-mini",
        temperature: float = 0.0
    ) -> str:
        """Genera una cache key determinista."""
        content = f"{model}|{temperature}|{prompt}"
        hash_hex = hashlib.sha256(content.encode()).hexdigest()
        return f"{self.prefix}{hash_hex}"
    
    def get(
        self,
        prompt: str,
        model: str = "gpt-4o-mini",
        temperature: float = 0.0
    ) -> Optional[str]:
        """Busca en caché. Retorna respuesta o None."""
        key = self._cache_key(prompt, model, temperature)
        cached = self.redis.get(key)
        
        if cached:
            self._hits += 1
            return json.loads(cached)
        
        self._misses += 1
        return None
    
    def set(
        self,
        prompt: str,
        response: str,
        model: str = "gpt-4o-mini",
        temperature: float = 0.0,
        ttl: Optional[int] = None
    ) -> None:
        """Guarda respuesta en caché."""
        key = self._cache_key(prompt, model, temperature)
        ttl_value = ttl if ttl is not None else self.ttl_default
        
        self.redis.setex(
            key,
            ttl_value,
            json.dumps(response)
        )
    
    def delete(
        self,
        prompt: str,
        model: str = "gpt-4o-mini",
        temperature: float = 0.0
    ) -> bool:
        """Invalida una entrada del caché."""
        key = self._cache_key(prompt, model, temperature)
        return bool(self.redis.delete(key))
    
    def invalidar_por_patron(self, patron: str) -> int:
        """
        Invalida todas las entradas que coincidan con un patrón.
        Útil para invalidar todo el caché de un prompt cuando cambia.
        """
        keys = list(self.redis.scan_iter(f"{self.prefix}*"))
        eliminados = 0
        for key in keys:
            valor = self.redis.get(key)
            if valor and patron in str(valor):
                self.redis.delete(key)
                eliminados += 1
        return eliminados
    
    def stats(self) -> dict:
        """Estadísticas de uso del caché."""
        total = self._hits + self._misses
        hit_rate = self._hits / total if total > 0 else 0.0
        
        return {
            "hits": self._hits,
            "misses": self._misses,
            "total_requests": total,
            "hit_rate": f"{hit_rate:.1%}",
            "keys_en_redis": self.redis.dbsize()
        }
    
    def call_with_cache(
        self,
        prompt: str,
        model: str = "gpt-4o-mini",
        ttl: Optional[int] = None,
        **kwargs
    ) -> dict:
        """
        Ejecuta el LLM con caché integrado.
        Si hay cache hit: retorna inmediatamente.
        Si no: llama a la API y cachea el resultado.
        """
        # Intentar caché primero
        cached = self.get(prompt, model)
        if cached:
            return {
                "output": cached,
                "source": "cache",
                "latencia_ms": 1  # < 1ms desde caché
            }
        
        # Cache miss: llamar a la API
        inicio = time.time()
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0,  # Importante: temperature=0 para reproducibilidad
            **kwargs
        )
        
        output = response.choices[0].message.content
        latencia_ms = (time.time() - inicio) * 1000
        
        # Guardar en caché
        self.set(prompt, output, model, ttl=ttl)
        
        return {
            "output": output,
            "source": "api",
            "latencia_ms": latencia_ms,
            "tokens": response.usage.total_tokens
        }


# Demo de uso:
cache = ExactMatchCache(ttl_default=3600)

# Primera llamada (cache miss)
r1 = cache.call_with_cache("¿Cuál es la capital de México?")
print(f"1ra llamada: {r1['source']} - {r1['latencia_ms']:.0f}ms")
# 1ra llamada: api - 743ms

# Segunda llamada (cache hit)
r2 = cache.call_with_cache("¿Cuál es la capital de México?")
print(f"2da llamada: {r2['source']} - {r2['latencia_ms']:.0f}ms")
# 2da llamada: cache - 1ms

print(cache.stats())
# {'hits': 1, 'misses': 1, 'hit_rate': '50.0%', ...}

Tipo 2: Semantic Cache

El semantic cache captura queries similares pero no idénticas:

import numpy as np
from openai import OpenAI
import redis
import json

client = OpenAI()

class SemanticCache:
    """
    Cache semántico que también encuentra respuestas para queries similares.
    
    Usa embeddings para comparar la similitud entre queries.
    """
    
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        threshold: float = 0.92,  # Similitud mínima para considerar match
        embedding_model: str = "text-embedding-3-small",
        ttl: int = 86400,          # 24 horas
        prefix: str = "sem_cache:"
    ):
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.threshold = threshold
        self.embedding_model = embedding_model
        self.ttl = ttl
        self.prefix = prefix
    
    def _get_embedding(self, texto: str) -> list[float]:
        """Obtiene el embedding de un texto."""
        response = client.embeddings.create(
            input=texto,
            model=self.embedding_model
        )
        return response.data[0].embedding
    
    def _cosine_similarity(self, v1: list[float], v2: list[float]) -> float:
        """Calcula la similitud coseno entre dos vectores."""
        a = np.array(v1)
        b = np.array(v2)
        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
    
    def get(self, query: str) -> Optional[dict]:
        """
        Busca una respuesta para la query usando similitud semántica.
        
        Returns: dict con {response, similarity, cached_query} o None
        """
        query_embedding = self._get_embedding(query)
        
        # Buscar todos los embeddings en Redis
        keys = list(self.redis.scan_iter(f"{self.prefix}emb:*"))
        
        mejor_match = None
        mejor_similarity = 0.0
        
        for key in keys:
            cached_emb_json = self.redis.get(key)
            if not cached_emb_json:
                continue
            
            cached_emb = json.loads(cached_emb_json)
            similarity = self._cosine_similarity(query_embedding, cached_emb["embedding"])
            
            if similarity > mejor_similarity:
                mejor_similarity = similarity
                mejor_match = {
                    "key_id": key.split(":")[-1],
                    "similarity": similarity,
                    "cached_query": cached_emb["query"]
                }
        
        if mejor_match and mejor_similarity >= self.threshold:
            # Obtener la respuesta asociada
            resp_key = f"{self.prefix}resp:{mejor_match['key_id']}"
            respuesta = self.redis.get(resp_key)
            
            if respuesta:
                return {
                    "response": json.loads(respuesta),
                    "similarity": mejor_similarity,
                    "cached_query": mejor_match["cached_query"],
                    "cache_hit": True
                }
        
        return None
    
    def set(self, query: str, response: str) -> str:
        """
        Guarda la query y su respuesta con embedding.
        Retorna el ID generado.
        """
        import time
        
        cache_id = hashlib.sha256(f"{query}:{time.time()}".encode()).hexdigest()[:16]
        embedding = self._get_embedding(query)
        
        # Guardar embedding
        emb_key = f"{self.prefix}emb:{cache_id}"
        self.redis.setex(
            emb_key,
            self.ttl,
            json.dumps({"embedding": embedding, "query": query})
        )
        
        # Guardar respuesta
        resp_key = f"{self.prefix}resp:{cache_id}"
        self.redis.setex(resp_key, self.ttl, json.dumps(response))
        
        return cache_id
    
    def call_with_semantic_cache(
        self,
        prompt_template: str,
        query: str,
        model: str = "gpt-4o-mini"
    ) -> dict:
        """
        Ejecuta con semantic cache.
        Primero busca respuesta a query similar, si no hay llama a la API.
        """
        # Buscar en caché semántico
        cached = self.get(query)
        if cached:
            return {
                "output": cached["response"],
                "source": "semantic_cache",
                "similarity": f"{cached['similarity']:.3f}",
                "cached_query": cached["cached_query"]
            }
        
        # Cache miss: llamar a la API
        prompt_completo = prompt_template.format(input=query)
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt_completo}],
            temperature=0
        )
        
        output = response.choices[0].message.content
        
        # Guardar en caché semántico
        self.set(query, output)
        
        return {
            "output": output,
            "source": "api",
            "similarity": None,
            "tokens": response.usage.total_tokens
        }


# Ejemplo de uso:
sem_cache = SemanticCache(threshold=0.92)

# Primera query
r1 = sem_cache.call_with_semantic_cache(
    "Responde la pregunta del usuario: {input}",
    "¿Cuál es el horario de atención del servicio al cliente?"
)
print(f"1ra: {r1['source']}")

# Query similar (diferente formulación)
r2 = sem_cache.call_with_semantic_cache(
    "Responde la pregunta del usuario: {input}",
    "¿A qué hora está disponible el soporte?"
)
print(f"2da: {r2['source']} (similitud: {r2.get('similarity', 'N/A')})")
# 2da: semantic_cache (similitud: 0.941) — ahorramos una llamada a la API

Cache In-Memory (Sin Redis)

Para proyectos simples o testing, un cache en memoria:

from functools import lru_cache
import time
from typing import Optional


class InMemoryCache:
    """Cache en memoria con TTL. Más simple que Redis, no persiste entre reinicios."""
    
    def __init__(self, ttl_default: int = 3600, max_size: int = 1000):
        self._cache: dict = {}  # {key: {"value": ..., "expires_at": ...}}
        self.ttl_default = ttl_default
        self.max_size = max_size
        self._hits = 0
        self._misses = 0
    
    def _evict_expired(self) -> None:
        """Elimina entradas expiradas."""
        now = time.time()
        keys_expiradas = [k for k, v in self._cache.items() if v["expires_at"] < now]
        for k in keys_expiradas:
            del self._cache[k]
    
    def _evict_lru(self) -> None:
        """Si está lleno, elimina la entrada menos reciente."""
        if len(self._cache) >= self.max_size:
            # Eliminar el más antiguo (con menor expires_at)
            oldest = min(self._cache, key=lambda k: self._cache[k]["expires_at"])
            del self._cache[oldest]
    
    def get(self, key: str) -> Optional[str]:
        self._evict_expired()
        entry = self._cache.get(key)
        
        if entry and entry["expires_at"] > time.time():
            self._hits += 1
            return entry["value"]
        
        self._misses += 1
        return None
    
    def set(self, key: str, value: str, ttl: Optional[int] = None) -> None:
        self._evict_lru()
        self._cache[key] = {
            "value": value,
            "expires_at": time.time() + (ttl or self.ttl_default)
        }
    
    def stats(self) -> dict:
        total = self._hits + self._misses
        return {
            "hits": self._hits,
            "misses": self._misses,
            "hit_rate": f"{self._hits/total*100:.1f}%" if total > 0 else "0%",
            "entries": len(self._cache)
        }


# Wrapper funcional simple con decorador:
_local_cache = InMemoryCache(ttl_default=1800, max_size=500)

def cached_llm_call(
    prompt: str,
    model: str = "gpt-4o-mini",
    ttl: int = 1800
) -> str:
    """Wrapper cacheado para llamadas LLM."""
    cache_key = hashlib.sha256(f"{model}:{prompt}".encode()).hexdigest()
    
    # Intentar cache
    cached = _local_cache.get(cache_key)
    if cached:
        return cached
    
    # API call
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    output = response.choices[0].message.content
    
    # Cachear
    _local_cache.set(cache_key, output, ttl=ttl)
    
    return output

TTL Strategies: Cuánto Tiempo Cachear

La duración del caché depende de la volatilidad del dato:

class TTLStrategy:
    """Estrategias de TTL según el tipo de contenido."""
    
    # Configuraciones predefinidas
    CONFIGS = {
        "estatico": {
            "ttl": 7 * 24 * 3600,    # 7 días
            "descripcion": "Datos que casi nunca cambian (definiciones, FAQs estáticas)",
            "ejemplos": ["¿Qué es el machine learning?", "Definición de términos"]
        },
        "semi_estatico": {
            "ttl": 24 * 3600,         # 24 horas
            "descripcion": "Cambia ocasionalmente (políticas, catálogos de producto)",
            "ejemplos": ["Política de devoluciones", "Lista de servicios disponibles"]
        },
        "dinamico": {
            "ttl": 3600,              # 1 hora
            "descripcion": "Cambia regularmente (inventario, disponibilidad)",
            "ejemplos": ["Stock disponible", "Horarios de apertura esta semana"]
        },
        "tiempo_real": {
            "ttl": 300,               # 5 minutos
            "descripcion": "Cambia frecuentemente (precios, noticias)",
            "ejemplos": ["Precio actual de acciones", "Noticias recientes"]
        },
        "no_cachear": {
            "ttl": 0,
            "descripcion": "No cachear — siempre diferente",
            "ejemplos": ["Conversaciones personalizadas", "Outputs creativos"]
        }
    }
    
    @classmethod
    def recomendar(cls, tipo_query: str) -> dict:
        """Recomienda TTL basado en el tipo de query."""
        palabras_tiempo_real = ["precio", "precio actual", "ahora", "hoy", "último"]
        palabras_dinamico = ["disponible", "stock", "horario", "esta semana"]
        palabras_estatico = ["qué es", "define", "explicar", "concepto"]
        
        query_lower = tipo_query.lower()
        
        for palabra in palabras_tiempo_real:
            if palabra in query_lower:
                return cls.CONFIGS["tiempo_real"]
        
        for palabra in palabras_dinamico:
            if palabra in query_lower:
                return cls.CONFIGS["dinamico"]
        
        for palabra in palabras_estatico:
            if palabra in query_lower:
                return cls.CONFIGS["estatico"]
        
        return cls.CONFIGS["semi_estatico"]  # Default


# Uso en el cache:
def call_with_smart_ttl(prompt: str, query: str) -> dict:
    """Determina TTL automáticamente según la naturaleza de la query."""
    ttl_config = TTLStrategy.recomendar(query)
    
    if ttl_config["ttl"] == 0:
        # No cachear
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt.format(input=query)}],
            temperature=0
        )
        return {"output": response.choices[0].message.content, "cached": False}
    
    # Cachear con TTL apropiado
    resultado = cache.call_with_cache(
        prompt.format(input=query),
        ttl=ttl_config["ttl"]
    )
    return {**resultado, "ttl_usado": ttl_config["ttl"]}

Cache Invalidation

La invalidación correcta evita servir datos obsoletos:

class CacheInvalidator:
    """Gestiona la invalidación del caché de forma ordenada."""
    
    def __init__(self, redis_client: redis.Redis, prefix: str = "llm_cache:"):
        self.redis = redis_client
        self.prefix = prefix
    
    def invalidar_prompt_version(self, prompt_name: str, old_version: str) -> int:
        """
        Invalida todo el caché cuando cambia la versión de un prompt.
        
        Estrategia: Include la versión del prompt en la cache key,
        al cambiar versión, las keys viejas son automáticamente inválidas.
        """
        # Si las keys incluyen la versión del prompt, simplemente cambiando
        # la versión activa el caché viejo queda huérfano y expirará por TTL
        patron = f"{self.prefix}*{prompt_name}:{old_version}*"
        
        keys = list(self.redis.scan_iter(patron))
        if keys:
            self.redis.delete(*keys)
        
        print(f"Invalidadas {len(keys)} entradas de caché para {prompt_name}:{old_version}")
        return len(keys)
    
    def invalidar_por_patron_query(self, patron: str) -> int:
        """Invalida entradas relacionadas con un patrón de query."""
        # Escanear y revisar contenido (solo funciona con decode_responses=True)
        keys_a_eliminar = []
        
        for key in self.redis.scan_iter(f"{self.prefix}*"):
            valor = self.redis.get(key)
            if valor and patron.lower() in valor.lower():
                keys_a_eliminar.append(key)
        
        if keys_a_eliminar:
            self.redis.delete(*keys_a_eliminar)
        
        return len(keys_a_eliminar)
    
    def invalidar_todo(self) -> int:
        """Limpia todo el caché (útil en deploys mayores)."""
        keys = list(self.redis.scan_iter(f"{self.prefix}*"))
        
        if keys:
            self.redis.delete(*keys)
        
        print(f"Caché limpiado: {len(keys)} entradas eliminadas")
        return len(keys)
    
    def invalidar_on_deploy(self, prompt_name: str, nueva_version: str) -> None:
        """
        Hook para ejecutar al hacer deploy de un nuevo prompt.
        Invalida caché específico del prompt cambiado.
        """
        print(f"Deploy detectado: {prompt_name} → {nueva_version}")
        
        # Estrategia 1: Invalidar todo el caché del prompt
        # invalidadas = self.invalidar_prompt_version(prompt_name, "*")
        
        # Estrategia 2: Dejar expirar por TTL (menos agresivo)
        # Solo invalidar si el cambio es breaking (MAJOR version change)
        is_breaking = nueva_version.startswith("v") and nueva_version[1] != "1"
        
        if is_breaking:
            invalidadas = self.invalidar_por_patron_query(prompt_name)
            print(f"Breaking change detectado: {invalidadas} entradas invalidadas")
        else:
            print("Cambio no-breaking: dejando caché expirar por TTL")

Métricas del Cache

class CacheMetrics:
    """Monitorea la efectividad del caché."""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    def reporte_completo(self, cache: ExactMatchCache) -> str:
        """Genera reporte de efectividad del caché."""
        stats = cache.stats()
        hits = cache._hits
        misses = cache._misses
        total = hits + misses
        hit_rate = hits / total if total > 0 else 0
        
        # Calcular ahorro estimado
        # Asumiendo $0.001 por request sin caché (prompt + completion)
        costo_sin_cache = total * 0.001
        costo_con_cache = misses * 0.001
        ahorro = costo_sin_cache - costo_con_cache
        
        return f"""
Cache Performance Report
━━━━━━━━━━━━━━━━━━━━━━━
Total requests:  {total:,}
Cache hits:      {hits:,} ({hit_rate:.1%})
Cache misses:    {misses:,}
Keys en Redis:   {stats['keys_en_redis']:,}

Estimación de ahorro:
Sin caché:   ${costo_sin_cache:.4f}
Con caché:   ${costo_con_cache:.4f}
Ahorro:      ${ahorro:.4f} ({hit_rate:.1%})

Recomendación:
{'✅ Excelente hit rate' if hit_rate > 0.5 else 
 '⚠️  Hit rate bajo — considera ajustar TTL o añadir semantic caching' if hit_rate > 0.2 else
 '❌ Hit rate muy bajo — revisar si el caché es apropiado para este caso de uso'}
"""

Comparación: Opciones de Caching

Troubleshooting

Problema 1: Cache hit rate muy bajo

Síntoma: Hit rate < 15% después de varios días.

Diagnóstico:

def diagnosticar_hit_rate_bajo(cache: ExactMatchCache) -> list[str]:
    """Identifica por qué el hit rate es bajo."""
    sugerencias = []
    
    # 1. Verificar si los prompts tienen variables dinámicas que rompen el cache
    sugerencias.append(
        "Verificar: ¿Los prompts incluyen timestamps, IDs de usuario, "
        "u otras variables dinámicas? Estos rompen el exact match."
    )
    
    # 2. TTL muy corto
    if cache.ttl_default < 300:
        sugerencias.append(
            f"TTL muy corto ({cache.ttl_default}s). "
            "Considera 3600s (1h) para queries semi-estáticas."
        )
    
    # 3. Si queries son todas únicas, exact match no aplica
    sugerencias.append(
        "¿Las queries son mayormente únicas (conversaciones personalizadas)? "
        "Si es así, exact match no aplica. Considerar semantic caching o no cachear."
    )
    
    return sugerencias

Problema 2: Datos obsoletos (stale data)

Síntoma: El caché sirve respuestas desactualizadas.

Solución:

# Incluir la versión del prompt en la cache key
def cache_key_con_version(prompt: str, prompt_version: str, model: str) -> str:
    content = f"{model}:{prompt_version}:{prompt}"
    return hashlib.sha256(content.encode()).hexdigest()

# Al cambiar el prompt a v1.2, las keys v1.1 no hacen match
# No necesitas invalidar explícitamente, simplemente expiran por TTL

Problema 3: Memoria Redis creciendo sin límite

Síntoma: Redis usa cada vez más memoria.

Solución:

# En redis.conf:
maxmemory 256mb
maxmemory-policy allkeys-lru  # Eliminar menos usadas cuando está lleno

# Configurar en la inicialización:
r = redis.Redis(host="localhost", port=6379)
r.config_set("maxmemory", "256mb")
r.config_set("maxmemory-policy", "allkeys-lru")

Ejercicios

Ejercicio 1: Implementar exact match cache sin Redis

Implementa un cache en memoria con TTL que funcione sin Redis (solo Python stdlib):

import hashlib
import time
from typing import Optional

class SimpleCache:
    """Cache en memoria sin dependencias externas."""
    
    def __init__(self, ttl_segundos: int = 3600, max_items: int = 500):
        self._store: dict = {}
        self.ttl = ttl_segundos
        self.max_items = max_items
    
    def _key(self, prompt: str) -> str:
        return hashlib.md5(prompt.encode()).hexdigest()
    
    def get(self, prompt: str) -> Optional[str]:
        key = self._key(prompt)
        if key in self._store:
            value, expires = self._store[key]
            if time.time() < expires:
                return value
            del self._store[key]
        return None
    
    def set(self, prompt: str, respuesta: str) -> None:
        if len(self._store) >= self.max_items:
            # Eliminar el más antiguo
            oldest_key = min(self._store, key=lambda k: self._store[k][1])
            del self._store[oldest_key]
        
        self._key(prompt)
        self._store[self._key(prompt)] = (respuesta, time.time() + self.ttl)
    
    def hit_rate(self, n_consultas: int = 0) -> float:
        return 0.0  # Simplificado — en producción trackear hits/misses

# Test:
from openai import OpenAI
client = OpenAI()
cache = SimpleCache(ttl_segundos=300)

def ask(pregunta: str) -> tuple[str, str]:
    """Retorna (respuesta, fuente)."""
    cached = cache.get(pregunta)
    if cached:
        return cached, "cache"
    
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": pregunta}],
        temperature=0
    )
    respuesta = r.choices[0].message.content
    cache.set(pregunta, respuesta)
    return respuesta, "api"

r1, s1 = ask("¿Qué es Python?")
r2, s2 = ask("¿Qué es Python?")  # Debe ser cache
print(f"1ra: {s1}, 2da: {s2}")

Ejercicio 2: Medir el ahorro con caching

Simula 100 requests donde el 40% son repeticiones y calcula el ahorro estimado:

import random
import time

def simular_caching(n_requests: int = 100, cache_hit_rate: float = 0.40) -> dict:
    """Simula el impacto del caching en costo y latencia."""
    
    costo_por_request_api = 0.001    # $0.001 por request a API
    latencia_api_ms = 800            # Latencia típica de API
    latencia_cache_ms = 2            # Latencia típica de cache hit
    
    n_hits = int(n_requests * cache_hit_rate)
    n_misses = n_requests - n_hits
    
    # Costo
    costo_sin_cache = n_requests * costo_por_request_api
    costo_con_cache = n_misses * costo_por_request_api
    
    # Latencia total
    latencia_sin_cache_ms = n_requests * latencia_api_ms
    latencia_con_cache_ms = n_misses * latencia_api_ms + n_hits * latencia_cache_ms
    
    return {
        "n_requests": n_requests,
        "cache_hits": n_hits,
        "cache_misses": n_misses,
        "hit_rate": f"{cache_hit_rate:.0%}",
        "costo_sin_cache": f"${costo_sin_cache:.4f}",
        "costo_con_cache": f"${costo_con_cache:.4f}",
        "ahorro_costo": f"${costo_sin_cache - costo_con_cache:.4f} ({cache_hit_rate:.0%})",
        "latencia_total_sin_cache": f"{latencia_sin_cache_ms/1000:.1f}s",
        "latencia_total_con_cache": f"{latencia_con_cache_ms/1000:.1f}s",
        "reduccion_latencia": f"{(1-latencia_con_cache_ms/latencia_sin_cache_ms):.0%}"
    }

resultado = simular_caching(100, 0.40)
for k, v in resultado.items():
    print(f"{k}: {v}")

Resumen

• Exact match cache: Hash del prompt como key en Redis — simple y efectivo para queries repetidas
• Semantic cache: Embeddings + cosine similarity — captura queries similares, 10-20% más hits
• In-memory cache: Sin dependencias externas — solo para desarrollo o un proceso
• TTL strategy: Ajustar según volatilidad: 7d para estático, 1h para dinámico, 5min para tiempo-real
• Cache invalidation: Con versión del prompt en la key, o invalidación explícita en deploy
• Hit rate objetivo: >30% para justify la complejidad; >50% para semantic cache

Recursos adicionales

1. Redis Documentation — Documentación completa de Redis
2. GPTCache — Caching library específica para LLMs
3. redis-py — Cliente Python para Redis
4. OpenAI Embeddings — Para semantic caching
5. Caching Best Practices — Guía de AWS sobre caching