Cápsula 06: Optimización de Costos y Rendimiento para RAG

Descripción de la cápsula

Esta cápsula te ayuda a mantener el equilibrio entre experiencia de usuario y costo operativo en sistemas RAG. La meta es mejorar eficiencia sin comprometer la calidad de las respuestas. Aprenderás palancas concretas (cache de embeddings, cache de resultados, batch operations, index tuning con PQ, selección de modelos, optimización de top_k), código Python listo para producción, números reales de costos y ejercicios con soluciones detalladas.

Tiempo estimado: 35-45 minutos

¿De dónde viene el costo en RAG?

Un pipeline RAG típico tiene tres fuentes principales de costo:

La optimización más rentable suele ser reducir llamadas repetidas: embeddings duplicados y respuestas ya generadas. En esta cápsula te enfocas en costos de embeddings, retrieval e infra; la optimización de tokens del LLM es otro tema.

1. Cache de embeddings (no re-embedear documentos existentes)

El problema

Cada vez que re-ingestas un documento, vuelves a llamar al API de embeddings. Si actualizas 1000 documentos diarios y cada uno tiene ~500 tokens, son 500K tokens/día solo en re-embedding. A $0.02/1M tokens, eso son $0.30/día = **$9/mes** solo en re-embedding innecesario.

La solución

Almacena el hash del contenido junto al embedding. Si el documento no cambió, no re-embedees.

import hashlib
import json
from typing import Optional

