Construyendo un pipeline de RAG desde cero

La Generación Aumentada por Recuperación (RAG) es el patrón más práctico que he encontrado para darle a un LLM acceso a tus propios datos. En lugar de realizar un fine-tuning del modelo con tus documentos —lo cual es costoso, lento y queda obsoleto en semanas—, recuperas el contenido adecuado al momento de la consulta y lo pasas como contexto. El concepto es simple. El pipeline que realmente funciona en producción no lo es.

Este artículo recorre cada componente, las decisiones que debes tomar en cada etapa y los modos de falla que encontrarás en el camino.

Arquitectura

Un pipeline de RAG tiene cuatro etapas:

También hay un pipeline de ingesta offline que procesa tus documentos de origen:

Analizaré cada pieza en el orden en que las construirás.

Ingesta de documentos y chunking

Tus documentos deben dividirse en fragmentos (chunks) antes de generar los embeddings. Aquí es donde la mayoría de los pipelines de RAG tienen éxito o fracasan, y también es el paso que recibe menos atención porque no es tan glamuroso.

Por qué fragmentar (chunking)

Los modelos de embedding tienen una ventana de contexto fija, usualmente de 512 a 8192 tokens. Incluso cuando el modelo soporta técnicamente una entrada larga, la calidad del embedding cae rápidamente en textos extensos; el vector resultante se convierte en un promedio de demasiados conceptos para ser útil en una recuperación precisa.

Estrategias de chunking

El chunking de tamaño fijo es lo más simple. Divide el texto en N tokens con M tokens de superposición (overlap).

def fixed_size_chunks(text, chunk_size=512, overlap=50):
    tokens = tokenizer.encode(text)
    chunks = []
    for i in range(0, len(tokens), chunk_size - overlap):
        chunk_tokens = tokens[i:i + chunk_size]
        chunks.append(tokenizer.decode(chunk_tokens))
    return chunks

Funciona, pero es rudimentario. Divide a mitad de una oración o párrafo, produciendo chunks que no mantienen un significado coherente por sí mismos.

La división recursiva por caracteres (RecursiveCharacterTextSplitter de LangChain) intenta dividir primero en límites naturales —párrafos, luego oraciones, luego palabras— y solo recurre a unidades más pequeñas cuando es estrictamente necesario.

El chunking semántico utiliza un modelo de embedding para detectar cambios de tema. Calculas los embeddings para cada oración y luego divides cuando la similitud de coseno entre oraciones consecutivas cae por debajo de un umbral. Los chunks terminan alineados con los límites reales del contenido en lugar de recuentos de tokens arbitrarios.

El chunking consciente de la estructura del documento (structure-aware) respeta la estructura del origen. Para Markdown, divide por encabezados. Para HTML, por etiquetas semánticas. Para PDFs, usa análisis de diseño (layout analysis) para identificar secciones. Es el que requiere más trabajo pero produce los mejores resultados cuando tus documentos fuente están bien estructurados.

Recomendaciones prácticas

Un tamaño de chunk de 256-512 tokens es el punto ideal para la mayoría de los casos de uso. Los chunks más pequeños mejoran la precisión de la recuperación, mientras que los más grandes le dan al LLM más contexto por cada elemento recuperado. Usa un overlap del 10-20% para evitar que la información se corte en los límites. Siempre almacena metadatos con cada chunk: documento fuente, encabezado de sección, número de página, fecha de creación. Los necesitarás para filtrar, citar y depurar la recuperación cuando las cosas salgan mal, algo que sucederá.

Modelos de embedding

El modelo de embedding convierte el texto en vectores densos que capturan el significado. La calidad del embedding es el factor determinante más importante de la calidad de la recuperación.

Opciones

Modelo	Dimensiones	Máx Tokens	Notas
OpenAI text-embedding-3-small	1536	8191	Buen equilibrio costo/calidad
OpenAI text-embedding-3-large	3072	8191	La mejor calidad de OpenAI
Sentence-BERT (all-MiniLM-L6-v2)	384	512	Gratis, rápido, ideal para prototipado
BGE-large-en-v1.5	1024	512	Opción robusta de código abierto
Gemini text-embedding-004	768	2048	Competitivo, se integra al stack de Google
Cohere embed-v3	1024	512	Excelente soporte multilingüe

Para producción, suelo usar los embeddings de OpenAI porque la relación calidad-esfuerzo es difícil de superar. Para casos sensibles al costo o de alto volumen, usar BGE o Sentence-BERT auto-alojado elimina por completo el costo por llamada.

Mejores prácticas

