Architecture RAG complète

Objectifs

• Comprendre l’architecture RAG de bout en bout
• Implémenter un pipeline RAG fonctionnel
• Identifier les composants à optimiser

Qu’est-ce que le RAG ?

RAG (Retrieval-Augmented Generation) combine la recherche documentaire avec la génération de texte par un LLM. Au lieu de se fier uniquement à ses connaissances internes, le modèle reçoit des documents pertinents comme contexte, puis rédige sa réponse à partir de ce contexte. Tout ce que vous avez construit depuis le début du cours — embeddings, index, recherche, évaluation — devient ici la moitié amont d’un système complet.

Pourquoi le RAG ?

Quatre raisons le justifient, et elles se cumulent. Les données propriétaires d’abord : votre politique de télétravail, vos procédures internes, votre catalogue produit n’ont jamais figuré dans les données d’entraînement du modèle, et aucun prompt ne les y fera apparaître. Les informations récentes ensuite : tout ce qui est postérieur à l’entraînement échappe au modèle, alors qu’un index se met à jour le jour même. La vérifiabilité compte tout autant, car chaque réponse est ancrée dans des sources identifiables que l’utilisateur peut ouvrir et vérifier — ce qui transforme la nature de la confiance qu’on peut accorder au système. Le coût, enfin, joue en sa faveur : ajouter des connaissances par le RAG revient bien moins cher que par le fine-tuning, et se corrige en supprimant un document plutôt qu’en réentraînant un modèle.

Les étapes du pipeline

Le RAG se lit en deux temps qu’il ne faut pas confondre. L’ingestion se déroule hors ligne, une fois pour toutes ou à chaque mise à jour du corpus :

Documents → Découpage → Embeddings → Stockage vectoriel

La requête se déroule en ligne, à chaque question posée, et doit rester rapide :

Question → Embedding → Recherche → Contexte + Question → LLM → Réponse

Cette séparation a une conséquence pratique : le coût de l’ingestion se paie une fois et peut être lourd, tandis que celui de la requête se paie à chaque appel et conditionne l’expérience utilisateur. On accepte donc volontiers un traitement d’ingestion sophistiqué, et l’on reste économe côté requête.

Implémentation complète

La classe suivante matérialise les trois phases dans trois méthodes distinctes, plus une quatrième qui les enchaîne. Cette séparation n’est pas cosmétique : elle vous permet de tester la recherche sans consommer de tokens de génération, et de rejouer une génération sur un contexte figé pour comparer deux prompts.

from openai import OpenAI
import numpy as np

client = OpenAI()

class PipelineRAG:
    def __init__(
        self,
        model_embedding: str = "text-embedding-3-large",
        model_generation: str = "gpt-5.6-terra"
    ):
        self.client = OpenAI()
        self.model_embedding = model_embedding
        self.model_generation = model_generation
        self.chunks: list[dict] = []
        self.embeddings: np.ndarray | None = None

    # --- Phase d'ingestion ---

    def ingerer(self, documents: list[dict]):
        """Ingère des documents {id, texte, metadata}."""
        self.chunks = documents
        textes = [d["texte"] for d in documents]

        # Générer les embeddings par lots
        tous_emb = []
        for i in range(0, len(textes), 100):
            batch = textes[i:i + 100]
            resp = self.client.embeddings.create(
                input=batch, model=self.model_embedding
            )
            batch_emb = sorted(resp.data, key=lambda x: x.index)
            tous_emb.extend([e.embedding for e in batch_emb])

        self.embeddings = np.array(tous_emb, dtype=np.float32)
        print(f"Ingestion terminée : {len(self.chunks)} chunks indexés")

    # --- Phase de retrieval ---

    def rechercher(self, question: str, k: int = 5) -> list[dict]:
        """Trouve les k chunks les plus pertinents."""
        query_emb = self.client.embeddings.create(
            input=question, model=self.model_embedding
        ).data[0].embedding

        scores = self.embeddings @ np.array(query_emb)
        top_k = np.argsort(scores)[-k:][::-1]

        return [
            {**self.chunks[i], "score": float(scores[i])}
            for i in top_k
        ]

    # --- Phase de génération ---

    def generer(self, question: str, contextes: list[dict]) -> str:
        """Génère une réponse basée sur les contextes retrouvés."""
        # Construire le prompt
        contexte_formate = "\n\n---\n\n".join(
            f"[Source : {c.get('id', 'inconnu')}]\n{c['texte']}"
            for c in contextes
        )

        messages = [
            {
                "role": "system",
                "content": (
                    "Vous êtes un assistant qui répond aux questions "
                    "en vous basant uniquement sur les documents fournis. "
                    "Si l'information n'est pas dans les documents, "
                    "dites-le clairement. Citez vos sources."
                )
            },
            {
                "role": "user",
                "content": (
                    f"Documents de référence :\n\n{contexte_formate}\n\n"
                    f"---\n\nQuestion : {question}"
                )
            }
        ]

        response = self.client.chat.completions.create(
            model=self.model_generation,
            messages=messages,
            temperature=0.2
        )

        return response.choices[0].message.content

    # --- Pipeline complet ---

    def repondre(self, question: str, k: int = 5) -> dict:
        """Pipeline RAG complet : recherche → génération."""
        contextes = self.rechercher(question, k=k)
        reponse = self.generer(question, contextes)

        return {
            "question": question,
            "reponse": reponse,
            "sources": [
                {"id": c["id"], "score": c["score"],
                 "extrait": c["texte"][:150]}
                for c in contextes
            ]
        }

