Arquitectura de Pipelines RAG: Estrategias de Chunking, Búsqueda Híbrida, Reranking y Frameworks de Evaluación

Por Qué la Mayoría de los Pipelines RAG Fallan

El tutorial RAG por defecto — dividir documentos, generar embeddings, recuperar top-k, insertar en el prompt — funciona para demos. Falla en producción por razones predecibles:

• El chunking destruye contexto. Los cortes de tamaño fijo rompen oraciones a mitad de pensamiento, separan encabezados de su contenido y pierden la estructura del documento.
• La búsqueda vectorial pierde keywords. Los embeddings capturan similitud semántica pero fallan en coincidencias exactas de entidades, acrónimos y terminología de dominio específica.
• La recuperación top-k es ruidosa. Recuperar 5 chunks por similitud coseno frecuentemente devuelve 2-3 resultados irrelevantes que diluyen el contexto del prompt.
• Sin evaluación no hay iteración. Sin métricas sistemáticas, los equipos optimizan por intuición y despliegan pipelines que se degradan silenciosamente.

Este artículo cubre la arquitectura que aborda cada uno de estos modos de falla.

Ingesta y Preprocesamiento de Documentos

Antes del chunking, el preprocesamiento de documentos determina la calidad del pipeline más de lo que la mayoría de los equipos creen.

Normalización de Formatos

Diferentes formatos fuente requieren diferentes estrategias de extracción:

# Extracción de PDF con detección de layout
from unstructured.partition.pdf import partition_pdf

elements = partition_pdf(
    filename="report.pdf",
    strategy="hi_res",           # OCR + detección de layout
    infer_table_structure=True,  # Extraer tablas como HTML
    extract_images_in_pdf=True,  # Extraer imágenes embebidas
)

# Separar tipos de elementos para procesamiento diferenciado
tables = [e for e in elements if e.category == "Table"]
narratives = [e for e in elements if e.category == "NarrativeText"]
titles = [e for e in elements if e.category == "Title"]

Decisiones clave en esta etapa:

• Tablas: Convertir a markdown o HTML, nunca dividir entre chunks. Un fragmento de tabla es peor que no tener tabla.
• Bloques de código: Mantener intactos. Dividir la definición de una función entre chunks hace que ambas mitades sean inútiles.
• Imágenes: Extraer captions y texto circundante. Para diagramas, considerar descripciones con modelos de visión.
• Encabezados: Preservar la jerarquía — se convierten en metadata que mejora el retrieval.

Extracción de Metadata

Adjuntar metadata en el momento de la ingesta, no después:

chunk_metadata = {
    "source": "engineering-handbook-v3.pdf",
    "section": "Capítulo 4: Operaciones de Base de Datos",
    "page_numbers": [42, 43],
    "document_type": "handbook",
    "last_updated": "2026-03-15",
    "audience": "backend-engineers",
}

Esta metadata habilita el retrieval filtrado — "encontrar chunks sobre operaciones de base de datos del handbook de ingeniería" — lo cual mejora dramáticamente la precisión.

Estrategias de Chunking

El chunking es donde la mayoría de los pipelines pierden calidad silenciosamente. La estrategia correcta depende de la estructura del documento.

Chunking de Tamaño Fijo

La línea base. Dividir cada N tokens con M tokens de overlap.

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,  # o tiktoken para conteo preciso de tokens
)
chunks = splitter.split_text(document_text)

Cuándo funciona: Texto uniforme sin estructura — transcripciones, logs en texto plano, historiales de chat.

Cuándo falla: Documentos estructurados donde dividir en posiciones arbitrarias destruye el significado.

Chunking Semántico

Agrupa oraciones por similitud de embedding, dividiendo donde los temas cambian:

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

chunker = SemanticChunker(
    OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=85,  # Dividir en el percentil 85 de distancia
)
chunks = chunker.split_text(document_text)

El chunking semántico produce chunks de tamaño variable que respetan los límites temáticos. El tradeoff es velocidad — generar embeddings de cada oración durante la ingesta es 10-50x más lento que la división por tamaño fijo.

