Créer les Embeddings

Encoder les chunks en vecteurs

Une fois vos documents découpés en chunks, l’étape suivante consiste à les transformer en embeddings via l’API mistral-embed. Ces vecteurs seront ensuite stockés dans un vector store pour la recherche.

Encodage simple

Pour un petit corpus (moins de 100 chunks), un appel direct suffit :

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 (batching)

Pour des corpus plus grands, vous devez découper en lots pour respecter les limites de l’API :

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

Gestion des erreurs et retry

En production, les appels API peuvent échouer. Ajoutez une logique de retry :

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)

Sauvegarde et chargement

Ne recalculez pas les embeddings à chaque démarrage. Sauvegardez-les sur disque :

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érification de la qualité

Avant de passer à l’étape suivante, vérifiez que vos embeddings sont corrects :

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 etre 1.0): {self_sim:.6f}")

verify_embeddings(chunks, embeddings)

Pipeline complet

Voici le résumé du pipeline d’encodage :

# 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