Générer des embeddings pour vos données — Embeddings, RAG et Recherche Sémantique

Objectifs

Préparer vos données textuelles pour l’embedding
Générer des embeddings efficacement à grande échelle
Stocker les vecteurs pour une utilisation ultérieure

Préparer vos données

La qualité de vos embeddings dépend directement de la qualité de vos données. Un texte truffé d’espaces parasites, de caractères de contrôle hérités d’un export mal configuré ou de retours à la ligne intempestifs produit un vecteur légèrement décalé, et ce décalage se paie ensuite sur chaque recherche. Le nettoyage n’est pas une coquetterie : c’est la première optimisation de pertinence, et la moins chère.

import re

def nettoyer_texte(texte: str) -> str:
    """Prépare un texte pour l'embedding."""
    # Supprimer les espaces multiples
    texte = re.sub(r"\s+", " ", texte)
    # Supprimer les caractères de contrôle
    texte = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", texte)
    # Trim
    texte = texte.strip()
    return texte

# Exemple
brut = "  Texte   avec   des   espaces\n\n\nen trop  "
propre = nettoyer_texte(brut)
print(propre)  # "Texte avec des espaces en trop"

Enrichir le contexte

Un extrait de documentation isolé de son titre perd une partie de son sens. Prenez un paragraphe qui commence par « Téléchargez le package depuis le site officiel » : hors contexte, rien n’indique qu’il s’agit de PostgreSQL, et le vecteur produit ne rapprochera jamais ce passage d’une requête portant sur PostgreSQL. En préfixant le contenu par son titre et sa section, vous réinjectez cette information dans le texte que le modèle voit.

def enrichir_document(titre: str, section: str, contenu: str) -> str:
    """Crée un texte enrichi pour un meilleur embedding."""
    return f"Titre : {titre}\nSection : {section}\n\n{contenu}"

texte_enrichi = enrichir_document(
    titre="Installation de PostgreSQL",
    section="Guide de démarrage",
    contenu="Téléchargez le package depuis le site officiel..."
)

Générer des embeddings en lots

Sur un corpus réel de centaines ou de milliers de documents, le traitement doit être découpé et capable de survivre à un incident réseau. La fonction ci-dessous parcourt les textes par tranches, trie systématiquement la réponse par index pour garantir que le vecteur n° 7 correspond bien au document n° 7, et réessaie une fois après une pause en cas d’erreur.

from openai import OpenAI
import time
import json

client = OpenAI()

def generer_embeddings(
    textes: list[str],
    model: str = "text-embedding-3-large",
    batch_size: int = 100
) -> list[list[float]]:
    """Génère des embeddings par lots avec gestion des erreurs."""
    tous_les_embeddings = []

    for i in range(0, len(textes), batch_size):
        batch = textes[i:i + batch_size]
        print(f"Traitement du lot {i // batch_size + 1}/"
              f"{(len(textes) - 1) // batch_size + 1}")

        try:
            response = client.embeddings.create(
                input=batch,
                model=model
            )
            # Trier par index pour garantir l'ordre
            batch_embeddings = sorted(
                response.data, key=lambda x: x.index
            )
            tous_les_embeddings.extend(
                [item.embedding for item in batch_embeddings]
            )
        except Exception as e:
            print(f"Erreur sur le lot {i}: {e}")
            # Retry après une pause
            time.sleep(5)
            response = client.embeddings.create(
                input=batch,
                model=model
            )
            batch_embeddings = sorted(
                response.data, key=lambda x: x.index
            )
            tous_les_embeddings.extend(
                [item.embedding for item in batch_embeddings]
            )

    return tous_les_embeddings

Le tri par index est la précaution à ne jamais négliger. Un décalage silencieux d’un seul rang suffit à associer chaque document au vecteur de son voisin, et le symptôme observé sera « la recherche renvoie n’importe quoi », sans aucune erreur technique pour vous mettre sur la piste.

Stocker les embeddings

Pour un petit corpus de quelques milliers de documents, un simple fichier JSON contenant textes, métadonnées et vecteurs fait parfaitement l’affaire, et il a l’avantage d’être lisible à l’œil nu quand vous cherchez à comprendre un résultat étrange.

import json

documents = [
    {"id": 1, "texte": "Premier document...", "metadata": {"source": "faq"}},
    {"id": 2, "texte": "Deuxième document...", "metadata": {"source": "docs"}},
]

