Embeddings et recherche vectorielle

Un utilisateur demande "où se reposent les chats ?", et ton carnet contient la phrase "The kitten lies on the carpet". Une recherche par mots-clés ne trouve ici aucun mot commun, alors que le sens est presque le même. Il nous faut un moyen de comparer le sens, pas les lettres. Ce moyen, ce sont les embeddings.

Les embeddings sont des représentations vectorielles du texte qui permettent aux machines de "comprendre" le sens des mots et des phrases. Vois-les comme une empreinte : chaque animal laisse une trace différente, et un pisteur reconnaît à sa forme qui est passé sur le sentier. Un embedding, c'est l'empreinte d'un texte - une suite de nombres qui décrit sa sémantique.

Que sont les embeddings ?

Un embedding est un vecteur de nombres réels qui représente un texte dans un espace sémantique. Des textes au sens proche reçoivent des vecteurs proches les uns des autres. Le plus simple pour en générer un, c'est l'API OpenAI, avec la méthode client.embeddings.create :

1from openai import OpenAI
2
3client = OpenAI()
4
5def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
6    """Génère un embedding pour le texte donné."""
7    response = client.embeddings.create(
8        model=model,
9        input=text
10    )
11    return response.data[0].embedding
12
13# Exemple
14text = "Python is a great programming language"
15embedding = get_embedding(text)
16
17print(f"Vector dimension: {len(embedding)}")  # 1536 for text-embedding-3-small
18print(f"First 5 values: {embedding[:5]}")

La fonction renvoie une simple liste de floats. Le modèle text-embedding-3-small produit par défaut des vecteurs de 1536 dimensions, et text-embedding-3-large en produit 3072. Le texte lui-même n'est stocké nulle part, on récupère seulement son empreinte numérique. Le processus suit toujours le même rythme : préparer le texte, l'envoyer au modèle d'embeddings, recevoir le vecteur et l'enregistrer dans une base vectorielle.

Modèles d'embeddings

Tous les campements n'ont pas le budget pour une API payante. La bibliothèque sentence-transformers exécute des modèles open source en local, sur ta propre machine. La classe ci-dessous réunit au même endroit un modèle local et un modèle OpenAI :

1from sentence_transformers import SentenceTransformer
2
3# Modèles open-source
4class EmbeddingModels:
5    """Divers modèles d'embedding."""
6
7    def __init__(self):
8        # Rapide et léger
9        self.minilm = SentenceTransformer('all-MiniLM-L6-v2')
10
11        # Plus grand, meilleur pour le multilingue
12        self.multilingual = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
13
14    def embed_with_openai(self, texts: list[str]) -> list[list[float]]:
15        """Embeddings OpenAI - la plus haute qualité."""
16        from openai import OpenAI
17        client = OpenAI()
18
19        response = client.embeddings.create(
20            model="text-embedding-3-large",  # 3072 dimensions
21            input=texts
22        )
23        return [item.embedding for item in response.data]
24
25    def embed_with_sentence_transformers(self, texts: list[str]) -> list[list[float]]:
26        """Embeddings locaux - sans coûts d'API."""
27        return self.minilm.encode(texts).tolist()

all-MiniLM-L6-v2 est rapide et léger (384 dimensions), paraphrase-multilingual-mpnet-base-v2 est plus gros et gère de nombreuses langues, dont le polonais. La méthode encode renvoie un tableau NumPy, c'est pour ça qu'on appelle .tolist().

Calculons maintenant les embeddings de deux phrases liées à l'IA et d'une phrase qui n'a rien à voir :

1# Comparaison des modèles
2models = EmbeddingModels()
3
4texts = [
5    "Machine learning is a field of AI",
6    "Deep learning uses neural networks",
7    "I like eating pizza"
8]
9
10# Embeddings open-source
11embeddings = models.embed_with_sentence_transformers(texts)
12print(f"Local embeddings: {len(embeddings[0])} dimensions")

La sortie affiche 384 dimensions. Remarque que la dimension dépend du modèle, pas de la longueur du texte : une phrase courte et un long paragraphe reçoivent des vecteurs de même longueur.

Métriques de similarité

On a des vecteurs, il faut donc les comparer. Voici les trois mesures les plus populaires, chacune en une ligne de NumPy :

