Intégrer un agent LangChain dans FastAPI : outils, recherche de manuels, flux en continu

Structure du projet

Chaque fonctionnalité dispose de son propre package : routage HTTP, authentification, logique de l’agent, outils et pipeline RAG. Cela permet à chaque fichier de rester compact et rend évident l’endroit où une nouvelle capacité doit être placée.

project/
│
├── main.py
├── .env
├── .gitignore
│
├── auth/
│   ├── __init__.py
│   └── dependencies.py
│
├── routers/
│   ├── __init__.py
│   └── chat.py
│
├── llm/
│   ├── __init__.py
│   ├── agent.py
│   ├── context.py
│   ├── orchestrator.py
│   ├── prompts.py
│   ├── provider.py
│   │
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── demo_tool.py
│   │   └── manual_tool.py
│   │
│   └── rag/
│       ├── __init__.py
│       ├── config.py
│       ├── context.py
│       │
│       ├── ingestion/
│       │   ├── __init__.py
│       │   ├── loader.py
│       │   ├── chunker.py
│       │   ├── chroma.py
│       │   └── indexer.py
│       │
│       └── retrieval/
│           ├── __init__.py
│           └── retriever.py
│
├── scripts/
│   ├── __init__.py
│   └── index_manual.py
│
├── docs/
│   └── manual.pdf
│
└── chroma_data/ (*generated locally, not commited or deployed)

Rôle de chaque partie :

• main.py crée l’application FastAPI.
• auth/ contient la dépendance d’authentification simulée.
• routers/ contient les points de terminaison HTTP.
• llm/ regroupe tout ce qui concerne l’agent.
• llm/tools/ contient les fonctions que l’agent peut appeler.

Vous ne créerez pas tout cela en même temps. L’ordre de construction est : la couche API, puis le modèle, ensuite les outils, suivie du pipeline RAG, puis le contexte, la diffusion en temps réel et l’historique.

Étape 1 : Un squelette FastAPI avec un utilisateur simulé

Commencez par installer tout ce dont le projet a besoin. La liste comprend le serveur web, LangChain et LangGraph, le client de chat compatible OpenAI, ChromaDB, les chargeurs de PDF et les séparateurs de texte, ainsi que python-dotenv pour la configuration.

pip install fastapi uvicorn langchain langgraph langchain-openai chromadb langchain-community langchain-text-splitters pypdf python-dotenv

Le modèle sera accessible via OpenRouter, il faut donc créer un fichier .env dans la racine du projet qui contiendra la clé API.

OPENROUTER_API_KEY=your_api_key_here

Ajoutez immédiatement .env à .gitignore. Une clé qui finit dans le contrôle de version doit être considérée comme divulguée.

Point d’entrée de l’application

main.py reste très simple. Il crée l’application et enregistre le routeur de chat ; rien concernant le modèle ou l’agent ne doit se trouver ici.

from fastapi import FastAPI
from routers.chat import chat_router

app = FastAPI(
    title="AI Agent Demo",
)
app.include_router(chat_router)

Un premier endpoint de chat

Dans routers/chat.py, définissez un routeur sous le préfixe /chat avec une seule route POST. Pour l’instant, il se contente de répéter le message reçu, ce qui suffit à vérifier que tout fonctionne avant d’intégrer une intelligence artificielle.

from fastapi import APIRouter

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
):
    return {
        "message": message,
    }

Notez que message: str sur une route POST sans modèle de corps fait en sorte que FastAPI le lit à partir de la chaîne de requête. Cela est pratique pour tester des fonctionnalités dans Swagger UI, mais pour un client réel, on accepterait normalement un corps JSON défini avec un modèle Pydantic, car les chaînes de requête finissent dans les journaux d’accès et présentent des limites de longueur pratiques.

Dépendance d’authentification simulée

Une application en production vérifierait une cookie de session ou un JWT et chargerait l’utilisateur depuis la base de données. Ici, une en-tête personnalisé remplace tout cela. Créez auth/dependencies.py avec une petite dataclass MockUser et une fonction get_current_user qui lit l’en-tête X-Demo-User et rejette la requête avec un code 401 s’il est manquant.

from dataclasses import dataclass
from fastapi import Header, HTTPException

@dataclass
class MockUser:
    id: str
    name: str

def get_current_user(
    x_demo_user: str | None = Header(default=None),
) -> MockUser:
    if x_demo_user is None:
        raise HTTPException(
            status_code=401,
            detail="Missing X-Demo-User header",
        )
    return MockUser(
        id=x_demo_user,
        name=x_demo_user,
    )

L’injection de dépendances de FastAPI transmet désormais l’utilisateur à l’endpoint. Il suffit de déclarer un paramètre avec Depends(get_current_user).

from fastapi import APIRouter, Depends
from auth.dependencies import (
    MockUser,
    get_current_user,
)

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    return {
        "user": current_user.name,
        "message": message,
    }

X-Demo-User: user-123

Au cours de chaque requête, FastAPI exécute d’abord get_current_user() et passe l’objet MockUser obtenu à ask_ai. Le point clé de la conception réside dans la division des tâches : FastAPI gère l’authentification, tandis que la couche d’intelligence artificielle reçoit simplement un objet utilisateur en qui elle peut avoir confiance. Plus tard, c’est cet objet utilisateur qui permet aux outils de retourner des données appartenant à la bonne personne. Remplacer le mock par une authentification réelle ne modifie que cette seule dépendance.

Étape 2 : Connecter un modèle via OpenRouter

Avec une extrémité de pointe fonctionnelle et un appelant connu, le service a besoin d’un modèle. OpenRouter expose une API compatible avec OpenAI, de sorte que la classe ChatOpenAI de LangChain fonctionne avec elle dès que vous pointez base_url vers OpenRouter et que vous transmettez votre clé OpenRouter.

Placez ce code dans llm/provider.py. Il charge le fichier .env, génère rapidement une erreur claire en l’absence de la clé, et enveloppe cette dernière dans SecretStr de Pydantic afin qu’elle ne soit pas affichée par erreur dans les journaux ou les représentations.