Genera los embeddings de las consultas y los documentos de la misma manera. Algunos modelos, especialmente BGE, soportan embeddings con prefijos de instrucción: antepones "Represent this document for retrieval:" a los documentos y "Represent this query for retrieval:" a las consultas. Cuando el modelo lo soporta, la mejora es real.

from openai import OpenAI

client = OpenAI()

def embed(texts, model="text-embedding-3-small"):
    response = client.embeddings.create(input=texts, model=model)
    return [item.embedding for item in response.data]

# Embedding por lotes para mayor eficiencia — hasta 2048 textos por llamada
doc_embeddings = embed(chunks)

Vector store

El vector store almacena los embeddings indexados y ejecuta la búsqueda de similitud. La elección depende de la escala y las restricciones de infraestructura.

Chroma es embebido, nativo de Python y no requiere configuración. Excelente para prototipado y aplicaciones con menos de 1 millón de documentos. Datos locales.

Pinecone es totalmente gestionado y serverless. Escala sin esfuerzo operativo, pero tiene un costo y dependes de un servicio externo.

pgvector es una extensión de PostgreSQL. Si ya usas Postgres, este es el paso lógico porque evita añadir otra base de datos al stack. Con un indexado HNSW adecuado, maneja millones de vectores cómodamente.

Weaviate, Qdrant y Milvus son bases de datos vectoriales dedicadas con búsqueda híbrida, filtrado, multi-tenancy, etc. Elige una de estas cuando la búsqueda vectorial sea el núcleo de tu aplicación y no solo una característica secundaria.

Para la mayoría de los proyectos, comienza con pgvector si ya usas Postgres o con Chroma si quieres el camino más rápido hacia un prototipo funcional.

# Ejemplo de pgvector con SQLAlchemy
from pgvector.sqlalchemy import Vector
from sqlalchemy import Column, Integer, Text
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Document(Base):
    __tablename__ = "documents"
    id = Column(Integer, primary_key=True)
    content = Column(Text)
    embedding = Column(Vector(1536))  # Debe coincidir con las dimensiones de tu modelo
    source = Column(Text)

Estrategias de recuperación

La recuperación top-K básica funciona en muchos casos, pero tiene modos de falla conocidos cuando se le exige más.

Similitud Top-K

Calcula la similitud de coseno entre el embedding de la consulta y cada embedding de documento, y devuelve los mejores K.

query_embedding = embed([user_query])[0]
results = vector_store.similarity_search(
    query_embedding,
    k=5
)

El problema con top-K es que a menudo devuelve resultados redundantes. Si tres de tus chunks son paráfrasis del mismo hecho, habrás desperdiciado el 60% de tu ventana de contexto en duplicados.

Relevancia Marginal Máxima (MMR)

MMR equilibra la relevancia con la diversidad. Selecciona iterativamente documentos que son similares a la consulta pero disímiles de los documentos ya elegidos.

results = vector_store.max_marginal_relevance_search(
    query_embedding,
    k=5,
    fetch_k=20,       # Obtiene 20 candidatos
    lambda_mult=0.7,   # 0=máxima diversidad, 1=máxima relevancia
)

Es un cambio simple que mejora notablemente la calidad de la respuesta en la mayoría de las aplicaciones.

Búsqueda híbrida

Combina la búsqueda de vectores densos con la búsqueda dispersa por palabras clave (BM25). La recuperación densa es excelente para la similitud semántica ("¿Qué causa la inflación?"). La recuperación dispersa maneja mejor las coincidencias exactas y términos raros ("código de error 0x80070005").

# Búsqueda híbrida: BM25 + similitud vectorial
bm25_results = bm25_index.search(query, k=10)
vector_results = vector_store.search(query_embedding, k=10)

# Reciprocal Rank Fusion
combined = reciprocal_rank_fusion(bm25_results, vector_results)
final_results = combined[:5]

La mayoría de los sistemas de RAG en producción que he visto terminan utilizando búsqueda híbrida. pgvector más la búsqueda de texto completo de Postgres te ofrece ambos en una sola base de datos sin piezas móviles adicionales.

Construcción del Prompt

Cómo presentas el contexto recuperado al modelo importa tanto como lo que recuperas.

def build_prompt(query, retrieved_docs):
    context = "\n\n---\n\n".join([
        f"Fuente: {{doc.metadata['source']}}\n{{doc.content}}"
        for doc in retrieved_docs
    ])

    return f"""Responde a la pregunta del usuario basándote en el contexto proporcionado.
Si el contexto no contiene suficiente información para responder completamente,
dilo en lugar de inventar información.

Contexto:
{{context}}

Pregunta: {{query}}

Respuesta:"""

