Montar un sistema RAG tiene mala fama de caro, y con razón. Cada vez que un usuario hace una pregunta, tu aplicación recupera varios fragmentos de documentos y los mete todos en el contexto del LLM. Multiplica eso por cientos o miles de consultas diarias y el gasto en tokens se va de las manos antes de que te des cuenta.
El problema no está en el concepto. RAG es una técnica sólida para dar respuestas basadas en tus propios documentos. El problema está en cómo se implementa por defecto: recuperar 10 fragmentos, mandarlos todos al LLM principal y esperar lo mejor. Hay dos patrones concretos que cambian esto sin complicar demasiado la arquitectura.
Por qué el LLM es el mayor gasto
En un pipeline RAG hay tres costes: los embeddings para indexar tus documentos, la búsqueda vectorial para recuperar fragmentos relevantes y el LLM que genera la respuesta final. Los dos primeros son baratos. El LLM domina el gasto, sobre todo si usas modelos como GPT-4o o Claude Sonnet.
El coste del LLM depende directamente del número de tokens que le mandas. Si recuperas 10 fragmentos de 500 tokens cada uno, ya llevas 5.000 tokens de contexto antes de añadir el historial de conversación y las instrucciones del sistema. Con un modelo que cobra 5 $ por millón de tokens de entrada, cada consulta te cuesta 2,5 céntimos solo en contexto. A 10.000 consultas diarias son 250 $ al día.
La solución no es cambiar de modelo o recortar documentos a ciegas. Es ser más inteligente con qué mandas y cuándo lo mandas.
Patrón 1: caché semántica
La idea es simple: si alguien preguntó algo muy parecido ayer y ya tienes la respuesta, no hace falta llamar al LLM otra vez. Devuelves la respuesta cacheada y listo.
La clave está en "muy parecido". No buscas coincidencia exacta de texto, sino similitud semántica. Para eso usas embeddings: conviertes la query del usuario en un vector numérico y lo comparas con las queries anteriores que tienes almacenadas. Si la similitud (cosine similarity) supera un umbral, hay un cache hit.
Implementación con Redis y RediSearch
El flujo es este: cuando llega una query, calculas su embedding y buscas el vecino más cercano en Redis. Si la similitud es mayor o igual a 0,92, devuelves la respuesta almacenada. Si no, llamas al LLM, generas la respuesta y la guardas junto al embedding y la query original.
import numpy as np
from redis import Redis
from redis.commands.search.field import VectorField, TextField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import Query
import openai
client = openai.OpenAI()
r = Redis(host='localhost', port=6379, decode_responses=False)
SIMILARITY_THRESHOLD = 0.92
EMBEDDING_DIM = 1536 # text-embedding-3-small
def get_embedding(text: str) -> list[float]:
response = client.embeddings.create(
model="text-embedding-3-small",
input=text
)
return response.data[0].embedding
def cosine_similarity(a: list[float], b: list[float]) -> float:
a, b = np.array(a), np.array(b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
def cache_lookup(query: str) -> str | None:
query_embedding = get_embedding(query)
query_vector = np.array(query_embedding, dtype=np.float32).tobytes()
q = (
Query("*=>[KNN 1 @embedding $vec AS score]")
.sort_by("score")
.return_fields("query", "response", "score")
.dialect(2)
)
results = r.ft("cache_idx").search(q, query_params={"vec": query_vector})
if results.total > 0:
doc = results.docs[0]
# RediSearch devuelve distancia L2; para cosine usamos vector normalizado
similarity = 1 - float(doc.score)
if similarity >= SIMILARITY_THRESHOLD:
return doc.response
return None
def cache_store(query: str, response: str):
embedding = get_embedding(query)
vector = np.array(embedding, dtype=np.float32).tobytes()
key = f"cache:{hash(query)}"
r.hset(key, mapping={
"query": query,
"response": response,
"embedding": vector
})
El umbral de 0,92 es un punto de partida razonable. Si ves que devuelves respuestas incorrectas para preguntas parecidas pero no iguales, súbelo a 0,95. Si el hit rate es demasiado bajo, bájalo a 0,88 y mide el impacto en calidad.
Alternativa: GPTCache
Si no quieres montar la infraestructura de Redis desde cero, GPTCache es una librería Python que hace exactamente esto con pocas líneas:
from gptcache import cache
from gptcache.adapter import openai
cache.init()
cache.set_openai_key()
# A partir de aquí, openai.ChatCompletion usa caché automáticamente
response = openai.ChatCompletion.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": "¿Qué es RAG?"}]
)
Con consultas repetitivas (preguntas sobre un mismo documento corporativo, por ejemplo), el ahorro típico está entre el 40% y el 60% del gasto en LLM.
Patrón 2: reescritura de consultas
El segundo problema es diferente. El usuario escribe cosas como "¿y el precio?" sin contexto, o "explícame lo de antes mejor". Tu sistema de búsqueda vectorial no tiene ni idea de a qué se refiere y recupera fragmentos irrelevantes. Para compensar, recuperas más fragmentos con la esperanza de que alguno sea útil. Más fragmentos, más tokens, más coste.
La solución es usar un LLM pequeño, barato y rápido para reescribir la query antes de buscar. GPT-4o-mini cuesta una décima parte de GPT-4o. Mistral 7B puedes ejecutarlo en local sin coste de API. El dinero que gastas en reescribir la query lo recuperas con creces al mandar menos contexto al LLM principal.
Multi-query: tres variaciones, una búsqueda
En lugar de buscar con la query original, generas tres variaciones que capturan diferentes aspectos de la misma pregunta:
from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
llm=llm
)
# Recupera documentos para las 3 variaciones y deduplica
docs = retriever.get_relevant_documents("¿cuál es la política de devoluciones?")
LangChain se encarga de generar las variaciones, recuperar para cada una y deduplicar los resultados. Acabas con menos fragmentos duplicados y más cobertura semántica.
HyDE: buscar con el documento hipotético
HyDE (Hypothetical Document Embeddings) es más sofisticado. En vez de buscar con la query del usuario, le pides al LLM pequeño que genere un fragmento hipotético de cómo sería la respuesta ideal. Luego usas ese fragmento como query de búsqueda.
El razonamiento: el embedding de una respuesta hipotética está más cerca en el espacio vectorial de los documentos reales que el embedding de una pregunta. Recuperas fragmentos más relevantes y el LLM principal necesita menos contexto para responder bien.
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.chains import HypotheticalDocumentEmbedder
base_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
hyde_embeddings = HypotheticalDocumentEmbedder.from_llm(
llm=llm,
embeddings=base_embeddings,
prompt_key="web_search" # plantilla incluida en LangChain
)
# Úsalo como si fuera un objeto de embeddings normal
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
docs = retriever.get_relevant_documents(
query,
embeddings=hyde_embeddings
)
Con HyDE sueles poder bajar el número de fragmentos recuperados de 10 a 3-5 sin perder calidad en la respuesta final.
Combinar los dos patrones
Los dos patrones se complementan bien. El flujo completo queda así:
- Llega la query del usuario.
- Buscas en la caché semántica. Si hay hit con similitud suficiente, devuelves la respuesta cacheada y terminas.
- Si no hay hit, reescribes la query con un LLM pequeño (multi-query o HyDE).
- Recuperas fragmentos con la query mejorada.
- Llamas al LLM principal con un contexto más reducido y relevante.
- Cacheas la respuesta antes de devolverla.
Un esquema básico con FastAPI y LangChain:
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class QueryRequest(BaseModel):
question: str
session_id: str = "default"
@app.post("/ask")
async def ask(req: QueryRequest):
# Paso 1: caché semántica
cached = cache_lookup(req.question)
if cached:
return {"answer": cached, "source": "cache"}
# Paso 2: reescribir y recuperar
docs = retriever.get_relevant_documents(req.question) # con HyDE o multi-query
# Paso 3: LLM principal
context = "nn".join(doc.page_content for doc in docs)
answer = llm_chain.run(question=req.question, context=context)
# Paso 4: cachear
cache_store(req.question, answer)
return {"answer": answer, "source": "llm"}
Otras técnicas que complementan bien
Reranking
Recupera 20 fragmentos con el buscador vectorial y luego usa un modelo de reranking (como Cohere Rerank o un cross-encoder local) para ordenarlos por relevancia real. Manda solo los 3 mejores al LLM principal. El reranker es mucho más barato que el LLM y mejora la precisión.
import cohere
co = cohere.Client("TU_API_KEY")
# Recuperas 20 candidatos del vectorstore
candidates = vectorstore.similarity_search(query, k=20)
# Reordenas con Cohere Rerank
reranked = co.rerank(
query=query,
documents=[doc.page_content for doc in candidates],
top_n=3,
model="rerank-multilingual-v3.0"
)
# Solo los 3 mejores van al LLM
top_docs = [candidates[r.index] for r in reranked.results]
Comprimir fragmentos antes de mandarlos
LangChain tiene ContextualCompressionRetriever: recupera fragmentos normales y luego pasa cada uno por un LLM pequeño para extraer solo la parte relevante para la query. Si un fragmento de 500 tokens solo tiene 80 tokens útiles, mandas 80.
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
compressor = LLMChainExtractor.from_llm(ChatOpenAI(model="gpt-4o-mini"))
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)
Elegir el modelo según la complejidad
No todas las consultas necesitan GPT-4o. Una pregunta de FAQ sobre horarios o precios la puede responder gpt-4o-mini perfectamente. Puedes añadir un clasificador simple que decida qué modelo usar según la complejidad estimada de la pregunta. Aquí también viene bien un interfaces para aplicaciones IA en Python ligera para visualizar métricas de uso por modelo.
Medir el ahorro real
Nada de esto vale si no lo mides. Usa tiktoken para contar los tokens que mandas antes y después de cada optimización:
import tiktoken
def count_tokens(text: str, model: str = "gpt-4o") -> int:
enc = tiktoken.encoding_for_model(model)
return len(enc.encode(text))
def estimate_cost(input_tokens: int, output_tokens: int, model: str = "gpt-4o") -> float:
# Precios en $ por millón de tokens (mayo 2026, verifica en la web de OpenAI)
prices = {
"gpt-4o": {"input": 2.50, "output": 10.00},
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
}
p = prices.get(model, prices["gpt-4o"])
return (input_tokens * p["input"] + output_tokens * p["output"]) / 1_000_000
# En cada llamada al LLM, registra:
context_tokens = count_tokens(context)
question_tokens = count_tokens(question)
total_input = context_tokens + question_tokens
cost = estimate_cost(total_input, output_tokens=300)
print(f"Tokens de entrada: {total_input} | Coste estimado: ${cost:.4f}")
Haz esto antes y después de activar la caché semántica en producción real, con tráfico real. Si el 40% de tus consultas son variaciones de las mismas 50 preguntas (lo habitual en sistemas de soporte o documentación corporativa), el ahorro se nota desde el primer día. Para explorar más sobre cómo Python para procesamiento de datos e IA puede encajar en este tipo de arquitecturas, tienes un buen punto de partida en ese artículo.
La caché semántica y la reescritura de consultas no son optimizaciones prematuras. Son decisiones de diseño que conviene tomar desde el principio, antes de que la factura de la API empiece a ser un problema.
Imagen: Pexels / Google DeepMind