Corrective RAG (CRAG) y Self-RAG con LangGraph: Cómo Evitar Alucinaciones en Producción

1. La Trampa del RAG Incondicional: Basura Entra, Basura Sale

La inmensa mayoría de las implementaciones de RAG en producción operan bajo una asunción ingenua: que la base de datos vectorial siempre encontrará información relevante para cualquier pregunta del usuario. En la realidad empresarial esto falla constantemente:

• El usuario pregunta por un tema que no existe en la base de datos corporativa.
• El motor de búsqueda vectorial recupera fragmentos con alta similitud coseno que hablan de temas tangenciales pero no resuelven la pregunta.
• El LLM, forzado por su prompt del sistema a responder en base al contexto aportado, inventa una conexión espuria provocando una alucinación destructiva.

2. Arquitectura de Estados de Corrective RAG con LangGraph

LangGraph permite modelar el flujo de RAG como una máquina de estados finita cíclica con ramas condicionales:

1. Nodo Recuperar (Retrieve): Extrae los mejores fragmentos de la base de datos vectorial.
2. Nodo Calificar Documentos (Grade Documents): Un modelo evaluador analiza cada fragmento y emite un veredicto binario: yes (relevante) o no (irrelevante).
3. Decisión Condicional:
   • Si todos los documentos son relevantes $ ightarrow$ Pasa directamente al nodo Generar.
   • Si algún documento es irrelevante $ ightarrow$ Pasa al nodo Transformar Consulta (Rewrite Query) y posteriormente al nodo Búsqueda Web (Web Search).
4. Nodo Generar (Generate): Redacta la respuesta final con las citas verificadas.

3. Tutorial Práctico: Implementación de CRAG en Python

from typing import List, Literal
from typing_extensions import TypedDict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, END

# 1. Definir el Estado del Grafo
class GraphState(TypedDict):
    question: str
    generation: str
    documents: List[str]
    web_search_needed: bool

# 2. Definir el evaluador de relevancia estructurado con Pydantic
class GradeDocuments(BaseModel):
    binary_score: Literal["yes", "no"] = Field(
        description="Indica si el documento es relevante para la pregunta: 'yes' o 'no'"
    )

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
evaluator = llm.with_structured_output(GradeDocuments)

prompt_evaluacion = ChatPromptTemplate.from_messages([
    ("system", "Eres un evaluador de relevancia. Determina si el siguiente documento es útil para responder a la pregunta del usuario. Responde estrictamente con 'yes' o 'no'."),
    ("human", "Pregunta: {question}

Documento:
{document}")
])

chain_evaluador = prompt_evaluacion | evaluator

# 3. Nodos del Grafo
def retrieve_node(state: GraphState):
    print("--- [NODO] Recuperando de la Base de Datos Vectorial ---")
    # Simulación de fragmentos recuperados
    return {"documents": ["Error 404: Ruta no encontrada en el enrutador Nginx."], "question": state["question"]}

def grade_documents_node(state: GraphState):
    print("--- [NODO] Calificando Relevancia de Documentos ---")
    question = state["question"]
    docs = state["documents"]
    
    docs_filtrados = []
    web_search = False
    
    for d in docs:
        res = chain_evaluador.invoke({"question": question, "document": d})
        if res.binary_score == "yes":
            docs_filtrados.append(d)
        else:
            print("Documento descartado por irrelevante.")
            web_search = True
            
    if len(docs_filtrados) == 0:
        web_search = True
        
    return {"documents": docs_filtrados, "web_search_needed": web_search}

def web_search_node(state: GraphState):
    print("--- [NODO] Ejecutando Fallback de Búsqueda Web Externa ---")
    # Aquí se integra Tavily Search o DuckDuckGo
    resultado_web = f"Información en vivo obtenida de la web para: {state['question']}"
    docs = state["documents"]
    docs.append(resultado_web)
    return {"documents": docs}

def generate_node(state: GraphState):
    print("--- [NODO] Generando Respuesta Verificada ---")
    prompt = f"Responde a '{state['question']}' usando este contexto: {state['documents']}"
    resp = llm.invoke(prompt)
    return {"generation": resp.content}

# 4. Decisión condicional
def decide_to_generate(state: GraphState):
    if state["web_search_needed"]:
        return "web_search"
    return "generate"

# 5. Construir y compilar el Grafo LangGraph
workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("grade_documents", grade_documents_node)
workflow.add_node("web_search", web_search_node)
workflow.add_node("generate", generate_node)

workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges("grade_documents", decide_to_generate, {
    "web_search": "web_search",
    "generate": "generate"
})
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)

app = workflow.compile()
print("Grafo Corrective RAG compilado con éxito.")

4. Comparativa de Fiabilidad en Entornos de Producción

5. Checklist para Producción con LangGraph

