Caché exacta con Redis
Almacene en caché las respuestas del LLM calculando un hash del prompt completo y guardando el resultado en Redis con un TTL, para servir al instante solicitudes idénticas sin realizar ninguna llamada a la API.
Caché exacta con Redis es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 1 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Engineering Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Engineering Academy incluye 4 lecciones en total.
¿Por qué almacenar en caché las respuestas de los LLM?
Las llamadas a las API de los LLM son caras: una sola solicitud a GPT-4o puede costar entre 0,005 y 0,15 $, según el número de tokens. En muchas aplicaciones, una parte significativa de las consultas entrantes son idénticas o casi idénticas a consultas anteriores; piense en bots de preguntas frecuentes, sistemas de atención al cliente o herramientas de revisión de código en las que los usuarios hacen repetidamente las mismas preguntas. El almacenamiento en caché puede eliminar entre el 20 y el 50 por ciento de las llamadas a la API en estos casos de uso, lo que reduce directamente los costes y la latencia.
Caché exacta: diseño de la clave de caché
Una caché exacta almacena las respuestas del LLM asociadas a un hash determinista de la entrada. La clave de caché debe incluir cada entrada que afecte al resultado: la matriz de mensajes, el nombre del modelo, la temperatura y cualquier otro parámetro que cambie la respuesta. Omitir cualquiera de estos elementos de la clave provoca colisiones de caché, de modo que se devuelve una respuesta almacenada para una solicitud efectiva diferente.
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()Conexión con Redis
Redis es la opción estándar para almacenar en caché las respuestas de los LLM gracias a su latencia de lectura inferior al milisegundo y a la compatibilidad integrada con TTL. Use la biblioteca redis-py para el acceso síncrono o aioredis (ahora integrado en redis-py como redis.asyncio) para el acceso asíncrono en aplicaciones FastAPI. Almacene la conexión de Redis como un singleton para evitar agotar el grupo de conexiones.
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 runningImplementación del patrón cache-aside
El patrón cache-aside es la estrategia de almacenamiento en caché estándar para las API de LLM. En cada solicitud: (1) calcule la clave de caché, (2) compruebe si hay una respuesta almacenada en Redis, (3) si la encuentra (acierto de caché), devuélvala inmediatamente, (4) si no la encuentra (fallo de caché), llame a la API del LLM, (5) almacene la respuesta en Redis con un TTL y (6) devuelva la respuesta. Este patrón mantiene la lógica de caché separada de la propia llamada al LLM.
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 resultCache-aside asíncrono para FastAPI
En una aplicación FastAPI asíncrona, use el cliente asíncrono de Redis para que las consultas de caché no bloqueen el bucle de eventos. El patrón es idéntico a la versión síncrona, pero utiliza await para todas las operaciones de Redis. Así, la capa de caché no bloquea en ningún momento y es compatible con el cliente asíncrono del LLM.
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 resultElegir el TTL adecuado
El TTL (tiempo de vida) controla durante cuánto tiempo siguen siendo válidas las respuestas almacenadas en caché. Para preguntas y respuestas factuales con bases de conocimiento estables, los TTL largos (de 24 a 72 horas) maximizan la tasa de aciertos de caché. Para respuestas que deben reflejar los datos más recientes (resúmenes de noticias, precios en tiempo real), son adecuados los TTL cortos (de 5 a 15 minutos) o no usar caché. En tareas creativas con una temperatura distinta de cero, las respuestas almacenadas pueden quedar obsoletas; considere almacenar en caché solo cuando temperature=0.
# 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 hourMétricas y supervisión de la caché
Realice un seguimiento de la tasa de aciertos de caché como métrica principal de reducción de costes. Una tasa de aciertos del 30 por ciento significa que se evita el 30 por ciento de las llamadas a la API. Almacene los recuentos de aciertos y fallos en la propia Redis mediante comandos INCR en contadores independientes. Exponga un endpoint /metrics en su aplicación FastAPI que informe de la tasa de aciertos actual, el total de solicitudes y el ahorro de costes estimado para cuantificar el ROI del almacenamiento en caché.
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}Estrategias de invalidación de la caché
Invalidar una caché exacta es sencillo porque las claves son deterministas. Para invalidar una entrada específica, vuelva a calcular su clave y llame a r.delete(key). Para invalidar todas las entradas de un patrón de prompt específico, use prefijos de clave de Redis con un escaneo mediante comodines. Para invalidar toda la caché tras una actualización importante de la base de conocimiento, llame a r.flushdb() (úselo con precaución: elimina todas las claves de la base de datos).
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')Serialización de respuestas complejas
Si su aplicación almacena en caché objetos de respuesta completos de la API (no solo el contenido de texto), serialícelos con cuidado. El objeto ChatCompletion completo incluye el uso de tokens, la versión del modelo y el motivo de finalización, datos útiles para el registro y el seguimiento de costes. Use el método .model_dump_json() del SDK para serializar objetos de respuesta de Pydantic como cadenas JSON y reconstruya los objetos con ChatCompletion.model_validate_json() al recuperar los datos de la caché.
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 responseAlmacenamiento en caché y no determinismo
El almacenamiento en caché exacto solo tiene sentido para solicitudes deterministas o casi deterministas. Con temperature=0 y top_p=1.0, la mayoría de los LLM producen la misma salida para la misma entrada (aunque no está garantizado debido al no determinismo de coma flotante). Con temperaturas más altas, las respuestas almacenadas pueden quedar obsoletas, ya que el modelo habría producido salidas diferentes. Almacene siempre en caché con temperature=0 o documente claramente en la clave de caché que las respuestas pueden variar.
Redis Cluster y configuración para producción
Para implementaciones en producción con grandes volúmenes de caché, use Redis Cluster para distribuir horizontalmente la carga entre varios nodos, o un servicio Redis administrado como AWS ElastiCache o Redis Cloud. Establezca una política de maxmemory (normalmente allkeys-lru, para expulsar las entradas menos utilizadas recientemente cuando la memoria está llena) a fin de evitar que Redis se quede sin memoria y gestionar automáticamente el tamaño de la caché.
# 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],
)Comprobación rápida
Compruebe su comprensión del almacenamiento en caché exacto de respuestas de LLM con Redis en esta lección.
Resumen de la lección
En esta lección ha aprendido que el almacenamiento en caché exacto aplica un hash a todas las entradas del LLM para producir una clave de caché determinista; el patrón cache-aside comprueba Redis antes de llamar a la API y almacena los resultados después de un fallo; y la selección del TTL debe reflejar la frecuencia con la que cambia el contenido: más largo para conocimientos estables y más corto para datos dinámicos. Supervise la tasa de aciertos de caché como métrica principal de reducción de costes. A continuación, crearemos una caché semántica para consultas similares pero no idénticas.
Preguntas frecuentes
¿La lección «Caché exacta con Redis» es gratis?
Sí — el texto completo de «Caché exacta con Redis» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Engineering Academy, actualiza a CoddyKit PRO. El curso de AI Engineering Academy incluye 4 lecciones en total.
¿Qué aprenderé en «Caché exacta con Redis»?
Almacene en caché las respuestas del LLM calculando un hash del prompt completo y guardando el resultado en Redis con un TTL, para servir al instante solicitudes idénticas sin realizar ninguna llamad… Practicas AI Engineering Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Engineering Academy?
No se requiere experiencia previa. AI Engineering Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 1 de 4.
¿Cuánto tiempo toma la lección «Caché exacta con Redis»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Engineering Academy?
Sí. Cada lección de AI Engineering Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- Caché exacta con Redis
- Caché semántica con embeddings
- Caché de prefijos de prompts de OpenAI
- Procesamiento por lotes, enrutamiento de modelos y paneles de costes