import os
from dotenv import load_dotenv
from pydantic import SecretStr
from langchain_openai import ChatOpenAI

load_dotenv()

api_key = os.getenv(
    "OPENROUTER_API_KEY",
)
if not api_key:
    raise RuntimeError(
        "OPENROUTER_API_KEY environment variable is not set."
    )

model = ChatOpenAI(
    model="YOUR_MODEL",
    api_key=SecretStr(api_key),
    base_url="https://openrouter.ai/api/v1",
)

Puisque la clé provient de l’environnement, elle n’apparaît jamais dans le code source. À ce stade, vous pourriez déjà envoyer des prompts au modèle et obtenir des réponses, mais il s’agirait alors d’une simple appel à un LLM. L’objectif est de disposer d’un agent capable de décider par lui-même quand il a besoin d’une outil.

Sélection d’un modèle

L’argument model n’est rien d’autre qu’un identifiant de modèle OpenRouter, ce qui vous permet de changer de modèle sans toucher au reste du code. Lors de la comparaison des modèles, vérifiez :

• le support des appels d’outils, sur lequel dépend l’agent ;
• le support du streaming ;
• la taille de la fenêtre de contexte ;
• les limites de vitesse ;
• la présence d’un plan gratuit.

OpenRouter propose certains modèles gratuitement, ce qui est pratique pour expérimenter. Le catalogue change régulièrement, il convient donc de consulter la liste actuelle et de filtrer les modèles gratuits plutôt que de compter sur des recommandations fixes. Ce que vous choisissez est directement passé au constructeur :

model = ChatOpenAI(
    model="YOUR_MODEL_ID",
    api_key=SecretStr(api_key),
    base_url="https://openrouter.ai/api/v1",
)

Par exemple, si le catalogue indique un identifiant comme celui ci-dessous (un exemple pris au moment de la rédaction ; il peut ne plus être disponible), vous transmettrez cette chaîne exacte en tant que model :

google/gemma-4-26b-a4b-it:free

N’oubliez pas que les modèles gratuits utilisent une capacité partagée. Ils peuvent faire l’objet de limitations de vitesse ou devenir indisponibles, souvent au pire moment lors d’une démonstration. Changer pour un autre modèle ou utiliser sa propre clé de fournisseur via OpenRouter résout généralement ce problème. Pour un environnement de production, choisissez en fonction de la fiabilité, des capacités, de la latence et du coût, et non uniquement du prix.

Étape 3 : Du modèle à l’agent

Une appel direct au modèle se fait en une seule étape : le texte de l’utilisateur est envoyé, puis une réponse est générée. Un agent introduit un cycle de décision. Le modèle examine la demande, détermine s’il peut répondre immédiatement ou s’il a besoin de quelque chose en premier, appelle un outil si nécessaire, lit le résultat, et ne produit la réponse finale qu’ensuite. En résumé :

• Appel simple : utilisateur, puis LLM, puis réponse ;

Le module agent

Dans llm/agent.py, la fonction create_agent de LangChain construit l’agent à partir d’un modèle et d’une instruction système. En sous-main, elle génère un graphe LangGraph qui exécute pour vous la boucle modèle-outils.

from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT

agent = create_agent(
    model=model,
    system_prompt=SYSTEM_PROMPT,
)

Cet agent ne dispose pas encore d’outils, il se comporte donc beaucoup comme le modèle pur. Tout d’abord, il faut lui donner des instructions.

L’instruction système

llm/prompts.py contient une courte instruction qui indique au modèle son utilité, interdit l’invention de données et précise quel type de question correspond à quel type de recherche.

SYSTEM_PROMPT = """
You are an AI assistant for our demo application.
You help users understand the application and navigate the system.
Never invent data.
When information about the demo system
is required, use the available application tools.
When answering questions about the application,
use the documentation search tool.
Always answer in clear, conversational language.
""".strip()

L’instruction définit deux sources de vérité :

• les données de l’application (ce qui se trouve dans le compte de l’utilisateur) proviennent des outils de l’application ;
• Connaissances relatives à l’application (fonctionnement du produit) proviennent de la recherche dans la documentation.

Une précision avant d’aller plus loin. Les outils et RAG sont présentés séparément ci-dessous car cela facilite leur compréhension, mais dans l’agent final, la recherche dans la documentation est elle-même un outil. Il n’existe pas de mécanisme supplémentaire : l’agent voit une liste de fonctions pouvant être appelées, et la recherche dans le manuel en fait partie.

Étape 4 : Le premier outil

Un outil est une fonction que l’agent a le droit d’appeler. C’est ce qui permet à l’architecture de s’échelonner : au lieu d’inclure tous les données de l’application dans le prompt, on expose des opérations spécifiques et on laisse le modèle les demander uniquement lorsque c’est nécessaire pour répondre à une question.

Pour la démonstration, llm/tools/demo_tool.py définit un outil qui renvoie un bloc fixe d’informations sur le projet.

from langchain.tools import tool

@tool
def get_my_demo_data() -> str:
    """
    Return information about the demonstration data.
    This is just for demo data. But in production, make a more detailed instruction.
    """

    return """
    Project: Aperture Analytics Dashboard
    Owner: Jordan Lee
    Status: In Progress
    Team size: 6
    Budget: $84,000
    Deadline: 2026-11-15
    Description: An internal dashboard for visualizing customer usage
    metrics, built with FastAPI and React, integrating with the
    company's data warehouse.
    """.strip()

Deux points sont importants ici. Le décorateur @tool transforme une fonction Python ordinaire en outil LangChain, en dérivant son nom et son schéma d’arguments de sa signature. De plus, la documentation devient la description de l’outil, c’est ce que le modèle lit pour décider s’il doit l’appeler ou non. Dans un système réel, cette description mérite une attention particulière : il faut préciser exactement ce que l’outil retourne, quand c’est approprié et quand ce n’est pas le cas. Une description vague est l’une des raisons les plus fréquentes pour lesquelles un agent appelle le mauvais outil ou aucun outil du tout.