Coloca el contexto antes de la pregunta. Los LLM prestan más atención al contenido cerca del final del prompt —el efecto "lost in the middle" (perdido en el medio)— por lo que la pregunta va al último. Incluye la atribución de fuentes para que el modelo pueda citar y para que tú puedas depurar problemas de recuperación. Establece límites explícitos cuando la base fáctica (grounding) sea crucial; indícale al modelo que no use nada fuera del contexto proporcionado. Y vigila la ventana de contexto. Cinco chunks de 512 tokens son 2500 tokens de contexto, y aún necesitas espacio para el prompt del sistema, la consulta y la respuesta.

Evaluación

El RAG necesita evaluación en dos niveles: calidad de la recuperación y calidad de la generación.

Métricas de recuperación

Recall@K es la fracción de documentos relevantes que aparecen en los top K resultados. MRR (Mean Reciprocal Rank) te indica qué tan arriba aparece el primer resultado relevante. Precision@K es la fracción de los resultados devueltos que son realmente relevantes.

Necesitas un conjunto de evaluación. Toma 50-100 consultas representativas, identifica manualmente los documentos fuente correctos para cada una y mide tu pipeline frente a esa verdad fundamental (ground truth). Sin esto, cada cambio que hagas será una suposición.

Métricas de generación

Fidelidad (faithfulness): ¿la respuesta solo utiliza información respaldada por el contexto recuperado? (Sin alucinaciones). Relevancia: ¿la respuesta realmente aborda la pregunta del usuario? Completitud: ¿cubre toda la información importante del contexto recuperado?

Frameworks como RAGAS automatizan esto usando un enfoque de "LLM como juez". Escala bien y te da una forma de medir la calidad a medida que iteras.

Modos de falla comunes

Recuperación de chunks incorrectos

El modelo da respuestas seguras pero incorrectas, o dice "no tengo información sobre eso" cuando la respuesta está en tu corpus. Registra los chunks recuperados para cada consulta. Verifica si la información correcta está en tu vector store y, si lo está, si el embedding de la consulta está realmente cerca de ella. La solución suele ser un mejor chunking, no un modelo de embedding diferente.

Chunks relevantes existen pero su ranking es bajo

Aumentar K de 5 a 20 lo soluciona, pero satura tu prompt. Prueba con búsqueda híbrida, re-ranking con cross-encoders o expansión de consultas (generar múltiples redacciones de la pregunta del usuario y recuperar para cada una).

El modelo ignora el contexto recuperado

La respuesta suena plausible pero no refleja los documentos recuperados. Refuerza las instrucciones en el prompt del sistema. Usa un modelo más capaz. Reduce la cantidad de contexto recuperado para que la relación señal-ruido sea mayor.

Los límites del chunk dividen información clave

La respuesta es parcialmente correcta y le faltan detalles que abarcan dos chunks. Aumenta el overlap. Usa chunking consciente de la estructura del documento. O usa un patrón de recuperación de documento padre (parent-document retriever): recupera chunks pequeños pero pasa su sección padre al LLM.

Uniendo todo

Un esquema de pipeline mínimo pero listo para producción:

# Esquema del pipeline completo
from openai import OpenAI
import psycopg2

client = OpenAI()

def ingest(documents):
    for doc in documents:
        chunks = recursive_split(doc.content, chunk_size=512, overlap=50)
        embeddings = embed(chunks)
        for chunk, emb in zip(chunks, embeddings):
            db.execute(
                "INSERT INTO documents (content, embedding, source) VALUES (%s, %s, %s)",
                (chunk, emb, doc.source)
            )

def query(user_question):
    q_emb = embed([user_question])[0]

    # Recuperación híbrida
    results = db.execute("""
        SELECT content, source,
               1 - (embedding <=> %s::vector) as similarity,
               ts_rank(to_tsvector(content), plainto_tsquery(%s)) as text_rank
        FROM documents
        ORDER BY (0.7 * (1 - (embedding <=> %s::vector)) + 0.3 * ts_rank(to_tsvector(content), plainto_tsquery(%s))) DESC
        LIMIT 5
    """, (q_emb, user_question, q_emb, user_question))

    prompt = build_prompt(user_question, results)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

Comienza aquí, mide con un conjunto de evaluación real e itera. La mayoría de las mejoras en RAG provienen de un mejor chunking y recuperación, no de cambiar de LLM o de modelo de embedding. Haz bien el pipeline de datos primero; todo lo demás viene después.