RAG Pipelines y Bases de Datos Vectoriales: Generación Aumentada por Recuperación en Producción

1. Arquitectura RAG: Visión General

La Generación Aumentada por Recuperación (RAG) es el patrón dominante para fundamentar las respuestas de LLMs en conocimiento externo. En vez de depender únicamente de la memoria paramétrica del modelo (que está congelada al momento del entrenamiento y propensa a alucinaciones), RAG recupera documentos relevantes de una base de conocimiento en tiempo de inferencia y los incluye en el contexto del prompt. Esto le da al modelo acceso a información actual, específica del dominio y verificable -- la diferencia entre un LLM que adivina y uno que cita fuentes.

Un sistema RAG de producción tiene tres fases: ingestión (procesar documentos en chunks buscables con embeddings almacenados en una base de datos vectorial), recuperación (encontrar los chunks más relevantes para una consulta usando similaridad vectorial, búsqueda por keywords o enfoques híbridos), y generación (alimentar el contexto recuperado a un LLM para producir una respuesta fundamentada). Cada fase tiene su propio conjunto de decisiones de ingeniería que se acumulan en la calidad general del sistema.

2. Procesamiento de Documentos

La calidad de tu sistema RAG está limitada por la calidad de tu procesamiento de documentos. Basura entra, basura sale aplica doblemente aquí: texto mal extraído lleva a chunks malos, que llevan a recuperaciones irrelevantes, que llevan a respuestas alucinadas. Invierte fuertemente en el procesamiento de documentos -- es la mejora de mayor apalancamiento que puedes hacer a un pipeline RAG.

3. Estrategias de Chunking

El chunking determina la granularidad de tu base de conocimiento. Chunks muy grandes diluyen el embedding con información irrelevante y desperdician tokens de ventana de contexto. Chunks muy pequeños pierden el contexto necesario para que el LLM produzca respuestas coherentes. El tamaño óptimo de chunk depende de tu tipo de documento, modelo de embedding y caso de uso -- pero la mayoría de sistemas de producción caen entre 256 y 1024 tokens por chunk.

4. Modelos de Embeddings

Los modelos de embeddings convierten texto en vectores densos que capturan significado semántico. La elección del modelo de embedding determina la calidad de recuperación: establece el techo de qué tan bien tu sistema RAG puede emparejar consultas con documentos relevantes. A junio de 2026, el panorama incluye modelos API propietarios (OpenAI, Cohere, Google, Voyage) y modelos open-source (BGE, E5, GTE, Jina) que rivalizan o superan a las opciones propietarias en benchmarks como MTEB.

5. Comparación de Bases de Datos Vectoriales

Las bases de datos vectoriales almacenan embeddings y habilitan búsqueda de similaridad rápida a escala. La elección depende de tu modelo de despliegue (gestionado vs self-hosted), requisitos de escala (miles vs miles de millones de vectores), patrones de consulta (vector puro vs búsqueda híbrida) e infraestructura existente. Todas las bases vectoriales de grado producción soportan indexación HNSW, filtrado por metadata y latencia de consulta sub-100ms para colecciones de millones de vectores.

6. Estrategias de Recuperación

La recuperación es donde se originan la mayoría de los problemas de calidad RAG. Un modelo de embedding perfecto con una estrategia naive de top-k tendrá peor rendimiento que un modelo decente con ingeniería de recuperación bien pensada. Las estrategias a continuación progresan de simples a avanzadas, y los sistemas de producción típicamente combinan múltiples enfoques.

7. Reclasificación (Reranking)

El reranking es la mejora post-recuperación de mayor apalancamiento. La recuperación inicial (similaridad de embeddings) es una primera pasada rápida pero imprecisa. Los rerankers usan modelos más costosos para re-puntuar y reordenar los top-k resultados basándose en la interacción profunda consulta-documento. Un bi-encoder (modelo de embedding) procesa la consulta y el documento de forma independiente; un cross-encoder los procesa conjuntamente, atendiendo a las interacciones entre los tokens de la consulta y del documento. Este procesamiento conjunto captura señales de relevancia que la codificación independiente pierde.

8. Patrones RAG Avanzados

