Introducción: Del RAG Estático al RAG que Piensa
Si has trabajado con Retrieval-Augmented Generation durante los últimos dos años, probablemente hayas vivido esta experiencia: construyes un pipeline RAG que funciona de maravilla en tus pruebas, lo despliegas en producción y, de repente, se rompe con preguntas que no encajan en el molde. El usuario hace una consulta ambigua, el retriever devuelve documentos irrelevantes y el LLM, con toda la confianza del mundo, genera una respuesta perfectamente articulada que es completamente incorrecta. Suena familiar, ¿verdad?
El problema no está en el modelo ni en tu base de datos vectorial. El problema es que el RAG tradicional es un pipeline estático — una secuencia lineal de pasos predefinidos que no tiene capacidad de adaptarse, corregirse o razonar sobre lo que está haciendo. Recupera, genera y entrega. Si lo recuperado es basura, la generación también lo será. Así de simple.
El RAG Agéntico cambia fundamentalmente esta dinámica. En lugar de seguir un pipeline rígido, coloca un agente inteligente — un LLM con capacidad de razonamiento y acceso a herramientas — en el centro del proceso de recuperación. Este agente puede decidir si necesita buscar más información, reescribir la consulta del usuario para obtener mejores resultados, verificar la relevancia de los documentos recuperados, combinar múltiples fuentes de datos y, crucialmente, autocorregirse cuando detecta que algo no va bien.
Bueno, en este artículo vamos a meternos de lleno en cómo construir sistemas RAG agénticos robustos y listos para producción. Cubriremos la arquitectura completa, los patrones de diseño fundamentales, implementaciones con código real usando LangGraph y las estrategias de chunking y recuperación que están definiendo el estado del arte en 2026. Si ya has leído nuestros artículos sobre ingeniería de contexto y sistemas multi-agente, considera este la pieza que conecta ambos mundos.
¿Por Qué el RAG Tradicional Se Queda Corto?
Antes de sumergirnos en las soluciones, vale la pena entender exactamente dónde fallan los pipelines RAG convencionales. No se trata de que no funcionen — funcionan, y bastante bien para muchos casos de uso simples. El problema surge cuando escalamos en complejidad.
Las cinco fragilidades del RAG estático
- Recuperación ciega: El pipeline siempre recupera documentos, incluso cuando la pregunta no lo requiere. Si el usuario pregunta "¿Qué hora es?", el sistema igualmente ejecuta una búsqueda vectorial, desperdiciando latencia y potencialmente contaminando el contexto con información irrelevante.
- Sin evaluación de calidad: Los documentos recuperados se pasan directamente al LLM sin ninguna verificación. No hay mecanismo para detectar que los resultados del retriever son de baja calidad o completamente irrelevantes. Es como entregar un examen sin revisarlo.
- Consultas complejas irresolubles: Preguntas que requieren múltiples pasos de razonamiento — como "Compara las estrategias de pricing de nuestros tres principales competidores y analiza las tendencias del último trimestre" — son intratables para un pipeline de un solo paso.
- Fuente única de conocimiento: El RAG estático típicamente consulta una sola base de datos vectorial. No tiene la capacidad de decidir dinámicamente si debería consultar una API, una base de datos SQL, un buscador web o una combinación de fuentes.
- Propagación de errores sin freno: Si el retriever falla, el error se propaga al generador sin ningún mecanismo de contención. No hay retry inteligente, no hay fallback, no hay autocorrección. Nada.
El cambio de paradigma agéntico
El RAG agéntico reconoce que la recuperación de información no es un paso mecánico sino una tarea de razonamiento. Igual que un investigador humano no se limita a buscar en una sola fuente y aceptar el primer resultado, un agente RAG puede:
- Planificar su estrategia de recuperación antes de ejecutarla.
- Evaluar la calidad y relevancia de lo recuperado.
- Reescribir las consultas cuando los resultados son insatisfactorios.
- Orquestar múltiples fuentes de datos de forma dinámica.
- Verificar que la respuesta generada está respaldada por la evidencia.
Y los números lo respaldan. Según datos recientes de la industria, las empresas que han migrado de RAG estático a RAG agéntico reportan mejoras del 25-40% en la precisión de las respuestas y reducciones significativas en las tasas de alucinación, particularmente en consultas complejas y multi-dominio. No son números menores.
Arquitectura del RAG Agéntico: Los Componentes Clave
Un sistema RAG agéntico se compone de varios módulos que trabajan dentro de un bucle de razonamiento controlado por un agente. Vamos a desglosar cada componente.
El Router: Primera línea de decisión
El router es el primer punto de contacto. Analiza la consulta del usuario y decide qué camino seguir: ¿necesita recuperación de documentos? ¿Es una pregunta conversacional que no requiere contexto externo? ¿Debe redirigirse a una herramienta específica?
from typing import Literal
from pydantic import BaseModel, Field
class DecisionRouter(BaseModel):
"""Modelo de decisión para el enrutamiento de consultas."""
accion: Literal["recuperar", "responder_directo", "buscar_web", "consultar_api"] = Field(
description="Acción a tomar basada en el análisis de la consulta"
)
razonamiento: str = Field(
description="Explicación breve de por qué se eligió esta acción"
)
consulta_reformulada: str = Field(
description="Versión optimizada de la consulta para recuperación"
)
PROMPT_ROUTER = """Eres un router inteligente para un sistema RAG.
Analiza la consulta del usuario y decide la mejor estrategia:
- "recuperar": La consulta requiere información de la base de conocimiento interna.
- "responder_directo": La consulta es conversacional o de conocimiento general.
- "buscar_web": La consulta requiere información actualizada no disponible internamente.
- "consultar_api": La consulta requiere datos en tiempo real de sistemas externos.
Si decides recuperar, reformula la consulta para maximizar la relevancia
de los resultados de búsqueda vectorial.
Consulta del usuario: {query}
"""El Retriever Híbrido: Más allá de la búsqueda vectorial
A estas alturas de 2026, la búsqueda puramente vectorial ya no es suficiente para sistemas de producción serios. Los retrievers más efectivos combinan múltiples estrategias de recuperación — lo que se conoce como recuperación híbrida.
La idea es bastante intuitiva: combinas búsqueda semántica (basada en embeddings vectoriales) con búsqueda léxica (BM25/TF-IDF). La búsqueda semántica captura el significado conceptual, mientras que la léxica es excelente para coincidencias exactas de términos técnicos, nombres propios y acrónimos que los modelos de embeddings a veces manejan mal (y créeme, lo manejan mal más a menudo de lo que pensarías).
from dataclasses import dataclass, field
@dataclass
class ResultadoRecuperacion:
"""Resultado individual de una búsqueda."""
contenido: str
fuente: str
puntuacion: float
metadatos: dict = field(default_factory=dict)
class RetrieverHibrido:
"""
Combina búsqueda vectorial semántica con búsqueda léxica BM25.
Usa Reciprocal Rank Fusion (RRF) para unificar rankings.
"""
def __init__(self, vector_store, bm25_index, peso_semantico: float = 0.6):
self.vector_store = vector_store
self.bm25_index = bm25_index
self.peso_semantico = peso_semantico
self.peso_lexico = 1.0 - peso_semantico
def recuperar(self, consulta: str, top_k: int = 10) -> list[ResultadoRecuperacion]:
"""Ejecuta búsqueda híbrida y fusiona resultados con RRF."""
# Búsqueda semántica
resultados_semanticos = self.vector_store.similarity_search(
consulta, k=top_k * 2
)
# Búsqueda léxica BM25
resultados_lexicos = self.bm25_index.search(consulta, top_k=top_k * 2)
# Fusión con Reciprocal Rank Fusion
return self._fusionar_rrf(
resultados_semanticos, resultados_lexicos, top_k
)
def _fusionar_rrf(self, sem: list, lex: list, top_k: int, k: int = 60) -> list:
"""
Reciprocal Rank Fusion: combina rankings de múltiples fuentes.
score(d) = sum(1 / (k + rank_i(d))) para cada sistema i
"""
puntuaciones = {}
for rank, doc in enumerate(sem):
doc_id = doc.metadata.get("id", doc.page_content[:50])
puntuaciones[doc_id] = puntuaciones.get(doc_id, 0)
puntuaciones[doc_id] += self.peso_semantico * (1.0 / (k + rank + 1))
for rank, doc in enumerate(lex):
doc_id = doc.get("id", doc["content"][:50])
puntuaciones[doc_id] = puntuaciones.get(doc_id, 0)
puntuaciones[doc_id] += self.peso_lexico * (1.0 / (k + rank + 1))
# Ordenar por puntuación fusionada y devolver top_k
ranking_final = sorted(
puntuaciones.items(), key=lambda x: x[1], reverse=True
)[:top_k]
return ranking_finalEl Evaluador de Relevancia: El filtro inteligente
Aquí es donde el RAG agéntico se diferencia radicalmente del estático. Después de recuperar documentos, un evaluador — otro LLM o un modelo especializado — examina cada documento y determina si realmente es relevante para la consulta. Esto es algo que el RAG clásico simplemente no hace, y honestamente, es un descuido bastante grande.
from pydantic import BaseModel, Field
class EvaluacionDocumento(BaseModel):
"""Resultado de la evaluación de relevancia de un documento."""
es_relevante: bool = Field(description="Si el documento es relevante para la consulta")
puntuacion_confianza: float = Field(
description="Nivel de confianza de 0.0 a 1.0", ge=0.0, le=1.0
)
justificacion: str = Field(description="Breve explicación de la evaluación")
PROMPT_EVALUADOR = """Eres un evaluador de relevancia documental.
Tu tarea es determinar si el documento recuperado es relevante
para responder la consulta del usuario.
Evalúa considerando:
1. ¿El documento contiene información directamente relacionada con la consulta?
2. ¿La información es suficientemente específica para ser útil?
3. ¿La información está actualizada y es confiable?
Consulta del usuario: {query}
Documento recuperado:
{document}
Proporciona tu evaluación con puntuación de confianza y justificación.
"""Este evaluador actúa como un filtro de calidad que previene la contaminación del contexto. Si ningún documento supera el umbral de relevancia, el sistema puede activar estrategias de corrección — reescribir la consulta, buscar en fuentes alternativas, o ambas cosas.
CRAG: RAG Correctivo — El Patrón de Autocorrección
El RAG Correctivo (CRAG), propuesto en un paper de investigación que se ha convertido en referencia obligada, es posiblemente el patrón más importante del RAG agéntico. Su principio es simple pero poderoso: antes de generar una respuesta, verifica si lo recuperado es suficientemente bueno; si no lo es, corrígelo.
Parece obvio cuando lo lees así, ¿no? Pero la realidad es que la inmensa mayoría de los sistemas RAG en producción todavía no implementan este paso.
Los tres escenarios de CRAG
El evaluador de recuperación clasifica cada conjunto de resultados en uno de tres escenarios:
- Correcto: Al menos un documento tiene alta relevancia. Se procede a la generación directamente, refinando el documento para extraer solo los fragmentos más relevantes.
- Incorrecto: Ningún documento alcanza el umbral de relevancia. El sistema activa una búsqueda web o consulta fuentes alternativas para obtener información relevante.
- Ambiguo: Los resultados son mixtos — algunos documentos parcialmente relevantes. Se combina el refinamiento de documentos internos con búsqueda web complementaria.
Refinamiento de conocimiento: Decompose-then-Recompose
CRAG no pasa documentos completos al generador. En su lugar, aplica un proceso de descomposición y recomposición: divide los documentos en fragmentos más pequeños, filtra los que no son relevantes para la consulta específica y reconstruye un contexto limpio y enfocado. En la práctica, esto reduce drásticamente el ruido y mejora la precisión de la generación de forma notable.
class RefinadorConocimiento:
"""
Implementa el patrón Decompose-then-Recompose de CRAG.
Descompone documentos, filtra fragmentos irrelevantes
y recompone un contexto optimizado.
"""
def __init__(self, llm_client, umbral_relevancia: float = 0.5):
self.llm = llm_client
self.umbral = umbral_relevancia
async def refinar(self, documentos: list[str], consulta: str) -> str:
"""Refina documentos eliminando fragmentos irrelevantes."""
fragmentos_relevantes = []
for doc in documentos:
# Descomponer en fragmentos granulares
fragmentos = self._descomponer(doc)
# Evaluar relevancia de cada fragmento
for fragmento in fragmentos:
relevancia = await self._evaluar_fragmento(fragmento, consulta)
if relevancia >= self.umbral:
fragmentos_relevantes.append(fragmento)
# Recomponer un contexto limpio y coherente
return self._recomponer(fragmentos_relevantes)
def _descomponer(self, documento: str) -> list[str]:
"""Divide un documento en proposiciones atómicas o párrafos."""
# Dividir por párrafos o proposiciones semánticas
parrafos = [p.strip() for p in documento.split("\n\n") if p.strip()]
fragmentos = []
for parrafo in parrafos:
# Si el párrafo es muy largo, subdividir por oraciones
if len(parrafo.split()) > 100:
oraciones = parrafo.split(". ")
fragmentos.extend(
[o.strip() + "." for o in oraciones if o.strip()]
)
else:
fragmentos.append(parrafo)
return fragmentos
def _recomponer(self, fragmentos: list[str]) -> str:
"""Recompone fragmentos filtrados en un contexto coherente."""
return "\n\n".join(fragmentos)Implementación Completa con LangGraph
LangGraph se ha consolidado como el framework de referencia para construir sistemas RAG agénticos en 2026. Su modelo basado en grafos de estado permite definir flujos complejos con ciclos, bifurcaciones y puntos de control de forma declarativa. Vamos a ver cómo se implementa paso a paso.
Paso 1: Definir el estado del grafo
El estado es la estructura de datos que fluye a través de todos los nodos del grafo. Cada nodo puede leer y modificar el estado — piensa en ello como una pizarra compartida donde cada componente del sistema anota sus resultados.
from typing import TypedDict, Annotated
from langchain_core.documents import Document
from langgraph.graph.message import add_messages
class EstadoRAGAgentico(TypedDict):
"""Estado compartido del sistema RAG agéntico."""
# Mensajes de la conversación (se acumulan)
mensajes: Annotated[list, add_messages]
# Consulta original del usuario
consulta: str
# Consulta reformulada para optimizar recuperación
consulta_optimizada: str
# Documentos recuperados
documentos: list[Document]
# Resultado de la evaluación de relevancia
evaluacion: str # "correcto", "incorrecto", "ambiguo"
# Respuesta generada
respuesta: str
# Contador de intentos de recuperación
intentos_recuperacion: int
# Límite máximo de reintentos
max_intentos: intPaso 2: Implementar los nodos del grafo
Cada nodo representa una función que transforma el estado. Los nodos clave son: el recuperador, el evaluador, el reescritor de consultas y el generador. Aquí es donde vive la lógica real del sistema.
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
# Inicializar el LLM
llm = ChatAnthropic(model="claude-sonnet-4-5-20250929", temperature=0)
def nodo_recuperar(estado: EstadoRAGAgentico) -> dict:
"""Recupera documentos de la base de conocimiento."""
consulta = estado.get("consulta_optimizada") or estado["consulta"]
# Usar el retriever híbrido configurado previamente
documentos = retriever.invoke(consulta)
return {
"documentos": documentos,
"intentos_recuperacion": estado.get("intentos_recuperacion", 0) + 1
}
def nodo_evaluar(estado: EstadoRAGAgentico) -> dict:
"""Evalúa la relevancia de los documentos recuperados."""
consulta = estado["consulta"]
documentos = estado["documentos"]
if not documentos:
return {"evaluacion": "incorrecto"}
prompt = ChatPromptTemplate.from_template(
"""Evalúa si los siguientes documentos son relevantes para la consulta.
Consulta: {consulta}
Documentos:
{documentos}
Responde SOLO con una palabra: "correcto", "incorrecto" o "ambiguo".
- "correcto": al menos un documento es altamente relevante
- "incorrecto": ningún documento es relevante
- "ambiguo": relevancia parcial o mixta"""
)
cadena = prompt | llm
resultado = cadena.invoke({
"consulta": consulta,
"documentos": "\n---\n".join([d.page_content for d in documentos])
})
evaluacion = resultado.content.strip().lower()
if evaluacion not in ("correcto", "incorrecto", "ambiguo"):
evaluacion = "ambiguo"
return {"evaluacion": evaluacion}
def nodo_reescribir(estado: EstadoRAGAgentico) -> dict:
"""Reescribe la consulta para mejorar la recuperación."""
prompt = ChatPromptTemplate.from_template(
"""La siguiente consulta no obtuvo resultados relevantes.
Reescríbela para mejorar los resultados de búsqueda.
Consulta original: {consulta}
Intento número: {intentos}
Genera una versión reformulada que:
- Use sinónimos o términos alternativos
- Sea más específica o más general según convenga
- Descomponga consultas complejas en conceptos clave"""
)
cadena = prompt | llm
resultado = cadena.invoke({
"consulta": estado["consulta"],
"intentos": estado["intentos_recuperacion"]
})
return {"consulta_optimizada": resultado.content.strip()}
def nodo_generar(estado: EstadoRAGAgentico) -> dict:
"""Genera la respuesta final basada en documentos verificados."""
prompt = ChatPromptTemplate.from_template(
"""Basándote EXCLUSIVAMENTE en los siguientes documentos,
responde la consulta del usuario. Si la información es insuficiente,
indícalo claramente.
Documentos verificados:
{documentos}
Consulta: {consulta}
Genera una respuesta precisa, citando las fuentes cuando sea posible."""
)
cadena = prompt | llm
resultado = cadena.invoke({
"consulta": estado["consulta"],
"documentos": "\n---\n".join(
[d.page_content for d in estado["documentos"]]
)
})
return {"respuesta": resultado.content}Paso 3: Construir el grafo con lógica condicional
Aquí viene la parte que hace que todo cobre sentido. La magia del RAG agéntico está en las aristas condicionales — las decisiones que determinan qué nodo se ejecuta a continuación basándose en el estado actual.
from langgraph.graph import StateGraph, START, END
def decidir_siguiente_paso(estado: EstadoRAGAgentico) -> str:
"""Decide el siguiente paso basándose en la evaluación."""
evaluacion = estado["evaluacion"]
intentos = estado.get("intentos_recuperacion", 0)
max_intentos = estado.get("max_intentos", 3)
if evaluacion == "correcto":
return "generar"
elif intentos >= max_intentos:
# Evitar bucles infinitos: generar con lo que hay
return "generar"
else:
# Reescribir y reintentar
return "reescribir"
# Construir el grafo
grafo = StateGraph(EstadoRAGAgentico)
# Agregar nodos
grafo.add_node("recuperar", nodo_recuperar)
grafo.add_node("evaluar", nodo_evaluar)
grafo.add_node("reescribir", nodo_reescribir)
grafo.add_node("generar", nodo_generar)
# Definir flujo
grafo.add_edge(START, "recuperar")
grafo.add_edge("recuperar", "evaluar")
grafo.add_conditional_edges(
"evaluar",
decidir_siguiente_paso,
{
"generar": "generar",
"reescribir": "reescribir"
}
)
grafo.add_edge("reescribir", "recuperar") # Bucle de corrección
grafo.add_edge("generar", END)
# Compilar con checkpoint para persistencia
from langgraph.checkpoint.memory import MemorySaver
app = grafo.compile(checkpointer=MemorySaver())Observa cómo el grafo implementa un bucle de corrección: si la evaluación determina que los documentos no son relevantes, el sistema reescribe la consulta y vuelve a recuperar. Y un detalle importante: este bucle tiene un límite configurable de intentos para evitar que se quede dando vueltas indefinidamente. Es un detalle pequeño, pero en producción marca la diferencia entre un sistema que funciona y uno que te genera facturas astronómicas.
Estrategias de Chunking: La Base que Todo lo Sostiene
Por muy sofisticado que sea tu agente, si los chunks en tu base de datos vectorial están mal diseñados, el sistema entero se resiente. El chunking es, literalmente, la base sobre la que se construye todo lo demás.
La paradoja del chunking en 2026
Cuando FloTorch publicó su estudio de benchmark de RAG en febrero de 2026, los resultados sacudieron a la comunidad. Después de probar siete estrategias de chunking diferentes sobre miles de documentos académicos, el ganador no fue el sofisticado chunking semántico ni el enfoque basado en proposiciones con LLMs. Las estrategias más simples — recursive character splitting y fixed-size chunking a 512 tokens — superaron consistentemente a los métodos más complejos, y además son dramáticamente más baratas de implementar y mantener.
Sí, leíste bien. Lo simple ganó.
¿Qué significa esto en la práctica? Que no deberías usar chunking semántico solo porque "suena más sofisticado". A menos que tengas evidencia específica de que tu corpus se beneficia de fronteras semánticas, comienza con lo simple y mide. Si recursive character splitting te da buenos resultados (y probablemente lo hará), no hay razón para complicarte la vida.
Las tres estrategias que debes conocer
1. Recursive Character Splitting — El caballo de batalla
Divide el texto recursivamente usando una jerarquía de separadores (párrafos, oraciones, palabras) hasta alcanzar el tamaño objetivo. Respeta las fronteras naturales del texto cuando es posible, lo cual es una gran ventaja frente al splitting fijo.
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=64, # ~12% de solapamiento
separators=["\n\n", "\n", ". ", " ", ""],
length_function=len,
)
chunks = splitter.split_documents(documentos)
print(f"Documentos originales: {len(documentos)}")
print(f"Chunks generados: {len(chunks)}")
print(f"Tamaño promedio: {sum(len(c.page_content) for c in chunks) / len(chunks):.0f} caracteres")2. Parent-Child (Indexación jerárquica) — Lo mejor de ambos mundos
Esta es mi estrategia favorita para documentos técnicos largos. La idea es crear dos niveles de chunks: los child chunks (pequeños, 128-256 tokens) se usan para la búsqueda, mientras que los parent chunks (grandes, 1024-2048 tokens) se usan para la generación. Cuando un child chunk coincide con una consulta, el sistema recupera el parent completo, proporcionando contexto suficiente al LLM.
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.storage import InMemoryStore
from langchain.retrievers import ParentDocumentRetriever
# Splitter para chunks padre (contexto amplio)
splitter_padre = RecursiveCharacterTextSplitter(chunk_size=2000)
# Splitter para chunks hijo (precisión en búsqueda)
splitter_hijo = RecursiveCharacterTextSplitter(chunk_size=256)
store = InMemoryStore()
retriever_jerarquico = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=store,
child_splitter=splitter_hijo,
parent_splitter=splitter_padre,
)
# Al indexar, se crean ambos niveles automáticamente
retriever_jerarquico.add_documents(documentos)3. Chunking con metadatos enriquecidos — Contexto sin costo de tokens
Cada chunk se enriquece con metadatos que el retriever puede usar para filtrar y priorizar resultados sin necesidad de cargar el contexto completo del LLM. Es una optimización sutil pero que escala muy bien.
def enriquecer_chunks(chunks: list, doc_original: dict) -> list:
"""Añade metadatos contextuales a cada chunk."""
total = len(chunks)
for i, chunk in enumerate(chunks):
chunk.metadata.update({
"titulo_documento": doc_original.get("titulo", ""),
"seccion": doc_original.get("seccion_actual", ""),
"posicion": f"{i+1}/{total}",
"fecha_creacion": doc_original.get("fecha", ""),
"categoria": doc_original.get("categoria", ""),
"tokens_estimados": len(chunk.page_content.split()),
})
return chunksConfiguración óptima de chunks
Basándonos en los benchmarks más recientes, estas son las recomendaciones que te daría para empezar:
- Tamaño de chunk: 256-512 tokens para documentos técnicos. 512-1024 para narrativas largas.
- Solapamiento: 10-15% del tamaño del chunk. Suficiente para mantener coherencia sin redundancia excesiva.
- Modelo de embeddings: En 2026, los modelos de la familia Qwen3 y los embeddings de OpenAI (text-embedding-3-large) ofrecen la mejor relación calidad-precio. Para casos de uso en español, los modelos multilinguales de Cohere funcionan particularmente bien.
Re-ranking: El Multiplicador de Precisión
Si tuvieras que elegir una sola técnica para mejorar tu pipeline RAG existente, el re-ranking sería probablemente la mejor inversión. Y no lo digo a la ligera.
El concepto es sencillo: después de la recuperación inicial (top-K amplio), un modelo de re-ranking reordena los resultados para priorizar los más relevantes.
¿Por qué funciona tan bien? Porque los modelos de recuperación (bi-encoders) están optimizados para velocidad — generan embeddings independientes para la consulta y los documentos. Los modelos de re-ranking (cross-encoders) son más lentos pero significativamente más precisos — procesan la consulta y el documento juntos, capturando interacciones semánticas que los bi-encoders simplemente no pueden detectar.
from sentence_transformers import CrossEncoder
class Reranker:
"""
Re-ranking de documentos usando un cross-encoder.
Aplicar después de la recuperación inicial para refinar el ranking.
"""
def __init__(self, modelo: str = "cross-encoder/ms-marco-MiniLM-L-12-v2"):
self.modelo = CrossEncoder(modelo)
def reordenar(
self,
consulta: str,
documentos: list,
top_k: int = 5
) -> list:
"""Reordena documentos por relevancia usando cross-encoder."""
pares = [(consulta, doc.page_content) for doc in documentos]
puntuaciones = self.modelo.predict(pares)
# Emparejar documentos con puntuaciones y ordenar
docs_puntuados = list(zip(documentos, puntuaciones))
docs_puntuados.sort(key=lambda x: x[1], reverse=True)
return [doc for doc, _ in docs_puntuados[:top_k]]
# Uso en el pipeline
reranker = Reranker()
docs_recuperados = retriever.invoke(consulta, k=20) # Recuperar amplio
docs_reranked = reranker.reordenar(consulta, docs_recuperados, top_k=5) # RefinarLa estrategia de recuperación en dos fases — recuperación amplia seguida de re-ranking — ya se considera práctica estándar en sistemas RAG de producción. Lo habitual es recuperar 20-50 documentos inicialmente y luego quedarte con los 3-5 más relevantes después del re-ranking.
Observabilidad y Monitoreo: No Vueles a Ciegas
Un sistema RAG en producción sin observabilidad es como pilotar un avión sin instrumentos. Necesitas visibilidad sobre cada paso del pipeline para diagnosticar problemas, optimizar el rendimiento y mantener la calidad a lo largo del tiempo. He visto equipos perder semanas enteras depurando problemas que habrían sido evidentes con las métricas adecuadas.
Métricas clave que debes rastrear
- Precisión de recuperación: ¿Los documentos recuperados son relevantes? Mide el porcentaje de documentos que pasan el evaluador de relevancia.
- Tasa de corrección: ¿Con qué frecuencia el sistema necesita reescribir consultas y reintentar? Una tasa alta indica problemas en el chunking o los embeddings.
- Latencia por componente: Desglosa el tiempo total en recuperación, evaluación, re-ranking y generación. Identifica los cuellos de botella antes de que tus usuarios los noten.
- Faithfulness (fidelidad): ¿La respuesta generada está respaldada por los documentos recuperados? Esta es, con diferencia, la métrica más importante para detectar alucinaciones.
- Throughput del vector store: Monitorea la latencia de consultas y la tasa de solicitudes. Con grandes volúmenes de vectores, mantener los índices en memoria puede ser sorprendentemente costoso.
import time
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class MetricasRAG:
"""Recolección de métricas para observabilidad del pipeline RAG."""
timestamp: datetime = field(default_factory=datetime.utcnow)
consulta: str = ""
latencia_recuperacion_ms: float = 0.0
latencia_evaluacion_ms: float = 0.0
latencia_reranking_ms: float = 0.0
latencia_generacion_ms: float = 0.0
documentos_recuperados: int = 0
documentos_relevantes: int = 0
intentos_recuperacion: int = 1
evaluacion_resultado: str = ""
@property
def latencia_total_ms(self) -> float:
return (
self.latencia_recuperacion_ms
+ self.latencia_evaluacion_ms
+ self.latencia_reranking_ms
+ self.latencia_generacion_ms
)
@property
def tasa_relevancia(self) -> float:
if self.documentos_recuperados == 0:
return 0.0
return self.documentos_relevantes / self.documentos_recuperados
def to_dict(self) -> dict:
return {
"timestamp": self.timestamp.isoformat(),
"consulta": self.consulta,
"latencia_total_ms": self.latencia_total_ms,
"tasa_relevancia": self.tasa_relevancia,
"intentos": self.intentos_recuperacion,
"evaluacion": self.evaluacion_resultado,
}Arquitecturas RAG Avanzadas para Casos Específicos
No todos los problemas se resuelven con la misma arquitectura. Dependiendo de la complejidad de tus datos y consultas, diferentes variantes del RAG agéntico pueden ser más apropiadas.
Graph RAG: Cuando las relaciones importan
Para dominios donde las relaciones entre entidades son fundamentales — organigramas empresariales, redes de conocimiento científico, cadenas de suministro — el Graph RAG combina búsqueda vectorial con traversal de grafos de conocimiento. Cuando un agente recupera un chunk que menciona "la adquisición", puede navegar el grafo para encontrar las empresas involucradas, la fecha, el valor y el contexto completo. Es el tipo de información que la recuperación pura de chunks simplemente no puede proporcionar.
RAG Modular: Componentes intercambiables
La arquitectura RAG Modular trata cada componente del pipeline como un bloque intercambiable. ¿Necesitas cambiar tu modelo de embeddings? ¿Añadir una nueva fuente de datos? ¿Actualizar la estrategia de re-ranking? Lo haces sin reconstruir todo el sistema. En 2026, frameworks como LlamaIndex y LangChain han adoptado este enfoque de lleno, permitiendo configuraciones declarativas donde los componentes se enchufan y se desenchufan según las necesidades del momento.
Multi-Agent RAG: Investigadores especializados
Para consultas realmente complejas que requieren información de múltiples dominios, el patrón Multi-Agent RAG despliega múltiples agentes especializados en paralelo. Cada agente recupera y procesa información de su dominio específico, y un agente sintetizador combina todos los resultados en una respuesta cohesiva. Si te interesa profundizar en este patrón, conecta directamente con los sistemas multi-agente que exploramos en artículos anteriores.
Consideraciones de Costos y Escalabilidad
Un punto que a menudo se pasa por alto al diseñar sistemas RAG agénticos es el costo. Y es algo que hay que hablar sin rodeos: cada bucle de corrección — cada reescritura de consulta, cada evaluación de relevancia — implica llamadas adicionales al LLM. En producción, esto escala más rápido de lo que imaginas.
Estrategias de optimización de costos
- Modelos escalonados: Usa modelos pequeños y rápidos (como Claude Haiku o GPT-4o mini) para tareas de evaluación y routing, y reserva los modelos más potentes para la generación final. La diferencia en costo puede ser de 10x o más.
- Caché de resultados: Implementa caché para consultas frecuentes. Si la misma pregunta (o una semánticamente similar) se ha respondido recientemente, sirve el resultado cacheado. Parece básico, pero te sorprendería cuántos sistemas en producción no lo implementan.
- Límites de bucle agresivos: Configura un máximo de 2-3 intentos de corrección. La evidencia empírica muestra que si el sistema no encuentra documentos relevantes después de 3 reescrituras, es probable que la información simplemente no exista en la base de conocimiento.
- Evaluación por lotes: En lugar de evaluar documentos uno por uno, envía lotes al evaluador en una sola llamada. Reduce la latencia total y el overhead de llamadas a la API.
Y aquí va un dato que poca gente menciona: según análisis de infraestructura publicados en 2026, el factor de costo más significativo en despliegues RAG no son las llamadas a la API del LLM ni el almacenamiento en la base de datos vectorial — es mantener los índices de memoria en RAM para lograr tiempos de consulta inferiores a 50ms. Este requisito escala con el número de vectores, no con su sofisticación.
Checklist para Producción
Antes de desplegar tu sistema RAG agéntico en producción, asegúrate de cubrir estos puntos. Te ahorrarán dolores de cabeza:
- Estrategia de chunking validada: Probaste al menos 2-3 estrategias con datos reales y métricas de recuperación.
- Búsqueda híbrida implementada: Combinas semántica con léxica, no dependes solo de vectores.
- Re-ranking activo: Los documentos pasan por un cross-encoder antes de llegar al generador.
- Límites de bucle configurados: El agente tiene un máximo de intentos de corrección para evitar ciclos infinitos.
- Observabilidad completa: Cada paso del pipeline tiene logging con métricas de latencia y relevancia.
- Fallback graceful: Si el sistema no puede recuperar información relevante, comunica esto claramente al usuario en lugar de alucinar.
- Tests de regresión: Tienes un conjunto de consultas de referencia con respuestas esperadas que validas después de cada cambio.
- Gestión de costos: Monitorizas el gasto por consulta y tienes alertas para anomalías.
Conclusión: El RAG como Infraestructura Inteligente
El RAG agéntico no es simplemente una mejora incremental sobre el RAG estático — es un cambio de paradigma en cómo pensamos sobre la recuperación de información para LLMs. Al colocar un agente inteligente en el centro del pipeline, transformamos un proceso mecánico y frágil en un sistema adaptativo capaz de razonar, corregirse y evolucionar.
Los puntos clave que me gustaría que te llevaras de este artículo:
- El RAG estático falla ante consultas complejas, ambiguas o multi-dominio. El RAG agéntico aborda estas limitaciones con bucles de corrección y razonamiento dinámico.
- CRAG (RAG Correctivo) es un patrón fundamental: evalúa antes de generar, y corrige cuando la recuperación es insuficiente.
- Las estrategias de chunking simples frecuentemente superan a las sofisticadas. No asumas que más complejidad equivale a mejores resultados.
- La recuperación híbrida (vectorial + léxica) con re-ranking es el estándar de producción en 2026.
- La observabilidad no es opcional. Sin métricas, estás navegando a ciegas.
En el contexto más amplio del ecosistema de IA, el RAG agéntico se sitúa en la intersección de tres disciplinas que hemos explorado en esta serie: la ingeniería de contexto determina qué información alimenta al agente, los sistemas multi-agente proporcionan los patrones de orquestación, y el RAG agéntico conecta ambos mundos con la infraestructura de recuperación inteligente. Juntas, estas tres disciplinas forman el stack completo para construir aplicaciones de IA que realmente funcionan en producción.