Indexer un corpus de documents — Embeddings, RAG et Recherche Sémantique

Objectifs

Construire un index de recherche sémantique complet
Gérer différents formats de documents (texte, PDF, Markdown)
Mettre à jour l’index incrémentalement

Architecture d’un index sémantique

Un index sémantique tient en trois éléments qu’il faut maintenir cohérents entre eux : les documents originaux stockés avec leurs métadonnées, les embeddings calculés à partir de leur texte, et le mapping qui relie chaque vecteur à son document source. Ce troisième point est le plus fragile — c’est presque toujours là que les index se corrompent, quand une suppression décale les indices d’un côté sans les décaler de l’autre.

La classe Document ci-dessous ajoute une empreinte MD5 calculée automatiquement à la création. Elle nous servira plus loin à détecter ce qui a déjà été indexé.

from dataclasses import dataclass, field
from openai import OpenAI
import numpy as np
import json
import hashlib

@dataclass
class Document:
    id: str
    texte: str
    metadata: dict = field(default_factory=dict)
    hash: str = ""

    def __post_init__(self):
        self.hash = hashlib.md5(self.texte.encode()).hexdigest()

Pipeline d’indexation

Étape 1 : Charger les documents

Le chargement dépend du format, et le format détermine la granularité. Un fichier texte brut devient un document ; un fichier Markdown, en revanche, contient souvent plusieurs sujets distincts, et l’indexer d’un bloc noierait chaque sujet dans la moyenne des autres. Le second chargeur découpe donc sur les titres de niveau 2, et réinjecte le titre de section dans le texte indexé — c’est la technique d’enrichissement vue en leçon 3.

from pathlib import Path

def charger_fichiers_texte(dossier: str) -> list[Document]:
    """Charge tous les fichiers texte d'un dossier."""
    documents = []
    for fichier in Path(dossier).rglob("*.txt"):
        contenu = fichier.read_text(encoding="utf-8").strip()
        if contenu:
            documents.append(Document(
                id=fichier.stem,
                texte=contenu,
                metadata={
                    "fichier": str(fichier),
                    "type": "txt",
                    "taille": len(contenu)
                }
            ))
    return documents

def charger_fichiers_markdown(dossier: str) -> list[Document]:
    """Charge des fichiers Markdown en séparant par sections."""
    documents = []
    for fichier in Path(dossier).rglob("*.md"):
        contenu = fichier.read_text(encoding="utf-8")
        sections = contenu.split("\n## ")

        for i, section in enumerate(sections):
            if not section.strip():
                continue
            titre = section.split("\n")[0].strip("# ")
            corps = "\n".join(section.split("\n")[1:]).strip()
            if corps:
                documents.append(Document(
                    id=f"{fichier.stem}_section_{i}",
                    texte=f"{titre}\n\n{corps}",
                    metadata={
                        "fichier": str(fichier),
                        "section": titre,
                        "type": "markdown"
                    }
                ))
    return documents

Étape 2 : Construire l’index

La classe IndexSemantique rassemble la construction par lots, la recherche avec seuil et la persistance sur disque. Le paramètre seuil de la méthode rechercher mérite l’attention : sans lui, une requête hors sujet renvoie quand même les cinq documents les moins mauvais, avec des scores de 0,15 qu’un appelant naïf traitera comme des réponses. Filtrer en dessous d’un score plancher permet de répondre honnêtement « je n’ai rien trouvé ».