Chunking Consciente de la Estructura del Documento

El enfoque más efectivo para contenido estructurado. Usa la jerarquía del documento para definir los límites de los chunks:

def structure_aware_chunk(elements, max_tokens=512):
    """Dividir por estructura del documento, respetando encabezados y secciones."""
    chunks = []
    current_chunk = []
    current_tokens = 0
    current_header_chain = []

    for element in elements:
        if element.category == "Title":
            # Nueva sección — volcar chunk actual
            if current_chunk:
                chunks.append({
                    "content": "\n".join(current_chunk),
                    "headers": list(current_header_chain),
                    "token_count": current_tokens,
                })
                current_chunk = []
                current_tokens = 0
            current_header_chain.append(element.text)

        elif element.category == "Table":
            # Las tablas van en su propio chunk, nunca se dividen
            if current_chunk:
                chunks.append({
                    "content": "\n".join(current_chunk),
                    "headers": list(current_header_chain),
                    "token_count": current_tokens,
                })
                current_chunk = []
                current_tokens = 0
            chunks.append({
                "content": element.metadata.text_as_html,
                "headers": list(current_header_chain),
                "token_count": count_tokens(element.text),
                "type": "table",
            })

        else:
            elem_tokens = count_tokens(element.text)
            if current_tokens + elem_tokens > max_tokens and current_chunk:
                chunks.append({
                    "content": "\n".join(current_chunk),
                    "headers": list(current_header_chain),
                    "token_count": current_tokens,
                })
                current_chunk = []
                current_tokens = 0
            current_chunk.append(element.text)
            current_tokens += elem_tokens

    if current_chunk:
        chunks.append({
            "content": "\n".join(current_chunk),
            "headers": list(current_header_chain),
            "token_count": current_tokens,
        })

    return chunks

Encabezados Contextuales en los Chunks

Anteponer la jerarquía de secciones a cada chunk antes de generar el embedding. Esta única técnica mejoró el recall de retrieval entre un 8-12% en nuestros benchmarks:

## Capítulo 4: Operaciones de Base de Datos > Estrategias de Backup > Recuperación Point-in-Time

Para realizar recuperación point-in-time, configurar el archivado continuo de WAL...

El embedding ahora captura tanto el contenido como su posición en la jerarquía del documento.

Estrategia de Embeddings

Selección del Modelo

El modelo de embedding determina el techo de la calidad de retrieval.

| Modelo | Dimensiones | MTEB Score | Latencia (p50) | Costo | |---|---|---|---|---| | text-embedding-3-small | 1536 | 62.3 | 12ms | $0.02/1M tokens | | text-embedding-3-large | 3072 | 64.6 | 18ms | $0.13/1M tokens | | voyage-3 | 1024 | 67.1 | 15ms | $0.06/1M tokens | | bge-m3 (self-hosted) | 1024 | 66.8 | 8ms | Costo de infraestructura | | Cohere embed-v4 | 1024 | 67.5 | 14ms | $0.10/1M tokens |

Recomendación práctica: Para workloads solo en inglés, voyage-3 o Cohere embed-v4 entregan la mejor calidad de retrieval. Para deployments multilingües o sensibles al costo, bge-m3 self-hosted en una sola GPU es difícil de superar.

Modelos de Interacción Tardía (ColBERT)

ColBERT produce embeddings por token en vez de un único vector, permitiendo matching más granular:

from ragatouille import RAGPretrainedModel

rag = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")
rag.index(
    collection=documents,
    index_name="knowledge-base",
    max_document_length=512,
)

results = rag.search(query="configuración de backup PITR", k=10)

ColBERT supera consistentemente a los modelos de vector único en calidad de retrieval pero requiere 10-50x más almacenamiento por documento. Usalo cuando la calidad de retrieval sea el cuello de botella y el almacenamiento sea barato.

Arquitectura de Búsqueda Híbrida

La búsqueda vectorial sola no es suficiente. Combinar búsqueda semántica y léxica cubre los puntos ciegos de cada método.

BM25 + Vector con Reciprocal Rank Fusion

from rank_bm25 import BM25Okapi
import numpy as np

