RAG agent dans LangGraph : évaluateurs, routage conditionnel et solution de secours par recherche web

Un pipeline RAG sémantique construit à partir des composants de base de LangChain, d’un backend FastAPI et d’une interface frontale de chatbot Streamlit, gère les téléchargements de PDF, transforme les documents en embeddings et répond aux questions via une interface de chat. Il présente cependant une limite fondamentale : les primitives de LangChain seules permettent des chaînes linéaires, un parcours direct du prompt à l’interpréteur de sortie en passant par le LLM. Tout ce qui nécessite des branches conditionnelles, des boucles, un routage dynamique ou des décisions prises par le LLM sort de ce cadre, et les simples instructions if/else ne permettent qu’un flux superficiel et prédéfini.

LangGraph résout ce problème. Le résultat est un flux de travail agentique présentant les propriétés suivantes :

• des branches conditionnelles et des boucles intégrées directement dans le pipeline
• un LLM capable de choisir entre la recherche sur le web et la récupération via RAG, sans règle codée à l’avance
• des transitions gérées au moyen de nœuds et d’arêtes LangGraph

Des évaluateurs basés sur des LLM qui jugent le contexte récupéré et les réponses générées

Un routage qui réagit à ces verdicts

L’état transmis lors des transitions par un StateGraph

Le graphique final intégré à un backend FastAPI pour alimenter un chatbot de type production

Finalement, on obtient un système RAG agentique qui répond soit à partir du stockage vectoriel, soit depuis le web.

Terminologie LangGraph

Quatre concepts couvrent le vocabulaire. Le StateGraph orchestre l’exécution. L’état est un dictionnaire typé qui stocke et met à jour les données du pipeline. Les nœuds sont des fonctions Python qui modifient l’état et servent de unités de calcul du graphique. Les arêtes représentent les transitions entre nœuds, et les arêtes conditionnelles permettent à la fois des branches et des boucles.

Configuration

Placer les clés API dans un fichier .env situé à la racine du projet :

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>

Puis les importez dans une classe 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")

Structure du projet

La structure des dossiers est la suivante :

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

Les chaînes de récupération, le service FastAPI et le client Streamlit proviennent du pipeline précédent. Ce qui est nouveau ici, c’est le graphe, sa classe d’état ainsi que les évaluateurs qui vérifient la pertinence des documents récupérés, détectent les hallucinations et déterminent si une réponse est pertinente par rapport à la question.

Les chaînes LLM

Ces chaînes se trouvent dans graphs/chains.py et utilisent le LangChain Expression Language (LCEL). Sept chaînes y sont définies :

• retrieve_documents récupère les K documents les plus similaires depuis Pinecone
• generate_answer génère une réponse basée sur la question ainsi que sur le contexte fourni
• retrival_grader décide si les documents récupérés sont pertinents
• web_search effectue une recherche sur Internet et retourne le contexte correspondant
• answer_grader vérifie que la réponse aborde bien la question

hallucination_grader vérifie que la réponse est ancrée dans le contexte

select_routes envoie des questions concernant les agents, l’ingénierie de prompts et les attaques adversariales à RAG, tandis que tout le reste est envoyé à une recherche web

Tests unitaires pour les chaînes

Les tests sont encore plus importants dans les systèmes agents que d’habitude, car la sortie des LLM est non déterministe et peut varier à mesure que les modèles sont mis à jour, que leurs fournisseurs les ajustent ou que les embeddings changent. Les tests dans test_chains.py utilisent pytest avec pytest.mark.asyncio pour les chaînes asynchrones. Un test positif pour la récupération via RAG ressemble à ceci :

@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

Celui-ci demande des informations sur l’ingénierie des prompts et transmet le premier document récupéré à retrieval_grader. Comme ce sujet est déjà abordé dans les documents de Pinecone, le correcteur doit renvoyer True. Le test permet d’éviter les pannes en cas de changement dans les embeddings ou le comportement du modèle, détecte tôt les dégradations dues aux mises à jour du modèle et vérifie la logique de routage entre RAG et la recherche web. Un test négatif correspondant confirme que le correcteur renvoie False et que le flux est redirigé vers la recherche web.

État et nœuds

