Bienvenue dans ce tutoriel complet dédié aux systèmes de recommandation basés sur l'intelligence artificielle. Si vous débutez en programmation et souhaitez créer un système capable de suggérer des contenus pertinents à vos utilisateurs, cet article est fait pour vous. Nous allons construire ensemble, pas à pas, un système de recommandation fonctionnel utilisant la technologie des embeddings vectoriels avec mise à jour en temps réel.
Comprendre les Fondamentaux
Avant de coder, expliquons simplement ce qu'est un système de recommandation. Imaginez une bibliothèque géante où chaque livre est accompagné de milliers de notes invisible. Ces notes décrivent le contenu du livre : son sujet, son style, son public cible. Plus les notes sont similaires entre deux livres, plus ils se ressemblent.
Les embeddings sont exactement ces « notes invisibles ». Ce sont des listes de nombres (vecteurs) qui représentent le sens d'un texte, d'une image ou de tout autre contenu. Quand vous recherchez « films d'action palpitants », le système convertit votre requête en vecteur et trouve les films dont les vecteurs sont les plus similaires au vôtre.
Pourquoi HolySheep AI ?
Pour générer ces embeddings, nous avons besoin d'une API d'intelligence artificielle performante. S'inscrire ici sur HolySheep AI vous donne accès à des avantages considérables : une latence inférieure à 50 millisecondes (contre souvent 200-500ms ailleurs), des tarifs réduits de plus de 85% par rapport aux solutions américaines, et des crédits gratuits pour commencer sans investir. Les prix 2026 par million de tokens sont particulièrement compétitifs : DeepSeek V3.2 à 0,42 dollar, Gemini 2.5 Flash à 2,50 dollars, contre 8 dollars pour GPT-4.1 ou 15 dollars pour Claude Sonnet 4.5 sur les plateformes traditionnelles.
Configuration de l'Environnement
Installation des Packages Nécessaires
Ouvrez votre terminal et installez les bibliothèques Python dont nous aurons besoin. Ces packages nous permettront de communiquer avec l'API et de manipuler des vecteurs efficacement.
pip install requests numpy faiss-cpu scikit-learn python-dotenvLe package faiss-cpu est essentiel : il permet de rechercher des vecteurs similaires parmi des millions d'éléments en quelques millisecondes seulement. Pour un système de recommandation, cette vitesse est cruciale.
Configuration de la Clé API
Créez un fichier nommé .env à la racine de votre projet et ajoutez votre clé API. Vous trouverez cette clé dans votre tableau de bord HolySheep après votre inscription.
HOLYSHEEP_API_KEY=votre_cle_api_iciCréation du Module d'Embeddings
Commençons par créer un module Python qui gérera toute la communication avec l'API HolySheep pour générer nos embeddings. Ce module sera le cœur de notre système de recommandation.
import requests
import numpy as np
from typing import List, Optional
from dotenv import load_dotenv
import os
load_dotenv()
class EmbeddingGenerator:
"""
Générateur d'embeddings utilisant l'API HolySheep AI.
Cette classe permet de convertir du texte en vecteurs numériques
utilisés pour la recherche de similarité.
"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
self.base_url = "https://api.holysheep.ai/v1"
self.model = "embedding-deepseek-v3"
if not self.api_key:
raise ValueError("Clé API HolySheep requise. "
"Obtenez-la sur https://www.holysheep.ai/register")
def generate_embedding(self, text: str) -> np.ndarray:
"""
Génère un embedding pour un texte unique.
Args:
text: Le texte à convertir en vecteur
Returns:
Vecteur numpy représentant le texte
"""
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"input": text,
"model": self.model
}
)
if response.status_code != 200:
raise Exception(f"Erreur API: {response.status_code} - {response.text}")
data = response.json()
embedding = np.array(data["data"][0]["embedding"])
return embedding
def generate_batch_embeddings(self, texts: List[str]) -> np.ndarray:
"""
Génère des embeddings pour plusieurs textes simultanément.
L'exécution par lots est beaucoup plus efficace.
Args:
texts: Liste de textes à convertir
Returns:
Matrice numpy où chaque ligne est un embedding
"""
response = requests.post(
f"{self.base_url}/embeddings",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"input": texts,
"model": self.model
}
)
if response.status_code != 200:
raise Exception(f"Erreur API: {response.status_code} - {response.text}")
data = response.json()
embeddings = np.array([item["embedding"] for item in data["data"]])
return embeddings
Exemple d'utilisation
if __name__ == "__main__":
generator = EmbeddingGenerator()
# Test avec un texte unique
texte_test = "Un film d'action palpitant avec des explosions"
embedding = generator.generate_embedding(texte_test)
print(f"Dimension de l'embedding: {embedding.shape}")
print(f"Extrait du vecteur: {embedding[:5]}")
Construction de l'Index Vectoriel
Maintenant que nous pouvons générer des embeddings, il faut les ranger dans une structure de données optimisée pour la recherche. FAISS (Facebook AI Search Similarity) est l'outil idéal pour cela. Notre index permettra de trouver rapidement les k-plus-proches voisins d'un vecteur de requête.
import faiss
import numpy as np
from typing import List, Dict, Tuple, Optional
import json
from datetime import datetime
class IncrementalVectorIndex:
"""
Index vectoriel avec mise à jour incrémentielle en temps réel.
Supporte l'ajout, la suppression et la mise à jour d'éléments
sans reconstruction complète de l'index.
"""
def __init__(self, dimension: int = 1536, index_type: str = "IVF"):
self.dimension = dimension
self.index_type = index_type
self.embeddings = []
self.metadata: List[Dict] = []
self.item_ids: List[str] = []
# Initialisation de l'index FAISS
if index_type == "Flat":
# Recherche exacte, lent mais précis
self.index = faiss.IndexFlatIP(dimension)
elif index_type == "IVF":
# Recherche approximative, rapide pour grands volumes
quantizer = faiss.IndexFlatIP(dimension)
self.index = faiss.IndexIVFFlat(quantizer, dimension, 100)
else:
raise ValueError(f"Type d'index inconnu: {index_type}")
def train(self, training_vectors: np.ndarray):
"""
Entraîne l'index IVF sur un échantillon de vecteurs.
À appeler avant d'ajouter des données si使用的是 IVF.
"""
if self.index_type == "IVF" and not self.index.is_trained:
self.index.train(training_vectors.astype('float32'))
def add_items(self, item_ids: List[str], embeddings: np.ndarray,
metadata: List[Dict]):
"""
Ajoute des éléments à l'index de manière incrémentale.
Args:
item_ids: Identifiants uniques des éléments
embeddings: Vecteurs d'embedding correspondants
metadata: Métadonnées associées (titre, catégorie, etc.)
"""
if len(item_ids) != len(embeddings) or len(embeddings) != len(metadata):
raise ValueError("Les listes doivent avoir la même longueur")
# Normalisation des vecteurs pour la similarité cosinus
normalized = embeddings / np.linalg.norm(embeddings, axis=1, keepdims=True)
# Conversion au format float32 requis par FAISS
vectors = normalized.astype('float32')
# Ajout à l'index
self.index.add(vectors)
# Stockage des métadonnées
self.item_ids.extend(item_ids)
self.embeddings.append(embeddings)
self.metadata.extend(metadata)
def search(self, query_embedding: np.ndarray, k: int = 10) -> List[Dict]:
"""
Recherche les k éléments les plus similaires.
Args:
query_embedding: Vecteur de requête
k: Nombre de résultats à retourner
Returns:
Liste des éléments similaires avec leur score
"""
normalized = query_embedding / np.linalg.norm(query_embedding)
query = normalized.reshape(1, -1).astype('float32')
distances, indices = self.index.search(query, k)
results = []
for dist, idx in zip(distances[0], indices[0]):
if idx < len(self.metadata):
result = {
"id": self.item_ids[idx],
"score": float(dist),
"metadata": self.metadata[idx]
}
results.append(result)
return results
def update_item(self, item_id: str, new_embedding: np.ndarray,
new_metadata: Dict):
"""
Met à jour un élément existant dans l'index.
"""
try:
idx = self.item_ids.index(item_id)
# Reconstruction simple : on recrée l'index
self.rebuild_index()
except ValueError:
print(f"Élément {item_id} non trouvé, ajout direct")
self.add_items([item_id], new_embedding.reshape(1, -1), [new_metadata])
def rebuild_index(self):
"""
Reconstruire l'index entier après modification.
Utilisé après suppression ou mise à jour.
"""
if not self.embeddings:
return
all_embeddings = np.vstack(self.embeddings)
normalized = all_embeddings / np.linalg.norm(all_embeddings, axis=1, keepdims=True)
# Réinitialisation de l'index
if self.index_type == "Flat":
self.index = faiss.IndexFlatIP(self.dimension)
else:
quantizer = faiss.IndexFlatIP(self.dimension)
self.index = faiss.IndexIVFFlat(quantizer, self.dimension, 100)
self.index.train(normalized.astype('float32'))
self.index.add(normalized.astype('float32'))
def save(self, filepath: str):
"""Sauvegarde l'index et les métadonnées sur disque."""
faiss.write_index(self.index, f"{filepath}.index")
with open(f"{filepath}_metadata.json", "w", encoding="utf-8") as f:
json.dump({
"metadata": self.metadata,
"item_ids": self.item_ids,
"dimension": self.dimension,
"index_type": self.index_type,
"saved_at": datetime.now().isoformat()
}, f, ensure_ascii=False, indent=2)
def load(self, filepath: str):
"""Charge un index existant depuis le disque."""
self.index = faiss.read_index(f"{filepath}.index")
with open(f"{filepath}_metadata.json", "r", encoding="utf-8") as f:
data = json.load(f)
self.metadata = data["metadata"]
self.item_ids = data["item_ids"]
self.dimension = data["dimension"]
self.index_type = data["index_type"]
Système de Recommandation Complet
Combinons maintenant nos composants pour créer un système de recommandation fonctionnel. Ce système gérera automatiquement les mises à jour en temps réel et construira un index incrémentiel au fur et à mesure que de nouveaux contenus sont ajoutés.
import time
from concurrent.futures import ThreadPoolExecutor
from threading import Lock
class RealTimeRecommender:
"""
Système de recommandation avec mise à jour temps réel.
Gère un index incrémental et répond aux requêtes utilisateur.
"""
def __init__(self, api_key: str, max_batch_size: int = 100):
self.embedding_generator = EmbeddingGenerator(api_key)
self.index = IncrementalVectorIndex(dimension=1536, index_type="Flat")
self.max_batch_size = max_batch_size
self.pending_items = []
self.lock = Lock()
self.stats = {
"total_requests": 0,
"total_recommendations": 0,
"avg_latency_ms": 0,
"cache_hits": 0
}
def add_content(self, content_id: str, text: str,
category: str = None, tags: List[str] = None):
"""
Ajoute un nouveau contenu au système de recommandation.
Args:
content_id: Identifiant unique du contenu
text: Description ou titre du contenu
category: Catégorie optionnelle
tags: Tags optionnels
"""
metadata = {
"id": content_id,
"text": text,
"category": category,
"tags": tags or [],
"added_at": time.time()
}
# Génération de l'embedding
embedding = self.embedding_generator.generate_embedding(text)
with self.lock:
self.index.add_items(
[content_id],
embedding.reshape(1, -1),
[metadata]
)
print(f"✓ Contenu ajouté: {content_id}")
return True
def add_content_batch(self, contents: List[Dict]):
"""
Ajoute plusieurs contenus simultanément pour optimiser les appels API.
Args:
contents: Liste de dictionnaires avec 'id', 'text', 'category', 'tags'
"""
texts = [c["text"] for c in contents]
# Un seul appel API pour tous les textes
embeddings = self.embedding_generator.generate_batch_embeddings(texts)
metadata_list = []
item_ids = []
for c in contents:
item_ids.append(c["id"])
metadata_list.append({
"id": c["id"],
"text": c["text"],
"category": c.get("category"),
"tags": c.get("tags", []),
"added_at": time.time()
})
with self.lock:
self.index.add_items(item_ids, embeddings, metadata_list)
print(f"✓ Batch ajouté: {len(contents)} contenus")
def recommend(self, user_query: str, top_k: int = 5) -> List[Dict]:
"""
Génère des recommandations basées sur une requête utilisateur.
Args:
user_query: Texte décrivant les intérêts de l'utilisateur
top_k: Nombre de recommandations à retourner
Returns:
Liste ordonnée des recommandations avec scores
"""
start_time = time.time()
# Conversion de la requête en embedding
query_embedding = self.embedding_generator.generate_embedding(user_query)
# Recherche dans l'index
results = self.index.search(query_embedding, k=top_k)
latency_ms = (time.time() - start_time) * 1000
# Mise à jour des statistiques
self._update_stats(latency_ms)
return results
def recommend_similar_to(self, content_id: str, top_k: int = 5) -> List[Dict]:
"""
Trouve des contenus similaires à un contenu existant.
Args:
content_id: ID du contenu de référence
top_k: Nombre de similaires à retourner
"""
try:
idx = self.index.item_ids.index(content_id)
all_embeddings = np.vstack(self.index.embeddings)
content_embedding = all_embeddings[idx]
return self.index.search(content_embedding, k=top_k + 1)
except (ValueError, IndexError):
print(f"Contenu {content_id} non trouvé")
return []
def _update_stats(self, latency_ms: float):
"""Met à jour les statistiques de performance."""
self.stats["total_requests"] += 1
n = self.stats["total_requests"]
current_avg = self.stats["avg_latency_ms"]
self.stats["avg_latency_ms"] = (current_avg * (n - 1) + latency_ms) / n
def get_stats(self) -> Dict:
"""Retourne les statistiques d'utilisation."""
return self.stats.copy()
Démonstration complète
if __name__ == "__main__":
API_KEY = os.getenv("HOLYSHEEP_API_KEY")
recommender = RealTimeRecommender(API_KEY)
# Ajout de contenus de démonstration
contenus_demo = [
{"id": "film_001", "text": "Avengers: Endgame - Film d'action super-héros",
"category": "Films", "tags": ["action", "super-héros", "marvel"]},
{"id": "film_002", "text": "Inception - Thriller de science-fiction complexe",
"category": "Films", "tags": ["sf", "thriller", "nolan"]},
{"id": "livre_001", "text": "Le Petit Prince - Conte philosophique touchant",
"category": "Livres", "tags": ["classique", "philosophie", "enfant"]},
{"id": "film_003", "text": "Titanic - Romance dramatique inoubliable",
"category": "Films", "tags": ["romance", "drame", "historique"]},
{"id": "serie_001", "text": "Breaking Bad - Série dramatique intense",
"category": "Séries", "tags": ["drame", "crime", "addiction"]},
]
recommender.add_content_batch(contenus_demo)
# Test de recommandation
print("\n📌 Recommandations pour 'films激动人心 d'action':")
resultats = recommender.recommend("films d'action palpitants avec des effets spéciaux", top_k=3)
for r in resultats:
print(f" → {r['metadata']['text']} (score: {r['score']:.3f})")
# Statistiques
print(f"\n📊 Statistiques: {recommender.get_stats()}")
Optimisation pour la Production
Pour un déploiement en production avec des milliers de requêtes par seconde, quelques optimisations sont essentielles. Nous allons implémenter un système de mise en cache et un gestionnaire de mise à jour incrémentielle.
Gestionnaire de Cache et Batch
from collections import OrderedDict
import hashlib
class EmbeddingCache:
"""Cache LRU pour les embeddings afin de réduire les appels API."""
def __init__(self, max_size: int = 10000):
self.cache = OrderedDict()
self.max_size = max_size
self.hits = 0
self.misses = 0
def _hash_text(self, text: str) -> str:
"""Génère un hash stable pour le texte."""
return hashlib.sha256(text.encode()).hexdigest()
def get(self, text: str) -> Optional[np.ndarray]:
"""Récupère un embedding depuis le cache."""
key = self._hash_text(text)
if key in self.cache:
self.hits += 1
self.cache.move_to_end(key)
return self.cache[key]
self.misses += 1
return None
def put(self, text: str, embedding: np.ndarray):
"""Stocke un embedding dans le cache."""
key = self._hash_text(text)
if key in self.cache:
self.cache.move_to_end(key)
else:
if len(self.cache) >= self.max_size:
self.cache.popitem(last=False)
self.cache[key] = embedding
def get_stats(self) -> Dict:
total = self.hits + self.misses
hit_rate = (self.hits / total * 100) if total > 0 else 0
return {"hits": self.hits, "misses": self.misses, "hit_rate": f"{hit_rate:.1f}%"}
class IncrementalUpdateManager:
"""
Gère les mises à jour incrémentielles de l'index.
Accumule les changements et les applique par lots.
"""
def __init__(self, recommender: RealTimeRecommender,
batch_interval_seconds: int = 60,
batch_size: int = 50):
self.recommender = recommender
self.batch_interval = batch_interval_seconds
self.batch_size = batch_size
self.pending_updates = []
self.last_flush = time.time()
def