LoopyLuci Building Recommendation Systems
Designing and implementing recommendation engines — from collaborative filtering through matrix factorization to neural recommenders with candidate generation and ranking.
When to Use
- Building product, content, or media recommendations
- Implementing personalized user experiences
- Designing two-stage (retrieval + ranking) recommendation pipelines
- Cold-start scenarios where user/item have no history
- Building real-time recommendation serving systems
System Architecture
Users → Candidate Generation → Ranking → Re-ranking → Recommendations
↓ ↓
Multiple Sources Deep Model
Collaborative Filtering
User-User and Item-Item
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class CollaborativeFiltering:
"""User-based and item-based collaborative filtering."""
def fit(self, user_item_matrix):
"""
user_item_matrix: shape (n_users, n_items), sparse
"""
self.user_item = user_item_matrix
self.user_similarity = cosine_similarity(user_item_matrix)
self.item_similarity = cosine_similarity(user_item_matrix.T)
def predict_user_based(self, user_id, item_id, k=20):
"""Predict rating using k most similar users who rated the item."""
# Users who rated this item
users_who_rated = np.where(self.user_item[:, item_id] > 0)[0]
if len(users_who_rated) == 0:
return self.user_item[user_id].mean()
# Similarities between target user and those who rated
sims = self.user_similarity[user_id, users_who_rated]
# Top-k most similar
top_k = np.argsort(sims)[-k:]
top_k_users = users_who_rated[top_k]
top_k_sims = sims[top_k]
# Weighted average
ratings = self.user_item[top_k_users, item_id]
if top_k_sims.sum() == 0:
return ratings.mean()
return np.dot(ratings, top_k_sims) / top_k_sims.sum()
Matrix Factorization (SVD)
class SVDRecommender:
"""Matrix factorization via SVD or ALS."""
def __init__(self, n_factors=100, n_epochs=20, lr=0.01, reg=0.02):
self.n_factors = n_factors
self.n_epochs = n_epochs
self.lr = lr
self.reg = reg
self.user_factors = None
self.item_factors = None
def fit(self, ratings):
"""
ratings: list of (user_id, item_id, rating) or DataFrame
"""
users = set(r[0] for r in ratings)
items = set(r[1] for r in ratings)
self.user_map = {u: i for i, u in enumerate(users)}
self.item_map = {it: i for i, it in enumerate(items)}
self.n_users = len(users)
self.n_items = len(items)
# Initialize factors
self.user_factors = np.random.normal(0, 0.1, (self.n_users, self.n_factors))
self.item_factors = np.random.normal(0, 0.1, (self.n_items, self.n_factors))
self.user_bias = np.zeros(self.n_users)
self.item_bias = np.zeros(self.n_items)
self.global_mean = np.mean([r[2] for r in ratings])
# SGD training
for epoch in range(self.n_epochs):
np.random.shuffle(ratings)
total_loss = 0
for user, item, rating in ratings:
u, i = self.user_map[user], self.item_map[item]
# Predict
pred = (self.global_mean + self.user_bias[u] + self.item_bias[i] +
np.dot(self.user_factors[u], self.item_factors[i]))
error = rating - pred
# Update
self.user_bias[u] += self.lr * (error - self.reg * self.user_bias[u])
self.item_bias[i] += self.lr * (error - self.reg * self.item_bias[i])
uf = self.user_factors[u].copy()
self.user_factors[u] += self.lr * (error * self.item_factors[i] - self.reg * self.user_factors[u])
self.item_factors[i] += self.lr * (error * uf - self.reg * self.item_factors[i])
total_loss += error ** 2
print(f"Epoch {epoch}: RMSE={np.sqrt(total_loss/len(ratings)):.4f}")
def predict(self, user, item):
u = self.user_map.get(user)
i = self.item_map.get(item)
if u is None or i is None:
return self.global_mean
return (self.global_mean + self.user_bias[u] + self.item_bias[i] +
np.dot(self.user_factors[u], self.item_factors[i]))
Neural Recommenders
Two-Tower Model (Retrieval)
import torch
import torch.nn as nn
class TwoTowerModel(nn.Module):
"""Two-tower neural network for candidate retrieval.
User tower + Item tower → dot product → relevance score."""
def __init__(self, num_users, num_items, n_factors=64):
super().__init__()
self.user_embedding = nn.Embedding(num_users, n_factors)
self.item_embedding = nn.Embedding(num_items, n_factors)
# Optional: add user/item features here
self.user_tower = nn.Sequential(
nn.Linear(n_factors, 128), nn.ReLU(),
nn.Linear(128, n_factors)
)
self.item_tower = nn.Sequential(
nn.Linear(n_factors, 128), nn.ReLU(),
nn.Linear(128, n_factors)
)
def forward(self, user_ids, item_ids):
user_emb = self.user_embedding(user_ids)
item_emb = self.item_embedding(item_ids)
user_vec = self.user_tower(user_emb)
item_vec = self.item_tower(item_emb)
# Normalize for cosine similarity
user_vec = nn.functional.normalize(user_vec, dim=1)
item_vec = nn.functional.normalize(item_vec, dim=1)
return (user_vec * item_vec).sum(dim=1) # Dot product
def get_all_item_vectors(self):
"""Pre-compute item vectors for fast ANN search."""
all_items = torch.arange(self.item_embedding.num_embeddings)
return self.item_tower(self.item_embedding(all_items)).detach()
Ranking Model (Deep Neural Network)
class RankingModel(nn.Module):
"""Deep neural ranking model with cross features."""
def __init__(self, n_users, n_items, n_categ_features, n_numerical_features):
super().__init__()
self.user_emb = nn.Embedding(n_users, 32)
self.item_emb = nn.Embedding(n_items, 32)
# Categorical feature embeddings
self.categ_embeddings = nn.ModuleList([
nn.Embedding(n, 16) for n in n_categ_features
])
total_features = 32 + 32 + len(n_categ_features) * 16 + n_numerical_features
self.deep_layers = nn.Sequential(
nn.Linear(total_features, 256), nn.ReLU(), nn.Dropout(0.2),
nn.Linear(256, 128), nn.ReLU(), nn.Dropout(0.1),
nn.Linear(128, 64), nn.ReLU(),
nn.Linear(64, 1)
)
def forward(self, user_ids, item_ids, categ_features, numerical_features):
u_emb = self.user_emb(user_ids)
i_emb = self.item_emb(item_ids)
c_embs = [emb(categ_features[:, i]) for i, emb in enumerate(self.categ_embeddings)]
features = torch.cat([u_emb, i_emb] + c_embs + [numerical_features], dim=1)
return self.deep_layers(features)
Two-Stage Pipeline
class RecommendationPipeline:
"""Candidate generation + ranking + re-ranking."""
def __init__(self, retriever, ranker):
self.retriever = retriever # TwoTowerModel
self.ranker = ranker # RankingModel
def recommend(self, user_id, n_candidates=500, n_final=10):
# Stage 1: Retrieve candidates
user_vector = self.retriever.user_tower(
self.retriever.user_embedding(torch.tensor([user_id]))
)
item_vectors = self.retriever.get_all_item_vectors()
# ANN search (simplified — use faiss in production)
scores = user_vector @ item_vectors.T
top_candidates = scores.topk(n_candidates).indices[0]
# Stage 2: Rank candidates
with torch.no_grad():
ranking_scores = self.ranker(
torch.full((n_candidates,), user_id),
top_candidates,
self._get_features(user_id, top_candidates),
self._get_numerical(user_id, top_candidates),
)
# Stage 3: Re-rank (diversity, business rules)
final_items = self._diversity_rerank(top_candidates, ranking_scores, n_final)
return final_items
Common Pitfalls
- Cold start — new users/items with no history; use content-based features as fallback
- Popularity bias — model recommends popular items that everyone already knows about; use debiasing
- Filter bubble — narrowing recommendations too much; add exploration via bandits
- Real-time serving latency — two-tower retrieval + ANN search is fast; avoid full-ranking every candidate
- Evaluation offline ≠ online — offline metrics don't always predict online A/B test results
- Feedback loop — recommending based on past recommendations can amplify bias; use random exploration
Verification Checklist
- Baseline model (popularity) established for comparison
- Matrix factorization beats collaborative filtering baseline
- Neural model beats matrix factorization on held-out data
- Two-stage pipeline (retrieval + ranking) tested for latency
- Cold-start strategy implemented for new users/items
- Diversity metric tracked alongside accuracy
- Online A/B test shows improvement over baseline
See Also
- embedding-models-patterns — training embeddings for retrieval
- nlp-techniques — content-based features for cold start
- data-augmentation-techniques — augmenting sparse interaction data
- ml-pipeline-design — serving pipeline architecture