L'assistant documentaire : le cœur de la stack souveraine

Le socle est en place : modèles, base vectorielle, passerelle. Cette deuxième étape du fil rouge bâtit le cœur de la stack — l'assistant documentaire. Il assemble, en un seul composant, tout le parcours RAG : l'ingestion d'un corpus dans Qdrant, la recherche hybride dense et lexicale, le re-ranking par cross-encoder, une boucle RAG agentique pilotée par LangGraph, deux étages de mémoire — court terme pour la conversation, long terme pour les faits durables sur l'utilisateur — et une réponse en streaming. Un routeur lui ouvre une seconde voie : les questions d'infrastructure ne passent pas par le RAG mais par le serveur MCP FinOps de la stack. Le tout exposé par une interface Chainlit, entièrement branché sur le socle. C'est l'étape qui transforme une infrastructure en un service qui répond. Public visé : développeur ayant monté le socle et voulant l'exploiter.

Ce que vous allez apprendre

Assembler en un assistant les briques RAG du parcours.
Ingérer un corpus dans le Qdrant du socle.
Combiner recherche hybride et re-ranking.
Orchestrer une boucle RAG agentique avec LangGraph.
Doter l'assistant d'une mémoire court terme et long terme.
Aiguiller les questions d'infrastructure vers le serveur MCP FinOps.
Diffuser la réponse en streaming dans une interface Chainlit.

Prérequis

Le socle de la stack démarré, modèles tirés.
Avoir parcouru le RAG agentique, le re-ranking et la mémoire des agents.
Python 3.10+.

L'assistant : ce qu'il assemble

L'assistant n'introduit aucune technique nouvelle. Il branche ensemble des briques déjà construites et validées, chacune dans son guide.

Brique	Rôle dans l'assistant	Guide d'origine
Chunking	Découper le corpus	Chunking
Embeddings + Qdrant	Indexer et stocker	Embeddings · Qdrant
Recherche hybride	Dense + BM25, fusion RRF	RAG avancé
Re-ranking	Reclasser les candidats	Re-ranking
RAG agentique	Boucle cherche / évalue / reformule	RAG agentique
Mémoire court et long terme	Conversation et faits durables	Mémoire des agents
Routeur + brique MCP FinOps	Aiguiller les questions d'infrastructure	Observabilité FinOps
Chainlit	Interface web et streaming	Chainlit

L'exercice du fil rouge est précisément là : vérifier que ces briques s'emboîtent. Un module unique, assistant.py, les enchaîne ; un second, app.py, l'expose. C'est tout.

Ingestion : du corpus à Qdrant

L'ingestion prépare le corpus une fois pour toutes : découper, vectoriser, indexer. Elle écrit dans le Qdrant du socle, et prépare en parallèle l'index lexical BM25.

def ingerer() -> int:
    """Découpe, vectorise et indexe le corpus ; prépare l'index BM25."""
    global _CHUNKS, _BM25
    _CHUNKS = [...]                      # découpage du corpus en chunks
    _BM25 = BM25Okapi([_tokeniser(c["texte"]) for c in _CHUNKS])
    if pas_déjà_indexé:
        QDRANT.create_collection(COLLECTION, vectors_config=...)
        vecteurs = _vectoriser([c["texte"] for c in _CHUNKS])
        QDRANT.upsert(COLLECTION, points=[...])
    return len(_CHUNKS)

Deux index sont construits côte à côte : les vecteurs dans Qdrant pour la recherche dense, l'index BM25 en mémoire pour la recherche lexicale. La recherche hybride a besoin des deux. L'ingestion est idempotente — si la collection est déjà peuplée, elle ne ré-indexe pas.

Recherche hybride et re-ranking

La recherche de l'assistant enchaîne deux étapes vues dans le guide sur le RAG avancé et celui sur le re-ranking — ici combinées.

D'abord la recherche hybride : la requête est cherchée en dense (Qdrant) et en lexicale (BM25), et les deux classements sont fusionnés par Reciprocal Rank Fusion. Puis le re-ranking : un cross-encoder note chaque candidat du lot fusionné et le reclasse.

