Aller au contenu principal

Créer les Embeddings

Mis à jour le 29 juillet 2026

Encoder les chunks en vecteurs

Vos documents sont découpés en chunks propres. L’étape suivante consiste à les transformer en embeddings via l’API mistral-embed, puis à ranger ces vecteurs dans un vector store qui servira de moteur de recherche. C’est le moment où votre corpus cesse d’être du texte pour devenir une géométrie interrogeable, et où quelques précautions d’ingénierie — lots, retry, sauvegarde — font la différence entre un prototype et un index sur lequel on peut s’appuyer.

Pour un petit corpus, disons moins d’une centaine de chunks, un appel direct suffit. Vous extrayez les textes de vos dictionnaires de chunks, vous les passez tels quels à l’API, et vous récupérez la matrice complète. La forme (87, 1024) affichée en fin d’exécution est votre premier contrôle : autant de lignes que de chunks, 1024 colonnes puisque c’est la dimension de mistral-embed. Si le nombre de lignes ne correspond pas, un chunk vide s’est glissé dans le lot.

import numpy as np
from mistralai import Mistral

client = Mistral(api_key=api_key)

# Supposons que chunks est la liste de nos chunks préparés
chunk_texts = [c["text"] for c in chunks]

# Encoder en un seul appel
response = client.embeddings.create(
    model="mistral-embed",
    inputs=chunk_texts,
)

embeddings = np.array([d.embedding for d in response.data])
print(f"Embeddings: {embeddings.shape}")
# Embeddings: (87, 1024) par exemple

Encodage par lots

Au-delà de quelques dizaines de chunks, l’appel unique se heurte aux limites de l’API. La parade consiste à découper la liste en lots et à concaténer les résultats. La fonction ci-dessous avance par tranches de 25 et affiche la progression, ce qui n’est pas cosmétique : sur un corpus de plusieurs milliers de chunks, l’encodage prend plusieurs minutes et vous voudrez savoir où en est le traitement plutôt que de fixer un terminal muet. L’ordre est préservé, chaque vecteur reste donc aligné sur le chunk correspondant dans votre liste de départ.

def encode_chunks(chunks, batch_size=25, model="mistral-embed"):
    """Encoder une liste de chunks par lots."""
    texts = [c["text"] for c in chunks]
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        print(f"  Encodage lot {i // batch_size + 1}/{(len(texts) - 1) // batch_size + 1} "
              f"({len(batch)} chunks)...")

        response = client.embeddings.create(
            model=model,
            inputs=batch,
        )
        all_embeddings.extend([d.embedding for d in response.data])

    return np.array(all_embeddings)

# Encoder tous les chunks
embeddings = encode_chunks(chunks, batch_size=25)
print(f"Total: {embeddings.shape[0]} vecteurs de {embeddings.shape[1]} dimensions")

Survivre aux erreurs réseau

En production, un appel API finit toujours par échouer : coupure réseau, limite de débit atteinte, indisponibilité passagère. Sans protection, votre indexation s’arrête au lot 180 sur 400 et vous repartez de zéro. La version robuste enveloppe chaque lot dans une boucle de tentatives avec backoff exponentiel — une seconde, puis deux, puis quatre. Cet espacement croissant est important : réessayer immédiatement après un rejet pour dépassement de quota ne fait qu’aggraver la situation. Après trois échecs consécutifs, la fonction lève une exception explicite qui nomme le lot fautif, plutôt que de retourner une matrice tronquée dont vous ne découvririez le trou qu’au moment des recherches.

import time

def encode_chunks_robust(chunks, batch_size=25, max_retries=3):
    """Encodage robuste avec gestion d'erreurs."""
    texts = [c["text"] for c in chunks]
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]

        for attempt in range(max_retries):
            try:
                response = client.embeddings.create(
                    model="mistral-embed",
                    inputs=batch,
                )
                all_embeddings.extend([d.embedding for d in response.data])
                break  # Succès, sortir de la boucle retry

            except Exception as e:
                wait_time = 2 ** attempt  # Backoff exponentiel : 1s, 2s, 4s
                print(f"  Erreur lot {i // batch_size + 1}, "
                      f"tentative {attempt + 1}/{max_retries}: {e}")
                if attempt < max_retries - 1:
                    print(f"  Attente {wait_time}s avant retry...")
                    time.sleep(wait_time)
                else:
                    raise RuntimeError(
                        f"Échec après {max_retries} tentatives pour le lot {i // batch_size + 1}"
                    )

    return np.array(all_embeddings)

Sauvegarder l’index sur disque