Enregistrez l’outil en le passant à create_agent :

from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import get_my_demo_data

agent = create_agent(
    model=model,
    tools=[
        get_my_demo_data,
    ],
    system_prompt=SYSTEM_PROMPT,
)

Votre code ne décide jamais quand la fonction doit s’exécuter. Lorsqu’un utilisateur demande "Quels sont les données de démonstration dont je dispose ?", le modèle reconnaît qu’il a besoin d’informations spécifiques au compte et appelle get_my_demo_data(). Si l’utilisateur demande "Qu’est-ce qu’une démonstration ?", aucune recherche n’est nécessaire et le modèle répond directement. Ce choix est effectué à chaque tour, au sein du cycle de l’agent.

Étape 5 : Création du pipeline RAG pour le manuel

L’agent peut désormais récupérer des données d’application, mais il ne sait toujours rien sur le fonctionnement du produit. Coller un manuel de 20 pages dans la prompt du système gaspillerait des tokens à chaque demande et rendrait difficile son mise à jour. RAG évite ces deux problèmes.

Il est important d’être précis quant à ce que RAG est et n’est pas. Aucun élément ne fait l’objet d’un entraînement ou d’un affinage sur la documentation. Lorsqu’une requête est soumise, le système recherche dans le manuel les passages les plus pertinents pour la question et fournit ces passages au modèle en tant que contexte, permettant ainsi au modèle de répondre à partir d’eux.

Le processus se déroule en deux phases :

1. Ingestion, qui s’exécute séparément de l’application web : charger le PDF, le diviser en fragments et stocker ces fragments avec leurs embeddings dans ChromaDB.
2. Récupération, qui a lieu au sein d’une requête : prendre la question, rechercher dans ChromaDB, sélectionner les meilleurs fragments et les transmettre au modèle.

Si vous souhaitez une vision plus large des concepts, l’aperçu du blog sur la récupération de connaissances fraîches sur demande les aborde en plus grande profondeur ; ici, nous nous concentrons sur l’implémentation.

Chargement du PDF

llm/rag/ingestion/loader.py utilise PyPDFLoader de LangChain, qui transforme chaque page du PDF en un Document.

from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader

PDF_PATH = Path("docs/manual.pdf")

def load_manual():
    loader = PyPDFLoader(
        str(PDF_PATH),
    )
    documents = loader.load()
    return documents

Document contient deux éléments : le texte extrait dans page_content, ainsi qu’un dictionnaire metadata décrivant l’origine de ce texte. Ce sont ces métadonnées qui permettent par la suite à une réponse de citer une page. Conceptuellement, chaque page chargée ressemble à ceci :

Document
├── page_content
│   └── "To create a new demo data..."
│
└── metadata
    ├── source: docs/manual.pdf
    └── page: 12

PyPDFLoader enregistre généralement à la fois un index de page basé zéro et une page_label lisible par l’homme. Le constructeur de contexte ci-dessous utilise page_label, qui correspond aux numéros de page que voient les lecteurs dans le PDF.

Découpage des pages en blocs

Rechercher des pages entières, sans parler du document dans son intégralité en un seul bloc, donne des résultats peu précis. llm/rag/ingestion/chunker.py divise les documents à l’aide de RecursiveCharacterTextSplitter.

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
)
from langchain_core.documents import Document

def chunk_documents(
    documents: list[Document],
) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=150,
        separators=[
            "\n\n",
            "\n",
            ". ",
            " ",
            "",
        ],
    )
    return splitter.split_documents(
        documents,
    )

Le séparateur vise des blocs d’environ 1 000 caractères avec une superposition de 150 caractères. La liste des séparateurs est testée dans l’ordre suivant : il préfère diviser aux limites des paragraphes, puis aux sauts de ligne, ensuite aux fins de phrase, puis aux espaces, et ne le fait qu’en dernier recours au milieu d’un mot. La superposition existe parce qu’un même fait peut se trouver à la frontière entre deux points de séparation ; en répétant un peu de texte des deux côtés, on diminue les chances que la phrase concernée soit coupée en deux.

Ces valeurs ne sont que des points de départ, pas des règles fixes. La taille idéale dépend de la manière dont vos documents sont rédigés et du niveau de précision requis pour la récupération des informations, il convient donc de les considérer comme des paramètres à ajuster en fonction des besoins réels. L’article du blog sur le découpage permettant de préserver les preuves aborde plus en détail ce compromis.

Une collection Chroma persistante

llm/rag/ingestion/chroma.py ouvre un PersistentClient qui stocke les données sur le disque et renvoie la collection du manuel, la créant en cas d’utilisation initiale.

import chromadb
from llm.rag.config import (
    CHROMA_PATH,
    MANUAL_COLLECTION_NAME,
)

def get_chroma_client():
    return chromadb.PersistentClient(
        path=CHROMA_PATH,
    )

def get_manual_collection():
    client = get_chroma_client()
    return client.get_or_create_collection(
        name=MANUAL_COLLECTION_NAME,
    )

Les chemins et les noms proviennent de llm/rag/config.py, qui lit les variables d’environnement et recourt à des valeurs par défaut appropriées en l’absence de celles-ci :

import os

CHROMA_PATH = os.getenv(
    "CHROMA_PATH",
    "./chroma_data",
)
MANUAL_PATH = os.getenv(
    "MANUAL_PATH",
    "docs/manual.pdf",
)
MANUAL_COLLECTION_NAME = os.getenv(
    "MANUAL_COLLECTION_NAME",
    "manual",
)

Ajoutez les entrées correspondantes dans .env :

CHROMA_PATH=./chroma_data
MANUAL_PATH=docs/manual.pdf
MANUAL_COLLECTION_NAME=manual

