RAG agente en LangGraph: evaluadores, enrutamiento condicional y solución de respaldo mediante búsqueda web

Un pipeline RAG semántico construido a partir de las piezas fundamentales de LangChain, un backend FastAPI y una interfaz frontal de chatbot Streamlit, maneja la carga de archivos PDF, convierte los documentos en embeddings y responde a preguntas a través de una interfaz de chat. También tiene un límite básico: las primitivas de LangChain por sí solas generan cadenas lineales, es decir, una línea recta desde el prompt pasando por el LLM hasta el analizador de salida. Todo lo que requiera ramas condicionales, bucles, enrutamiento dinámico o decisiones tomadas por el LLM queda fuera de este modelo, y las simples instrucciones if/else solo permiten un flujo superficial y predefinido.

LangGraph soluciona este problema. El resultado es un flujo de trabajo basado en agentes con las siguientes características:

• ramas y bucles condicionales integrados directamente en el pipeline
• un LLM que elige entre búsqueda en la web y recuperación mediante RAG, sin reglas codificadas de antemano
• transiciones gestionadas a través de nodos y aristas de LangGraph

Evaluadores basados en LLM que analizan el contexto recuperado y las respuestas generadas

Ruteo que reacciona a dichos resultados

Estado transportado entre transiciones mediante un StateGraph

El grafo final conectado a un backend FastAPI para alimentar un chatbot de tipo producción

Al final se obtiene un sistema RAG agente que responde ya sea desde el almacén de vectores o desde la web.

Términos de LangGraph

Hay cuatro conceptos que conforman el vocabulario. El StateGraph orquesta la ejecución. El estado es un diccionario tipado que almacena y actualiza los datos del proceso. Los nodos son funciones en Python que modifican el estado y funcionan como unidades de cálculo del grafo. Las aristas representan las transiciones entre nodos, y las aristas condicionales permiten tanto ramificaciones como bucles.

Configuración

Ponga las claves API en un archivo .env en la raíz del proyecto:

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
Hugging_Face_TOKEN=<your huggingface token>
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>

Luego léalos en una clase BaseSettings de Pydantic:

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

    """Loads the dotenv file. Including this is necessary to get
    pydantic to load a .env file."""
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

Estructura del proyecto

La estructura de carpetas es la siguiente:

project/
├── rag_process/
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── graphs
│   │── chains.py       # llm chains
│   ├── nodes.py        # graph nodes. Defined as Python functions
│   ├── graph.py        # graph workflow
│   ├── consts.py       # define node names as constants
│   └── state.py        # typed dictionary to store graph states
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── dependencies.py     # dependency inject to fastAPI endpoint function
├── test_chains.py      # pytests
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

Las cadenas de recuperación, el servicio FastAPI y el cliente Streamlit provienen del pipeline anterior. Lo nuevo aquí es el grafo, su clase de estado y los evaluadores que verifican la relevancia de los documentos recuperados, detectan alucinaciones y determinan si una respuesta es pertinente para la pregunta.

Las cadenas de LLM

Las cadenas se encuentran en graphs/chains.py y utilizan el Lenguaje de Expresiones de LangChain (LCEL). Se han definido siete:

• retrieve_documents obtiene los K documentos más similares de Pinecone
• generate_answer escribe una respuesta basada en la pregunta junto con el contexto
• retrival_grader decide si los documentos recuperados son relevantes
• web_search realiza una búsqueda en Internet y devuelve el contexto
• answer_grader verifica que la respuesta aborde la pregunta

hallucination_grader verifica que la respuesta esté basada en el contexto

select_routes envía preguntas sobre agentes, ingeniería de prompts y ataques adversariales a RAG, y todo lo demás a búsquedas en la web

Pruebas unitarias para cadenas

Las pruebas son aún más importantes en los sistemas agentes porque la salida de los LLM es no determinística y puede desviarse a medida que se actualizan los modelos, los proveedores los ajustan o cambian las incrustaciones. Las pruebas en test_chains.py utilizan pytest con pytest.mark.asyncio para las cadenas asíncronas. Una prueba positiva para la recuperación mediante RAG se ve así:

@pytest.mark.asyncio
async def test_rag_retrieval_yes() -> None:
    prompt = "what is prompt engineer ?"
    searched_documents = await retrieve_documents(prompt)
    searched_document = searched_documents[0]
    # logger.info(f"searched_content: {searched_document.page_content}")
    retrieval_grade = await retrieval_grader(searched_document, prompt)
    assert retrieval_grade

Pregunta sobre la ingeniería de prompts y pasa el primer documento recuperado a retrieval_grader. Dado que ese tema ya está cubierto por documentos en Pinecone, el evaluador debería devolver True. La prueba previene fallos cuando cambian los embeddings o el comportamiento del modelo, detecta tempranamente regresiones en las actualizaciones del modelo y verifica la lógica de enrutamiento entre RAG y la búsqueda en la web. Una prueba negativa correspondiente confirma que el evaluador devuelve False y que el flujo se dirige a la búsqueda en la web.

Estado y nodos

El estado es un diccionario tipado que se guarda en graphs/state.py:

class GraphState(TypedDict):

    question: str
    generation: str
    web_search: bool
    documents: list[str]

