Générer des embeddings pour vos données
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. Avant de générer des vecteurs, nettoyez et normalisez vos textes.
Nettoyage de base
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
Pour améliorer la qualité des embeddings, ajoutez du contexte au texte brut. Par exemple, pour des articles de documentation :
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
Pour un corpus réel, vous aurez des centaines ou des milliers de documents. Voici un pipeline robuste :
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_embeddingsStocker les embeddings
Format JSON simple
Pour un petit corpus (quelques milliers de documents), un fichier JSON suffit :
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)Format NumPy pour la performance
Pour de plus gros volumes, NumPy est plus efficace :
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
Voici un exemple complet qui lit des fichiers, génère les embeddings et les stocke :
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/")Estimer les coûts
Avant de lancer un gros traitement, estimez le nombre de tokens :
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, coutRé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