Aucun modèle d’incorporation n’est configuré nulle part, et c’est intentionnel pour la démonstration. Lorsqu’une collection est créée sans fonction d’incorporation explicite, Chroma utilise son modèle par défaut intégré : chaque fois que vous ajoutez des documents, Chroma calcule leurs embeddings par lui-même et les stocke à côté du texte et des métadonnées. Le modèle par défaut s’exécute localement et est téléchargé lors de la première utilisation, ce qui peut provoquer un arrêt temporaire lors du premier indexage en attendant son téléchargement.

Le résultat est un stockage de vecteurs local et persistant dans le répertoire chroma_data/. Comme il est entièrement dérivé du PDF, ajoutez-le à .gitignore aux côtés de .env.

Tâche d’indexation

llm/rag/ingestion/indexer.py relie les étapes d’ingestion entre elles.

from pathlib import Path
from llm.rag.config import MANUAL_PATH
from llm.rag.ingestion.loader import load_manual
from llm.rag.ingestion.chunker import chunk_documents
from llm.rag.ingestion.chroma import get_manual_collection

def index_manual():
    collection = get_manual_collection()
    if collection.count() > 0:
        print(
            f"Manual already indexed "
            f"({collection.count()} chunks)."
        )
        return
    manual_path = Path(
        MANUAL_PATH,
    )
    if not manual_path.exists():
        raise FileNotFoundError(
            f"Manual not found: {manual_path}"
        )
    documents = load_manual()
    print(
        f"Loaded {len(documents)} pages."
    )
    chunks = chunk_documents(
        documents,
    )
    print(
        f"Created {len(chunks)} chunks."
    )
    collection.add(
        ids=[
            f"manual-chunk-{i}"
            for i in range(len(chunks))
        ],
        documents=[
            chunk.page_content
            for chunk in chunks
        ],
        metadatas=[
            chunk.metadata
            for chunk in chunks
        ],
    )
    print(
        f"Stored {len(chunks)} chunks."
    )

Décortiquons ce qu’il fait. Il ouvre la collection et renvoie immédiatement s’il contient déjà des blocs, ce qui rend les exécutions répétées inoffensives. Il vérifie que le PDF existe, sinon il génère une erreur claire. Ensuite, il charge les pages, les divise en blocs, et ajoute tout à Chroma en une seule opération, avec des identifiants stables (manual-chunk-0, manual-chunk-1, etc.), les textes des blocs et leurs métadonnées. Les messages de progression indiquent le nombre de pages et de blocs traités.

Un piège découle de ce retour prématuré : si vous modifiez le manuel et exécutez à nouveau le script, rien ne se passe, car la collection n’est pas vide. Pour prendre en compte les modifications, vous devez supprimer la collection (ou le répertoire chroma_data/) avant de réindexer, ou remplacer la logique de protection par une autre qui effectue des mises à jour ou une reconstruction délibérée.

L’affichage ne montre pas scripts/index_manual.py ; il suffit qu’il importe index_manual et l’appelle. Exécutez-le une fois en tant que module depuis la racine du projet :

python -m scripts.index_manual

Cette seule exécution lit le PDF, le divise en parties, insère ces parties et les stocke. La sortie du terminal indique les comptes. Dans la démo simplifiée, le manuel est un PDF d’une seule page contenant une règle unique : « Les données de la démo ne peuvent être fournies qu’aux utilisateurs administrateurs », ce qui suffit pour vérifier si la récupération fonctionne. Par la suite, le démarrage de FastAPI ne touche pas du tout au PDF, car les vecteurs ont déjà été persistés.

Requête vers la collection

L’indexation seule ne donne rien à l’agent ; il a besoin d’un moyen de rechercher. Le fichier llm/rag/retrieval/retriever.py encapsule l’API de requête de Chroma.

from dataclasses import dataclass
from typing import Any
from llm.rag.ingestion.chroma import (
    get_manual_collection,
)

@dataclass
class RetrievedChunk:
    content: str
    metadata: dict[str, Any]
    distance: float

def retrieve_manual(
    query: str,
    n_results: int = 5,
) -> list[RetrievedChunk]:
    collection = get_manual_collection()
    results = collection.query(
        query_texts=[query],
        n_results=n_results,
        include=[
            "documents",
            "metadatas",
            "distances",
        ],
    )
    documents = results["documents"] or []
    metadatas = results["metadatas"] or []
    distances = results["distances"] or []
    retrieved_chunks = []
    for document, metadata, distance in zip(
        documents[0],
        metadatas[0],
        distances[0],
    ):
        retrieved_chunks.append(
            RetrievedChunk(
                content=document,
                metadata=dict(metadata)
                if metadata else {},
                distance=distance,
            )
        )
    return retrieved_chunks

La fonction envoie la question sous forme de query_texts, demande jusqu’à cinq résultats, et sollicite les documents, leurs métadonnées ainsi que leurs distances. Chroma renvoie une liste par requête, c’est pourquoi le code lit l’index [0] de chaque champ ; les solutions de secours or [] permettent de faire face à l’absence de champs. Chaque résultat est emballé dans une classe data RetrievedChunk afin que le reste du code ne dépende pas de la structure des réponses de Chroma.

Puisque la requête est intégrée au même modèle que les fragments stockés, la correspondance est sémantique. Une question comme « Comment ajouter de nouveaux données de démonstration ? » trouve des passages sur la création ou l’attribution de données de démonstration, même si ces derniers n’utilisent jamais les mots « ajouter de nouveaux ». La distance indique à quel point chaque résultat est proche ; une valeur plus basse signifie une similarité plus grande. Cette valeur est utile par la suite si vous souhaitez éliminer les correspondances faibles au lieu d’envoyer toujours cinq fragments au modèle.

Avec cela en place, la partie de récupération de RAG fonctionne. Il ne reste plus qu’à transmettre les résultats au modèle.

Transformer les fragments en contexte

llm/rag/context.py formate les résultats en une seule chaîne que le modèle peut lire.

from llm.rag.retrieval.retriever import (
    RetrievedChunk,
)