def rechercher(question: str, k: int = 3) -> list[dict]:
    """Recherche hybride (dense + BM25), fusion RRF, puis re-ranking."""
    dense = [p.id for p in QDRANT.query_points(COLLECTION, query=v, limit=6).points]
    lexical = meilleurs_indices_bm25(question, n=6)
    fusion = fusion_rrf([dense, lexical])[:6]
    candidats = [_CHUNKS[i] for i in fusion]
    notes = _reranker().predict([(question, c["texte"]) for c in candidats])
    return [c for c, _ in classer_par_note(candidats, notes)[:k]]

Le schéma est celui, éprouvé, des guides : une présélection large par la recherche hybride, un reclassement fin par le cross-encoder. L'assistant ne fait que les chaîner.

Le RAG agentique : la boucle de recherche

La recherche, aussi bonne soit-elle, peut échouer sur une question mal formulée. L'assistant l'enveloppe donc dans une boucle agentique — un graphe LangGraph qui se corrige.

Point important : ce graphe ne génère pas la réponse. Il s'arrête dès qu'il tient un contexte jugé suffisant. La génération, elle, vient ensuite — pour pouvoir être diffusée en streaming.

class EtatRecherche(TypedDict):
    question: str
    requete: str
    contexte: list[dict]
    suffisant: bool
    tentatives: int

Trois nœuds composent le graphe. rechercher applique la recherche hybride et le re-ranking. evaluer demande au modèle si le contexte récupéré suffit. reformuler réécrit la question quand il ne suffit pas. Un routeur boucle de evaluer vers reformuler tant que le contexte est insuffisant — puis termine, dans la limite d'un plafond de tentatives.

constructeur.add_edge("rechercher", "evaluer")
constructeur.add_conditional_edges("evaluer", router, ["reformuler", END])
constructeur.add_edge("reformuler", "rechercher")   # la boucle de correction

C'est le pattern du guide sur le RAG agentique — recentré ici sur sa fonction propre : fournir le bon contexte. La réponse est une étape séparée.

Le garde-fou : répondre, ou dire qu'on ne sait pas

La boucle agentique améliore la recherche, mais ne la rend pas infaillible. Une question hors du corpus ne trouvera jamais de bon contexte, même reformulée trois fois. Le graphe finit alors par s'arrêter sur son plafond de tentatives, avec un contexte marqué suffisant: false.

Que faire de ce cas ? Le piège serait de générer quand même : un modèle privé de contexte utile invente — c'est l'hallucination, et pour un assistant de documentation, c'est le pire des défauts. L'assistant fait donc l'inverse, et deux mesures l'y obligent.

D'abord le prompt système, qui pose la règle : répondre uniquement à partir du contexte fourni, et dire franchement quand l'information n'y est pas.

SYSTEME = ("Tu es un assistant documentaire. Réponds en français, "
           "uniquement à partir du contexte documentaire fourni. Si ce "
           "contexte ne contient pas la réponse, dis-le franchement "
           "plutôt que d'inventer. ...")

Ensuite, quand le graphe sort sur suffisant: false, l'assistant ajoute une note explicite au prompt : il signale au modèle que la recherche n'a rien trouvé de probant, et lui demande de l'admettre.

if not ctx.get("suffisant", True):
    blocs.append("Note : la recherche n'a pas trouvé de passage "
                 "clairement pertinent. Si le contexte ci-dessus ne "
                 "répond pas à la question, dis-le franchement.")

Le résultat : à une question hors corpus, l'assistant répond « le contexte documentaire ne contient pas cette information » au lieu d'un paragraphe inventé. Pour un assistant de documentation, savoir dire « je ne sais pas » vaut mieux qu'une réponse fausse — c'est une exigence de production, pas un détail de confort.

Deux étages de mémoire

Un assistant sans mémoire traite chaque question isolément — il ne suit pas une conversation et ne se souvient de rien. Comme le détaille le guide sur la mémoire des agents, on distingue deux étages, et l'assistant les implémente tous les deux.

La mémoire court terme est l'historique de la conversation en cours. L'interface la tient dans la session de l'utilisateur ; à chaque question, les derniers tours sont réinjectés dans le prompt. C'est ce qui permet de comprendre une question de suivi : « et plusieurs conteneurs peuvent-ils le partager ? » n'a de sens que si l'assistant a gardé le tour précédent.

La mémoire long terme retient des faits durables sur l'utilisateur, d'une session à l'autre. Elle vit dans une collection Qdrant séparée — distincte du corpus, et persistante.