def content_hash(content: str, metadata: Optional[dict] = None) -> str:
    """Genera un hash único para detectar si el documento cambió."""
    payload = content
    if metadata:
        payload += json.dumps(metadata, sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()

def should_reembed(
    doc_id: str,
    content: str,
    metadata: Optional[dict],
    cache: dict  # {doc_id: {"hash": str, "embedding": list}}
) -> bool:
    """Devuelve True solo si el documento cambió o no está en cache."""
    new_hash = content_hash(content, metadata)
    if doc_id not in cache:
        return True
    return cache[doc_id]["hash"] != new_hash

# Uso en pipeline de ingesta
def ingest_with_embedding_cache(
    doc_id: str,
    content: str,
    metadata: Optional[dict],
    embedding_cache: dict,
    embed_fn,
    vector_db
):
    if not should_reembed(doc_id, content, metadata, embedding_cache):
        # Recuperar embedding existente de la base (o de un store separado)
        embedding = vector_db.get_embedding_by_id(doc_id)
        if embedding is not None:
            vector_db.upsert(doc_id, embedding, metadata)
            return  # Sin llamada a API
    embedding = embed_fn(content)
    embedding_cache[doc_id] = {"hash": content_hash(content, metadata), "embedding": embedding}
    vector_db.upsert(doc_id, embedding, metadata)

Costo ahorrado

• Sin cache: 1000 docs/día × 500 tokens × $0.02/1M = $10/día ≈ $300/mes.
• Con cache (70% sin cambios): 300 docs/día × 500 tokens × $0.02/1M = $90/mes.
• Ahorro: ~$210/mes solo en embeddings de ingesta.

2. Cache de resultados con TTL (respuestas y retrieval)

El problema

Muchas queries son repetidas o muy parecidas. Sin cache, cada una paga embedding de la query + búsqueda vectorial. Ejemplo:

• Sin cache: 1000 queries/día × $0.00002/query (embedding) = $0.60/mes solo en embeddings de query.
• Si incluyes costo de retrieval y LLM, una query completa puede costar ~$0.001-0.01.

La solución

Cache de resultados con TTL. Clave: hash de la query normalizada. Si el contenido no cambia, reutilizas la respuesta.

import redis
import json
import hashlib
import pickle
from datetime import timedelta
from typing import Any, Optional

class RAGResultCache:
    """Cache de resultados RAG con TTL."""
    
    def __init__(self, redis_url: str = "redis://localhost:6379", ttl_seconds: int = 3600):
        self.client = redis.from_url(redis_url)
        self.ttl = ttl_seconds
    
    def _cache_key(self, query: str, top_k: int, filters: Optional[dict] = None) -> str:
        payload = f"{query.strip().lower()}|{top_k}|{json.dumps(filters or {}, sort_keys=True)}"
        return f"rag:result:{hashlib.sha256(payload.encode()).hexdigest()}"
    
    def get(self, query: str, top_k: int, filters: Optional[dict] = None) -> Optional[dict]:
        key = self._cache_key(query, top_k, filters)
        data = self.client.get(key)
        if data is None:
            return None
        return pickle.loads(data)
    
    def set(self, query: str, top_k: int, result: dict, filters: Optional[dict] = None) -> None:
        key = self._cache_key(query, top_k, filters)
        self.client.setex(key, self.ttl, pickle.dumps(result))

# Uso en endpoint
def rag_query(query: str, top_k: int = 5):
    cache = RAGResultCache(ttl_seconds=1800)  # 30 min
    cached = cache.get(query, top_k)
    if cached:
        return cached
    result = do_full_rag_pipeline(query, top_k)
    cache.set(query, top_k, result)
    return result

Números concretos

• Sin cache: 1000 queries/día × 30 días × $0.00002/query (embedding) = $0.60/mes.
• Con cache 30% hit rate: 70% × 30,000 = 21,000 queries pagadas = $0.42/mes.
• Con cache 60% hit rate: 40% × 30,000 = 12,000 queries pagadas = $0.24/mes.

Si cada query completa (embedding + retrieval + LLM) cuesta ~$0.005:

• Sin cache: 30,000 × $0.005 = $150/mes.
• Con 30% cache: 21,000 × $0.005 = $105/mes (ahorro $45/mes).
• Con 60% cache: 12,000 × $0.005 = $60/mes (ahorro $90/mes).

3. Batch operations en ingesta

El problema

Llamar al API de embeddings documento por documento es lento y a veces más caro. Muchos proveedores ofrecen descuentos por batch y limitan requests/segundo.

La solución

Agrupa documentos en lotes (ej. 100) y embedea en una sola llamada.

def batch_embed(texts: list[str], embed_fn, batch_size: int = 100) -> list[list[float]]:
    """Embedea en lotes para reducir llamadas y latencia."""
    results = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        embeddings = embed_fn(batch)  # API acepta array
        results.extend(embeddings)
    return results

# ChromaDB acepta add con documentos en batch
def batch_ingest(chroma_collection, documents: list[dict], embed_fn, batch_size: int = 100):
    for i in range(0, len(documents), batch_size):
        batch = documents[i : i + batch_size]
        ids = [d["id"] for d in batch]
        texts = [d["text"] for d in batch]
        metadatas = [d.get("metadata", {}) for d in batch]
        embeddings = batch_embed(texts, embed_fn, batch_size)
        chroma_collection.add(ids=ids, embeddings=embeddings, metadatas=metadatas)

Impacto

• Menos round-trips → menor latencia.
• Algunos proveedores cobran por request; batching reduce el número de requests.
• Típicamente 2-5x más rápido en ingesta masiva.

4. Index tuning con Product Quantization (PQ)

¿Qué es PQ?

Product Quantization comprime vectores en subespacios de menor dimensionalidad. Reduce uso de memoria y acelera la búsqueda, con un pequeño trade-off en recall.

Cuándo usarlo

• Índices grandes (>100K vectores).
• Restricciones de memoria.
• Latencia crítica.

# Qdrant con PQ
from qdrant_client.models import VectorParams, Distance, QuantizationConfig, ScalarQuantization

# Configurar colección con compresión
client.create_collection(
    collection_name="docs",
    vectors_config=VectorParams(
        size=1536,
        distance=Distance.COSINE,
        on_disk=True,
    ),
    quantization_config=ScalarQuantization(
        scalar=ScalarQuantization(
            type="int8",
            quantile=0.99,
            always_ram=True,
        )
    ),
)

Impacto típico

• Memoria: ~4x reducción (float32 → int8).
• Velocidad: 1.5-2x más rápido en muchos workloads.
• Recall: Pérdida de 1-3% en benchmarks típicos; compensable con top_k mayor en algunos casos.

5. Selección de modelos por entorno

Regla simple

Código para cambio por entorno

import os

def get_embedding_model(env: str = None):
    env = env or os.getenv("ENV", "development")
    if env == "production":
        return "text-embedding-3-small"  # OpenAI
    return "all-MiniLM-L6-v2"  # Local, gratis

def get_llm_model(env: str = None):
    env = env or os.getenv("ENV", "development")
    if env == "production":
        return "gpt-4o-mini"
    return "gpt-3.5-turbo"

Ahorro

• Local en dev: $0 vs ~$0.02/1M tokens.
• En un equipo de 5 desarrolladores haciendo 50K tokens/día en dev: ~$30/mes ahorrados.

6. Optimización de top_k

El problema

top_k alto = más documentos recuperados = más tokens al LLM = mayor costo y latencia. top_k bajo = riesgo de perder contexto relevante.

Estrategia

Ajusta por caso de uso:

Experimento controlado

Hipótesis: bajar top_k de 8 a 5 reduce latencia sin perder calidad.

1. Ejecuta A/B interno durante 3-5 días.
2. Compara latencia (p50, p95), costo por query y feedback de calidad.
3. Conserva el cambio solo si cumple umbrales (ej. recall >95%, p95 <500ms).

def optimize_top_k(current: int, candidate: int, rag_fn, eval_queries: list) -> dict:
    """Compara top_k actual vs candidato en recall aproximado y latencia."""
    results = {"current": [], "candidate": []}
    for q in eval_queries:
        r_curr = rag_fn(q, top_k=current)
        r_cand = rag_fn(q, top_k=candidate)
        results["current"].append({"latency_ms": r_curr["latency_ms"], "docs": r_curr["doc_ids"]})
        results["candidate"].append({"latency_ms": r_cand["latency_ms"], "docs": r_cand["doc_ids"]})
    # Calcular métricas y decidir
    return results

7. Capa de cache y tracking de costos en Python

Implementación completa de una capa que combina cache de embeddings de query, cache de resultados y tracking de costos:

import hashlib
import time
import redis
import pickle
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class CostTracker:
    """Tracking de costos por operación."""
    embedding_calls: int = 0
    embedding_tokens: int = 0
    cache_hits: int = 0
    cache_misses: int = 0
    
    def cost_embedding(self) -> float:
        # $0.02/1M tokens (text-embedding-3-small)
        return (self.embedding_tokens / 1_000_000) * 0.02
    
    def cache_hit_rate(self) -> float:
        total = self.cache_hits + self.cache_misses
        return (self.cache_hits / total * 100) if total > 0 else 0.0

class RAGCachingLayer:
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        result_ttl: int = 1800,
        embed_fn=None,
        vector_db=None,
    ):
        self.redis = redis.from_url(redis_url)
        self.result_ttl = result_ttl
        self.embed_fn = embed_fn
        self.vector_db = vector_db
        self.cost_tracker = CostTracker()
    
    def _query_embedding_key(self, query: str) -> str:
        return f"rag:embed:{hashlib.sha256(query.encode()).hexdigest()}"
    
    def _result_key(self, query: str, top_k: int) -> str:
        return f"rag:result:{hashlib.sha256(f'{query}|{top_k}'.encode()).hexdigest()}"
    
    def get_or_create_embedding(self, query: str) -> list[float]:
        key = self._query_embedding_key(query)
        cached = self.redis.get(key)
        if cached:
            self.cost_tracker.cache_hits += 1
            return pickle.loads(cached)
        self.cost_tracker.cache_misses += 1
        emb = self.embed_fn(query)
        self.cost_tracker.embedding_calls += 1
        self.cost_tracker.embedding_tokens += len(query.split()) * 2  # aprox
        self.redis.setex(key, self.result_ttl * 2, pickle.dumps(emb))  # embeddings TTL más largo
        return emb
    
    def get_or_create_result(self, query: str, top_k: int, retrieve_fn) -> dict:
        key = self._result_key(query, top_k)
        cached = self.redis.get(key)
        if cached:
            self.cost_tracker.cache_hits += 1
            return pickle.loads(cached)
        self.cost_tracker.cache_misses += 1
        embedding = self.get_or_create_embedding(query)
        start = time.perf_counter()
        docs = self.vector_db.query(embedding, top_k=top_k)
        result = {"docs": docs, "latency_ms": (time.perf_counter() - start) * 1000}
        self.redis.setex(key, self.result_ttl, pickle.dumps(result))
        return result
    
    def cost_summary(self) -> dict:
        return {
            "embedding_cost_usd": round(self.cost_tracker.cost_embedding(), 6),
            "cache_hit_rate_pct": round(self.cost_tracker.cache_hit_rate(), 2),
            "embedding_calls": self.cost_tracker.embedding_calls,
            "cache_hits": self.cost_tracker.cache_hits,
            "cache_misses": self.cost_tracker.cache_misses,
        }