def build_context(
    chunks: list[RetrievedChunk],
) -> str:
    context_parts = []
    for chunk in chunks:
        page = chunk.metadata.get(
            "page_label",
        )
        context_parts.append(
            f"Source: User Guide, page {page}\n"
            f"{chunk.content}"
        )
    return "\n\n---\n\n".join(
        context_parts,
    )

Chaque morceau est précédé d’une ligne de source indiquant le guide utilisateur et son numéro de page, et les morceaux sont séparés par un séparateur. C’est cette ligne de source qui permet au modèle de préciser d’où provient une réponse, et elle offre aux utilisateurs un moyen de vérifier cela.

Étape 6 : Mettre à disposition le manuel en tant qu’outil

Le pipeline est complet : les pages sont chargées et divisées en morceaux, ces derniers se trouvent dans ChromaDB, et il est possible de les retrouver et de les formater. Cependant, l’agent n’a aucune idée que tout cela existe. C’est là que la note architecturale mentionnée précédemment s’avère utile : la recherche de documentation devient alors un simple outil parmi d’autres.

llm/tools/manual_tool.py définit search_user_manual, qui prend une requête, récupère cinq extraits et les renvoie sous forme de contexte formaté. S’il n’y a rien en retour, il affiche un message indiquant clairement que le manuel ne couvre pas cette question, ce qui permet au modèle de transmettre une information honnête plutôt qu’une chaîne vide.

from langchain.tools import tool
from llm.rag.context import build_context
from llm.rag.retrieval.retriever import retrieve_manual

@tool
def search_user_manual(
    query: str,
) -> str:
    """
    Search the application user manual.
    Use this tool when the user asks about application
    behavior, instructions, rules, limitations, or
    how something works.
    """
    chunks = retrieve_manual(
        query=query,
        n_results=5,
    )
    if not chunks:
        return (
            "The manual does not contain enough "
            "information to answer this question."
        )
    return build_context(
        chunks,
    )

Comme précédemment, la documentation est l’annonce de l’outil auprès du modèle. Elle indique d’utiliser cet outil pour les questions concernant le comportement, les instructions, les règles, les limites et le fonctionnement des choses, ce qui correspond au prompt du système.

Enregistrez maintenant les deux outils auprès de l’agent :

from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import (
    get_my_solar_system,
)
from llm.tools.manual_tool import (
    search_user_manual,
)

agent = create_agent(
    model=model,
    tools=[
        get_my_solar_system,
        search_user_manual,
    ],
    system_prompt=SYSTEM_PROMPT,
)

Prêtez attention à l’import dans cette liste : il fait référence à get_my_solar_system, un nom provenant de l’application complète, tandis que le module d’outil de démonstration définit get_my_demo_data. Utilisez get_my_demo_data tant dans l’import que dans la liste tools, sinon le module ne pourra pas être importé.

Pourquoi une conception basée sur des outils reste efficace à mesure que l’application grandit

L’agent n’a jamais besoin d’une seule instruction détaillée décrivant tout ce que l’application sait. Il dispose plutôt de capacités ciblées et contrôlées. Ajouter une fonctionnalité à l’assistant signifie écrire un nouvel outil et le enregistrer ; la couche HTTP ne change pas. La démonstration conserve un seul outil de données et un seul outil de documentation afin que le schéma reste facile à suivre, mais la même structure permet d’intégrer un ensemble bien plus important d’outils dans l’application complète.

Étape 7 : Transmettre l’utilisateur authentifié à l’agent

Outil de démonstration : il renvoie toujours des données codées en dur. Un véritable outil doit savoir qui pose la demande, et l’application le sait déjà : FastAPI a identifié l’utilisateur via la dépendance d’authentification. Ce qui manque, c’est de transmettre cet utilisateur lors du lancement de l’agent. LangChain appelle cela le contexte d’exécution.

Définissez la structure du contexte dans llm/context.py :

from dataclasses import dataclass
from auth.dependencies import MockUser

@dataclass
class AgentContext:
    user: MockUser

Cet objet est fourni lorsque l’agent est invoqué, et cette distinction est importante. L’utilisateur représente des informations liées à une requête spécifique. Elles appartiennent à la demande HTTP en cours, et non à la conversation, et ne doivent jamais être stockées sous forme de message que le modèle pourrait lire ou réécrire. Le fait de les exclure de l’historique des messages signifie également qu’une instruction ne peut pas amener l’agent à agir comme un utilisateur différent. Dans l’application complète, le même objet de contexte contient également des éléments tels que la session de base de données et l’ID du projet en cours d’édition.

Le code de démonstration s’arrête à la définition de la classe, il vous reste donc deux connexions à établir, et il est utile de consulter la documentation actuelle de LangChain pour connaître l’API exacte. Tout d’abord, déclarez le schéma lors de la création de l’agent, généralement en utilisant l’argument context_schema=AgentContext avec la fonction create_agent. Ensuite, faites en sorte que les outils le lisent : dans LangChain 1.x, un outil peut accepter un paramètre de exécution (par exemple annoté comme ToolRuntime[AgentContext]) et lire les informations de l’utilisateur depuis son attribut context, qui reste caché aux yeux du modèle concernant les arguments de l’outil. C’est là qu’une véritable fonction get_my_demo_data chercherait les enregistrements correspondant à user.id.

Étape 8 : Une couche d’orchestration qui streame

Au lieu d’appeler l’agent depuis l’intérieur du routeur, placez l’interaction dans llm/orchestrator.py. Le routeur reste ainsi concentré sur HTTP, tandis que l’orchestrateur gère la transformation d’un message en exécution d’un agent.

from collections.abc import Iterator
from langchain_core.messages import (
    AIMessage,
    AIMessageChunk,
    BaseMessage,
    ToolMessage,
)
from langchain_core.runnables import RunnableConfig
from llm.agent import agent
from llm.context import AgentContext