def rappeler(user_id: str, question: str, k: int = 3) -> list[str]:
    """Retrouve les faits long terme pertinents pour la question."""
    vecteur = _vectoriser([question])[0]
    filtre = Filter(must=[FieldCondition(key="user_id",
                                         match=MatchValue(value=user_id))])
    points = QDRANT.query_points(MEMOIRE, query=vecteur, limit=k,
                                 query_filter=filtre, with_payload=True).points
    return [p.payload["fait"] for p in points]

def memoriser(user_id: str, question: str, reponse: str) -> str | None:
    """Extrait un fait durable de l'échange et le range en mémoire long terme."""
    extrait = LLM.chat.completions.create(model=MODELE_CHAT, messages=[...])
    # ... si l'échange contient un fait durable, on le vectorise et l'indexe.

Le mécanisme est exactement celui du guide sur la mémoire : on extrait un fait avec le modèle, on le vectorise, on le range dans Qdrant ; à la question suivante, on le retrouve par similarité, filtré par user_id. C'est un RAG — appliqué non au corpus, mais à l'historique de l'utilisateur. Les souvenirs retrouvés sont injectés dans le prompt aux côtés du contexte documentaire.

La réponse en streaming

Faire patienter l'utilisateur devant un écran figé pendant que le modèle rédige donne une impression de lenteur. La réponse doit s'afficher au fil de sa génération — en streaming.

C'est la raison pour laquelle le graphe agentique ne génère pas la réponse : la génération est isolée, pour être diffusée. On sépare donc deux temps. D'abord recuperer_contexte : il lance le graphe agentique et la mémoire long terme, et renvoie le contexte. Ensuite la génération en streaming, à partir de ce contexte.

def recuperer_contexte(question: str, user_id: str = "demo") -> dict:
    """Lance le RAG agentique et la mémoire long terme ; renvoie le contexte."""
    souvenirs = rappeler(user_id, question)
    etat = GRAPHE.invoke({"question": question, "requete": question})
    return {"contexte": etat["contexte"], "souvenirs": souvenirs,
            "sources": sorted({c["source"] for c in etat["contexte"]})}

La génération utilise un client asynchrone et l'option stream=True : la réponse arrive en fragments, transmis un à un à l'interface.

flux = await LLM_ASYNC.chat.completions.create(
    model=MODELE_CHAT, messages=messages, stream=True)
async for fragment in flux:
    token = fragment.choices[0].delta.content or ""
    if token:
        await msg.stream_token(token)

Ce découpage — graphe pour le contexte, génération streamée pour la réponse — est ce qui rend l'assistant vivant : le RAG agentique fait son travail en coulisses, puis la réponse se compose sous les yeux de l'utilisateur.

Tout passe par la passerelle

Un point d'architecture traverse tout l'assistant : il ne parle jamais directement à Ollama. Embeddings, évaluations, générations — streamées ou non — passent par la passerelle LiteLLM du socle.

# Deux clients, tous deux pointés sur la passerelle — jamais sur Ollama.
LLM = OpenAI(base_url="http://localhost:4000/v1", api_key=CLE_LITELLM)
LLM_ASYNC = AsyncOpenAI(base_url="http://localhost:4000/v1", api_key=CLE_LITELLM)

L'assistant est ainsi découplé du détail des modèles, l'accès est gouverné par la clé maître du proxy, et la stack reste cohérente : un point d'entrée unique aux modèles. Le client synchrone sert la logique ; l'asynchrone, le streaming.

Deux voies : documentation et infrastructure

Jusqu'ici, l'assistant ne sait répondre que sur le corpus documentaire. Mais une stack souveraine se supervise aussi : « combien de conteneurs tournent ? », « quel est le coût de la stack ? ». Ces questions n'ont rien à voir avec le RAG — elles interrogent l'infrastructure, pas la documentation.

L'assistant gère les deux par un routeur. La fonction aiguiller classe chaque question avant tout traitement : une question documentaire suit le RAG agentique ; une question d'infrastructure ouvre une session MCP vers le serveur MCP FinOps du fil rouge.

def repondre(question, historique=None, user_id="demo"):
    """Le routeur choisit la voie ; les deux convergent vers une réponse."""
    if aiguiller(question, historique) == "finops":
        prep = asyncio.run(preparer_finops(question))   # voie infrastructure
        messages = prep["messages"]
    else:
        ctx = recuperer_contexte(question, user_id)     # voie documentaire
        messages = construire_messages(question, ctx, historique)
    # ... puis génération de la réponse à partir de `messages`