• Utiliza modelos pequeños con temperatura cero (como GPT-4o-mini o Claude 3.5 Haiku) para el nodo calificador de documentos para mantener el coste por debajo de 0.001$ por petición.
• Fija límites máximos de iteración en el grafo de LangGraph (recursion_limit=5) para evitar bucles infinitos de reescritura de consultas.

7. Persistencia de Estados y Checkpointing en LangGraph con PostgreSQL

En aplicaciones de misión crítica, si el servidor se reinicia o se pierde la conexión mientras el agente está calificando documentos o esperando una respuesta de búsqueda web, no se puede perder el estado de la conversación. LangGraph resuelve esto integrando PostgresSaver:

from langgraph.checkpoint.postgres import PostgresSaver

# Conectar al pool de base de datos para almacenar el historial de estados
with PostgresSaver.from_conn_string("postgresql://user:pass@localhost:5432/agent_db") as checkpointer:
    checkpointer.setup()
    app = workflow.compile(checkpointer=checkpointer)

    # Cada conversación tiene un thread_id único que permite pausar y reanudar
    config = {"configurable": {"thread_id": "sesion_usuario_12894"}}
    salida = app.invoke({"question": "¿Cuál es la última versión de CUDA soportada por PyTorch 2.5?"}, config=config)
    print(salida["generation"])

8. Integración de Motores de Búsqueda Web para Fallback (Tavily AI)

Para alimentar el nodo de búsqueda externa cuando la base de datos interna carezca de respuestas, Tavily Search API está optimizada específicamente para agentes de IA: en lugar de devolver código HTML desordenado y anuncios publicitarios, devuelve fragmentos de texto limpios, sintetizados y puntuados por relevancia directa para el LLM.

9. Automatización de Tests Unitarios para Agentes CRAG con Pytest

Para asegurar que las actualizaciones de modelos o prompts no degraden la fiabilidad del agente en producción, se deben implementar tests automatizados de regresión:

import pytest

@pytest.mark.asyncio
async def test_crag_activa_fallback_cuando_contexto_es_falso():
    # Consulta sobre un hecho falso que no existe en el corpus interno
    consulta_trampa = "¿Cuál es la capital del planeta Marte según el informe interno de 2024?"
    
    estado_final = app.invoke({"question": consulta_trampa})
    
    # El nodo calificador debe marcar que se requiere búsqueda externa o rechazar
    assert estado_final["web_search_needed"] is True, "El evaluador debió descartar los documentos internos irrelevantes."
    assert "no se encontró información" in estado_final["generation"].lower() or len(estado_final["documents"]) > 0

10. Métricas de Observabilidad para el Grafo de Estados

• Ratio de Activación de Búsqueda Web: Porcentaje de consultas que requieren acudir a Internet. Si supera el 40%, indica que el corpus documental interno tiene lagunas informativas graves.
• Latencia por Nodo: Registra el tiempo consumido en cada transición: el nodo de calificación de documentos no debe tardar más de 300 ms en emitir su veredicto.

11. Implementación de Self-RAG con Tokens de Reflexión Especiales

Para llevar la fiabilidad al siguiente nivel, Self-RAG entrena al modelo generador para emitir tokens de introspección antes de formular la respuesta final:

• [Retrieve]: El modelo decide activamente si necesita consultar la base de datos o si su conocimiento interno es suficiente.
• [IsRel]: Evalúa si el pasaje devuelto es verdaderamente relevante.
• [IsSup]: Comprueba si cada frase de la respuesta final está estrictamente respaldada por el texto fuente.
• [IsUse]: Califica la utilidad percibida de la respuesta completa del 1 al 5.

En frameworks abiertos como LangGraph, estos tokens se emulan mediante analizadores de salida estructurados con Pydantic, garantizando que el agente se autocensura y corrige antes de emitir cualquier dato no respaldado.

12. Checklist de Puesta en Marcha en Producción

• Configura timeouts de 5 segundos en las llamadas a motores de búsqueda web externos para evitar que peticiones lentas degraden la experiencia de usuario.
• Almacena todas las trazas del grafo en una base de datos PostgreSQL para auditoría forense ante cualquier queja de alucinación de clientes.

13. Estrategias de Fallback Jerárquico: Vector Store -> Grafo -> Web Search

En arquitecturas empresariales complejas, el nodo correctivo no salta directamente a la búsqueda en la web pública (lo que podría filtrar intenciones de negocio confidenciales a terceros). En su lugar, implementa un enrutamiento escalonado de 3 niveles:

1. Nivel 1: Base Vectorial Interna (Local): Recuperación primaria de fragmentos con alta similitud coseno.
2. Nivel 2: Grafo de Conocimiento Corporativo (Neo4j): Si los fragmentos no superan el umbral de relevancia, se consulta el grafo relacional en busca de entidades conectadas.
3. Nivel 3: Búsqueda Web Externa (Tavily): Únicamente si la consulta trata sobre competidores externos, noticias de actualidad o cotizaciones bursátiles públicas.