class IndexSemantique:
    def __init__(self, model: str = "text-embedding-3-large"):
        self.client = OpenAI()
        self.model = model
        self.documents: list[Document] = []
        self.embeddings: np.ndarray | None = None

    def construire(self, documents: list[Document], batch_size: int = 100):
        """Construit l'index à partir d'une liste de documents."""
        self.documents = documents
        textes = [d.texte for d in documents]
        tous_embeddings = []

        for i in range(0, len(textes), batch_size):
            batch = textes[i:i + batch_size]
            response = self.client.embeddings.create(
                input=batch,
                model=self.model
            )
            batch_sorted = sorted(response.data, key=lambda x: x.index)
            tous_embeddings.extend([e.embedding for e in batch_sorted])
            print(f"  Lot {i // batch_size + 1} traité "
                  f"({min(i + batch_size, len(textes))}/{len(textes)})")

        self.embeddings = np.array(tous_embeddings, dtype=np.float32)
        print(f"Index construit : {len(self.documents)} documents, "
              f"{self.embeddings.shape[1]} dimensions")

    def rechercher(self, requete: str, k: int = 5,
                   seuil: float = 0.0) -> list[dict]:
        """Recherche les k documents les plus pertinents."""
        query_emb = self.client.embeddings.create(
            input=requete,
            model=self.model
        ).data[0].embedding

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

        resultats = []
        for idx in top_indices:
            score = float(scores[idx])
            if score >= seuil:
                doc = self.documents[idx]
                resultats.append({
                    "id": doc.id,
                    "texte": doc.texte[:200],
                    "score": score,
                    "metadata": doc.metadata
                })
        return resultats

    def sauvegarder(self, chemin: str):
        """Sauvegarde l'index sur disque."""
        np.save(f"{chemin}_embeddings.npy", self.embeddings)
        docs_serialisables = [
            {"id": d.id, "texte": d.texte,
             "metadata": d.metadata, "hash": d.hash}
            for d in self.documents
        ]
        with open(f"{chemin}_documents.json", "w") as f:
            json.dump(docs_serialisables, f, ensure_ascii=False)

    def charger(self, chemin: str):
        """Charge un index depuis le disque."""
        self.embeddings = np.load(f"{chemin}_embeddings.npy")
        with open(f"{chemin}_documents.json") as f:
            docs = json.load(f)
        self.documents = [
            Document(id=d["id"], texte=d["texte"],
                     metadata=d["metadata"])
            for d in docs
        ]

Les méthodes sauvegarder et charger écrivent deux fichiers portant le même préfixe. Traitez-les comme un couple indissociable : copier l’un sans l’autre, ou restaurer une sauvegarde partielle, produit un index dont les vecteurs ne correspondent plus aux documents — une panne silencieuse qui ne se manifestera que par des résultats absurdes.

Mise à jour incrémentale

Un corpus vit : trois documents modifiés le lundi, dix ajoutés le mercredi. Recalculer les quinze mille vecteurs à chaque changement coûte du temps et de l’argent pour rien. C’est ici que l’empreinte MD5 sert : elle identifie le contenu, pas le fichier, donc un document renommé mais inchangé ne sera pas réindexé, tandis qu’un document modifié produira une empreinte différente.

def mettre_a_jour(self, nouveaux_docs: list[Document]):
    """Ajoute des documents à l'index existant."""
    # Filtrer les documents déjà indexés
    hashes_existants = {d.hash for d in self.documents}
    docs_a_ajouter = [
        d for d in nouveaux_docs if d.hash not in hashes_existants
    ]

    if not docs_a_ajouter:
        print("Aucun nouveau document à indexer")
        return

    textes = [d.texte for d in docs_a_ajouter]
    response = self.client.embeddings.create(
        input=textes,
        model=self.model
    )

    nouveaux_vecteurs = np.array(
        [e.embedding for e in sorted(response.data, key=lambda x: x.index)],
        dtype=np.float32
    )

    self.documents.extend(docs_a_ajouter)
    self.embeddings = np.vstack([self.embeddings, nouveaux_vecteurs])
    print(f"{len(docs_a_ajouter)} documents ajoutés. "
          f"Total : {len(self.documents)}")

Le np.vstack empile les nouveaux vecteurs sous les anciens dans le même ordre que le extend sur la liste de documents : c’est cette symétrie qui préserve le mapping. Notez toutefois la limite de cette implémentation — elle ajoute, mais ne supprime ni ne remplace. Un document modifié se retrouve donc présent deux fois, dans son ancienne et sa nouvelle version. Gérer les suppressions demande de retirer la ligne correspondante de la matrice, opération coûteuse en NumPy, et c’est l’une des raisons qui poussent vers une vraie base vectorielle.

Utilisation complète

Bout à bout, la construction d’un index sur une documentation tient en quelques lignes.

# Créer et peupler l'index
index = IndexSemantique()

docs = charger_fichiers_markdown("./documentation/")
print(f"{len(docs)} sections chargées")

index.construire(docs)
index.sauvegarder("mon_index")

# Rechercher
resultats = index.rechercher("comment configurer l'authentification", k=3)
for r in resultats:
    print(f"[{r['score']:.3f}] {r['id']}")
    print(f"  {r['texte'][:100]}...")

Lancez-le sur votre propre documentation et posez trois questions dont vous connaissez la réponse. Si le bon document n’arrive pas en tête, le coupable est presque toujours le découpage — sections trop longues ou titres non repris — plutôt que le modèle d’embedding.

Résumé

Un index sémantique associe documents, embeddings et métadonnées
Chargez les documents depuis différents formats (texte, Markdown, etc.)
Construisez l’index par lots pour gérer les gros corpus
Sauvegardez et mettez à jour incrémentalement pour éviter de tout recalculer