Los nodos leen y actualizan este estado. Definidos en graphs/nodes.py, son funciones ordinarias de Python que leen el estado, llaman a cadenas LLM y devuelven los campos actualizados. Un nodo utilizado en un borde condicional devuelve el nombre del siguiente nodo a ejecutar. Hay siete.

• generate llama a generate_answer y devuelve la respuesta.
• retrieve llama a retrieve_documents y devuelve los documentos.
• search_web llama a web_search y agrega los resultados a los documentos.
• grade_documents conserva los documentos relevantes y establece web_search=True cuando es necesario.
• grade_generation ejecuta los evaluadores de alucinaciones y respuestas y devuelve “útil”, “no útil” o “no soportado”.

decide_to_generate elige entre WEBSEARCH o GENERATE según el valor de web_search.

route_question envía las preguntas relacionadas con el agente a RAG y las demás a la búsqueda en la web.

Construcción de StateGraph

El grafo en sí se crea en graphs/graph.py, y el flujo de trabajo final tiene la apariencia de un proceso ramificado entre recuperación de información, evaluación, generación y búsqueda. Su construcción implica varios pasos.

Primero, se asigna a cada nodo un identificador de tipo cadena. Al mantener esos nombres como constantes en graphs/consts.py, el grafo resulta más fácil de leer y se evitan errores tipográficos:

RETRIEVE = "retrieve"
GRADE_DOCUMENTS = "grade_documents"
GENERATE = "generate"
WEBSEARCH = "websearch"

A continuación, cree el StateGraph y registre cada nodo con su función:

workflow = StateGraph(GraphState)

workflow.add_node(RETRIEVE, retrieve)
workflow.add_node(GRADE_DOCUMENTS, grade_documents)
workflow.add_node(GENERATE, generate)
workflow.add_node(WEBSEARCH, search_web)

Cada llamada a add_node asocia un nombre a una función, y esa separación entre nombres y comportamientos hace que el flujo de trabajo sea más fácil de mantener, visualizar y extender.

En tercer lugar, establezca un punto de entrada condicional. En lugar de codificar manualmente un nodo de inicio, LangGraph permite que una función examine la pregunta y elija la primera ruta. Aquí route_question decide si comenzar con una búsqueda en la web o con recuperación de información mediante RAG:

workflow.set_conditional_entry_point(
    route_question,
    {
        WEBSEARCH: WEBSEARCH,
        RETRIEVE: RETRIEVE,
    }
)

Esa función se ejecuta primero y devuelve "WEBSEARCH" o "RETRIEVE" según el tema, y LangGraph utiliza su valor de retorno para elegir el nodo de partida. Las preguntas relacionadas con agentes van a RAG, mientras que las de conocimiento general comienzan con una búsqueda en la web.

Cuarto, conecta los nodos con aristas. Las aristas estáticas siempre siguen un mismo camino, mientras que las aristas condicionales eligen el siguiente nodo según la decisión de un LLM:

workflow.add_edge(RETRIEVE, GRADE_DOCUMENTS)
workflow.add_conditional_edges(
    GRADE_DOCUMENTS,
    decide_to_generate,
    {
        WEBSEARCH: WEBSEARCH,
        GENERATE: GENERATE,
    },
)

workflow.add_conditional_edges(
    GENERATE,
    grade_generation,
    {
        "not supported": GENERATE,
        "useful": END,
        "not useful": WEBSEARCH,
    }
)

workflow.add_edge(WEBSEARCH, GENERATE)
workflow.add_edge(GENERATE, END)

Juntos, estos elementos forman un flujo de trabajo dinámico en el que el LLM elige la ruta inicial, la relevancia del documento decide entre RAG y búsqueda en la web, y la calidad de la respuesta determina si se debe refinar, recurrir a otra opción o finalizar. Esa estructura convierte un pipeline RAG simple en un sistema agente más robusto.

Finalmente, compile el grafo y, opcionalmente, córtelo:

app = workflow.compile()

app.get_graph().draw_mermaid_png(output_file_path="graphs.png")

Servirlo mediante FastAPI

En dependencies.py, una función importa el grafo compilado e lo invoca de forma asíncrona:

from graphs.graph import app


async def get_generation(body: RAGRequest=Body(...)) -> dict:
    try:
        generation = await app.ainvoke(body.dict())

        return {
            "answer": generation.get("generation", ""),
            "web_search": generation.get("web_search", False)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

FastAPI lo inyecta como una dependencia, y devuelve la respuesta generada junto con una marca que indica si proviene de RAG o de una búsqueda en la web.

Probándolo

El front end de Streamlit del pipeline anterior puede reutilizarse tal como está. Al preguntar cómo convertirse en millonario, la solicitud se dirige a una búsqueda en la web, mientras que al preguntar qué es la ingeniería de prompts, la solicitud se dirige al camino semántico de RAG. Esto demuestra un enrutamiento dinámico junto con una procedencia clara para cada respuesta.

En resumen, LangGraph StateGraph ahora gestiona el estado; el flujo de trabajo cuenta con transiciones dinámicas, bucles y ramificaciones a través de bordes condicionales; los evaluadores de LLM se encargan de la relevancia de los documentos, la detección de alucinaciones y la calidad de las respuestas; y la API informa al usuario si una respuesta proviene de RAG o de la web. El siguiente paso en este trabajo es el uso de herramientas y el framework ReAct, para lograr un comportamiento aún más agente.