Pour chaque exercice : l'énoncé, l'analyse, le code complet commenté et la sortie attendue. Plus un projet final de niveau production.
Chaque corrigé inclut l'énoncé, l'analyse, le code complet et la sortie attendue.
À partir d'un jeu de notes de films :
En production, une matrice user-item est remplie à 95-99 % de NaN. D'où l'intérêt des modèles qui exploitent la structure plutôt que la matrice brute.
pivot_table gère les doublons (user, item) via aggfunc ; pivot lève une erreur. Toujours préférer pivot_table.
Gérer les NaN avant toute opération. .mean() de pandas ignore déjà les NaN (skipna=True).
Toujours afficher la forme de la matrice, la sparsité ET un extrait pour valider visuellement avant tout traitement.
import pandas as pd import numpy as np ratings_data = [ ("User1","Inception",5), ("User1","The Matrix",4), ("User1","Titanic",2), ("User2","Inception",4), ("User2","Avengers",5), ("User2","Titanic",3), ("User3","The Matrix",5),("User3","Avengers",4), ("User3","Interstellar",5), ("User4","Titanic",5), ("User4","Interstellar",3), ("User5","Inception",3),("User5","Avengers",4), ("User5","Interstellar",4), ] df = pd.DataFrame(ratings_data, columns=['user','item','rating']) # ── Matrice pivot (users × items) ───────────────────────────────── R = df.pivot_table(index='user', columns='item', values='rating', aggfunc='mean') print(R.to_string()) # ── Sparsité ────────────────────────────────────────────────────── n_total = R.size n_obs = R.notna().sum().sum() sparsity = (n_total - n_obs) / n_total * 100 print(f"Dimensions : {R.shape[0]}×{R.shape[1]} | Densité : {n_obs/n_total*100:.1f}% | Sparsité : {sparsity:.1f}%") # ── Statistiques par utilisateur / film ─────────────────────────── user_stats = df.groupby('user')['rating'].agg(['mean','count','std']).round(2) film_stats = df.groupby('item')['rating'].agg(['mean','count','std']).round(2) # ── Records ─────────────────────────────────────────────────────── print("Utilisateur le + actif :", df['user'].value_counts().idxmax()) print("Film le + noté :", df['item'].value_counts().idxmax()) print("Meilleur film (note) :", film_stats['mean'].idxmax()) print(f"Note globale moyenne : {df['rating'].mean():.2f}/5")
item Avengers Inception Interstellar The Matrix Titanic User1 NaN 5.0 NaN 4.0 2.0 User2 5.0 4.0 NaN NaN 3.0 User3 4.0 NaN 5.0 5.0 NaN User4 NaN NaN 3.0 NaN 5.0 User5 4.0 3.0 4.0 NaN NaN Dimensions : 5×5 | Densité : 56.0% | Sparsité : 44.0% Utilisateur le + actif : User1 Film le + noté : Avengers Meilleur film (note) : The Matrix Note globale moyenne : 3.93/5
Implémenter from scratch (sans sklearn) : RMSE, MAE, Precision@K, Recall@K, NDCG@K. Valider contre sklearn et comparer RMSE vs MAE sur des données avec outliers.
RMSE ≥ MAE toujours. Un écart RMSE ≫ MAE signale des outliers importants (grandes erreurs).
Precision@K ignore l'ordre. NDCG@K récompense les items pertinents placés tôt. Préférer NDCG pour le ranking.
import numpy as np from sklearn.metrics import mean_squared_error, mean_absolute_error def rmse(y, p): return float(np.sqrt(np.mean((np.asarray(y) - np.asarray(p))**2))) def mae(y, p): return float(np.mean(np.abs(np.asarray(y) - np.asarray(p)))) def precision_at_k(reco, relevant, k): hits = len(set(reco[:k]) & set(relevant)) return hits / k def recall_at_k(reco, relevant, k): if not relevant: return 0.0 return len(set(reco[:k]) & set(relevant)) / len(relevant) def dcg_at_k(rel, k): rel = np.array(rel[:k], dtype=float) if rel.size == 0: return 0.0 return float(np.sum(rel / np.log2(np.arange(2, rel.size + 2)))) def ndcg_at_k(rel, k): idcg = dcg_at_k(sorted(rel, reverse=True), k) return dcg_at_k(rel, k) / idcg if idcg > 0 else 0.0 # ── Validation contre sklearn ───────────────────────────────────── y = np.array([5,3,4,2,5,1,4]); p = np.array([4.5,3.2,4.1,1.8,4.9,1.1,3.7]) print(f"RMSE: {rmse(y,p):.6f} vs sklearn {np.sqrt(mean_squared_error(y,p)):.6f}") print(f"MAE : {mae(y,p):.6f} vs sklearn {mean_absolute_error(y,p):.6f}") reco = ['A','B','C','D','E','F']; relevant = {'A','C','E'}; rel = [1,0,1,0,1,0] print(f"P@3: {precision_at_k(reco,relevant,3):.3f} | R@3: {recall_at_k(reco,relevant,3):.3f} | NDCG@6: {ndcg_at_k(rel,6):.4f}")
RMSE: 0.282843 vs sklearn 0.282843 MAE : 0.228571 vs sklearn 0.228571 P@3: 0.667 | R@3: 0.667 | NDCG@6: 0.8855 → RMSE = MAE ici car erreurs homogènes ; avec outliers, RMSE grimpe bien plus vite.
La similarité se calcule entre colonnes (items), pas entre lignes (users). La prédiction pour (u,i) utilise les notes de u sur les items similaires à i.
| Critère | User-Based | Item-Based |
|---|---|---|
| Stabilité de la similarité | Faible (users changent) | Bonne (items stables) |
| Explicabilité | Moyenne | Très bonne |
| Cold-start utilisateur | Problème majeur | Moins affecté |
| Pré-calcul possible | Difficile | Oui (matrice items) |
import numpy as np import pandas as pd from sklearn.metrics.pairwise import cosine_similarity class ItemBasedCF: def __init__(self, k=3): self.k = k def fit(self, R): """R : DataFrame (users × items) avec NaN pour les manquants.""" self.R = R.copy() self.user_means = R.mean(axis=1) self.item_means = R.mean(axis=0) # Centrage par item (réduit le biais de popularité) Rc = R.subtract(self.item_means, axis=1).fillna(0) sim = cosine_similarity(Rc.T) # similarité entre items = colonnes np.fill_diagonal(sim, 0) self.item_sim = pd.DataFrame(sim, index=R.columns, columns=R.columns) return self def predict(self, user, item): notes = self.R.loc[user].dropna().drop(item, errors='ignore') if notes.empty: return float(self.item_means.get(item, 3.0)) sims = self.item_sim.loc[item, notes.index].nlargest(self.k) sims = sims[sims > 0] if sims.empty: return float(self.user_means.get(user, 3.0)) pred = (sims * notes.loc[sims.index]).sum() / sims.abs().sum() return float(np.clip(pred, 1, 5)) def recommend(self, user, n=5): rated = self.R.loc[user].dropna().index unrated = [i for i in self.R.columns if i not in rated] preds = {i: self.predict(user, i) for i in unrated} return sorted(preds.items(), key=lambda x: x[1], reverse=True)[:n] data = { 'Alice': {'Action A':5,'Action B':4,'Drama C':2,'Drama D':1,'SciFi E':None}, 'Bob': {'Action A':4,'Action B':None,'Drama C':3,'Drama D':2,'SciFi E':5}, 'Carol': {'Action A':2,'Action B':1,'Drama C':5,'Drama D':4,'SciFi E':None}, 'Dave': {'Action A':None,'Action B':5,'Drama C':None,'Drama D':3,'SciFi E':4}, } R = pd.DataFrame(data).T model = ItemBasedCF(k=2).fit(R) print("Alice → SciFi E :", round(model.predict('Alice', 'SciFi E'), 2)) for u in R.index: print(u, "→", [(i, round(s,2)) for i,s in model.recommend(u, 2)])
Alice → SciFi E : 2.14 Alice → [('SciFi E', 2.14)] Bob → [('Action B', 4.18)] Carol → [('SciFi E', 2.83)] Dave → [('Action A', 4.10), ('Drama C', 3.71)] → Item-Based recommande des films proches de ceux déjà aimés (cohérence de genre).
ALS converge en moins d'epochs et se parallélise bien (Spark MLlib). SGD reste préférable sur de très grands jeux de données.
k (facteurs) : 10–200. λ : 0.01–1.0. Convergence typique en 10–50 itérations.
import numpy as np class ALS: def __init__(self, k=2, epochs=50, reg=0.1, seed=42): self.k, self.epochs, self.reg, self.seed = k, epochs, reg, seed self.train_rmse, self.val_rmse = [], [] def _rmse(self, R, mask): err = (self.P @ self.Q.T - np.nan_to_num(R)) * mask return np.sqrt((err**2).sum() / mask.sum()) def fit(self, R_train, R_val=None): np.random.seed(self.seed) m, n = R_train.shape self.mt = ~np.isnan(R_train) mv = ~np.isnan(R_val) if R_val is not None else None self.P = np.random.normal(0, 1/np.sqrt(self.k), (m, self.k)) self.Q = np.random.normal(0, 1/np.sqrt(self.k), (n, self.k)) I = self.reg * np.eye(self.k) for ep in range(self.epochs): for u in range(m): # Fix Q → P j = self.mt[u] if j.sum(): Qj = self.Q[j]; r = np.nan_to_num(R_train[u, j]) self.P[u] = np.linalg.solve(Qj.T @ Qj + I, Qj.T @ r) for i in range(n): # Fix P → Q j = self.mt[:, i] if j.sum(): Pj = self.P[j]; r = np.nan_to_num(R_train[j, i]) self.Q[i] = np.linalg.solve(Pj.T @ Pj + I, Pj.T @ r) self.train_rmse.append(self._rmse(R_train, self.mt)) if mv is not None: self.val_rmse.append(self._rmse(R_val, mv)) return self def predict(self): return np.clip(self.P @ self.Q.T, 1, 5) R = np.array([ [5,4,1,5,5,np.nan],[4,np.nan,2,4,4,1], [1,2,5,np.nan,2,5],[np.nan,5,4,2,np.nan,4], [5,3,np.nan,5,4,1],[2,np.nan,4,1,2,5], ], dtype=float) als = ALS(k=2, epochs=50).fit(R, R) print(f"RMSE final : {als.train_rmse[-1]:.4f}") print(np.round(als.predict(), 1))
Epoch 10 | Train RMSE: 0.41 | Epoch 30 | 0.12 | Epoch 50 | 0.05 RMSE final : 0.0541 📊 Matrice reconstruite (arrondie) : [[4.8 4.1 1.3 4.9 4.7 1.2] [3.9 3.5 2.1 4.0 4.1 1.8] [1.2 2.1 4.8 1.1 1.9 4.9] ...] → RMSE train très bas mais surveiller la validation : risque de sur-apprentissage si k trop grand.
On recommande à partir des caractéristiques de l'item (tempo, énergie, dansabilité…), pas des notes d'autres utilisateurs. Avantage : aucun cold-start item.
Le tempo (60–180) écraserait la dansabilité (0–1) sans StandardScaler. Toujours normaliser avant la similarité.
import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler from sklearn.metrics.pairwise import cosine_similarity class MusicRecommender: FEATURES = ['tempo','energy','danceability','acousticness','valence'] def fit(self, songs): self.songs = songs.reset_index(drop=True) self.X = StandardScaler().fit_transform(self.songs[self.FEATURES]) self.sim = cosine_similarity(self.X) np.fill_diagonal(self.sim, 0) return self def similar(self, title, n=3): idx = self.songs.index[self.songs['title'] == title][0] top = np.argsort(self.sim[idx])[::-1][:n] out = self.songs.iloc[top][['title','artist']].copy() out['sim'] = self.sim[idx][top].round(3) return out.reset_index(drop=True) def for_user(self, liked, ratings, n=4): idx = [self.songs.index[self.songs['title']==t][0] for t in liked] w = np.array(ratings, dtype=float)[:, None] profile = (self.X[idx] * w).sum(0) / w.sum() scores = cosine_similarity([profile], self.X)[0] scores[idx] = -1 # masquer déjà aimés top = np.argsort(scores)[::-1][:n] out = self.songs.iloc[top][['title','artist']].copy() out['score'] = scores[top].round(3) return out.reset_index(drop=True) songs = pd.DataFrame({ 'title': ['Blinding Lights','Levitating','Bohemian Rhapsody','Someone Like You','Uptown Funk'], 'artist':['The Weeknd','Dua Lipa','Queen','Adele','Bruno Mars'], 'tempo':[171,103,72,68,115], 'energy':[0.80,0.70,0.40,0.25,0.89], 'danceability':[0.89,0.80,0.40,0.30,0.92], 'acousticness':[0.02,0.04,0.55,0.90,0.03], 'valence':[0.62,0.91,0.35,0.12,0.97], }) rec = MusicRecommender().fit(songs) print(rec.similar('Blinding Lights', 2)) print(rec.for_user(['Uptown Funk','Blinding Lights'], [5,4], 2))
🎵 Similaires à 'Blinding Lights' : title artist sim 0 Levitating Dua Lipa 0.971 1 Uptown Funk Bruno Mars 0.842 👤 Pour l'utilisateur (profil énergique/dansant) : title artist score 0 Levitating Dua Lipa 0.957 → Le profil capture le goût « dansant », pas un genre figé.
| Stratégie | Atout | Limite |
|---|---|---|
| Weighted | Lisse, robuste | α à régler |
| Switching | Idéal en cold-start | Seuil à choisir |
import numpy as np class HybridRecommender: """Combine un modèle CF et un modèle Content-Based.""" def __init__(self, cf, cb, alpha=0.6, min_ratings=5): self.cf, self.cb = cf, cb self.alpha, self.min_ratings = alpha, min_ratings def weighted(self, u, i): return self.alpha * self.cf.predict(u, i) + (1 - self.alpha) * self.cb.predict(u, i) def switching(self, u, i): n = self.cf.n_ratings(u) return self.cf.predict(u, i) if n >= self.min_ratings else self.cb.predict(u, i) def eval_cold_start(hybrid, R, n_held=3, seed=99): """Masque des notes pour simuler des utilisateurs peu actifs.""" np.random.seed(seed) ew, es = [], [] for u in R.index: notes = R.loc[u].dropna() if len(notes) < n_held + 2: continue hidden = notes.sample(len(notes) - n_held).index for i in hidden: t = R.loc[u, i] ew.append((t - hybrid.weighted(u, i))**2) es.append((t - hybrid.switching(u, i))**2) return np.sqrt(np.mean(ew)), np.sqrt(np.mean(es)) # Balayage de α pour trouver le meilleur compromis for a in [0.0, 0.2, 0.4, 0.6, 0.8, 1.0]: h = HybridRecommender(cf, cb, alpha=a) rw, rs = eval_cold_start(h, R_full) print(f"α={a:.1f} | Weighted RMSE={rw:.3f} | Switching RMSE={rs:.3f}")
α=0.0 | Weighted RMSE=1.118 | Switching RMSE=0.987 α=0.2 | Weighted RMSE=1.052 | Switching RMSE=0.987 α=0.4 | Weighted RMSE=0.971 | Switching RMSE=0.987 α=0.6 | Weighted RMSE=1.024 | Switching RMSE=0.987 α=1.0 | Weighted RMSE=1.207 | Switching RMSE=0.987 ✅ Meilleur α = 0.4 (RMSE 0.971) → En cold-start, le Switching est souvent le plus stable car il évite de mélanger un CF peu fiable au début.
Factorisation linéaire r̂ = μ + bᵤ + bᵢ + pᵤ·qᵢ. Rapide, robuste, excellent sur données denses.
Remplace le produit scalaire par un MLP qui apprend des interactions non-linéaires. Gourmand en données.
Le deep learning ne bat pas toujours SVD : sur petits jeux de données, SVD reste souvent meilleur ET bien plus rapide.
import torch, torch.nn as nn, numpy as np from torch.utils.data import DataLoader, TensorDataset class NCF(nn.Module): """Neural Collaborative Filtering : GMF + MLP.""" def __init__(self, n_users, n_items, emb=32): super().__init__() self.u = nn.Embedding(n_users, emb) self.i = nn.Embedding(n_items, emb) self.mlp = nn.Sequential( nn.Linear(2*emb, 64), nn.ReLU(), nn.Dropout(0.2), nn.Linear(64, 16), nn.ReLU(), nn.Linear(16, 1)) def forward(self, u, i): x = torch.cat([self.u(u), self.i(i)], dim=-1) return torch.sigmoid(self.mlp(x)).squeeze() * 4 + 1 # note ∈ [1,5] def train_ncf(u, i, r, n_users, n_items, epochs=15): model = NCF(n_users, n_items) opt = torch.optim.Adam(model.parameters(), lr=2e-3) loss_fn = nn.MSELoss() dl = DataLoader(TensorDataset(torch.LongTensor(u), torch.LongTensor(i), torch.FloatTensor(r)), batch_size=256, shuffle=True) for _ in range(epochs): for bu, bi, br in dl: opt.zero_grad() loss = loss_fn(model(bu, bi), br) loss.backward(); opt.step() return model # ── SVD via Surprise (comparaison) ──────────────────────────────── from surprise import SVD, Dataset, Reader, accuracy from surprise.model_selection import train_test_split # ... chargement des données, split 80/20 ... svd = SVD(n_factors=50, n_epochs=20, random_state=42) svd.fit(trainset) rmse_svd = accuracy.rmse(svd.test(testset), verbose=False) ncf = train_ncf(u_tr, i_tr, r_tr, N_USERS, N_ITEMS) with torch.no_grad(): pred = ncf(torch.LongTensor(u_te), torch.LongTensor(i_te)).numpy() rmse_ncf = np.sqrt(np.mean((pred - r_te)**2)) print(f"SVD RMSE = {rmse_svd:.4f} | NCF RMSE = {rmse_ncf:.4f}")
📊 BENCHMARK SVD vs NCF ================================================= Modèle RMSE MAE Temps (s) SVD 1.0102 0.8056 0.3 NCF 1.0247 0.8123 8.7 🔍 ANALYSE • Sur un dataset dense et modeste, SVD reste très compétitif et ~25× plus rapide. • NCF ne prend l'avantage qu'avec beaucoup d'interactions et des features riches. → Le bon modèle dépend du volume de données, pas de la mode du moment.
Un système hybride complet, de la génération de candidats au serving : retrieval à deux tours, ranking fin, re-ranking diversité et évaluation offline rigoureuse.
Concevoir un moteur de recommandation production-ready capable de servir des recommandations personnalisées parmi un catalogue de 100 000+ films en quelques millisecondes. Le système combine un étage de retrieval (deep two-tower) pour générer rapidement des candidats, un étage de ranking pour affiner, puis un re-ranking pour la diversité et la nouveauté.
Interactions implicites (clics, visionnages) + features users/items. Split temporel.
Maximiser Recall@10 et NDCG@10 tout en gardant couverture et nouveauté élevées.
PyTorch (two-tower), FAISS (ANN), NumPy. Serving < 50 ms par requête.
Modèle entraîné, index ANN, fonction de reco, rapport d'évaluation offline.
Construire les paires positives (user, item) à partir des interactions ; séparer train/val/test par date pour éviter toute fuite du futur.
Deux encodeurs indépendants (user / item) projettent dans un espace commun. Entraînement par négatifs in-batch (sampled softmax).
Encoder tout le catalogue une fois, indexer les vecteurs items dans FAISS pour une recherche approchée en quelques millisecondes.
Re-scorer les ~200 candidats avec un modèle plus riche (NCF / GBDT) intégrant des features contextuelles.
Équilibrer pertinence et diversité pour éviter les listes redondantes et améliorer la couverture du catalogue.
Recall@K, NDCG@K, couverture catalogue et nouveauté — sur le hold-out temporel.
Requête utilisateur (uid + contexte)
│
▼
┌────────────────────────┐
│ USER TOWER (deep) │ → vecteur u (d=64, L2-normalisé)
└────────────────────────┘
│
▼ recherche ANN (FAISS, produit scalaire)
╔════════════════════════╗ Index items pré-calculé
║ RETRIEVAL ~200 cand. ║ ◄── item_vecs = ITEM TOWER(catalogue)
╚════════════════════════╝
│
▼
┌────────────────────────┐
│ RANKING (NCF / GBDT) │ → score fin + features contextuelles
└────────────────────────┘
│
▼
┌────────────────────────┐
│ RE-RANKING (MMR) │ → Top-10 pertinent ET diversifié
└────────────────────────┘
│
▼
Recommandations
| Critère | Retrieval (Two-Tower) | Ranking (NCF/GBDT) |
|---|---|---|
| Rôle | Réduire 100k → 200 candidats | Ordonner 200 → 10 |
| Vitesse requise | Très rapide (ANN) | Modérée (200 scores) |
| Interaction user-item | Tardive (produit scalaire) | Précoce (croisée) |
| Features contextuelles | Limitées | Riches |
| Pré-calcul possible | Oui (index items) | Non (par requête) |
Scorer 100 000 items par un modèle riche à chaque requête est impossible en temps réel. Le two-tower permet de pré-calculer les vecteurs items et de ne faire qu'une recherche ANN ultra-rapide ; le ranking, lui, ne s'applique qu'à une petite liste de candidats.
import torch import torch.nn as nn import torch.nn.functional as F class Tower(nn.Module): """Encodeur : embedding ID + features → vecteur latent L2-normalisé.""" def __init__(self, n_ids, n_feats, dim=64, hidden=128): super().__init__() self.emb = nn.Embedding(n_ids, dim) self.mlp = nn.Sequential( nn.Linear(dim + n_feats, hidden), nn.ReLU(), nn.Dropout(0.2), nn.Linear(hidden, dim) ) def forward(self, ids, feats): x = torch.cat([self.emb(ids), feats], dim=-1) return F.normalize(self.mlp(x), dim=-1) # cosinus = produit scalaire class TwoTower(nn.Module): """Retrieval à deux tours indépendantes (user / item).""" def __init__(self, n_users, n_items, nu_feats, ni_feats, dim=64): super().__init__() self.user_tower = Tower(n_users, nu_feats, dim) self.item_tower = Tower(n_items, ni_feats, dim) def forward(self, uid, ufeat, iid, ifeat): u = self.user_tower(uid, ufeat) # (B, d) v = self.item_tower(iid, ifeat) # (B, d) return u, v
On n'a que des interactions positives. L'astuce : dans un batch de B paires (u, i⁺), chaque item des autres paires sert de négatif. Une seule passe fournit B positifs et B×(B−1) négatifs — gratuit et efficace.
def in_batch_softmax(u, v, temperature=0.05, logq=None): """ u, v : (B, d) L2-normalisés. Les positifs sont sur la diagonale. logq : correction de popularité (sampled softmax) — optionnel. """ logits = u @ v.t() / temperature # (B, B) if logq is not None: logits = logits - logq # correction logQ (anti-popularité) labels = torch.arange(u.size(0), device=u.device) return F.cross_entropy(logits, labels) # ── Boucle d'entraînement ───────────────────────────────────────── model = TwoTower(n_users, n_items, nu_feats, ni_feats, dim=64) opt = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-6) for epoch in range(20): model.train() for uid, ufeat, iid, ifeat in train_loader: # uniquement des paires positives u, v = model(uid, ufeat, iid, ifeat) loss = in_batch_softmax(u, v, temperature=0.05) opt.zero_grad(); loss.backward(); opt.step() print(f"epoch {epoch+1:2d} | loss={loss.item():.4f}")
Les vecteurs items ne dépendent pas de l'utilisateur : on les calcule une seule fois et on les indexe dans FAISS. À chaque requête, seul le vecteur user est calculé, puis une recherche approchée renvoie les meilleurs candidats en quelques millisecondes.
import faiss import numpy as np # 1. Indexer tout le catalogue (hors-ligne, une seule fois) item_vecs = model.item_tower(all_ids, all_feats).detach().cpu().numpy().astype('float32') index = faiss.IndexFlatIP(item_vecs.shape[1]) # produit scalaire = cosinus (vecteurs normés) index.add(item_vecs) def mmr(user_vec, cand_vecs, cand_ids, k=10, lam=0.7): """Maximal Marginal Relevance : équilibre pertinence et diversité.""" selected, selected_vecs = [], [] rel = cand_vecs @ user_vec # pertinence de chaque candidat while len(selected) < k and len(selected) < len(cand_ids): if not selected: j = int(np.argmax(rel)) else: div = cand_vecs @ np.array(selected_vecs).T # similarité aux déjà choisis score = lam * rel - (1 - lam) * div.max(axis=1) score[selected] = -1e9 # ne pas re-sélectionner j = int(np.argmax(score)) selected.append(j); selected_vecs.append(cand_vecs[j]) return [cand_ids[j] for j in selected] def recommend(user_vec, k_retrieval=200, k_final=10): user_vec = user_vec.astype('float32') scores, cand = index.search(user_vec[None, :], k_retrieval) # ANN rapide cand = cand[0] return mmr(user_vec, item_vecs[cand], list(cand), k=k_final, lam=0.7)
FAISS IndexFlatIP est exact mais O(n). Pour 100k+ items, passer à IndexIVFFlat ou IndexHNSWFlat (approché) pour rester sous 50 ms.
import numpy as np def recall_at_k(recommended, relevant, k): if not relevant: return 0.0 return len(set(recommended[:k]) & set(relevant)) / len(relevant) def ndcg_at_k(recommended, relevant, k): dcg = sum(1 / np.log2(i + 2) for i, it in enumerate(recommended[:k]) if it in relevant) idcg = sum(1 / np.log2(i + 2) for i in range(min(len(relevant), k))) return dcg / idcg if idcg > 0 else 0.0 def catalog_coverage(all_recos, n_items): """Part du catalogue effectivement recommandée (anti-popularité).""" recommended_items = set(i for recos in all_recos for i in recos) return len(recommended_items) / n_items def novelty(all_recos, item_pop): """Nouveauté moyenne = -log2(popularité). Élevé = items de niche.""" eps = 1e-9 vals = [-np.log2(item_pop[i] + eps) for recos in all_recos for i in recos] return float(np.mean(vals))
📊 ÉVALUATION OFFLINE ========================================================== Modèle Recall@10 NDCG@10 Couvert. Nouveauté Populaire (baseline) 0.082 0.071 0.04 3.1 Two-Tower (retrieval) 0.193 0.164 0.38 6.8 + Ranking NCF 0.241 0.221 0.33 6.1 + Re-ranking MMR 0.236 0.214 0.52 7.9 → Le MMR sacrifie ~2% de NDCG pour +57% de couverture catalogue.
Un modèle entraîné sur les clics recommande surtout les items déjà populaires, qui reçoivent encore plus de clics… La correction logQ (sampled softmax) et le suivi de la couverture limitent cet emballement.
Le moindre écart de features entre l'entraînement et le serving dégrade silencieusement les performances. Partager le même code de featurisation des deux côtés.
Le two-tower gère mieux le cold-start que le CF pur : un nouvel item sans interaction est quand même encodable via ses features de contenu (genre, synopsis, acteurs).
Un meilleur NDCG offline ne garantit pas plus d'engagement réel. Valider en A/B test ; l'offline ne sert qu'à présélectionner les candidats à tester.
Un split aléatoire fait fuiter le futur dans le passé et gonfle les scores. Toujours évaluer sur des interactions postérieures à la période d'entraînement.