Le routeur reçoit l'historique de conversation. Une question comme « quel est le nom de mon conteneur ? » s'appuie sur un tour précédent : il la classe alors en documentaire, pas en infrastructure. La mémoire court terme sert ici à orienter, pas seulement à répondre.

Les deux voies convergent : chacune produit une liste de messages, et la génération streamée qui suit est identique. Le détail de la voie FinOps — session MCP, choix des outils, appels — est traité dans le guide observabilité FinOps par MCP. C'est lui qui fait de l'assistant un agent qui interroge des outils, au-delà du seul RAG.

L'interface Chainlit

L'interface app.py assemble tout : elle tient la mémoire court terme, aiguille la question vers la bonne voie, diffuse la réponse en streaming.

@cl.on_message
async def repondre(message: cl.Message):
    historique = cl.user_session.get("historique")
    # 1. Le routeur choisit la voie ; chacune prépare la liste `messages`.
    if await cl.make_async(assistant.aiguiller)(message.content, historique) == "finops":
        messages = (await assistant.preparer_finops(message.content))["messages"]
    else:
        ctx = await cl.make_async(assistant.recuperer_contexte)(message.content, USER_ID)
        messages = assistant.construire_messages(message.content, ctx, historique)
    # 2. Réponse en streaming — identique quelle que soit la voie.
    msg = cl.Message(content="")
    flux = await assistant.LLM_ASYNC.chat.completions.create(
        model=assistant.MODELE_CHAT, messages=messages, stream=True)
    reponse = ""
    async for fragment in flux:
        token = fragment.choices[0].delta.content or ""
        if token:
            reponse += token
            await msg.stream_token(token)
    await msg.update()
    # 3. Mémoires : court terme (historique) et long terme (faits durables).
    historique += [{"role": "user", "content": message.content},
                   {"role": "assistant", "content": reponse}]
    await cl.make_async(assistant.memoriser)(USER_ID, message.content, reponse)

L'interface orchestre les trois temps : aiguiller puis préparer le contexte, diffuser la réponse, alimenter les mémoires. La logique, elle, reste dans assistant.py — testable sans Chainlit. Le cl.make_async exécute les fonctions synchrones hors de la boucle d'événements ; le streaming, lui, est nativement asynchrone.

Dépannage

Symptôme	Cause probable	Solution
`WRONG_VERSION_NUMBER` côté Qdrant	Client en HTTPS, serveur en HTTP	Passer `https=False` au client
Requêtes Qdrant qui expirent	Versions client/serveur trop éloignées	Aligner `qdrant-client` et le serveur
La réponse arrive d'un bloc	Génération non streamée	Client asynchrone + `stream=True`
L'assistant ignore le tour précédent	Historique non réinjecté	Tenir l'historique dans `cl.user_session`
La mémoire long terme se mélange	Recherche non filtrée	Filtrer la collection mémoire par `user_id`
L'assistant invente sur une question hors corpus	Garde-fou absent du prompt	Instruire le modèle d'admettre l'absence d'information
Une question d'infrastructure traitée comme documentaire	Routeur privé de l'historique	Passer l'historique à `aiguiller`

À retenir

L'assistant assemble les briques RAG du parcours — il n'en invente aucune.
Le graphe agentique ne fournit que le contexte ; la réponse est une étape séparée.
La mémoire court terme réinjecte l'historique de conversation — elle résout les questions de suivi.
La mémoire long terme range les faits durables dans une collection Qdrant séparée, filtrée par user_id.
La réponse est streamée : un client asynchrone, stream=True, des tokens transmis un à un.
Tout appel de modèle passe par la passerelle ; la logique reste testable sans l'interface.
Un routeur sépare deux voies : le RAG pour la documentation, le serveur MCP FinOps pour l'infrastructure.
Un garde-fou fait dire « je ne sais pas » plutôt qu'inventer quand le contexte ne suffit pas.

Prochaines étapes

Observabilité FinOps par MCP Étape 3 : un serveur MCP qui expose le coût et les ressources de la stack.

Le socle de la stack Étape précédente : l'infrastructure sur laquelle l'assistant tourne.

La mémoire des agents Les deux étages de mémoire, court et long terme, en détail.

RAG agentique La boucle de retrieval qui se corrige, au cœur de l'assistant.