Ejemplo de uso y números

# 1000 queries/día durante 30 días
# Sin cache: 30,000 embedding calls
# Con 30% cache: 21,000 embedding calls

queries_per_day = 1000
days = 30
cost_per_embedding = 0.00002  # aprox para query corta

without_cache = queries_per_day * days * cost_per_embedding
with_30_cache = queries_per_day * days * 0.7 * cost_per_embedding

print(f"Sin cache: ${without_cache:.2f}/mes")
print(f"Con 30% cache: ${with_30_cache:.2f}/mes (ahorro ${without_cache - with_30_cache:.2f})")
# Sin cache: $0.60/mes
# Con 30% cache: $0.42/mes (ahorro $0.18)

Mini framework de decisión

1. Identifica el mayor costo: embeddings, generación LLM o infra.
2. Aplica una optimización a la vez: embedding cache, result cache, top_k, etc.
3. Mide impacto en latencia, recall y costo.
4. Conserva solo los cambios con beneficio claro y sin degradar calidad.

Métricas para decidir si una optimización se queda

• Mejora de p95 de latencia (objetivo cuantitativo, ej. <500ms).
• Reducción de costo por consulta.
• Impacto en precisión de retrieval (recall, MRR).
• Impacto en complejidad operativa (mantenimiento, deps).