def chat_stream(
    user_message: str,
    user,
) -> Iterator[str]:
    config: RunnableConfig = {
        "configurable": {
            "thread_id": f"user:{user.id}",
        }
    }
    context = AgentContext(
        user=user,
    )
    for chunk, metadata in agent.stream(
        {
            "messages": [
                {
                    "role": "user",
                    "content": user_message,
                }
            ]
        },
        config=config,
        context=context,
        stream_mode="messages",
    ):
        if not isinstance(
            chunk,
            BaseMessage,
        ):
            continue
        if isinstance(
            chunk,
            ToolMessage,
        ):
            continue
        if not isinstance(
            chunk,
            (
                AIMessage,
                AIMessageChunk,
            ),
        ):
            continue
        if isinstance(
            chunk.content,
            str,
        ):
            yield chunk.content

Il y a beaucoup de choses dans cette fonction, alors abordez-la morceau par morceau.

L’ID de thread sélectionne la conversation

Le premier bloc construit la configuration d’exécution :

config = {
    "configurable": {
        "thread_id": f"user:{user.id}",
    }
}

thread_id comme clé. Chaque exécution avec le même thread ID poursuit la même conversation, et ses messages peuvent être lus ultérieurement. Ici, le thread ID est dérivé de l’ID de l’utilisateur, ce qui signifie que chaque utilisateur dispose d’une seule conversation. Cela convient pour une démonstration ; une application réelle créerait des IDs de conversation appropriés, permettrait plusieurs conversations par utilisateur et vérifierait à chaque requête que l’appelant possède bien le thread auquel on accède.

La description suppose l’existence d’un checkpointer, mais aucun des extraits d’agent ne le satisfait. Sans celui-ci, thread_id n’a aucun effet et rien n’est mémorisé entre les requêtes. Créez un seul InMemorySaver dans llm/agent.py et transmettez-le à create_agent via son argument checkpointer, afin que tant l’agent que les fonctions d’historique puissent importer la même instance.

Le contexte de exécution est transmis avec l’exécution

Ensuite, l’orchestrateur enveloppe l’utilisateur dans l’objet contextuel :

context = AgentContext(
    user=user,
)

Cet objet est transmis en tant que context= à agent.stream(). C’est par ce biais que l’identité établie dans FastAPI atteint l’agent, et par lui, les outils.

Filtrage du flux

agent.stream() est appelé avec stream_mode="messages", ce qui génère des paires composées d’un morceau de message et de métadonnées au fur et à mesure que le modèle produit des tokens. Tous les éléments de ce flux ne doivent pas parvenir à l’utilisateur. La boucle ignore tout ce qui n’est pas un message LangChain, élimine les objets ToolMessage (les résultats bruts d’outils tels que du texte manuel récupéré), ne conserve que les messages et morceaux de messages générés par l’IA, et affiche leur contenu lorsqu’il s’agit d’une chaîne de caractères simple. Le contenu fourni par certains fournisseurs sous forme de liste d’éléments est silencieusement ignoré lors de cette dernière vérification ; si vous changez de modèle et que les réponses sont vides, c’est là qu’il faut chercher.

Étape 9 : Retourner une réponse en flux

Il a été délibéré de faire de chat_stream() un générateur. Attendre la réponse complète avant d’envoyer un octet oblige l’utilisateur à regarder un indicateur de chargement, et les réponses des LLM peuvent prendre plusieurs secondes. Le streaming affiche les premiers mots presque immédiatement, ce qui donne l’impression à l’assistant d’être bien plus réactif.

StreamingResponse de FastAPI accepte directement un générateur. Mettez à jour routers/chat.py :

from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from auth.dependencies import (
    MockUser,
    get_current_user,
)
from llm.orchestrator import chat_stream

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    return StreamingResponse(
        chat_stream(
            user_message=message,
            user=current_user,
        ),
        media_type="text/plain",
    )

La réponse est envoyée en format text/plain, chaque élément généré étant écrit sur la connexion dès sa production. Comme chat_stream est un générateur ordinaire (synchrone), Starlette l’itère dans un thread de travail, ce qui évite de bloquer le boucle d’événements. Si vous avez plus tard besoin d’événements structurés côté client (par exemple, pour afficher « recherche du manuel... » pendant l’exécution d’une fonction), les Server-Sent Events constituent une étape naturelle suivante.

Le chemin complet de la requête est maintenant le suivant : le client envoie une demande vers /chat, FastAPI authentifie l’utilisateur, chat_stream() lance l’exécution de l’agent, le modèle décide s’il faut appeler une outil, l’outil correspondant s’exécute et renvoie son résultat, le modèle écrit la réponse, puis les tokens sont renvoyés au client.

La caractéristique essentielle est que FastAPI ne lance jamais le modèle lui-même. Le routeur gère les communications HTTP, l’orchestrateur pilote l’agent, c’est l’agent qui détermine quelles informations il a besoin, et les outils s’occupent des récupérations. Chaque couche peut être modifiée sans affecter les autres.

Étape 10 : Récupération de l’historique des conversations

Puisque l’agent est sauvegardé à intervalles réguliers, son état est stocké après chaque interaction. Cela permet de montrer à un utilisateur revenant ses conversations précédentes. Deux fonctions d’aide s’occupent de cette tâche.

Lecture du fichier de sauvegarde brut

La première fonction charge le dernier point de contrôle d’un thread et renvoie le canal messages, ou une liste vide si le thread n’a jamais été utilisé :

def get_conversation_messages(
    thread_id: str,
) -> list[BaseMessage]:
config: RunnableConfig = {
        "configurable": {
            "thread_id": thread_id,
        }
    }
    checkpoint = checkpointer.get(
        config,
    )
    if checkpoint is None:
        return []
    return checkpoint[
        "channel_values"
    ].get(
        "messages",
        [],
    )

Si vous copiez ce code, corrigez l’indentation de l’affectation config : elle doit être indentée à l’intérieur du corps de la fonction, sinon Python génère une erreur. La fonction doit également importer BaseMessage, RunnableConfig ainsi que l’instance partagée checkpointer.

