Pipeline RAG Complet — Embeddings, RAG et Recherche Sémantique

Tout assembler

Les leçons précédentes ont traité chaque maillon séparément : le chargement, le chunking, l’encodage, l’indexation, la recherche, la génération. Il est temps de les souder en un programme unique, complet et fonctionnel de bout en bout, que vous pourrez lancer tel quel puis adapter à vos propres documents. Ce script tient en un fichier et ne demande que quatre dépendances.

pip install mistralai faiss-cpu numpy requests

Le pipeline complet

Deux partis pris structurent ce code. Le premier est la configuration groupée en tête de fichier : modèles, taille de chunk, taille de lot et TOP_K sont des constantes nommées, ce qui vous évite d’aller chasser une valeur codée en dur au milieu d’une fonction quand vous voudrez régler le comportement. Le second est le découpage en six fonctions indépendantes, chacune correspondant à une étape et affichant sa progression sous la forme [n/6]. Cette trace n’est pas décorative : quand le pipeline échoue ou rend une réponse étrange, elle vous dit immédiatement si le problème vient du chargement, du chunking ou du retrieval. Chaque fonction est aussi remplaçable isolément — vous pouvez substituer un chargeur PDF à load_data sans toucher au reste.

Le main enchaîne les étapes puis interroge le corpus avec trois questions portant sur l’essai de Paul Graham utilisé comme jeu de démonstration. Remarquez que le retrieval et la génération sont appelés dans la boucle, alors que l’indexation a lieu une seule fois avant : encoder le corpus à chaque question serait absurde en coût comme en latence.

import os
import json
import numpy as np
import faiss
import requests
from mistralai import Mistral

# ============================================================
# Configuration
# ============================================================
MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY")
EMBED_MODEL = "mistral-embed"
CHAT_MODEL = "mistral-large-latest"
CHUNK_SIZE = 2048
BATCH_SIZE = 25
TOP_K = 3

client = Mistral(api_key=MISTRAL_API_KEY)


# ============================================================
# Étape 1 : Charger les données
# ============================================================
def load_data(url):
    """Charger un document texte depuis une URL."""
    print("[1/6] Chargement des données...")
    response = requests.get(url)
    response.raise_for_status()
    text = response.text
    print(f"  -> {len(text)} caractères chargés")
    return text


# ============================================================
# Étape 2 : Découper en chunks
# ============================================================
def create_chunks(text, chunk_size=CHUNK_SIZE):
    """Découper le texte en chunks."""
    print("[2/6] Découpage en chunks...")
    chunks = []
    paragraphs = text.split("\n\n")
    current = ""

    for para in paragraphs:
        para = para.strip()
        if not para:
            continue
        if len(current) + len(para) + 2 <= chunk_size:
            current = current + "\n\n" + para if current else para
        else:
            if current:
                chunks.append(current)
            current = para

    if current:
        chunks.append(current)

    print(f"  -> {len(chunks)} chunks créés (taille moyenne: {sum(len(c) for c in chunks) // len(chunks)} chars)")
    return chunks


# ============================================================
# Étape 3 : Générer les embeddings
# ============================================================
def create_embeddings(chunks, batch_size=BATCH_SIZE):
    """Encoder les chunks en embeddings via mistral-embed."""
    print("[3/6] Génération des embeddings...")
    all_embeddings = []

    for i in range(0, len(chunks), batch_size):
        batch = chunks[i:i + batch_size]
        response = client.embeddings.create(
            model=EMBED_MODEL,
            inputs=batch,
        )
        all_embeddings.extend([d.embedding for d in response.data])
        print(f"  -> Lot {i // batch_size + 1}/{(len(chunks) - 1) // batch_size + 1} encodé")

    embeddings = np.array(all_embeddings, dtype=np.float32)
    print(f"  -> Matrice: {embeddings.shape}")
    return embeddings


# ============================================================
# Etape 4 : Créer l'index FAISS
# ============================================================
def create_index(embeddings):
    """Créer et peupler un index FAISS."""
    print("[4/6] Création de l'index FAISS...")
    dimension = embeddings.shape[1]
    index = faiss.IndexFlatL2(dimension)
    index.add(embeddings)
    print(f"  -> Index: {index.ntotal} vecteurs, dimension {dimension}")
    return index


# ============================================================
# Étape 5 : Rechercher les chunks pertinents
# ============================================================
def retrieve(question, index, chunks, k=TOP_K):
    """Retrouver les chunks les plus pertinents."""
    print(f"[5/6] Recherche pour: '{question[:50]}...'")
    response = client.embeddings.create(
        model=EMBED_MODEL,
        inputs=[question],
    )
    query_vector = np.array([response.data[0].embedding], dtype=np.float32)

    distances, indices = index.search(query_vector, k)

    results = []
    for dist, idx in zip(distances[0], indices[0]):
        if idx >= 0:
            results.append({
                "text": chunks[idx],
                "distance": float(dist),
            })

    print(f"  -> {len(results)} chunks retrouvés")
    return results