Troubleshooting

1. "Optimicé costo y bajó la calidad"

Síntoma: Tras reducir top_k o activar cache agresivo, las respuestas empeoran.

Solución: Restaura la configuración anterior. Busca otra palanca menos agresiva (ej. cache de embeddings pero no de resultados completos, o subir top_k un punto).

2. "No vemos impacto de la caché"

Síntoma: Hit rate muy bajo (<10%) aunque hay queries repetidas.

Solución: Revisa TTL (puede ser demasiado corto), la clave de cache (¿normalizas la query? ¿incluyes top_k/filtros?) y el patrón real de consultas. Considera semantic caching si las queries varían mucho en redacción.

3. "La latencia subió al activar Redis"

Síntoma: Añadiste cache y la p95 empeoró.

Solución: Verifica que Redis esté en la misma región/VPC que la app. Revisa serialización (pickle vs msgpack). Si el payload es grande, considera compresión.

4. "Demasiadas iniciativas de optimización en paralelo"

Síntoma: No sabes qué cambio causó qué efecto.

Solución: Prioriza por impacto × esfuerzo. Implementa una optimización a la vez, mide 3-5 días y documenta. Luego pasa a la siguiente.

5. "El cache crece sin control"

Síntoma: Redis usa más memoria de la esperada.

Solución: Revisa TTL (que no sea 0 o muy alto). Usa maxmemory y política allkeys-lru. Monitorea redis-cli INFO memory.

Ejercicios

Ejercicio 1: Calcular ahorro con cache de resultados

Tienes 2000 queries/día. Cada query cuesta $0.00002 en embedding. Sin cache pagas el 100%. Con 40% hit rate, ¿cuánto ahorras al mes?

queries_per_day = 2000
days = 30
cost_per_query = 0.00002
hit_rate = 0.40

total_queries = queries_per_day * days  # 60,000
without_cache = total_queries * cost_per_query  # $1.20
with_cache = total_queries * (1 - hit_rate) * cost_per_query  # 36,000 * 0.00002 = $0.72
savings = without_cache - with_cache  # $0.48/mes

print(f"Sin cache: ${without_cache:.2f}/mes")
print(f"Con 40% cache: ${with_cache:.2f}/mes")
print(f"Ahorro: ${savings:.2f}/mes ({savings/without_cache*100:.0f}%)")

Ejercicio 2: Implementar should_reembed con store persistente

Implementa una versión de should_reembed que use Redis para persistir el hash de cada documento, en lugar de un diccionario en memoria.

import redis
import hashlib
import json

def content_hash(content: str, metadata: dict = None) -> str:
    payload = content + (json.dumps(metadata or {}, sort_keys=True) if metadata else "")
    return hashlib.sha256(payload.encode()).hexdigest()

def should_reembed_redis(doc_id: str, content: str, metadata: dict, r: redis.Redis) -> bool:
    key = f"doc_hash:{doc_id}"
    new_hash = content_hash(content, metadata)
    stored = r.get(key)
    if stored is None:
        return True  # Nunca visto
    return stored.decode() != new_hash

def update_doc_hash(doc_id: str, content: str, metadata: dict, r: redis.Redis, ttl: int = 86400 * 90):
    key = f"doc_hash:{doc_id}"
    h = content_hash(content, metadata)
    r.setex(key, ttl, h)

Ejercicio 3: Batch embedding con manejo de errores