Ce qui est renvoyé correspond à l’état brut de l’agent, et celui-ci inclut bien plus que le contenu du chat mémorisé par l’utilisateur. Lorsque l’agent appelle une outil, LangGraph enregistre un message d’IA contenant l’appel à l’outil ainsi qu’un message distinct indiquant le résultat. Il s’agit de détails d’implémentation que l’interface utilisateur n’a pas besoin d’interpréter.

Formater les messages pour l’affichage

La deuxième fonction crée l’affichage destiné à l’utilisateur :

def get_conversation_messages_for_display(
    thread_id: str,
) -> list[dict[str, str]]:
    display = []
    for message in get_conversation_messages(
        thread_id,
    ):
        if isinstance(
            message,
            HumanMessage,
        ):
            content = _extract_text_content(
                message.content,
            )
            if content.strip():
                display.append(
                    {
                        "type": "human",
                        "content": content,
                    }
                )
            continue
        if isinstance(
            message,
            AIMessage,
        ):
            content = _extract_text_content(
                message.content,
            )
            if content.strip():
                display.append(
                    {
                        "type": "ai",
                        "content": content,
                    }
                )
    return display

Elle ne conserve que les objets HumanMessage et AIMessage, extrait leur texte, élimine ceux qui sont vides, et renvoie des dictionnaires simples contenant un type et un content. Les messages d’outil n’apparaissent jamais car ils ne correspondent à aucun des deux types acceptés. Les messages d’IA vides sont également éliminés, ce qui est important car un message d’IA qui ne fait que demander une appel à outil n’a généralement pas de texte.

Cette fonctionnalité repose sur une aide, _extract_text_content, qui n’est pas montrée. Son rôle est de renvoyer le contenu tel quel s’il s’agit d’une chaîne, et, s’il s’agit d’une liste de parties de contenu, de les assembler ensemble. Il faut également importer HumanMessage et AIMessage.

L’endpoint d’historique

Exposez la vue d’affichage via une route GET dans routers/chat.py. Elle dérive le même identifiant de thread utilisé par l’endpoint de chat et le renvoie accompagné des messages.

