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

Tout assembler

Dans cette leçon, vous allez construire un pipeline RAG complet et fonctionnel de bout en bout, depuis le chargement des données jusqu’à la génération de réponses. Ce code est prêt à être adapté pour vos propres projets.

Prérequis

pip install mistralai faiss-cpu numpy requests

Le pipeline complet

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)


# ============================================================
# Etape 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


# ============================================================
# Etape 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


# ============================================================
# Etape 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


# ============================================================
# Etape 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


# ============================================================
# Etape 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 etes 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()

Tester le pipeline

Sauvegardez le code ci-dessus dans un fichier rag_pipeline.py et exécutez-le :

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

Améliorations possibles

Ce pipeline de base peut être étendu de plusieurs manières :

# 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