def hybrid_search(query, corpus_chunks, embeddings_index, k=20, alpha=0.5):
    """
    Combinar BM25 y búsqueda vectorial usando Reciprocal Rank Fusion.
    alpha controla el balance: 0.5 = peso igual.
    """
    # Búsqueda vectorial
    query_embedding = embed(query)
    vector_scores = embeddings_index.search(query_embedding, k=k * 2)

    # Búsqueda BM25
    tokenized_corpus = [chunk.split() for chunk in corpus_chunks]
    bm25 = BM25Okapi(tokenized_corpus)
    bm25_scores = bm25.get_scores(query.split())
    bm25_top = np.argsort(bm25_scores)[-k * 2:][::-1]

    # Reciprocal Rank Fusion
    rrf_scores = {}
    rrf_k = 60  # Constante RRF estándar

    for rank, doc_id in enumerate(vector_scores.ids):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + alpha / (rrf_k + rank + 1)

    for rank, doc_id in enumerate(bm25_top):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0) + (1 - alpha) / (rrf_k + rank + 1)

    # Ordenar por score fusionado
    ranked = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
    return ranked[:k]

Por Qué Funciona la Búsqueda Híbrida

Considerá la consulta: "¿Cuál es el SLA del cluster Redis en EU-WEST-1?"

• Búsqueda vectorial encuentra chunks sobre SLAs y clusters Redis en general — semánticamente similares pero potencialmente de la región incorrecta.
• BM25 encuentra chunks que contienen la cadena exacta "EU-WEST-1" — coincidencia léxica.
• Híbrida surfacea el chunk que menciona tanto la política de SLA como la región específica.

Búsqueda Híbrida Nativa en Base de Datos

PostgreSQL con pgvector 0.8+ soporta ambas en una sola consulta:

-- Búsqueda híbrida con pgvector + tsvector
WITH vector_results AS (
    SELECT id, content, 1 - (embedding <=> $1::vector) AS vector_score
    FROM documents
    WHERE metadata->>'department' = 'engineering'
    ORDER BY embedding <=> $1::vector
    LIMIT 40
),
text_results AS (
    SELECT id, content, ts_rank(search_vector, plainto_tsquery($2)) AS text_score
    FROM documents
    WHERE search_vector @@ plainto_tsquery($2)
      AND metadata->>'department' = 'engineering'
    ORDER BY text_score DESC
    LIMIT 40
),
combined AS (
    SELECT
        COALESCE(v.id, t.id) AS id,
        COALESCE(v.content, t.content) AS content,
        COALESCE(v.vector_score, 0) * 0.5 + COALESCE(t.text_score, 0) * 0.5 AS hybrid_score
    FROM vector_results v
    FULL OUTER JOIN text_results t ON v.id = t.id
)
SELECT * FROM combined ORDER BY hybrid_score DESC LIMIT 10;

Reranking

El reranking es la mejora de mayor impacto que podés hacer a un pipeline RAG. Los embeddings de bi-encoder son rápidos pero imprecisos. Los rerankers cross-encoder son lentos pero precisos. La arquitectura usa ambos:

1. Recuperar 50-100 candidatos con búsqueda híbrida rápida (~10ms)
2. Rerankear los top 20-30 con un cross-encoder (~50-80ms)
3. Seleccionar los top 5-8 para el contexto del LLM

Reranking con Cross-Encoder

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)

def rerank(query, candidates, top_k=5):
    """Rerankear candidatos usando cross-encoder."""
    pairs = [[query, candidate["content"]] for candidate in candidates]
    scores = reranker.predict(pairs)

    for i, candidate in enumerate(candidates):
        candidate["rerank_score"] = float(scores[i])

    ranked = sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)
    return ranked[:top_k]

API de Rerank de Cohere

Para equipos que prefieren infraestructura gestionada:

import cohere

co = cohere.ClientV2()

results = co.rerank(
    model="rerank-v3.5",
    query="configuración de backup PITR para PostgreSQL",
    documents=[chunk["content"] for chunk in candidates],
    top_n=5,
    return_documents=True,
)