@chat_router.get("/current")
def get_current_conversation(
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    thread_id = (
        f"user:{current_user.id}"
    )
    messages = (
        get_conversation_messages_for_display(
            thread_id,
        )
    )
    return {
        "thread_id": thread_id,
        "messages": messages,
    }

Une interface utilisateur de chat peut appeler cette fonction lors de l’ouverture de la page afin d’afficher la conversation existante avant même que l’utilisateur n’écrive quoi que ce soit :

GET /chat/current

La réponse a l’air de ceci :

{
    "thread_id": "user:user-123",
    "messages": [
        {
            "type": "human",
            "content": "How much demo data do I have?"
        },
        {
            "type": "ai",
            "content": "You currently have 18 demo data."
        }
    ]
}

Pourquoi ne pas renvoyer l’état brut ?

L’état interne de l’agent et la conversation que voit l’utilisateur sont deux choses distinctes. À mesure que de nouvelles fonctionnalités sont ajoutées, l’état accumule des appels d’outils, leurs résultats, des étapes intermédiaires, des métadonnées du modèle ainsi que d’autres informations de suivi. Renvoyer tout cela lierait l’interface utilisateur aux composants internes de l’agent et pourrait divulguer des résultats d’outils que vous ne souhaitiez pas afficher. Le backend doit définir quelles sont les informations de conversation publiques et n’en renvoyer que celles-ci.

Comment s’effectue le flux d’une seule requête de bout en bout

Lorsque tous les éléments sont en place, il est important de préciser un point : le modèle ne touche jamais à votre base de données ni à vos fichiers PDF. Il ne peut que demander l’exécution d’une fonction. Cette fonction, exécutée en tant que code Python ordinaire avec des contrôles d’accès standard, effectue l’opération et renvoie du texte, que le modèle utilise ensuite pour composer sa réponse. C’est cette frontière qui permet d’intégrer l’assistant en toute sécurité dans une application utilisant des données réelles.

Essai dans Swagger UI

FastAPI génère automatiquement une documentation interactive, il n’est donc pas nécessaire d’utiliser un client distinct pour les tests. Démarrez le serveur :

python -m uvicorn main:app --reload

Ensuite, ouvrez la documentation API interactive disponible à l’adresse /docs, configurez l’en-tête X-Demo-User, et essayez trois interactions :

• Une question liée aux données, comme demander quelles données de démonstration on dispose. L’agent doit déterminer qu’il a besoin de l’outil de démonstration, l’appeler et répondre en se basant sur les détails du projet retournés. Dans l’application complète, le même type d’outil interroge les enregistrements réels de l’utilisateur.
• Une question liée à la documentation, comme demander qui peut recevoir des données de démonstration. L’agent doit effectuer une recherche manuelle, récupérer la règle indexée correspondante et répondre que seuls les utilisateurs administrateurs le peuvent.
• Le point de terminaison historique, GET /chat/current, qui doit retourner les échanges de l’humain et de l’IA concernant les deux questions précédentes, sans aucun message provenant d’un outil.

Si les deux premières questions produisent des réponses basées sur les résultats de l’outil et que la troisième affiche une transcription claire, chaque niveau fonctionne correctement.

Au moment de le mettre en production

La démo simplifie délibérément plusieurs composants. Ce sont ceux qu’il convient de revoir avant que des utilisateurs réels ne dépendent du service.

État de conversation durable

InMemorySaver est parfait pour le développement, mais tout ce qu’il contient disparaît lorsque le processus redémarre, et il ne peut pas être partagé entre plusieurs instances API derrière un load balancer. Utilisez un point de contrôle soutenu par une base de données ou un autre système de stockage durable afin que les conversations survivent aux déploiements et que chaque instance voie le même état. Pour savoir ce que stocke réellement l’optimiseur en mémoire et comment, consultez le guide du blog sur la manière dont InMemorySaver organise les points de contrôle, les écritures et les blobs.

Déploiement réel d’un stockage vectoriel

Un répertoire Chroma local convient bien pour démontrer le pipeline, mais ce n’est pas une infrastructure de production. Exécutez Chroma en tant que service persistant, ou passez à une base de données vectorielle gérée adaptée à votre stack. Ce que vous choisirez doit être persistant, sauvegardé et accessible depuis chaque instance d’application.

L’indexation reste en dehors de l’API

La démonstration prend déjà une décision importante correctement : l’indexation est un script distinct que vous exécutez séparément, tandis que l’API se contente de récupérer des données.

python -m scripts.index_manual

Le serveur ne charge jamais le PDF, ne le divise pas en parties ni ne calcule de embeddings au démarrage. L’indexation s’effectue en mode hors ligne ; la récupération fait partie du traitement d’une demande. Les séparer signifie que l’API n’a pas besoin de détecter des modifications dans la documentation ni de reconstruire quoi que ce soit.

Dans un environnement de production, passez à l’étape suivante en exécutant le même code d’indexation comme une tâche dédiée d’ingestion, déclenchée depuis un pipeline de déploiement, selon un calendrier prédéfini, ou par un travailleur dès que de nouvelle documentation est téléchargée. L’architecture reste identique ; la tâche devient simplement automatisée, reproductible et pouvant être déployée de manière indépendante. Les responsabilités se divisent alors clairement :

• Tâche d’ingestion : charger les documents, les diviser en blocs, les intégrer, mettre à jour le stockage vectoriel.
• Service FastAPI : recevoir les questions, récupérer les blocs pertinents, générer des réponses.
• Stockage vectoriel : conserver les représentations indexées utilisées lors des requêtes.

N’oubliez pas la protection de retour précoce présente dans l’indexeur lorsqu’on le met en automatique ; une tâche qui saute silencieusement la réindexation est pire que pas de tâche du tout.

Un modèle d’incorporation explicite

En s’appuyant sur la fonction d’incorporation par défaut de Chroma, la démonstration n’exige aucune configuration supplémentaire, mais un système de production doit choisir et configurer explicitement son modèle d’incorporation. Cela permet d’obtenir des résultats reproductibles et de contrôler la qualité, le coût, la latence ainsi que l’endroit où les incorporations sont calculées. Une règle est incontournable : le même modèle d’incorporation doit être utilisé tant pour l’indexation que pour les requêtes. En changer signifie devoir réindexer tout le système.

Observabilité et gestion des erreurs

Lorsqu’un agent est en service, savoir ce qu’il a fait est tout aussi important que de le faire fonctionner correctement. Une seule requête peut impliquer plusieurs appels à des modèles, un ou plusieurs appels à des outils et une étape de récupération avant d’obtenir la réponse finale. Ne consigner que cette réponse finale ne vous dit presque rien en cas de problème. Instrumentez l’ensemble du flux :

• Appels aux outils : quels outils ont été utilisés, avec quels arguments, et combien de temps chaque appel a duré.

L’objectif est que l’agent ne soit jamais une boîte noire. Pour chaque exécution, vous devez pouvoir indiquer ce qu’il a fait, quels outils il a utilisés, combien de temps chaque étape a duré et où il a échoué. Les outils spécifiques dépendent de votre stack technique ; le principe, lui, reste identique.

Points clés

• Vous n’avez pas besoin d’une plateforme d’IA complexe pour ajouter un assistant utile à une application lourde en fonctionnalités. Commencez avec le plus petit ensemble d’éléments capable de résoudre un problème réel.
• Considérez la recherche de documentation comme un outil parmi d’autres. L’agent dispose ainsi d’une méthode unique et uniforme pour accéder à la fois aux connaissances sur le produit et aux données des utilisateurs.
• Gardez l’identité dans le contexte en temps de exécution, et non dans les messages. L’utilisateur fait partie de la demande, et les outils doivent y accéder directement.
• Séparez HTTP, orchestration, agent et outils. Chaque couche doit rester simple, et ajouter une nouvelle fonctionnalité signifie simplement ajouter un outil.

Lectures complémentaires

• Mémoire hybride d’agent : Fusion de BM25 et recherche vectorielle avec RRF en Python — Découvrez pourquoi la recherche vectorielle pure échoue comme mémoire d’agent, comment la fusion de rangs réciproques combine BM25 et les résultats densifs en Python, et quand les résumés GraphRAG sont utiles.
• Concevoir une mémoire agente en quatre niveaux avec LangGraph et Amazon Bedrock — Apprenez à fournir aux agents LLM une mémoire épisodique, sémantique et procédurale fonctionnelle sur Bedrock et LangGraph, ainsi qu’à les protéger contre le poisonage, les fuites de PII et les problèmes liés aux utilisateurs.
• Permettre à Gemini de choisir la source : FAISS, Tavily et réponses directes dans LangGraph — Créez un petit flux de travail LangGraph dans lequel Gemini dirige chaque question vers une base de connaissances FAISS, une recherche web Tavily ou une réponse directe, avec un routage des résultats structurés.
• Routage des questions entre outils SQL et recherche web avec un agent Gemini — Comment un agent de appel d’outils LangChain sur Vertex AI choisit parmi trois outils SQLite texte vers SQL et la recherche web en temps réel, ainsi que les bugs liés aux données, aux dépendances et à l’authentification à anticiper.
• Diffusion des agents LangGraph vers React sans fuites d’informations internes des outils — Comment serialiser la sortie astream_events de LangGraph pour Server-Sent Events, masquer les données sensibles des appels d’outils dans FastAPI et afficher des badges d’outil conviviaux dans React.
• Évaluer le comportement des agents : résultats, trajectoires et retour d’information sur la production — Comment juger les systèmes agents au-delà de la réponse finale : associer LangChain, LangGraph et LangSmith à des rôles, évaluer les trajectoires et le contexte, ainsi que prendre en compte les échecs de production dans les analyses.