El RAG estándar (recuperar-luego-generar) alcanza un techo para preguntas complejas que requieren razonamiento multi-paso, auto-corrección o fuentes de datos heterogéneas. Los patrones RAG avanzados extienden la arquitectura básica con comportamiento agéntico, conocimiento basado en grafos, recuperación multi-modal y loops de auto-evaluación. Estos patrones aumentan la complejidad del sistema pero desbloquean capacidades que el RAG simple no puede lograr.

9. Evaluación (Framework RAGAS)

No puedes mejorar lo que no mides. La evaluación RAG es más difícil que la evaluación ML estándar porque necesitas evaluar tanto la calidad de recuperación como la calidad de generación de forma independiente, y su interacción. El framework RAGAS (Retrieval Augmented Generation Assessment) es el toolkit de evaluación más adoptado, proveyendo métricas automatizadas que correlacionan bien con los juicios humanos sin requerir etiquetas de ground-truth.

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset

# Prepare evaluation dataset
eval_data = Dataset.from_dict({
    "question": questions,
    "answer": generated_answers,
    "contexts": retrieved_contexts,      # list[list[str]]
    "ground_truth": reference_answers     # for context_recall
})

# Run evaluation
results = evaluate(
    dataset=eval_data,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    llm=ChatOpenAI(model="gpt-5.5"),      # evaluator LLM
    embeddings=OpenAIEmbeddings()
)
print(results)
# {'faithfulness': 0.87, 'answer_relevancy': 0.91,
#  'context_precision': 0.78, 'context_recall': 0.82}

10. Despliegue en Producción

Mover RAG de un notebook a producción involucra caching, streaming, multi-tenancy, monitoreo y optimización de costos. La diferencia entre un demo y un sistema de producción no es el algoritmo de recuperación -- es la infraestructura que hace al algoritmo confiable, rápido y económico a escala.

11. Ejemplos de Código

Tres enfoques de implementación desde frameworks de alto nivel hasta llamadas directas a APIs. LangChain provee el camino más rápido a un sistema RAG funcional con la mayor cantidad de abstracciones. LlamaIndex está construido específicamente para RAG con primitivas de indexación más profundas. Las llamadas directas a API dan máximo control y mínimas dependencias para equipos que prefieren código explícito sobre la magia del framework.

Pipeline RAG con LangChain

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Qdrant
from langchain_community.document_loaders import PyMuPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

# 1. Load and chunk documents
loader = PyMuPDFLoader("technical_manual.pdf")
docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512, chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = splitter.split_documents(docs)

# 2. Embed and store in Qdrant
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Qdrant.from_documents(
    chunks, embeddings,
    url="http://localhost:6333",
    collection_name="tech_manual"
)

# 3. Create retrieval chain with reranking
retriever = vectorstore.as_retriever(
    search_type="mmr",          # Maximum Marginal Relevance
    search_kwargs={"k": 10, "fetch_k": 25, "lambda_mult": 0.7}
)

prompt = ChatPromptTemplate.from_messages([
    ("system", """Answer based on the context below. Cite sources.
If the context doesn't contain the answer, say so.

Context: {context}"""),
    ("human", "{input}")
])

llm = ChatOpenAI(model="gpt-5.5", temperature=0)
chain = create_retrieval_chain(
    retriever,
    create_stuff_documents_chain(llm, prompt)
)

# 4. Query
result = chain.invoke({"input": "How do I configure failover?"})
print(result["answer"])

Pipeline RAG con LlamaIndex

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.pinecone import PineconeVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.anthropic import Anthropic
from llama_index.core.postprocessor import SentenceTransformerRerank
from pinecone import Pinecone

# 1. Configure models
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-large")
Settings.llm = Anthropic(model="claude-sonnet-4-6", temperature=0)

# 2. Load documents with metadata
documents = SimpleDirectoryReader(
    input_dir="./knowledge_base",
    recursive=True,
    filename_as_id=True
).load_data()

# 3. Parse with sentence-aware splitting
parser = SentenceSplitter(chunk_size=512, chunk_overlap=50)
nodes = parser.get_nodes_from_documents(documents)

# 4. Create Pinecone index
pc = Pinecone(api_key="your-api-key")
pinecone_index = pc.Index("rag-production")
vector_store = PineconeVectorStore(pinecone_index=pinecone_index)

index = VectorStoreIndex.from_documents(
    documents, vector_store=vector_store
)

