Indexer un corpus de documents

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 se compose de trois éléments :

1. Les documents originaux : stockés avec leurs métadonnées
2. Les embeddings : vecteurs numériques calculés à partir du texte
3. Le mapping : lien entre chaque embedding et son document source

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

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

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
        ]

Mise à jour incrémentale

Ne recalculez pas tout l’index quand vous ajoutez ou modifiez des documents :

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)}")

Utilisation complète

# 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]}...")

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