L’état est un dictionnaire typé conservé dans graphs/state.py:

class GraphState(TypedDict):

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

Les nœuds lisent et mettent à jour cet état. Définis dans graphs/nodes.py, il s’agit de fonctions Python ordinaires qui lisent l’état, appellent des chaînes LLM et retournent les champs mis à jour. Un nœud utilisé dans une arête conditionnelle renvoie le nom du nœud suivant à exécuter. Il y en a sept.

• generate appelle generate_answer et retourne la réponse.
• retrieve appelle retrieve_documents et retourne les documents.
• search_web appelle web_search et ajoute les résultats aux documents.
• grade_documents conserve les documents pertinents et définit web_search=True lorsque c’est nécessaire.
• grade_generation exécute les outils d’évaluation des hallucinations et des réponses, puis retourne "utile", "pas utile" ou "non pris en charge">.

decide_to_generate choisit entre WEBSEARCH ou GENERATE en fonction de web_search.

route_question envoie les questions relatives à l’agent à RAG et les autres à la recherche sur le web.

Construction de StateGraph

Le graphique lui-même est assemblé dans graphs/graph.py, et le flux de travail final ressemble à un parcours ramifié entre la récupération d’informations, l’évaluation, la génération et la recherche. Sa construction nécessite plusieurs étapes.

Tout d’abord, il faut attribuer à chaque nœud un identifiant sous forme de chaîne de caractères. En conservant ces noms en tant que constantes dans graphs/consts.py, on facilite la lecture du graphique et on évite les fautes d’orthographe :

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

Ensuite, créez le StateGraph et enregistrez chaque nœud avec sa fonction :

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)

Chaque appel à add_node associe un nom à une fonction, et cette séparation entre les noms et le comportement rend le flux de travail plus facile à maintenir, à visualiser et à étendre.

Troisièmement, définissez un point d’entrée conditionnel. Au lieu de coder manuellement un nœud de départ, LangGraph permet à une fonction d’examiner la question et de choisir le premier itinéraire. Ici, route_question décide s’il faut commencer par une recherche sur le web ou par une récupération via RAG :

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

Cette fonction s’exécute en premier et renvoie soit "WEBSEARCH" soit "RETRIEVE" en fonction du sujet ; LangGraph utilise sa valeur de retour pour choisir le nœud de départ. Les questions relatives aux agents sont acheminées vers RAG, tandis que les questions de connaissance générale commencent par une recherche sur le web.

Quatrièmement, connectez les nœuds avec des arêtes. Les arêtes statiques suivent toujours un même chemin, tandis que les arêtes conditionnelles choisissent le nœud suivant en fonction d’une décision de l’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)

Ces éléments ensemble forment un flux de travail dynamique dans lequel le LLM choisit la route initiale, la pertinence du document décide entre RAG et recherche sur le web, et la qualité de la réponse détermine s’il faut affiner, revenir en arrière ou finaliser. Cette structure transforme un simple pipeline RAG en un système agent plus solide.

Finalement, compilez le graphe et, si vous le souhaitez, dessinez-le :

app = workflow.compile()

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

L’exécution via FastAPI

Dans dependencies.py, une fonction importe le graphe compilé et l’appelle de manière asynchrone :

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 l’injecte en tant que dépendance, et il renvoie la réponse générée ainsi qu’un indicateur indiquant si elle provient de RAG ou d’une recherche sur le web.

Essayer tout ça

L’interface Streamlit issue du pipeline précédent peut être réutilisée telle quelle. Demander comment devenir millionnaire redirige vers une recherche sur le web, tandis que demander ce qu’est l’ingénierie des prompts redirige vers le chemin RAG sémantique. Cela montre un routage dynamique ainsi qu’une provenance claire pour chaque réponse.

En résumé, LangGraph StateGraph gère désormais l’état, le flux de travail comporte des transitions dynamiques avec des boucles et des branches via des arêtes conditionnelles, les évaluateurs de LLM assurent la pertinence des documents, la détection des hallucinations et la qualité des réponses, et l’API informe l’utilisateur si une réponse provient de RAG ou du web. La prochaine étape dans ce travail concerne l’utilisation d’outils et le framework ReAct, afin d’obtenir encore plus de comportements agents.