# Générer les embeddings
textes = [d["texte"] for d in documents]
embeddings = generer_embeddings(textes)

# Ajouter les vecteurs aux documents
for doc, emb in zip(documents, embeddings):
    doc["embedding"] = emb

# Sauvegarder
with open("corpus_embeddings.json", "w") as f:
    json.dump(documents, f)

Au-delà, le JSON devient pénalisant : chaque nombre y est écrit en toutes lettres, ce qui gonfle le fichier et impose un parsing coûteux à chaque chargement. NumPy stocke les mêmes valeurs en binaire sur 4 octets et se recharge instantanément.

import numpy as np

# Convertir en matrice NumPy
matrice = np.array(embeddings, dtype=np.float32)
print(f"Shape : {matrice.shape}")  # (n_docs, 3072)

# Sauvegarder au format binaire
np.save("embeddings.npy", matrice)

# Charger
matrice_chargee = np.load("embeddings.npy")

Pipeline complet : de fichiers texte à embeddings

Rassemblons ces éléments dans une fonction qui lit un dossier, nettoie chaque fichier, génère les vecteurs et sépare le stockage en deux : les métadonnées en JSON, lisibles et modifiables, et la matrice en NumPy, compacte et rapide.

from openai import OpenAI
from pathlib import Path
import json
import numpy as np

client = OpenAI()

def pipeline_embeddings(dossier: str, model: str = "text-embedding-3-large"):
    """Pipeline complet : fichiers → embeddings → stockage."""
    dossier_path = Path(dossier)

    # 1. Lire les fichiers
    documents = []
    for fichier in sorted(dossier_path.glob("*.txt")):
        contenu = fichier.read_text(encoding="utf-8")
        contenu = nettoyer_texte(contenu)
        if contenu:
            documents.append({
                "id": fichier.stem,
                "fichier": fichier.name,
                "texte": contenu
            })

    print(f"{len(documents)} documents trouvés")

    # 2. Générer les embeddings
    textes = [d["texte"] for d in documents]
    embeddings = generer_embeddings(textes, model=model)

    # 3. Sauvegarder
    # Métadonnées en JSON
    meta = [{"id": d["id"], "fichier": d["fichier"]} for d in documents]
    with open("metadata.json", "w") as f:
        json.dump(meta, f, ensure_ascii=False, indent=2)

    # Vecteurs en NumPy
    np.save("embeddings.npy", np.array(embeddings, dtype=np.float32))

    print(f"Embeddings sauvegardés : {len(embeddings)} vecteurs "
          f"de {len(embeddings[0])} dimensions")

    return documents, embeddings

# Utilisation
documents, embeddings = pipeline_embeddings("./mes_documents/")

Notez le sorted() sur le glob : sans lui, l’ordre des fichiers dépend du système de fichiers, et vos deux fichiers de sortie risquent de ne plus correspondre l’un à l’autre après un simple changement de machine.

Estimer les coûts

Avant de lancer un traitement massif, comptez les tokens localement avec tiktoken. L’opération prend quelques secondes et vous évite de découvrir après coup qu’un corpus mal filtré contenait dix fois le volume prévu.

import tiktoken

encoder = tiktoken.get_encoding("cl100k_base")

def estimer_cout(textes: list[str], model: str = "text-embedding-3-large"):
    """Estime le coût d'embedding pour une liste de textes."""
    total_tokens = sum(len(encoder.encode(t)) for t in textes)

    # Prix approximatifs par million de tokens (vérifiez les tarifs actuels)
    prix = {
        "text-embedding-3-large": 0.13,
        "text-embedding-3-small": 0.02,
    }

    cout = total_tokens / 1_000_000 * prix.get(model, 0.13)
    print(f"Tokens estimés : {total_tokens:,}")
    print(f"Coût estimé : ${cout:.4f}")
    return total_tokens, cout

Les prix inscrits dans le dictionnaire sont indicatifs et vieillissent : vérifiez toujours les tarifs en vigueur avant de présenter un chiffrage à un client.

Résumé

Nettoyez et enrichissez vos textes avant de générer des embeddings
Utilisez le traitement par lots (batch) pour les gros corpus
Stockez les métadonnées en JSON et les vecteurs en NumPy
Estimez vos coûts avant de lancer un traitement massif