1import numpy as np
2from typing import Callable
3
4def cosine_similarity(vec1: np.ndarray, vec2: np.ndarray) -> float:
5    """Similarité cosinus - la plus utilisée."""
6    return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
7
8def euclidean_distance(vec1: np.ndarray, vec2: np.ndarray) -> float:
9    """Distance euclidienne."""
10    return np.linalg.norm(vec1 - vec2)
11
12def dot_product(vec1: np.ndarray, vec2: np.ndarray) -> float:
13    """Produit scalaire."""
14    return np.dot(vec1, vec2)

La similarité cosinus ne regarde que l'angle entre les vecteurs, et c'est la métrique la plus utilisée pour comparer des embeddings. La distance euclidienne mesure un écart, donc ici un nombre plus petit signifie plus de similarité. Le produit scalaire dépend à la fois de l'angle et de la longueur des vecteurs.

Vérifions-le sur trois phrases, dont deux parlent de chats :

1# Démonstration
2def demonstrate_similarity():
3    """Montre comment fonctionnent les métriques de similarité."""
4    from sentence_transformers import SentenceTransformer
5
6    model = SentenceTransformer('all-MiniLM-L6-v2')
7
8    sentences = [
9        "The cat sits on the mat",              # 0
10        "The kitten lies on the carpet",         # 1 - sens similaire
11        "Programming in Python",                  # 2 - sens différent
12    ]
13
14    embeddings = model.encode(sentences)
15
16    print("Cosine similarity:")
17    print(f"  Sentence 0 vs 1: {cosine_similarity(embeddings[0], embeddings[1]):.3f}")
18    print(f"  Sentence 0 vs 2: {cosine_similarity(embeddings[0], embeddings[2]):.3f}")
19    print(f"  Sentence 1 vs 2: {cosine_similarity(embeddings[1], embeddings[2]):.3f}")
20
21demonstrate_similarity()

Les phrases 0 et 1 devraient obtenir un score nettement plus élevé que les paires avec la phrase sur Python, même si "cat" et "kitten" sont des mots différents. Je n'écris pas de valeurs exactes, car elles dépendent de la version du modèle.

Encore une option utile. Si tu demandes au modèle des vecteurs normalisés (de longueur 1), le produit scalaire devient exactement la similarité cosinus :

1from sentence_transformers import SentenceTransformer
2
3model = SentenceTransformer('all-MiniLM-L6-v2')
4document_text = "Lions hunt most often at dusk"
5
6# Vecteur de longueur 1 - le produit scalaire donne directement la similarité cosinus
7embedding = model.encode(document_text, normalize_embeddings=True)
8print(embedding.shape)  # (384,)

Le paramètre s'appelle normalize_embeddings, et une seule chaîne donne un tableau à une dimension. Le vecteur pointe dans la même direction que sans normalisation, seule sa longueur change.

Recherche sémantique

Assemblons ces pièces en un moteur de recherche sémantique. D'abord un petit conteneur pour un résultat, créé avec le décorateur @dataclass, et une classe qui indexe les documents une seule fois :

1from dataclasses import dataclass
2import numpy as np
3
4@dataclass
5class SearchResult:
6    """Résultat de recherche sémantique."""
7    text: str
8    score: float
9    index: int
10
11class SemanticSearch:
12    """Implémentation simple de recherche sémantique."""
13
14    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
15        from sentence_transformers import SentenceTransformer
16        self.model = SentenceTransformer(model_name)
17        self.documents: list[str] = []
18        self.embeddings: np.ndarray | None = None
19
20    def index_documents(self, documents: list[str]) -> None:
21        """Indexe les documents."""
22        self.documents = documents
23        self.embeddings = self.model.encode(documents)
24        print(f"Indexed {len(documents)} documents")

index_documents calcule les embeddings de tous les documents en un seul appel à encode et les garde dans une matrice. Ça corrige la faiblesse de la leçon précédente : on calcule les embeddings des documents une fois, pas à chaque question.

La méthode search calcule la similarité de la question avec toute la matrice d'un coup, sans boucle :

1    def search(self, query: str, top_k: int = 5) -> list[SearchResult]:
2        """Recherche les documents les plus similaires."""
3        if self.embeddings is None:
4            raise ValueError("No indexed documents!")
5
6        query_embedding = self.model.encode([query])[0]
7
8        # Calcule les similarités
9        similarities = np.dot(self.embeddings, query_embedding)
10        similarities /= np.linalg.norm(self.embeddings, axis=1)
11        similarities /= np.linalg.norm(query_embedding)
12
13        # Résultats top-K
14        top_indices = np.argsort(similarities)[::-1][:top_k]
15
16        results = []
17        for idx in top_indices:
18            results.append(SearchResult(
19                text=self.documents[idx],
20                score=float(similarities[idx]),
21                index=int(idx)
22            ))
23
24        return results