reranked = [
    {"content": r.document.text, "relevance_score": r.relevance_score}
    for r in results.results
]

Medición de Impacto

En nuestro pipeline de producción (2.3M chunks, 10K consultas de prueba):

| Configuración | Recall@5 | MRR | Relevancia de Respuesta | |---|---|---|---| | Solo vectorial | 0.62 | 0.54 | 0.71 | | Híbrida (BM25 + vector) | 0.74 | 0.63 | 0.78 | | Híbrida + reranker | 0.83 | 0.76 | 0.89 | | Híbrida + reranker + encabezados contextuales | 0.87 | 0.81 | 0.91 |

El reranker solo representa una mejora de 12 puntos en recall. Combinado con búsqueda híbrida y encabezados contextuales, el pipeline recupera el 87% de los chunks relevantes en los top 5 resultados.

Construcción del Prompt

Después del retrieval y reranking, cómo construís el prompt del LLM importa más de lo que la mayoría de los equipos creen.

Gestión de la Ventana de Contexto

def build_rag_prompt(query, retrieved_chunks, max_context_tokens=6000):
    """Construir prompt con inyección de contexto consciente del presupuesto de tokens."""
    context_parts = []
    token_count = 0

    for chunk in retrieved_chunks:
        chunk_tokens = count_tokens(chunk["content"])
        if token_count + chunk_tokens > max_context_tokens:
            break
        context_parts.append(
            f"[Fuente: {chunk['metadata']['source']}, "
            f"Sección: {chunk['metadata'].get('section', 'N/A')}]\n"
            f"{chunk['content']}"
        )
        token_count += chunk_tokens

    context = "\n\n---\n\n".join(context_parts)

    return f"""Respondé la siguiente pregunta basándote en el contexto proporcionado.
Si el contexto no contiene suficiente información para responder completamente, decilo explícitamente.
Citá el documento fuente para cada afirmación.

Contexto:
{context}

Pregunta: {query}

Respuesta:"""

Grounding con Citaciones

Forzá al modelo a citar fuentes requiriendo referencias inline:

Para cada afirmación factual en tu respuesta, incluí una cita en el formato [Fuente: nombre_archivo, Sección: nombre_sección].
Si no podés encontrar evidencia de soporte en el contexto para una afirmación, prefijala con [Sin soporte].

Esto hace que la detección de alucinaciones sea trivial — cualquier afirmación marcada como [Sin soporte] o sin cita es sospechosa.

Framework de Evaluación

Un pipeline RAG sin evaluación es una demo, no un producto. La evaluación ocurre en tres niveles.

Nivel 1: Calidad de Retrieval

Medida independientemente del LLM:

def evaluate_retrieval(test_queries, ground_truth, retriever, k=5):
    """Evaluar retrieval con métricas estándar de IR."""
    recalls, mrrs, ndcgs = [], [], []

    for query, relevant_ids in zip(test_queries, ground_truth):
        retrieved = retriever.search(query, k=k)
        retrieved_ids = [r["id"] for r in retrieved]

        # Recall@k
        hits = len(set(retrieved_ids) & set(relevant_ids))
        recalls.append(hits / len(relevant_ids))

        # MRR
        for rank, rid in enumerate(retrieved_ids, 1):
            if rid in relevant_ids:
                mrrs.append(1.0 / rank)
                break
        else:
            mrrs.append(0.0)

    return {
        "recall@k": np.mean(recalls),
        "mrr": np.mean(mrrs),
    }

Nivel 2: Calidad de Generación con RAGAS

RAGAS evalúa el pipeline completo — retrieval y generación juntos:

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)

results = evaluate(
    dataset=eval_dataset,  # Preguntas + respuestas ground truth + contextos
    metrics=[
        faithfulness,        # ¿La respuesta se mantiene fiel al contexto recuperado?
        answer_relevancy,    # ¿La respuesta es relevante a la pregunta?
        context_precision,   # ¿Los chunks recuperados son relevantes?
        context_recall,      # ¿Se recuperaron todos los chunks necesarios?
    ],
)

