Capítulo 2: Caché y Rendimiento

Tipos de Caché

Tipo de caché	Qué se almacena	Beneficio
Embedding	Embeddings de consulta	Reduce llamadas a la API
Semántica	Resultados de consultas similares	Omite la recuperación
Respuesta	Respuestas completas de RAG	Más rápido

Caché de Embeddings

from functools import lru_cache
import hashlib

class CachedEmbeddings:
    def __init__(self, embeddings, cache_size=10000):
        self.embeddings = embeddings
        self._cache = {}
    
    def embed_query(self, text):
        key = hashlib.md5(text.encode()).hexdigest()
        if key not in self._cache:
            self._cache[key] = self.embeddings.embed_query(text)
        return self._cache[key]

Caché Semántica

Caché basada en la similitud de consultas:

from langchain.cache import GPTCache
import gptcache

def get_similar_query(query):
    """Find cached result for similar query"""
    query_embedding = embed(query)
    
    # Search cache vector store
    results = cache_store.similarity_search(
        query_embedding, 
        threshold=0.95  # Very similar queries
    )
    
    if results:
        return results[0].metadata["response"]
    return None

Caché de Respuestas con Redis

import redis
import json

r = redis.Redis()
CACHE_TTL = 3600  # 1 hour

def cached_rag(query, rag_chain):
    # Check cache
    cached = r.get(f"rag:{query}")
    if cached:
        return json.loads(cached)
    
    # Generate and cache
    response = rag_chain.invoke(query)
    r.setex(f"rag:{query}", CACHE_TTL, json.dumps(response))
    
    return response

Invalidación de Caché

def invalidate_on_update(document_id):
    """Invalidate caches when documents change"""
    # Clear related response caches
    r.delete_pattern(f"rag:*doc:{document_id}*")
    
    # Remove from semantic cache
    semantic_cache.delete(document_id)

Comparación de Rendimiento

Escenario	Latencia	Costo
Sin caché	2-5s	$0.01/consulta
Caché de embeddings	1-3s	$0.005/consulta
Éxito en caché semántica	<100ms	~$0
Éxito en caché de respuesta	<50ms	~$0