# Recommender Systems Building > Use when building recommendation engine architectures. - Skill: `loopyluci/recommender-systems-building` (Agent Skill) - Install (CLI): `npx skillmds@latest add loopyluci/recommender-systems-building` - Raw SKILL.md: https://api.skillmd.com/api/skills/loopyluci/recommender-systems-building/raw - Safety review: pending (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: LoopyLuci (https://skillmd.com/u/loopyluci) - Updated: 2026-09-08 - Page: https://skillmd.com/skills/loopyluci/recommender-systems-building --- # 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 ```python 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) ```python 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) ```python 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) ```python 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 ```python 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 1. **Cold start** — new users/items with no history; use content-based features as fallback 2. **Popularity bias** — model recommends popular items that everyone already knows about; use debiasing 3. **Filter bubble** — narrowing recommendations too much; add exploration via bandits 4. **Real-time serving latency** — two-tower retrieval + ANN search is fast; avoid full-ranking every candidate 5. **Evaluation offline ≠ online** — offline metrics don't always predict online A/B test results 6. **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