# ============================================================
# Étape 6 : Générer la réponse
# ============================================================
def generate(question, context_chunks):
    """Générer la réponse avec le contexte retrouvé."""
    print("[6/6] Génération de la réponse...")

    context = "\n\n---\n\n".join([c["text"] for c in context_chunks])

    prompt = f"""Contexte informatif :

{context}

---

En vous basant UNIQUEMENT sur le contexte ci-dessus, répondez à la question suivante de manière précise et complète.

Question : {question}"""

    response = client.chat.complete(
        model=CHAT_MODEL,
        messages=[
            {"role": "system", "content": "Vous êtes un assistant qui répond en se basant sur le contexte fourni."},
            {"role": "user", "content": prompt},
        ],
        temperature=0.1,
    )

    answer = response.choices[0].message.content
    print(f"  -> Réponse générée ({len(answer)} caractères)")
    return answer


# ============================================================
# Pipeline principal
# ============================================================
def main():
    """Exécuter le pipeline RAG complet."""
    # 1. Charger
    url = "https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt"
    text = load_data(url)

    # 2. Chunker
    chunks = create_chunks(text)

    # 3. Encoder
    embeddings = create_embeddings(chunks)

    # 4. Indexer
    index = create_index(embeddings)

    # 5+6. Question -> Réponse
    questions = [
        "What were the two main things the author worked on before college?",
        "What did the author do during their time at Y Combinator?",
        "What programming language did the author learn first?",
    ]

    for question in questions:
        print(f"\n{'='*60}")
        context = retrieve(question, index, chunks)
        answer = generate(question, context)
        print(f"\nQ: {question}")
        print(f"A: {answer}")


if __name__ == "__main__":
    main()

Lancer le pipeline

Sauvegardez ce code dans un fichier rag_pipeline.py, exportez votre clé API dans l’environnement — jamais en dur dans le fichier, qui finira tôt ou tard dans un dépôt — puis exécutez le script. La sortie déroule les six étapes puis les trois questions avec leurs réponses ; si l’une d’elles paraît hors sujet, comparez-la aux chunks affichés à l’étape 5 avant de mettre en cause le modèle de génération.

export MISTRAL_API_KEY="votre-clé-api"
python rag_pipeline.py

Cinq directions pour aller plus loin

Ce pipeline est un socle, pas un aboutissement, et chacune de ses limites ouvre une piste. La plus immédiate est la persistance : dans sa forme actuelle, le script réencode tout le corpus à chaque lancement, alors que sauvegarder l’index FAISS et les chunks rendrait les exécutions suivantes quasi instantanées. Le re-ranking constitue le deuxième palier, en faisant repasser les résultats de FAISS devant un cross-encoder qui juge la pertinence bien plus finement qu’une distance vectorielle. Le filtrage par métadonnées répond à un autre besoin, celui de contraindre la recherche par date, catégorie ou auteur en complément de la similarité. Restent deux améliorations tournées vers l’expérience utilisateur : le streaming, qui affiche la réponse au fil de sa génération plutôt qu’après plusieurs secondes de silence, et l’historique de conversation, qui permet à l’utilisateur d’enchaîner « et pour la version 2 ? » sans reformuler toute sa question.

# 1. Sauvegarder l'index pour ne pas recalculer
faiss.write_index(index, "my_rag.index")
np.save("my_chunks.npy", np.array(chunks, dtype=object))

# 2. Ajouter un re-ranking
# Après le retrieval FAISS, utiliser un modèle cross-encoder
# pour ré-ordonner les résultats par pertinence fine

# 3. Filtrage par métadonnées
# Ajouter des filtres (date, catégorie, auteur) en plus
# de la similarité vectorielle

# 4. Streaming de la réponse
# Utiliser client.chat.stream() pour afficher la réponse
# au fur et à mesure de la génération

# 5. Historique de conversation
# Ajouter les échanges précédents dans le prompt
# pour un chatbot RAG contextuel

Points clés à retenir

Le pipeline RAG complet se décompose en 6 étapes : charger, chunker, encoder, indexer, rechercher, générer
Chaque étape est une fonction indépendante, facilement testable et remplaçable
Utilisez temperature=0.1 pour des réponses factuelles
Sauvegardez l’index FAISS et les embeddings pour ne pas recalculer à chaque lancement
Ce pipeline est un point de départ solide, à enrichir selon vos besoins