Introduction : Le Défi des Embeddings dans un Système RAG E-commerce
En tant qu'ingénieur senior qui a déployé une dizaines de systèmes RAG en production, je me souviens d'un cas particulièrement instructif : notre plateforme e-commerce traitait 50 000 requêtes quotidiennes de recherche de produits. Le système initial utilisait des embeddings génériques avec un recall de 62% — autrement dit, 38% des produits pertinents échappaient complètement aux utilisateurs.
Après trois mois d'optimisation intensive, nous avons atteint un recall de 94% tout en réduisant la latence de 280ms à 67ms. Cet article condense les techniques qui ont fait la différence, avec du code production-ready utilisant l'API HolySheep AI.
Comprendre le Pipeline d'Embedding
Avant d'optimiser, visualisons le pipeline complet :
- Tokenisation : Conversion du texte en tokens numériques
- Encodage : Transformation via le modèle d'embedding (dimension typiquement 384 à 3072)
- Indexation : Stockage dans une base vectorielle (FAISS, Milvus, Pinecone)
- Requête : Embedding de la requête → recherche du plus proche voisinnage
- Reranking : Réordonnancement des résultats par cross-attention
Technique 1 : Choix du Modèle d'Embedding Adapté
Le modèle d'embedding est la fondation de tout le système. Utiliser un modèle générique sur des données domain-specific génère des pertes considérables.
# Installation des dépendances
pip install sentence-transformers httpx numpy
import httpx
import json
from typing import List, Dict
class HolySheepEmbeddingClient:
"""Client optimisé pour les embeddings HolySheep AI"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def embed_documents(self, texts: List[str], model: str = "embedding-3") -> List[List[float]]:
"""Génère des embeddings pour des documents avec gestion du batch"""
embeddings = []
batch_size = 32 # Optimisé pour le throughput
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
payload = {
"model": model,
"input": batch,
"encoding_format": "float"
}
with httpx.Client(timeout=30.0) as client:
response = client.post(
f"{self.base_url}/embeddings",
headers=self.headers,
json=payload
)
response.raise_for_status()
data = response.json()
embeddings.extend([item["embedding"] for item in data["data"]])
return embeddings
Utilisation
client = HolySheepEmbeddingClient("YOUR_HOLYSHEEP_API_KEY")
documents = [
"iPhone 15 Pro Max 256GB Titane naturel",
"Samsung Galaxy S24 Ultra 512GB Phantom Black",
"AirPods Pro 2ème génération avec USB-C"
]
embeddings = client.embed_documents(documents)
print(f"Dimension de l'embedding : {len(embeddings[0])}")Technique 2 : Chunking Stratégique pour la Récupération
La taille des chunks impacte directement la granularité de la récupération. Trop grands = dilution sémantique. Trop petits = perte de contexte.
import re
from typing import List, Tuple
class SmartChunker:
"""
Stratégie de chunking hybride combinant:
- Séparation sémantique (paragraphes)
- Limite de tokens configurable
- Chevauchement pour la continuité contextuelle
"""
def __init__(self, max_tokens: int = 512, overlap_tokens: int = 64):
self.max_tokens = max_tokens
self.overlap_tokens = overlap_tokens
def chunk_text(self, text: str) -> List[Dict[str, any]]:
# Séparation par paragraphs preserve la structure sémantique
paragraphs = re.split(r'\n\n+', text)
chunks = []
current_chunk = ""
current_tokens = 0
for para in paragraphs:
para_tokens = len(para.split()) * 1.3 # Approximation tokens
if current_tokens + para_tokens > self.max_tokens:
# Sauvegarder le chunk courant
if current_chunk:
chunks.append({
"text": current_chunk.strip(),
"tokens": current_tokens,
"chunk_id": len(chunks)
})
# Gestion du chevauchement
words = current_chunk.split()
overlap_words = words[-(self.overlap_tokens // 2):]
current_chunk = " ".join(overlap_words) + " " + para
current_tokens = len(current_chunk.split()) * 1.3
else:
current_chunk += " " + para
current_tokens += para_tokens
# Dernier chunk
if current_chunk.strip():
chunks.append({
"text": current_chunk.strip(),
"tokens": current_tokens,
"chunk_id": len(chunks)
})
return chunks
Exemple d'utilisation pour un catalogue e-commerce
chunker = SmartChunker(max_tokens=384, overlap_tokens=48)
produit_description = """
iPhone 15 Pro Max - Le summum de la technologie mobile
Écran Super Retina XDR de 6.7 pouces avec ProMotion 120Hz.
Puce A17 Pro avec Neural Engine 16 cœurs pour des performances
inégalées en intelligence artificielle.
Système de caméra Pro avec:
- Capteur principal 48MP avec pixels photosites 2.24μm
- Téléobjectif 5x avec stabilisation par décalage de capteur
- Ultra grand-angle 120° avec mode macro
Finition Titane Grade 5, la même qualité utilisée en aerospace.
Batterie jusqu'à 29 heures de lecture vidéo.
"""
chunks = chunker.chunk_text(produit_description)
for chunk in chunks:
print(f"Chunk {chunk['chunk_id']}: {chunk['tokens']} tokens")
print(f" {chunk['text'][:80]}...")Technique 3 : Indexation Hybride et Recherche
Pour les catalogues volumineux, combinez recherche vectorielle dense et recherche BM25 sparse. HolySheep offre une latence moyenne de 47ms sur les embeddings, ce qui permet desindexations en temps réel.
from scipy.spatial.distance import cosine
import numpy as np
class HybridSearchEngine:
"""
Moteur de recherche hybride combinant:
1. Recherche vectorielle (similarité cosinus)
2. Pondération par score BM25
3. Reranking par cross-encoder
"""
def __init__(self, embedding_client: HolySheepEmbeddingClient,
alpha: float = 0.7):
self.client = embedding_client
self.alpha = alpha # Pondération vectorielle (1-alpha pour BM25)
self.document_embeddings = {}