Mejora la función batch_embed para que, si un batch falla, reintente con sub-batches más pequeños en lugar de fallar todo.

def batch_embed_resilient(texts: list[str], embed_fn, batch_size: int = 100) -> list[list[float]]:
    results = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        try:
            embeddings = embed_fn(batch)
            results.extend(embeddings)
        except Exception as e:
            if len(batch) == 1:
                raise
            mid = len(batch) // 2
            left = batch_embed_resilient(batch[:mid], embed_fn, batch_size)
            right = batch_embed_resilient(batch[mid:], embed_fn, batch_size)
            results.extend(left)
            results.extend(right)
    return results

Ejercicio 4: Cache hit rate y cost tracking

Escribe una función que, dado un CostTracker con cache_hits y cache_misses, devuelva el hit rate en % y el costo estimado de embeddings ahorrado asumiendo $0.00002 por embedding evitado.

def cache_stats(tracker: CostTracker, cost_per_embedding: float = 0.00002) -> dict:
    total = tracker.cache_hits + tracker.cache_misses
    hit_rate = (tracker.cache_hits / total * 100) if total > 0 else 0.0
    saved_cost = tracker.cache_hits * cost_per_embedding
    return {
        "hit_rate_pct": round(hit_rate, 2),
        "saved_embeddings": tracker.cache_hits,
        "saved_cost_usd": round(saved_cost, 4),
    }

Ejercicio 5: Decidir top_k según recall

Tienes 20 queries de evaluación. Con top_k=8 recuperas el documento correcto en 18 de 20. Con top_k=5 lo recuperas en 16 de 20. ¿Qué top_k elegirías si el costo por documento adicional es relevante?

def choose_top_k(results: list[tuple[int, float]]) -> int:
    """results = [(top_k, recall), ...]"""
    # Objetivo: recall >= 0.85 con mínimo top_k
    for top_k, recall in sorted(results, key=lambda x: x[0]):
        if recall >= 0.85:
            return top_k
    return results[-1][0]  # fallback al mayor

Ejercicio 6: Proyectar costos con y sin cache a 6 meses

Partiendo de 500 queries/día con 10% de crecimiento mensual, proyecta el costo de embeddings (a $0.00002/query) a 6 meses: (a) sin cache, (b) con 30% cache estable.

base_queries = 500
growth = 0.10
cost_per = 0.00002
cache_rate = 0.30
months = 6

without, with_cache = 0, 0
q = base_queries
for m in range(months):
    days = 30
    monthly = q * days
    without += monthly * cost_per
    with_cache += monthly * (1 - cache_rate) * cost_per
    q *= (1 + growth)

print(f"Sin cache (6 meses): ${without:.2f}")
print(f"Con 30% cache (6 meses): ${with_cache:.2f}")
print(f"Ahorro: ${without - with_cache:.2f}")

Resumen

• Embedding cache: No re-embedees documentos sin cambios; usa hash de contenido y store (Redis o DB).
• Result cache con TTL: Cachea respuestas de RAG por query; 30-60% hit rate es común y reduce costo y latencia.
• Batch operations: Agrupa embeddings en lotes para menos llamadas y mejor throughput.
• Index tuning (PQ): Product Quantization reduce memoria y puede mejorar velocidad con pequeño trade-off en recall.
• Modelo por entorno: Dev con modelos locales/gratis; prod con modelos managed + cache agresivo.
• top_k: Ajusta según caso de uso (3-5 para FAQ, 5-8 para doc, 8-12 para investigación); mide recall antes de bajar.
• Optimizar bien es medir: Una optimización a la vez, métricas claras, conservar solo lo que aporta beneficio neto.
• Tracking de costos: Implementa CostTracker y cache_hit_rate para tomar decisiones con datos.

Recursos adicionales

1. AWS Caching Best Practices — Patrones de cache en producción.
2. OpenAI Embeddings Pricing — Precios actuales de embeddings.
3. Qdrant Quantization — PQ y otras técnicas en Qdrant.
4. ChromaDB Batch Operations — Ingesta eficiente.
5. Redis Caching Patterns — TTL, eviction, buenas prácticas.
6. Semantic Caching for RAG — Cache por similitud semántica.
7. LangChain Caching — Cache en pipelines LLM.
8. Cost Optimization & Caching Guide — Guía interna de optimización de costos AI.

Tiempo estimado: 35-45 minutos
Siguiente: 07-migracion-sin-downtime.md