Introducción
La Generación Aumentada por Recuperación comenzó como una idea sencilla: en lugar de depender únicamente de la memoria paramétrica de un modelo de lenguaje, recuperar documentos relevantes de una base de conocimiento externa e incluirlos en el contexto del prompt. Este patrón de RAG naíve, que consiste en embeber-recuperar-generar, funcionó sorprendentemente bien para responder preguntas simples sobre colecciones de documentos estructurados. Pero a medida que los equipos llevaron RAG a producción para casos de uso más complejos, sus limitaciones se hicieron dolorosamente evidentes. Las consultas que requerían razonamiento a través de múltiples documentos, las preguntas ambiguas que necesitaban clarificación y las bases de conocimiento con tipos de contenido heterogéneos expusieron la fragilidad del pipeline de recuperar-y-generar.
El RAG agéntico representa un cambio fundamental en cómo arquitectamos los sistemas de recuperación. En lugar de un pipeline fijo donde cada paso alimenta linealmente al siguiente, el RAG agéntico le da al modelo de lenguaje la capacidad de planificar su estrategia de recuperación, evaluar la calidad de los resultados recuperados, reformular consultas cuando la recuperación inicial falla y decidir cuándo tiene suficiente información para generar una respuesta final. El modelo se convierte en un participante activo del proceso de recuperación en lugar de un consumidor pasivo del contexto recuperado.
Esta guía cubre los patrones de arquitectura, frameworks de implementación, métodos de evaluación y consideraciones de producción para construir sistemas RAG agénticos. Nos basamos en despliegues del mundo real en bases de conocimiento empresariales, sistemas de atención al cliente y plataformas de documentación técnica donde estos patrones han sido validados a escala.
De RAG Naíve a RAG Agéntico: Qué Cambió
El RAG naíve sigue un pipeline determinístico de tres pasos: la consulta del usuario se embebe, una búsqueda de similitud vectorial recupera los top-k fragmentos de documentos más similares, y esos fragmentos se concatenan en un prompt que el LLM utiliza para generar una respuesta. Este pipeline tiene tres debilidades fundamentales que el RAG agéntico aborda.
Primero, el RAG naíve asume que la consulta del usuario ya está bien formulada para la recuperación. En la práctica, las consultas de los usuarios son frecuentemente vagas, multifacéticas o usan terminología que no coincide con el vocabulario de los documentos indexados. Un usuario preguntando sobre fallos de despliegue podría necesitar documentos sobre manejo de errores, configuración de infraestructura y pipelines de CI/CD, pero una sola búsqueda vectorial puede arrojar solo una de estas facetas.
Segundo, el RAG naíve no tiene puerta de calidad. Si los documentos recuperados son irrelevantes, obsoletos o insuficientes, el pipeline continúa de todos modos y el LLM genera una respuesta a partir de un contexto deficiente. No hay mecanismo para que el sistema reconozca un fallo de recuperación e intente de nuevo.
Tercero, el RAG naíve trata todas las consultas de forma idéntica. Una pregunta de búsqueda factual, una pregunta analítica compleja y una pregunta que requiere síntesis de múltiples fuentes todas pasan por el mismo pipeline de recuperar-generar. El RAG agéntico introduce lógica condicional que selecciona diferentes estrategias de recuperación y generación basándose en las características de la consulta.
La transición de RAG naíve a agéntico implica añadir tres capacidades: planificación de consultas (descomponer consultas complejas en sub-consultas), evaluación de recuperación (evaluar si el contexto recuperado es suficiente) y refinamiento iterativo (reformular consultas y re-recuperar cuando los resultados son insuficientes). Estas capacidades transforman un pipeline estático en un sistema dinámico y autocorrectivo.
Patrones de Arquitectura Central
Los sistemas RAG agénticos se construyen a partir de un pequeño número de patrones componibles. Comprender estos patrones te permite diseñar sistemas adaptados a tu caso de uso específico.
Enrutamiento
El patrón agéntico más simple enruta las consultas a diferentes backends de recuperación basándose en la clasificación de la consulta. Un agente enrutador analiza la consulta entrante y la dirige a la fuente de conocimiento más apropiada:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
router_prompt = ChatPromptTemplate.from_messages([
("system", """Classify the user query into one of these categories:
- technical_docs: Questions about API usage, configuration, or code
- policy: Questions about company policies, procedures, or compliance
- support: Questions about troubleshooting or known issues
- general: General questions that don't fit other categories
Respond with ONLY the category name."""),
("human", "{query}")
])
router_chain = router_prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
El enrutamiento es valioso cuando tu base de conocimiento abarca múltiples dominios con diferentes estrategias de indexación. La documentación técnica podría estar indexada con fragmentación consciente del código y embeddings, mientras que los documentos de políticas usan fragmentación semántica con filtros de metadatos.
Descomposición de Consultas
Las preguntas complejas frecuentemente requieren información de múltiples fragmentos de documentos que no aparecerían juntos en una sola recuperación. La descomposición de consultas divide una consulta compleja en sub-consultas independientes, recupera para cada una y sintetiza los resultados:
decomposition_prompt = ChatPromptTemplate.from_messages([
("system", """Break the following complex question into 2-4 simpler
sub-questions that, when answered together, provide a complete answer
to the original question. Return as a JSON array of strings."""),
("human", "{query}")
])
# Example: "How does our API rate limiting compare to competitors
# and what are customers saying about it?"
# Decomposes to:
# ["What are our current API rate limiting policies?",
# "What rate limiting do our main competitors use?",
# "What customer feedback have we received about rate limiting?"]
Autocorrección
La autocorrección es el patrón que más distingue al RAG agéntico del RAG naíve. Después de la recuperación, un paso de calificación evalúa si los documentos recuperados son relevantes y suficientes. Si no lo son, el sistema reformula la consulta e intenta de nuevo:
grading_prompt = ChatPromptTemplate.from_messages([
("system", """You are a retrieval quality grader. Given a user question
and a set of retrieved documents, determine:
1. Are the documents relevant to the question? (yes/no)
2. Do the documents contain sufficient information to answer? (yes/no)
3. If insufficient, suggest a reformulated query.
Respond as JSON with keys: relevant, sufficient, reformulated_query"""),
("human", "Question: {query}\n\nDocuments: {documents}")
])
Este patrón típicamente permite 2-3 intentos de recuperación antes de recurrir a un mensaje de fallo elegante. Las consultas reformuladas frecuentemente cambian de búsqueda por similitud semántica a búsqueda basada en palabras clave, amplían el alcance o apuntan a campos de metadatos específicos.
LangGraph para Orquestación de Agentes con Estado
LangGraph se ha convertido en el framework estándar para construir sistemas RAG agénticos porque proporciona gestión de estado explícita, enrutamiento condicional y soporte de ciclos que las abstracciones más simples basadas en cadenas de LangChain no pueden expresar.
Un sistema RAG agéntico basado en LangGraph se define como un grafo de estados donde los nodos representan pasos de procesamiento y las aristas representan transiciones entre pasos. Las aristas condicionales permiten que el grafo se ramifique basándose en resultados intermedios:
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
from operator import add
class AgentState(TypedDict):
query: str
sub_queries: List[str]
documents: List[dict]
generation: str
retry_count: int
retrieval_grade: str
def route_query(state: AgentState) -> AgentState:
"""Classify and route the incoming query."""
query = state["query"]
classification = router_chain.invoke({"query": query})
state["route"] = classification
return state
def retrieve(state: AgentState) -> AgentState:
"""Retrieve documents based on query and route."""
query = state["query"]
docs = retriever.invoke(query)
state["documents"] = docs
return state
def grade_documents(state: AgentState) -> AgentState:
"""Grade retrieved documents for relevance and sufficiency."""
grade = grading_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["retrieval_grade"] = grade["sufficient"]
state["retry_count"] = state.get("retry_count", 0) + 1
if not grade["sufficient"] and grade.get("reformulated_query"):
state["query"] = grade["reformulated_query"]
return state
def generate(state: AgentState) -> AgentState:
"""Generate final answer from retrieved context."""
answer = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["generation"] = answer
return state
def should_retry(state: AgentState) -> str:
"""Decide whether to retry retrieval or proceed to generation."""
if state["retrieval_grade"] == "yes":
return "generate"
if state["retry_count"] >= 3:
return "generate" # Give up and generate with what we have
return "retrieve"
# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("route", route_query)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.set_entry_point("route")
workflow.add_edge("route", "retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", should_retry, {
"retrieve": "retrieve",
"generate": "generate"
})
workflow.add_edge("generate", END)
app = workflow.compile()
La estructura del grafo hace que el flujo de control sea explícito y depurable. Puedes visualizar el grafo, rastrear ejecuciones a través de cada nodo y entender exactamente por qué el sistema tomó un camino particular para cualquier consulta dada. Esta transparencia es crítica para sistemas de producción donde necesitas diagnosticar fallos y explicar el comportamiento.
LangGraph también soporta patrones más avanzados como recuperación paralela a través de múltiples índices, puntos de control con humano-en-el-bucle donde el agente se pausa para confirmación del usuario, y estado persistente que sobrevive entre conversaciones.
Estrategias de Recuperación
La capa de recuperación en un sistema RAG agéntico es típicamente más sofisticada que una sola búsqueda en un vector store. Los sistemas de producción combinan múltiples estrategias de recuperación.
Búsqueda Híbrida
La búsqueda híbrida combina la recuperación vectorial densa con la recuperación dispersa por palabras clave, dándote la comprensión semántica de los embeddings con la precisión del matching de palabras clave BM25:
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_chroma import Chroma
vectorstore = Chroma(
collection_name="documents",
embedding_function=embedding_model,
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 10
hybrid_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4],
)
Re-ranking
Después de la recuperación inicial, un re-ranker de codificador cruzado puntúa cada documento contra la consulta con mucha mayor precisión que la similitud de codificador bi-direccional. Esto es computacionalmente costoso pero mejora dramáticamente la precisión:
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
reranker = CohereRerank(model="rerank-v3.5", top_n=5)
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=hybrid_retriever,
)
Recuperación Multi-Índice
Las bases de conocimiento de producción frecuentemente abarcan múltiples índices con diferentes esquemas, modelos de embeddings y tipos de documentos. Un sistema agéntico puede consultar múltiples índices en paralelo y fusionar resultados:
async def multi_index_retrieve(query: str, indexes: list) -> list:
"""Retrieve from multiple indexes in parallel."""
import asyncio
async def retrieve_from_index(index, query):
return await index.aretrieve(query)
tasks = [retrieve_from_index(idx, query) for idx in indexes]
results = await asyncio.gather(*tasks)
# Flatten and deduplicate
all_docs = []
seen_ids = set()
for result_set in results:
for doc in result_set:
if doc.metadata["id"] not in seen_ids:
all_docs.append(doc)
seen_ids.add(doc.metadata["id"])
return all_docs
Evaluación con RAGAS y DeepEval
Evaluar sistemas RAG agénticos requiere medir múltiples dimensiones de calidad. RAGAS y DeepEval son los dos frameworks de evaluación más ampliamente adoptados, cada uno con fortalezas distintas.
RAGAS proporciona un conjunto de métricas sin referencia que evalúan la calidad del RAG sin requerir respuestas de verdad base para cada consulta de prueba:
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
eval_dataset = Dataset.from_dict({
"question": questions,
"answer": generated_answers,
"contexts": retrieved_contexts,
"ground_truth": reference_answers,
})
results = evaluate(
dataset=eval_dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall,
],
)
print(results)
La fidelidad mide si la respuesta generada está respaldada por el contexto recuperado, detectando alucinaciones. La relevancia de la respuesta mide si la respuesta aborda la pregunta. La precisión del contexto mide si los documentos recuperados son relevantes, y la cobertura del contexto mide si la recuperación capturó toda la información necesaria.
DeepEval extiende la evaluación con métricas adicionales particularmente relevantes para sistemas agénticos:
from deepeval import evaluate
from deepeval.metrics import (
FaithfulnessMetric,
AnswerRelevancyMetric,
ContextualRelevancyMetric,
HallucinationMetric,
)
from deepeval.test_case import LLMTestCase
test_case = LLMTestCase(
input="How do I configure rate limiting?",
actual_output=generated_answer,
expected_output=reference_answer,
retrieval_context=retrieved_chunks,
)
metrics = [
FaithfulnessMetric(threshold=0.8),
AnswerRelevancyMetric(threshold=0.7),
ContextualRelevancyMetric(threshold=0.7),
HallucinationMetric(threshold=0.5),
]
evaluate(test_cases=[test_case], metrics=metrics)
Para RAG agéntico específicamente, también deberías medir la eficiencia de recuperación: cuántas rondas de recuperación necesita el sistema en promedio, qué porcentaje de consultas requiere reformulación y cómo se degrada la calidad de la respuesta a través de los intentos de reintento. Estas métricas son específicas del bucle agéntico y no están cubiertas por los frameworks estándar de evaluación de RAG.
Patrones de Producción
Mover el RAG agéntico de prototipo a producción requiere abordar preocupaciones de fiabilidad, seguridad y operaciones que no emergen durante el desarrollo.
Guardarraíles
Todo sistema RAG de producción necesita guardarraíles que prevengan que el LLM genere respuestas dañinas, fuera de tema o sin respaldo factual:
from guardrails import Guard
from guardrails.hub import ToxicLanguage, CompetitorCheck
guard = Guard().use_many(
ToxicLanguage(on_fail="exception"),
CompetitorCheck(
competitors=["competitor_a", "competitor_b"],
on_fail="fix"
),
)
raw_response = generation_chain.invoke({"query": query, "documents": docs})
validated_response = guard.validate(raw_response)
Cadenas de Respaldo
Cuando el bucle agéntico no logra encontrar contexto suficiente después del máximo de reintentos, el sistema necesita degradación elegante en lugar de generar una respuesta poco confiable:
def generate_with_fallback(state: AgentState) -> AgentState:
if state["retrieval_grade"] != "yes" and state["retry_count"] >= 3:
state["generation"] = (
"I wasn't able to find sufficient information in our knowledge base "
"to fully answer your question. Here's what I found:\n\n"
f"{partial_answer_from_context(state['documents'])}\n\n"
"For a complete answer, I'd recommend contacting the support team."
)
state["confidence"] = "low"
else:
state["generation"] = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["confidence"] = "high"
return state
Humano-en-el-Bucle
Para aplicaciones de alto riesgo, LangGraph soporta puntos de interrupción donde el agente pausa la ejecución y espera la aprobación humana:
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
# Add interrupt before generation for sensitive queries
workflow.add_node("human_review", lambda state: state)
workflow.add_conditional_edges(
"grade",
lambda state: "human_review" if state.get("sensitive") else "generate",
{"human_review": "human_review", "generate": "generate"}
)
app = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["human_review"]
)
Observabilidad con LangSmith y Phoenix
La observabilidad no es negociable para sistemas agénticos en producción. La naturaleza no determinística del flujo de control impulsado por LLM significa que no puedes predecir ni probar cada camino de ejecución posible. Necesitas rastreo integral para entender qué está haciendo tu sistema en producción.
LangSmith proporciona rastreo de extremo a extremo para aplicaciones LangGraph:
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "agentic-rag-production"
# All LangGraph invocations are automatically traced
result = app.invoke({"query": "How do I configure rate limiting?"})
Cada traza muestra la ruta de ejecución completa a través del grafo: qué nodos fueron visitados, cuáles fueron las entradas y salidas del LLM en cada paso, resultados de recuperación, decisiones de calificación y conteos de tokens. Esto es esencial para depurar casos donde el agente toma un camino inesperado o genera una respuesta deficiente.
Arize Phoenix proporciona una alternativa de código abierto con enfoque en el monitoreo de calidad de recuperación:
import phoenix as px
from phoenix.trace.langchain import LangChainInstrumentor
px.launch_app()
LangChainInstrumentor().instrument()
# Traces are now collected in Phoenix
result = app.invoke({"query": user_query})
Phoenix es particularmente valioso para monitorear la calidad de recuperación a lo largo del tiempo. Rastrea la deriva de embeddings, distribuciones de relevancia de recuperación y puede alertar cuando la calidad de recuperación se degrada, lo que frecuentemente indica que la base de conocimiento se ha desviado de la distribución de entrenamiento del modelo de embeddings y necesita re-indexación.
Métricas clave para monitorear en producción:
# Custom metrics to track
metrics = {
"avg_retrieval_rounds": [],
"reformulation_rate": [],
"fallback_rate": [],
"latency_p50_ms": [],
"latency_p99_ms": [],
"token_cost_per_query": [],
"faithfulness_score": [],
}
Optimización de Costo y Latencia
Los sistemas RAG agénticos son inherentemente más costosos y lentos que el RAG naíve porque realizan múltiples llamadas al LLM por consulta. Optimizar el costo y la latencia sin sacrificar la calidad requiere decisiones arquitectónicas cuidadosas.
Usa modelos más pequeños para el enrutamiento y la calificación. El enrutador y el calificador de documentos no necesitan la capacidad completa de razonamiento de un modelo de frontera. GPT-4o-mini o Claude 3.5 Haiku pueden manejar estas tareas a una fracción del costo y la latencia:
# Use a small, fast model for routing and grading
routing_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
grading_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Use a larger model only for final generation
generation_model = ChatOpenAI(model="gpt-4o", temperature=0.1)
Cachea los resultados de recuperación agresivamente. Muchas consultas de usuarios son variaciones de la misma pregunta subyacente. Un caché semántico que coincida con consultas dentro de un umbral de similitud puede eliminar recuperaciones y llamadas al LLM redundantes:
from langchain_community.cache import RedisSemanticCache
set_llm_cache(RedisSemanticCache(
redis_url="redis://localhost:6379",
embedding=embedding_model,
score_threshold=0.95,
))
Establece timeouts agresivos en cada paso para prevenir latencia descontrolada. Una sola llamada lenta al LLM no debería causar que toda la solicitud se agote:
from asyncio import wait_for, TimeoutError
async def retrieve_with_timeout(state, timeout_seconds=5):
try:
return await wait_for(retrieve(state), timeout=timeout_seconds)
except TimeoutError:
state["documents"] = []
state["retrieval_grade"] = "no"
return state
Casos de Estudio del Mundo Real y Anti-Patrones
El patrón de fallo más común en despliegues de RAG agéntico es el bucle de reintentos infinito. Sin un límite estricto en los intentos de recuperación y un comportamiento de respaldo claro, el agente puede ciclar a través de reformulaciones indefinidamente, quemando tokens y latencia. Siempre establece límites de reintento explícitos y mide tu tasa de reformulación. Si más del 20% de las consultas requieren reformulación, el problema probablemente está en tu capa de recuperación (fragmentación deficiente, modelo de embeddings incorrecto, índice desactualizado) en lugar de en la lógica del agente.
Otro anti-patrón frecuente es el sobre-enrutamiento. Los equipos a veces construyen grafos de enrutamiento complejos con docenas de índices especializados cuando un solo recuperador híbrido bien diseñado tendría mejor rendimiento. Comienza con la arquitectura más simple que cumpla tus requisitos y añade complejidad solo cuando tengas métricas que muestren que los enfoques más simples son insuficientes.
El relleno de ventana de contexto es un tercer anti-patrón. Recuperar 20 fragmentos de documentos y meterlos todos en el prompt desperdicia tokens y puede realmente degradar la calidad de la respuesta porque el LLM tiene dificultades para identificar la información relevante en un mar de contexto marginalmente relacionado. El re-ranking a un pequeño número de fragmentos de alta calidad (típicamente 3-5) supera consistentemente a las ventanas de contexto más grandes en benchmarks de evaluación.
Finalmente, cuidado con la sobre-optimización impulsada por evaluación. Los equipos que optimizan únicamente para puntuaciones RAGAS pueden construir sistemas que puntúan bien en benchmarks pero producen respuestas excesivamente cautelosas y evasivas que los usuarios encuentran inútiles. Equilibra la evaluación cuantitativa con la revisión cualitativa de conversaciones reales de usuarios.
El Futuro: Sistemas RAG Multi-Agente
La siguiente frontera en RAG agéntico son los sistemas multi-agente donde agentes especializados colaboran en tareas complejas de recuperación y síntesis. En lugar de un solo agente manejando todo el pipeline de recuperación-generación, despliegas un equipo de agentes con diferentes especializaciones.
Un agente de investigación maneja la descomposición de consultas y la recuperación multi-salto a través de grandes colecciones de documentos. Un agente de verificación de hechos verifica las afirmaciones generadas contra los documentos fuente. Un agente de síntesis combina los hallazgos de múltiples sub-consultas en una respuesta coherente y bien estructurada. Un agente editor revisa la salida final en busca de claridad, precisión y tono.
from langgraph.graph import StateGraph
# Multi-agent RAG architecture
multi_agent = StateGraph(MultiAgentState)
multi_agent.add_node("planner", planner_agent)
multi_agent.add_node("researcher", researcher_agent)
multi_agent.add_node("fact_checker", fact_check_agent)
multi_agent.add_node("synthesizer", synthesis_agent)
multi_agent.set_entry_point("planner")
multi_agent.add_edge("planner", "researcher")
multi_agent.add_edge("researcher", "fact_checker")
multi_agent.add_conditional_edges(
"fact_checker",
lambda s: "researcher" if s["needs_more_research"] else "synthesizer",
)
multi_agent.add_edge("synthesizer", END)
Esta arquitectura es más costosa por consulta pero maneja preguntas de investigación complejas con las que los sistemas de agente único tienen dificultades. El desafío de diseño clave es definir interfaces claras entre agentes y asegurar que la sobrecarga de la comunicación inter-agente no niegue los beneficios de la especialización.
El ecosistema de herramientas está convergiendo para soportar patrones multi-agente de forma nativa. La composición de sub-grafos de LangGraph, la capa de orquestación de agentes de LlamaIndex y CrewAI proporcionan todas las primitivas para construir sistemas RAG multi-agente. A medida que estos frameworks maduran y los costos continúan bajando, el RAG multi-agente se volverá práctico para un rango más amplio de aplicaciones de producción. Los principios cubiertos en esta guía — gestión de estado explícita, evaluación de calidad de recuperación, respaldos elegantes y observabilidad integral — siguen siendo esenciales independientemente de si despliegas un solo agente o un equipo de agentes especializados.