Chaque encodage coûte du temps et des appels facturés. Recalculer les vecteurs à chaque démarrage de votre application est donc un gaspillage pur, d’autant que le corpus, lui, n’a pas bougé. La sauvegarde se fait en deux fichiers complémentaires : les vecteurs partent en NumPy, format binaire compact et rechargé instantanément, tandis que les métadonnées — texte, source, position — partent en JSON lisible que vous pouvez ouvrir pour inspecter un chunk suspect. Les deux fichiers sont solidaires et l’ordre des lignes les relie : ne régénérez jamais l’un sans l’autre, sous peine de faire pointer vos résultats de recherche sur les mauvais textes.

import json

def save_index(chunks, embeddings, filepath="rag_index"):
    """Sauvegarder les chunks et embeddings."""
    # Sauvegarder les embeddings (NumPy)
    np.save(f"{filepath}_embeddings.npy", embeddings)

    # Sauvegarder les métadonnées (JSON)
    metadata = []
    for c in chunks:
        metadata.append({
            "text": c["text"],
            "source": c.get("source", ""),
            "chunk_index": c.get("chunk_index", 0),
        })
    with open(f"{filepath}_metadata.json", "w", encoding="utf-8") as f:
        json.dump(metadata, f, ensure_ascii=False, indent=2)

    print(f"Index sauvegardé: {len(chunks)} chunks, {embeddings.shape}")


def load_index(filepath="rag_index"):
    """Charger les chunks et embeddings depuis le disque."""
    embeddings = np.load(f"{filepath}_embeddings.npy")

    with open(f"{filepath}_metadata.json", "r", encoding="utf-8") as f:
        metadata = json.load(f)

    print(f"Index chargé: {len(metadata)} chunks, {embeddings.shape}")
    return metadata, embeddings

# Sauvegarder
save_index(chunks, embeddings)

# Charger (au prochain démarrage)
chunks_loaded, embeddings_loaded = load_index()

Vérifier avant de continuer

Un index silencieusement corrompu se remarque tard, généralement quand un utilisateur signale une réponse absurde. Quelques contrôles à froid coûtent trente secondes et vous épargnent cette découverte. La fonction de vérification affiche d’abord les grandeurs de base — nombre de chunks, dimension, norme moyenne — puis mesure la similarité entre chunks voisins : issus du même document, ils devraient afficher des valeurs nettement plus élevées que deux passages tirés au hasard. Le dernier test est le plus simple et le plus révélateur : un chunk comparé à lui-même doit donner exactement 1.0. Toute autre valeur signale un problème d’alignement ou de conversion numérique.

from sklearn.metrics.pairwise import cosine_similarity

def verify_embeddings(chunks, embeddings, n_tests=5):
    """Vérifier la qualité des embeddings avec des tests simples."""
    print("=== Vérification des embeddings ===")
    print(f"Nombre de chunks: {len(chunks)}")
    print(f"Dimensions: {embeddings.shape[1]}")
    print(f"Norme moyenne: {np.linalg.norm(embeddings, axis=1).mean():.4f}")

    # Test : les chunks du même document devraient être plus proches
    for i in range(min(n_tests, len(chunks) - 1)):
        sim = cosine_similarity([embeddings[i]], [embeddings[i + 1]])[0][0]
        print(f"  Similarité chunk {i} <-> {i+1}: {sim:.4f}")

    # Test : un chunk comparé à lui-même = 1.0
    self_sim = cosine_similarity([embeddings[0]], [embeddings[0]])[0][0]
    print(f"  Auto-similarité (doit être 1.0): {self_sim:.6f}")

verify_embeddings(chunks, embeddings)

Mis bout à bout, l’enchaînement tient en quatre lignes utiles. Vous préparez les chunks avec le pipeline de la leçon précédente, vous encodez en mode robuste, vous vérifiez, vous sauvegardez. C’est ce script que vous relancerez à chaque mise à jour du corpus.

# 1. Préparer les chunks (leçon précédente)
chunks = prepare_documents(raw_docs)

# 2. Encoder
embeddings = encode_chunks_robust(chunks, batch_size=25)

# 3. Vérifier
verify_embeddings(chunks, embeddings)

# 4. Sauvegarder
save_index(chunks, embeddings)

print("Index prêt pour la recherche !")

Points clés à retenir

  • Utilisez le batching pour encoder de gros corpus (lots de 25-50 chunks)
  • Ajoutez une logique de retry avec backoff exponentiel pour la robustesse
  • Sauvegardez les embeddings sur disque (NumPy) pour éviter de recalculer
  • Conservez les métadonnées (texte, source, position) en JSON séparé
  • Vérifiez la qualité des embeddings avant de les utiliser en production
← Lecon precedentePréparer les DonnéesLecon suivante →Vector Store avec FAISS