np.argsort trie par ordre croissant, donc [::-1] inverse l'ordre et [:top_k] prend les meilleurs résultats. Les deux divisions par les normes transforment le produit scalaire en similarité cosinus.

Place au test sur cinq documents :

1# Exemple d'utilisation
2search = SemanticSearch()
3
4documents = [
5    "Python is a general-purpose programming language",
6    "Machine Learning uses algorithms to learn from data",
7    "RAG combines retrieval with text generation",
8    "FastAPI is a modern framework for building APIs",
9    "Docker containerizes applications",
10]
11
12search.index_documents(documents)
13
14results = search.search("How to build web applications?")
15for r in results:
16    print(f"Score: {r.score:.3f} | {r.text}")

Pour la question sur les applications web, le document sur FastAPI devrait arriver en tête, alors qu'il ne partage aucun mot-clé avec la question.

Traitement par lots

Mille documents envoyés un par un, ce sont mille requêtes HTTP. L'API OpenAI accepte une liste de textes dans le paramètre input, donc on les envoie par lots :

1def batch_embed(texts: list[str], batch_size: int = 100) -> list[list[float]]:
2    """Crée les embeddings de grandes quantités de texte par lots."""
3    from openai import OpenAI
4    client = OpenAI()
5
6    all_embeddings = []
7
8    for i in range(0, len(texts), batch_size):
9        batch = texts[i:i + batch_size]
10
11        response = client.embeddings.create(
12            model="text-embedding-3-small",
13            input=batch
14        )
15
16        batch_embeddings = [item.embedding for item in response.data]
17        all_embeddings.extend(batch_embeddings)
18
19        print(f"Processed {min(i + batch_size, len(texts))}/{len(texts)}")
20
21    return all_embeddings

L'ordre des résultats dans response.data correspond à l'ordre des textes du lot, c'est pourquoi on peut simplement les ajouter avec extend.

La deuxième façon d'économiser, c'est le cache. Le même texte avec le même modèle donne toujours le même embedding, donc on enregistre le résultat sur disque sous une clé de hachage :

1# Mise en cache des embeddings
2import hashlib
3import json
4from pathlib import Path
5
6class EmbeddingCache:
7    """Cache pour les embeddings."""
8
9    def __init__(self, cache_dir: str = ".embedding_cache"):
10        self.cache_dir = Path(cache_dir)
11        self.cache_dir.mkdir(exist_ok=True)
12
13    def _get_cache_key(self, text: str, model: str) -> str:
14        """Génère une clé de cache."""
15        content = f"{model}:{text}"
16        return hashlib.md5(content.encode()).hexdigest()
17
18    def get(self, text: str, model: str) -> list[float] | None:
19        """Récupère un embedding depuis le cache."""
20        key = self._get_cache_key(text, model)
21        cache_file = self.cache_dir / f"{key}.json"
22
23        if cache_file.exists():
24            return json.loads(cache_file.read_text())
25        return None
26
27    def set(self, text: str, model: str, embedding: list[float]) -> None:
28        """Enregistre un embedding dans le cache."""
29        key = self._get_cache_key(text, model)
30        cache_file = self.cache_dir / f"{key}.json"
31        cache_file.write_text(json.dumps(embedding))

MD5 sert ici uniquement d'identifiant de fichier rapide, pas de protection, donc ses faiblesses cryptographiques n'ont aucune importance. Le modèle fait partie de la clé exprès : après un changement de modèle, les anciens vecteurs ne correspondent plus aux nouveaux. Je te conseille de mettre les embeddings en cache dès le premier jour, car une réindexation fait alors gagner du temps et de l'argent.

Les embeddings sont la base de tout système RAG. Dans la prochaine leçon, tu vas découvrir les bases de données vectorielles - des bases spécialisées pour stocker et rechercher des vecteurs - qui font ce que fait notre matrice NumPy, mais pour des millions de documents.

Retiens : un embedding, c'est l'empreinte d'un texte, et des sens proches laissent des traces proches.