Trois détails de la phase de génération portent l’essentiel de la qualité. Le message système impose de répondre uniquement à partir des documents fournis et d’admettre l’absence d’information : sans cette consigne, le modèle comblera les trous avec ses connaissances générales, et vous obtiendrez une réponse plausible mais inventée. Le préfixe [Source : id] devant chaque extrait donne au modèle de quoi citer précisément, plutôt que de renvoyer un vague « d’après les documents ». La temperature à 0,2, enfin, réduit la variabilité : sur une question factuelle posée à une base documentaire, on cherche la constance, pas la créativité.

Utilisation

Testons le pipeline sur un cas d’entreprise minimal — trois politiques internes, une question qui ne recopie aucun de leurs termes exacts.

# Préparer les documents
documents = [
    {
        "id": "politique-conges",
        "texte": "Les employés bénéficient de 25 jours de congés payés "
                 "par an. Les congés doivent être posés au moins 2 semaines "
                 "à l'avance via le portail RH.",
        "metadata": {"type": "rh", "date": "2026-01"}
    },
    {
        "id": "politique-teletravail",
        "texte": "Le télétravail est autorisé jusqu'à 3 jours par semaine. "
                 "Les mardi et jeudi sont des jours de présence obligatoire. "
                 "Un accord écrit avec le manager est requis.",
        "metadata": {"type": "rh", "date": "2026-01"}
    },
    {
        "id": "procedure-remboursement",
        "texte": "Les notes de frais doivent être soumises dans les 30 jours "
                 "suivant la dépense. Joindre le justificatif original. "
                 "Validation par le N+1 requise au-delà de 200 euros.",
        "metadata": {"type": "finance", "date": "2026-01"}
    },
]

# Créer le pipeline
rag = PipelineRAG()
rag.ingerer(documents)

# Poser une question
resultat = rag.repondre("Combien de jours de télétravail par semaine ?")
print(resultat["reponse"])
print("\nSources utilisées :")
for s in resultat["sources"]:
    print(f"  [{s['score']:.3f}] {s['id']}: {s['extrait']}")

Poussez l’exercice plus loin en posant une question à laquelle aucun document ne répond, par exemple sur la mutuelle d’entreprise. Le pipeline retournera quand même trois sources — la recherche renvoie toujours ses k meilleurs candidats — et c’est le message système qui doit empêcher le modèle d’inventer. Vérifier ce comportement avant la mise en production vaut mieux que le découvrir par une réclamation.

Les points d’optimisation

Chaque étape du pipeline peut être améliorée indépendamment, ce qui permet de progresser par petites touches mesurables.

Les leçons suivantes détaillent chacune de ces optimisations. Abordez-les dans cet ordre : le découpage détermine ce qui pourra être retrouvé, et aucun reranking ne rattrapera un chunk mal découpé.

Résumé

• RAG = Recherche + Contexte + Génération par LLM
• Le pipeline se décompose en ingestion (offline) et requête (online)
• Chaque composant (chunking, embedding, recherche, prompt, génération) peut être optimisé indépendamment
• Le RAG est plus flexible et moins coûteux que le fine-tuning pour ajouter des connaissances