# 5. Query with reranking
reranker = SentenceTransformerRerank(
    model="BAAI/bge-reranker-v2-m3", top_n=5
)
query_engine = index.as_query_engine(
    similarity_top_k=20,
    node_postprocessors=[reranker],
    response_mode="tree_summarize"
)

response = query_engine.query("Explain the backup recovery procedure")
print(response)
for node in response.source_nodes:
    print(f"  [{node.score:.3f}] {node.metadata['file_name']}")

API Directa: Pinecone + OpenAI

import openai
from pinecone import Pinecone

client = openai.OpenAI()
pc = Pinecone(api_key="your-api-key")
index = pc.Index("rag-production")

def embed(text: str, input_type: str = "search_query") -> list[float]:
    """Generate embedding for text."""
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=text,
        dimensions=1536  # Matryoshka shortening for cost savings
    )
    return response.data[0].embedding

def retrieve(query: str, top_k: int = 10, filters: dict = None) -> list[dict]:
    """Retrieve relevant chunks from Pinecone."""
    query_vector = embed(query)
    results = index.query(
        vector=query_vector,
        top_k=top_k,
        include_metadata=True,
        filter=filters  # e.g., {"source": "engineering-docs", "year": {"$gte": 2025}}
    )
    return [
        {"text": m.metadata["text"], "source": m.metadata["source"], "score": m.score}
        for m in results.matches
    ]

def generate(query: str, contexts: list[dict]) -> str:
    """Generate answer grounded in retrieved context."""
    context_str = "\n\n---\n\n".join(
        f"[Source: {c['source']}] {c['text']}" for c in contexts
    )
    response = client.chat.completions.create(
        model="gpt-5.5",
        temperature=0,
        messages=[
            {"role": "system", "content": f"""Answer based on the context below.
Cite sources using [Source: ...] format.
If context is insufficient, say "I don't have enough information."

Context:
{context_str}"""},
            {"role": "user", "content": query}
        ],
        stream=True
    )
    chunks = []
    for chunk in response:
        if chunk.choices[0].delta.content:
            chunks.append(chunk.choices[0].delta.content)
            print(chunk.choices[0].delta.content, end="", flush=True)
    return "".join(chunks)

# RAG query
query = "What are the SLA requirements for the payment service?"
contexts = retrieve(query, top_k=10, filters={"team": "payments"})
answer = generate(query, contexts)

pgvector con Python (SQL Directo)

import psycopg2
import openai
import json

client = openai.OpenAI()
conn = psycopg2.connect("postgresql://user:pass@localhost:5432/ragdb")

# Setup: create table with vector column
with conn.cursor() as cur:
    cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
    cur.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            id SERIAL PRIMARY KEY,
            content TEXT NOT NULL,
            source VARCHAR(255),
            embedding vector(1536),
            metadata JSONB DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ DEFAULT NOW()
        )
    """)
    cur.execute("""
        CREATE INDEX IF NOT EXISTS docs_embedding_idx
        ON documents USING hnsw (embedding vector_cosine_ops)
        WITH (m = 16, ef_construction = 200)
    """)
    conn.commit()

def ingest(content: str, source: str, metadata: dict = None):
    """Embed and store a document chunk."""
    resp = client.embeddings.create(
        model="text-embedding-3-large", input=content, dimensions=1536
    )
    embedding = resp.data[0].embedding
    with conn.cursor() as cur:
        cur.execute(
            """INSERT INTO documents (content, source, embedding, metadata)
               VALUES (%s, %s, %s::vector, %s::jsonb)""",
            (content, source, str(embedding), json.dumps(metadata or {}))
        )
    conn.commit()

def search(query: str, top_k: int = 5, source_filter: str = None) -> list:
    """Hybrid search: vector similarity + optional source filter."""
    q_emb = client.embeddings.create(
        model="text-embedding-3-large", input=query, dimensions=1536
    ).data[0].embedding

    sql = """
        SELECT content, source,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents
        WHERE 1=1
    """
    params = [str(q_emb)]
    if source_filter:
        sql += " AND source = %s"
        params.append(source_filter)
    sql += " ORDER BY embedding <=> %s::vector LIMIT %s"
    params.extend([str(q_emb), top_k])

    with conn.cursor() as cur:
        cur.execute(sql, params)
        return [
            {"content": r[0], "source": r[1], "score": r[2]}
            for r in cur.fetchall()
        ]