print(results)
# {'faithfulness': 0.89, 'answer_relevancy': 0.91,
#  'context_precision': 0.82, 'context_recall': 0.87}

Faithfulness es la métrica más importante — mide alucinación. Un score de faithfulness por debajo de 0.85 significa que el pipeline está generando afirmaciones no respaldadas por el contexto recuperado.

Nivel 3: Evaluación Humana

Automatizar la recolección, evaluar manualmente:

def sample_for_human_eval(production_queries, n=200):
    """Muestrear consultas estratificadas por dificultad y tema."""
    # Clasificar consultas por dificultad estimada
    easy = [q for q in production_queries if q["retrieval_confidence"] > 0.8]
    medium = [q for q in production_queries if 0.5 < q["retrieval_confidence"] <= 0.8]
    hard = [q for q in production_queries if q["retrieval_confidence"] <= 0.5]

    sample = (
        random.sample(easy, min(80, len(easy))) +
        random.sample(medium, min(80, len(medium))) +
        random.sample(hard, min(40, len(hard)))
    )

    return sample

Los evaluadores humanos califican en tres ejes: correctitud (factualmente precisa), completitud (cubre la pregunta en su totalidad) y fundamentación (cita contexto, no alucina).

Arquitectura de Producción

Vista General del Pipeline

Documentos → Preprocesamiento → Chunking → Embedding → Vector Store
                                                            ↓
Consulta → Expansión de Query → Búsqueda Híbrida → Reranker → Constructor de Prompt → LLM → Respuesta
                                                                                              ↓
                                                                                        Logger de Evaluación

Expansión de Consultas

Reescribir la consulta del usuario antes del retrieval para mejorar el recall:

def expand_query(query, llm):
    """Generar reformulaciones alternativas para mejorar el retrieval."""
    expansion_prompt = f"""Generá 3 reformulaciones alternativas de esta consulta de búsqueda.
Cada una debe capturar la misma intención pero usar terminología diferente.
Devolvé solo las consultas, una por línea.

Consulta: {query}"""

    alternatives = llm.generate(expansion_prompt).strip().split("\n")
    return [query] + alternatives[:3]

Ejecutar búsqueda híbrida sobre todas las consultas expandidas y fusionar resultados con RRF. Esto consistentemente agrega 3-5% de mejora en recall.

Capa de Caché

Cachear en dos niveles para reducir latencia y costo:

1. Caché de embeddings: Hashear el texto de entrada, cachear el vector de embedding. Evita re-generar embeddings de consultas idénticas.
2. Caché semántico: Para consultas con similitud coseno > 0.95 respecto a una consulta cacheada, devolver la respuesta cacheada. Reduce llamadas al LLM en un 20-40% en casos de uso de soporte donde preguntas similares se repiten.

Monitoreo

Rastrear estas métricas en producción:

• Latencia de retrieval (p50, p95, p99) — objetivo < 100ms para híbrida + rerank
• Score de faithfulness sobre una muestra rotativa — alertar si cae por debajo de 0.85
• Tasa de retrieval vacío — consultas donde ningún chunk supera el umbral de relevancia
• Señales de feedback del usuario — thumbs up/down, eventos de copia, preguntas de seguimiento

Cuándo RAG No Es Suficiente

RAG tiene límites bien definidos:

• Razonamiento multi-hop: Cuando la respuesta requiere sintetizar información de 5+ documentos con pasos inferenciales entre ellos, el retrieval de RAG frecuentemente pierde documentos intermedios.
• Razonamiento temporal: "¿Qué cambió entre Q3 y Q4?" requiere recuperar y comparar dos conjuntos de documentos específicos en el tiempo — el retrieval top-k estándar no está diseñado para esto.
• Cómputo sobre datos: "¿Cuál es el SLA promedio en todas las regiones?" requiere una consulta estructurada, no recuperación de texto.

Para estos casos, considerá RAG agéntico — un agente que planifica pasos de retrieval, ejecuta múltiples búsquedas y sintetiza resultados programáticamente — o enfoques híbridos que combinen RAG con consultas SQL y llamadas a APIs.