Two-Stage Fashion Recommender
Build a production retrieval-and-ranking system on real H&M fashion data.
Introduction
30 Second Summary
When you open Instagram Explore or see a "you might also like" suggestion on an online store, a recommendation system is working behind the scenes to match you with items you are most likely to want. These systems process millions of products and users in real time, making them one of the most impactful applications of machine learning in the real world.
In this project, you will build a production-style two-stage fashion recommendation engine using real data from H&M. You will train a neural retrieval model, enhance it with LLM-generated commonsense knowledge inspired by Amazon's COSMO system, and deploy an interactive demo to the web.
What You'll Build
A two-stage fashion recommender that uses a two-tower neural network for fast candidate retrieval, Phi-4-mini for commonsense knowledge generation, LightGBM for precision ranking, and Gradio for an interactive demo deployed to Hugging Face Spaces.
By the end of this project, you'll have:
- A two-tower retrieval model with multi-modal embeddings (ResNet-50 visual + Sentence-BERT text) and DropoutNet cold-start handling for new users and new items.
- An LLM-powered commonsense knowledge layer inspired by Amazon's COSMO system that generates intent triples explaining why people buy fashion items.
- A LightGBM ranking model with Multi-Objective Ranking Score (MORS) balancing style relevance and eco-friendliness, enhanced with COSMO intent features and validated with MAP@12, Recall@K, and NDCG metrics.
- An interactive Gradio demo deployed to Hugging Face Spaces with a recruiter-ready metrics dashboard, COSMO intent explanations, and a MORS alpha slider for real-time style-vs-sustainability tuning.
- Secret Mission: Add temporal decay weighting and A/B test simulation to make the project research-paper-ready.
Are there any prerequisites?
You will need a Kaggle account to download the H&M Personalized Fashion Recommendations dataset, and a Hugging Face account for deploying your demo. You should also have Python installed and be comfortable running terminal commands.
This is a hard project that covers deep learning, information retrieval, and LLM integration. Familiarity with PyTorch basics and neural network concepts will help you get the most out of it.
Set Up Your Development Environment
A two-stage recommendation system has a lot of moving parts. You need PyTorch for deep learning, FAISS for fast similarity search, a LightGBM ranker, an LLM for knowledge generation, and a Gradio UI for deployment.
Before any of that can work, you need a clean, isolated environment with the right library versions. In this step, you will create a Python virtual environment, install all dependencies, organize your project folder, verify the H&M dataset, and write a central configuration file that controls the entire pipeline.
In this step, get ready to:
- Create a Python virtual environment and install all project dependencies.
- Organize the project directory and verify the H&M dataset files.
- Create a configuration file that controls the entire pipeline.
Create your Python virtual environment and install dependencies
A virtual environment keeps this project's libraries separate from your system Python. This prevents version conflicts with other projects on your machine.
- Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type Terminal and press Enter to open it.
- Move to your Desktop by running this command:
cd ~/Desktop- Create your project folder and move into it by running these commands:
mkdir hm_recommender
cd hm_recommender- Confirm the folder exists by running:
ls ~/Desktop | grep hm_recommenderYou should see hm_recommender in the output.
- Create a Python virtual environment by running this command:
python3 -m venv venv- Activate the virtual environment by running:
source venv/bin/activateYour terminal prompt should now start with (venv). This confirms you are working inside the virtual environment.
Don't see (venv) in your prompt?
Make sure you ran the source command (not just venv/bin/activate by itself). If you are on Windows, use venv\Scripts\activate instead.
help me activate my Python virtual environment
This project depends on over a dozen libraries. You will define them all in a requirements.txt file so they can be installed in one go.
- Create a file called requirements.txt inside your hm_recommender folder and paste the following contents:
torch>=2.11.0
torchvision>=0.26.0
torchaudio
transformers>=4.40.0
sentence-transformers>=5.4.0
faiss-cpu>=1.13.0
lightgbm>=4.6.0
gradio>=6.13.0
pandas>=2.0.0
numpy>=1.24.0
scikit-learn>=1.3.0
matplotlib>=3.7.0
tqdm>=4.65.0
Pillow>=10.0.0What are all these libraries for?
- torch, torchvision, torchaudio power the two-tower neural network and visual feature extraction with ResNet-50.
- transformers loads Phi-4-mini for COSMO-inspired commonsense knowledge generation.
- sentence-transformers provides Sentence-BERT for encoding text into dense embeddings.
- faiss-cpu enables approximate nearest neighbor search for fast candidate retrieval.
- lightgbm trains the ranking model that re-scores retrieved candidates.
- gradio builds the interactive UI that deploys to Hugging Face Spaces.
- Save requirements.txt.
- Install all the dependencies by running:
pip install -r requirements.txtThis takes a few minutes. You should see a stream of download and installation messages ending with Successfully installed followed by a list of packages.
Seeing installation errors?
Make sure your virtual environment is active. You should see (venv) in your terminal prompt. If not, run source venv/bin/activate again.
If you see errors about faiss-cpu or lightgbm, make sure you are running Python 3.10 or higher. Check with python3 --version.
help me fix my pip install errors
Your Mac M4 has Apple's MPS (Metal Performance Shaders) GPU backend. This accelerates PyTorch training significantly compared to CPU-only training.
- Verify MPS is available by running:
python3 -c "import torch; print(f'MPS available: {torch.backends.mps.is_available()}'); print(f'PyTorch version: {torch.__version__}')"You should see MPS available: True and a PyTorch version of 2.11.0 or higher.
MPS available: False?
On Apple Silicon Macs (M1 through M4), MPS should be available automatically with PyTorch 2.11.0+. If it shows False, make sure you installed the standard PyTorch package (no special index URL is needed for MPS support on Mac).
The project will fall back to CPU if MPS is not available, so you can continue either way.
help me enable MPS on my Mac
✔️ Awesome, I've got everything!
Great. Make sure you have saved requirements.txt and let's keep going.
ⓧ I'd like to double check the full code
Here is the complete requirements.txt for comparison.
torch>=2.11.0
torchvision>=0.26.0
torchaudio
transformers>=4.40.0
sentence-transformers>=5.4.0
faiss-cpu>=1.13.0
lightgbm>=4.6.0
gradio>=6.13.0
pandas>=2.0.0
numpy>=1.24.0
scikit-learn>=1.3.0
matplotlib>=3.7.0
tqdm>=4.65.0
Pillow>=10.0.0Organize the project directory and verify the H&M dataset
Your project needs a clear folder structure to keep embeddings, trained models, knowledge triples, and raw data organized. You also need to confirm the H&M dataset files are in place before building the pipeline.
- Create all the project subdirectories by running this command:
mkdir -p data models embeddings artifacts knowledge app- Confirm the directories exist by running:
lsYou should see app, artifacts, data, embeddings, knowledge, models, requirements.txt, and venv listed in the output.
What is each folder for?
- data/ holds the H&M CSV files and optional product images.
- embeddings/ stores pre-computed visual, text, intent, and trained embeddings.
- models/ stores trained model checkpoints (two-tower and LightGBM).
- artifacts/ stores the FAISS index and evaluation results.
- knowledge/ stores the COSMO-generated commonsense triples.
- app/ holds the Gradio UI code for deployment.
Now you need to place the H&M dataset files into the data/ folder. These are the files you downloaded from Kaggle.
- Copy the three H&M CSV files into your data/ directory by running these commands (adjust the source path to match where you downloaded the dataset):
cp [[HM_DATASET_PATH="path/to/your/h-and-m-personalized-fashion-recommendations"]]/articles.csv data/
cp [[HM_DATASET_PATH="path/to/your/h-and-m-personalized-fashion-recommendations"]]/customers.csv data/
cp [[HM_DATASET_PATH="path/to/your/h-and-m-personalized-fashion-recommendations"]]/transactions_train.csv data/Have the images directory too?
If you downloaded the H&M images from Kaggle, copy the entire images/ folder into data/ as well. The project will use real product images for visual embeddings if available. If not, it falls back to placeholder embeddings.
- Verify the CSV files are in place and check their row counts by running:
wc -l data/articles.csv data/customers.csv data/transactions_train.csvYou should see approximately 105,542 lines for articles.csv, 1,371,980 for customers.csv, and 31,788,324 for transactions_train.csv. These counts include the header row.
Seeing "No such file or directory"?
Make sure your terminal is inside the hm_recommender folder. Run pwd to check. You should see a path ending in hm_recommender.
Double check that the source path in your cp commands points to where Kaggle actually saved the dataset files on your machine.
help me find and copy my H&M dataset files
Create the configuration file with all project constants
Every module in this project reads from a single config.py file. Centralizing constants like embedding dimensions, learning rates, and file paths here means you only need to change one file to tune the entire pipeline.
- Create a file called config.py in the root of your hm_recommender folder.
- Start by adding the device detection and file path configuration:
import torch
import os
# Device configuration
if torch.backends.mps.is_available():
DEVICE = torch.device("mps")
elif torch.cuda.is_available():
DEVICE = torch.device("cuda")
else:
DEVICE = torch.device("cpu")
print(f"Using device: {DEVICE}")
# Paths
DATA_DIR = "data"
ARTICLES_PATH = os.path.join(DATA_DIR, "articles.csv")
CUSTOMERS_PATH = os.path.join(DATA_DIR, "customers.csv")
TRANSACTIONS_PATH = os.path.join(DATA_DIR, "transactions_train.csv")
IMAGES_DIR = os.path.join(DATA_DIR, "images")
EMBEDDINGS_DIR = "embeddings"
MODELS_DIR = "models"
ARTIFACTS_DIR = "artifacts"
KNOWLEDGE_DIR = "knowledge"
# Data sampling
NUM_WEEKS = 5
MIN_USER_TRANSACTIONS = 3
TRAIN_WEEKS = 3
VAL_WEEK = 4
TEST_WEEK = 5What does this code do?
- The device detection block picks the best available hardware accelerator. On your M4 Mac, it selects MPS. On a cloud GPU it would select CUDA. On anything else it falls back to CPU.
- The path constants ensure every module reads data from the same locations. All paths are relative to the project root.
- The data sampling constants control how much of the H&M dataset you use. You will sample the last 5 weeks and split them into 3 weeks for training, 1 for validation, and 1 for testing.
- Below the data sampling section, add the model architecture and training hyperparameters:
# Model architecture
EMBEDDING_DIM = 64
VISUAL_DIM = 2048
TEXT_DIM = 384
INTENT_DIM = 384 # COSMO intent embedding dimension (same as SBERT)
MULTIMODAL_DIM = VISUAL_DIM + TEXT_DIM # 2432
USER_FEATURE_DIM = 10 # demographics + behavioral features
HIDDEN_DIM = 256
DROPOUT_RATE = 0.3
DROPOUTNET_P = 0.5 # probability of dropping interaction history
# Training
BATCH_SIZE = 512
LEARNING_RATE = 1e-3
WEIGHT_DECAY = 1e-5
NUM_EPOCHS = 20
GRAD_CLIP = 1.0
# Retrieval
TOP_K_RETRIEVAL = 50
FAISS_INDEX_PATH = os.path.join(ARTIFACTS_DIR, "faiss_index.bin")
# Ranking
MORS_ALPHA = 0.8 # style vs eco tradeoff
ECO_KEYWORDS = ["organic", "recycled", "sustainable", "bci", "cotton", "linen", "tencel"]
# Evaluation
MAP_K = 12
RECALL_K_VALUES = [10, 50]
NDCG_K = 12
# Sentence-BERT model
SBERT_MODEL_NAME = "all-MiniLM-L6-v2"What do these hyperparameters control?
EMBEDDING_DIM = 64 is the size of the final user and item vectors. Both towers compress their inputs down to 64 dimensions so they can be compared with a dot product.
DROPOUTNET_P = 0.5 is the core DropoutNet cold-start mechanism. During training, interaction history is zeroed out 50% of the time. This forces the model to rely on content features alone, so it still works for brand-new users with no purchase history.
MORS_ALPHA = 0.8 controls the Multi-Objective Ranking Score tradeoff. A value of 0.8 means 80% weight on style relevance and 20% on eco-friendliness.
- Finally, add the LLM configuration and directory creation at the bottom of config.py:
# LLM Configuration (COSMO knowledge generation)
LLM_MODEL_NAME = "microsoft/Phi-4-mini-instruct"
LLM_MAX_NEW_TOKENS = 256
LLM_TEMPERATURE = 0.7
LLM_BATCH_SIZE = 1 # process one article at a time for quality
COSMO_TRIPLES_PATH = os.path.join(KNOWLEDGE_DIR, "cosmo_triples.json")
COSMO_RELATIONS = ["used_for_function", "used_for_event", "used_for_audience", "used_for_season"]
# Deployment
MAX_DISPLAY_ITEMS = 12
# Create directories
for d in [EMBEDDINGS_DIR, MODELS_DIR, ARTIFACTS_DIR, KNOWLEDGE_DIR]:
os.makedirs(d, exist_ok=True)What does this code do?
The LLM section configures Phi-4-mini (3.8 billion parameters) for generating COSMO-style commonsense knowledge triples. The four relation types capture why people buy fashion items: their function, the occasion, the target audience, and the season.
The os.makedirs loop at the bottom ensures the output directories exist every time any module imports config.py. This prevents "directory not found" errors later.
- Save config.py.
- Test the configuration by running:
python3 config.pyYou should see Using device: mps printed to your terminal. This confirms the config loaded successfully and detected your M4 GPU.
Seeing an import error or wrong device?
If you see ModuleNotFoundError: No module named 'torch', your virtual environment is not active. Run source venv/bin/activate first.
If you see Using device: cpu instead of mps, the project will still work. CPU training is slower but produces identical results.
help me fix my config.py errors
✔️ Awesome, I've got everything!
Great. Double check you have saved config.py and that it printed the correct device.
ⓧ I'd like to double check the full code
Here is the complete config.py for comparison.
import torch
import os
# Device configuration
if torch.backends.mps.is_available():
DEVICE = torch.device("mps")
elif torch.cuda.is_available():
DEVICE = torch.device("cuda")
else:
DEVICE = torch.device("cpu")
print(f"Using device: {DEVICE}")
# Paths
DATA_DIR = "data"
ARTICLES_PATH = os.path.join(DATA_DIR, "articles.csv")
CUSTOMERS_PATH = os.path.join(DATA_DIR, "customers.csv")
TRANSACTIONS_PATH = os.path.join(DATA_DIR, "transactions_train.csv")
IMAGES_DIR = os.path.join(DATA_DIR, "images")
EMBEDDINGS_DIR = "embeddings"
MODELS_DIR = "models"
ARTIFACTS_DIR = "artifacts"
KNOWLEDGE_DIR = "knowledge"
# Data sampling
NUM_WEEKS = 5
MIN_USER_TRANSACTIONS = 3
TRAIN_WEEKS = 3
VAL_WEEK = 4
TEST_WEEK = 5
# Model architecture
EMBEDDING_DIM = 64
VISUAL_DIM = 2048
TEXT_DIM = 384
INTENT_DIM = 384 # COSMO intent embedding dimension (same as SBERT)
MULTIMODAL_DIM = VISUAL_DIM + TEXT_DIM # 2432
USER_FEATURE_DIM = 10 # demographics + behavioral features
HIDDEN_DIM = 256
DROPOUT_RATE = 0.3
DROPOUTNET_P = 0.5 # probability of dropping interaction history
# Training
BATCH_SIZE = 512
LEARNING_RATE = 1e-3
WEIGHT_DECAY = 1e-5
NUM_EPOCHS = 20
GRAD_CLIP = 1.0
# Retrieval
TOP_K_RETRIEVAL = 50
FAISS_INDEX_PATH = os.path.join(ARTIFACTS_DIR, "faiss_index.bin")
# Ranking
MORS_ALPHA = 0.8 # style vs eco tradeoff
ECO_KEYWORDS = ["organic", "recycled", "sustainable", "bci", "cotton", "linen", "tencel"]
# Evaluation
MAP_K = 12
RECALL_K_VALUES = [10, 50]
NDCG_K = 12
# Sentence-BERT model
SBERT_MODEL_NAME = "all-MiniLM-L6-v2"
# LLM Configuration (COSMO knowledge generation)
LLM_MODEL_NAME = "microsoft/Phi-4-mini-instruct"
LLM_MAX_NEW_TOKENS = 256
LLM_TEMPERATURE = 0.7
LLM_BATCH_SIZE = 1 # process one article at a time for quality
COSMO_TRIPLES_PATH = os.path.join(KNOWLEDGE_DIR, "cosmo_triples.json")
COSMO_RELATIONS = ["used_for_function", "used_for_event", "used_for_audience", "used_for_season"]
# Deployment
MAX_DISPLAY_ITEMS = 12
# Create directories
for d in [EMBEDDINGS_DIR, MODELS_DIR, ARTIFACTS_DIR, KNOWLEDGE_DIR]:
os.makedirs(d, exist_ok=True)Your environment is ready. You have a clean virtual environment, all the libraries you need, the H&M dataset in place, and a central configuration file. Next up, you will build the data pipeline that loads, samples, and splits the transaction data for training.
Build the Data Pipeline
Your environment is set up and the H&M dataset is ready. Now you need to transform 28.8 million raw transactions into clean, structured data your recommendation model can learn from.
The full dataset is far too large for 16GB of RAM. You will sample the most recent 5 weeks of transactions, filter to active customers, split by time for train/validation/test, and engineer features for both users and items. You will also carve out dedicated cold-start holdout sets for new users and new items that the model has never seen during training.
In this step, get ready to:
- Load and intelligently sample the H&M transaction data.
- Create time-based train/validation/test splits with cold-start holdout sets.
- Engineer user and item features and build interaction history.
Load and sample the H&M data intelligently
The H&M dataset contains over 105,000 articles, 1.37 million customers, and 28.8 million transactions spanning two years. Loading everything would blow past your memory. Kaggle silver medalists addressed this by sampling only the most recent weeks of transactions.
- Create a new file called data_pipeline.py in the root of your hm_recommender folder.
- Add the imports and the load_raw_data() function by pasting this code:
import pandas as pd
import numpy as np
from datetime import timedelta
from collections import defaultdict
import config
def load_raw_data():
print("Loading raw data...")
articles = pd.read_csv(config.ARTICLES_PATH, dtype={"article_id": str})
customers = pd.read_csv(config.CUSTOMERS_PATH, dtype={"customer_id": str})
transactions = pd.read_csv(
config.TRANSACTIONS_PATH,
dtype={"customer_id": str, "article_id": str},
parse_dates=["t_dat"]
)
print(f" Articles: {len(articles):,}")
print(f" Customers: {len(customers):,}")
print(f" Transactions: {len(transactions):,}")
return articles, customers, transactionsWhat does this code do?
The imports bring in pandas for DataFrames, numpy for numerical operations, and your project's config module for all path and parameter constants.
The load_raw_data() function reads the three H&M CSVs. Passing dtype={"article_id": str} ensures IDs are treated as strings (they have leading zeros that would be stripped as integers). The parse_dates parameter converts the transaction date column to proper datetime objects for time-based operations.
- Save data_pipeline.py.
You will not run this file yet. The next functions need to be added first.
- Add the sample_recent_weeks() and filter_active_users() functions below load_raw_data() by pasting this code:
def sample_recent_weeks(transactions, num_weeks=5):
max_date = transactions["t_dat"].max()
start_date = max_date - timedelta(weeks=num_weeks)
sampled = transactions[transactions["t_dat"] >= start_date].copy()
print(f" Sampled {len(sampled):,} transactions from last {num_weeks} weeks")
return sampled
def filter_active_users(transactions, min_txn=3):
user_counts = transactions["customer_id"].value_counts()
active_users = user_counts[user_counts >= min_txn].index
filtered = transactions[transactions["customer_id"].isin(active_users)]
print(f" Active users (>={min_txn} txns): {len(active_users):,}")
return filteredWhat does this code do?
These two functions pare the dataset down to a manageable size. sample_recent_weeks() finds the latest transaction date, subtracts 5 weeks, and keeps only transactions in that window. This captures recent trends while staying within your 16GB RAM.
filter_active_users() keeps only customers who made at least 3 purchases in the window. Users with too few transactions don't give the model enough signal to learn meaningful preferences.
- Save data_pipeline.py.
You won't see a change yet. The next substep adds the splitting logic that makes these functions usable.
Create train/validation/test splits with cold-start holdout sets
Recommendation systems must be evaluated on future data, not random samples. You will split by time so the model trains on older weeks and gets tested on the newest week.
Why time-based splits instead of random splits?
Random splits leak future information into training. A customer who bought a winter coat in week 5 could appear in training, teaching the model about a trend it should be predicting, not memorizing. Time-based splits mimic how a real production recommender operates: it trains on past data and predicts the future.
The cold-start holdout sets go further. "New items" are articles that appear only in the test week. "New users" are customers who appear only in the test week. These let you measure how well your model handles the real-world scenario where brand-new products or first-time shoppers arrive with zero history.
- Add the assign_week_numbers() and create_splits() functions below filter_active_users() by pasting this code:
def assign_week_numbers(transactions):
max_date = transactions["t_dat"].max()
transactions = transactions.copy()
# Calculate how many weeks ago each transaction occurred
transactions["week"] = ((max_date - transactions["t_dat"]).dt.days // 7) + 1
# Flip so week 1 is oldest, week 5 is newest
transactions["week"] = config.NUM_WEEKS + 1 - transactions["week"]
return transactions
def create_splits(transactions):
# Split by week number: weeks 1-3 train, week 4 val, week 5 test
train = transactions[transactions["week"] <= config.TRAIN_WEEKS]
val = transactions[transactions["week"] == config.VAL_WEEK]
test = transactions[transactions["week"] == config.TEST_WEEK]
# Identify items and users seen during training
train_items = set(train["article_id"].unique())
train_users = set(train["customer_id"].unique())
# Cold-start holdouts: items/users that only appear in test
test_new_items = test[~test["article_id"].isin(train_items)]
test_new_users = test[~test["customer_id"].isin(train_users)]
print(f" Train: {len(train):,} | Val: {len(val):,} | Test: {len(test):,}")
print(f" Cold-start new items in test: {test_new_items['article_id'].nunique():,}")
print(f" Cold-start new users in test: {test_new_users['customer_id'].nunique():,}")
return train, val, test, test_new_items, test_new_usersWhat does this code do?
assign_week_numbers() labels each transaction with a week number from 1 (oldest) to 5 (newest). The math first calculates how many weeks ago a transaction occurred, then flips the number so newer weeks have higher numbers.
create_splits() uses those week numbers to carve out three sets: training (weeks 1-3), validation (week 4), and test (week 5). It then identifies articles and customers that appear only in the test set. These cold-start holdouts let you measure performance on completely unseen entities later.
- Save data_pipeline.py.
You won't see a change yet. The next substep adds the feature engineering functions and the orchestrator that ties everything together.
Seeing a syntax error?
Make sure each function is at the top level of the file (no extra indentation). Check that create_splits appears after assign_week_numbers since it depends on the week column that function creates.
Help me debug a syntax error in data_pipeline.py
Engineer user and item features
Raw IDs and dates are not enough for a neural network. You need numerical features that describe each user's behavior and each item's attributes. The user features combine demographics (age, club status) with computed behavioral signals (purchase frequency, price sensitivity).
- Add the engineer_user_features() function below create_splits() by pasting this code:
def engineer_user_features(customers, transactions):
# Compute behavioral features from transaction history
user_stats = transactions.groupby("customer_id").agg(
txn_count=("t_dat", "count"),
avg_price=("price", "mean"),
std_price=("price", "std"),
unique_articles=("article_id", "nunique"),
days_active=("t_dat", lambda x: (x.max() - x.min()).days),
).reset_index()
user_stats["std_price"] = user_stats["std_price"].fillna(0)
# Extract demographic features and encode categoricals
users = customers[["customer_id", "age", "club_member_status", "fashion_news_frequency"]].copy()
users["age"] = users["age"].fillna(users["age"].median())
users["club_member_status"] = users["club_member_status"].map(
{"ACTIVE": 1, "PRE-CREATE": 0.5, "LEFT CLUB": 0}
).fillna(0)
users["fashion_news_frequency"] = users["fashion_news_frequency"].map(
{"Regularly": 1, "Monthly": 0.5, "NONE": 0}
).fillna(0)
# Merge demographics with behavioral stats
users = users.merge(user_stats, on="customer_id", how="left")
users = users.fillna(0)
feature_cols = ["age", "club_member_status", "fashion_news_frequency",
"txn_count", "avg_price", "std_price", "unique_articles", "days_active"]
# Normalize all features to [0, 1] range
for col in feature_cols:
col_max = users[col].max()
if col_max > 0:
users[col] = users[col] / col_max
return users, feature_colsWhat does this code do?
This function builds an 8-feature profile for each customer. The behavioral features (transaction count, average price, price variance, unique articles purchased, and days active) are aggregated from the training transactions. The demographic features (age, club status, news frequency) come from the customers table.
Categorical columns like club_member_status are mapped to ordinal numbers. All features are then normalized to the [0, 1] range so no single feature dominates the neural network's learning.
- Save data_pipeline.py.
You won't see a change yet. The item features and orchestrator function come next.
- Add the engineer_item_features() and build_user_interaction_history() functions below engineer_user_features() by pasting this code:
def engineer_item_features(articles):
# Select the metadata columns used for text embeddings and ranking
items = articles[["article_id", "product_type_name", "colour_group_name",
"department_name", "section_name", "garment_group_name",
"prod_name", "detail_desc"]].copy()
items["detail_desc"] = items["detail_desc"].fillna("")
items["prod_name"] = items["prod_name"].fillna("")
return items
def build_user_interaction_history(transactions, max_history=20):
# Build a dict mapping each user to their most recent purchased article IDs
history = defaultdict(list)
sorted_txn = transactions.sort_values("t_dat", ascending=False)
for _, row in sorted_txn.iterrows():
uid = row["customer_id"]
if len(history[uid]) < max_history:
history[uid].append(row["article_id"])
return dict(history)What does this code do?
engineer_item_features() extracts the metadata columns you will use later for text embeddings and ranking features. It fills missing descriptions with empty strings to avoid errors during text encoding.
build_user_interaction_history() creates a dictionary mapping each customer to their 20 most recent purchases. This history is what the User Tower will mean-pool into a single vector. Capping at 20 keeps memory bounded while still capturing recent preferences.
- Save data_pipeline.py.
You won't see a change yet. The final piece is the orchestrator function that calls everything in order.
- Add the prepare_data() function and the __main__ block at the bottom of data_pipeline.py by pasting this code:
def prepare_data():
articles, customers, transactions = load_raw_data()
print("\nSampling and filtering...")
transactions = sample_recent_weeks(transactions, config.NUM_WEEKS)
transactions = filter_active_users(transactions, config.MIN_USER_TRANSACTIONS)
transactions = assign_week_numbers(transactions)
# Keep only articles and customers that appear in the filtered transactions
active_article_ids = set(transactions["article_id"].unique())
active_customer_ids = set(transactions["customer_id"].unique())
articles = articles[articles["article_id"].isin(active_article_ids)]
customers = customers[customers["customer_id"].isin(active_customer_ids)]
print("\nCreating splits...")
train, val, test, test_new_items, test_new_users = create_splits(transactions)
print("\nEngineering features...")
users_df, user_feature_cols = engineer_user_features(customers, train)
items_df = engineer_item_features(articles)
print("\nBuilding interaction history...")
user_history = build_user_interaction_history(train)
return {
"articles": articles,
"customers": customers,
"train": train,
"val": val,
"test": test,
"test_new_items": test_new_items,
"test_new_users": test_new_users,
"users_df": users_df,
"user_feature_cols": user_feature_cols,
"items_df": items_df,
"user_history": user_history,
}
if __name__ == "__main__":
data = prepare_data()
print("\nData pipeline complete!")
print(f" Users: {len(data['users_df']):,}")
print(f" Items: {len(data['items_df']):,}")What does this code do?
prepare_data() is the orchestrator. It calls every function in the right order: load, sample, filter, split, engineer features, and build history. It also filters the articles and customers DataFrames to match only the IDs that survived the transaction sampling.
The function returns a single dictionary with all the data objects your later steps need. Every other module in this project will call prepare_data() to get a consistent, ready-to-use dataset.
- Save data_pipeline.py.
- Run the pipeline by executing this command in your terminal:
python data_pipeline.pyYou should see printed output showing the number of articles, customers, and transactions loaded. Then the sampling, filtering, splitting, and feature engineering steps print their progress and counts.
Seeing a FileNotFoundError?
Make sure your H&M CSV files (articles.csv, customers.csv, transactions_train.csv) are inside the data/ folder within your hm_recommender directory. Also make sure you are running the command from inside the hm_recommender folder, not from a parent directory.
Help me fix a FileNotFoundError when running data_pipeline.py for my H&M recommender project
Running out of memory?
If your machine freezes or the script is killed, the full transactions file may be too large. Verify that NUM_WEEKS in config.py is set to 5. Reducing it to 3 will shrink the data further if needed.
Help me fix a memory issue when loading the H&M dataset in my recommender project
✔️ Awesome, I've got everything!
Double check you have saved data_pipeline.py and that your terminal output shows counts for train, val, test, and cold-start sets.
ⓧ I'd like to double check the full code
Here is the complete data_pipeline.py file. Compare it against your version.
import pandas as pd
import numpy as np
from datetime import timedelta
from collections import defaultdict
import config
def load_raw_data():
print("Loading raw data...")
articles = pd.read_csv(config.ARTICLES_PATH, dtype={"article_id": str})
customers = pd.read_csv(config.CUSTOMERS_PATH, dtype={"customer_id": str})
transactions = pd.read_csv(
config.TRANSACTIONS_PATH,
dtype={"customer_id": str, "article_id": str},
parse_dates=["t_dat"]
)
print(f" Articles: {len(articles):,}")
print(f" Customers: {len(customers):,}")
print(f" Transactions: {len(transactions):,}")
return articles, customers, transactions
def sample_recent_weeks(transactions, num_weeks=5):
max_date = transactions["t_dat"].max()
start_date = max_date - timedelta(weeks=num_weeks)
sampled = transactions[transactions["t_dat"] >= start_date].copy()
print(f" Sampled {len(sampled):,} transactions from last {num_weeks} weeks")
return sampled
def filter_active_users(transactions, min_txn=3):
user_counts = transactions["customer_id"].value_counts()
active_users = user_counts[user_counts >= min_txn].index
filtered = transactions[transactions["customer_id"].isin(active_users)]
print(f" Active users (>={min_txn} txns): {len(active_users):,}")
return filtered
def assign_week_numbers(transactions):
max_date = transactions["t_dat"].max()
transactions = transactions.copy()
transactions["week"] = ((max_date - transactions["t_dat"]).dt.days // 7) + 1
transactions["week"] = config.NUM_WEEKS + 1 - transactions["week"]
return transactions
def create_splits(transactions):
train = transactions[transactions["week"] <= config.TRAIN_WEEKS]
val = transactions[transactions["week"] == config.VAL_WEEK]
test = transactions[transactions["week"] == config.TEST_WEEK]
train_items = set(train["article_id"].unique())
train_users = set(train["customer_id"].unique())
test_new_items = test[~test["article_id"].isin(train_items)]
test_new_users = test[~test["customer_id"].isin(train_users)]
print(f" Train: {len(train):,} | Val: {len(val):,} | Test: {len(test):,}")
print(f" Cold-start new items in test: {test_new_items['article_id'].nunique():,}")
print(f" Cold-start new users in test: {test_new_users['customer_id'].nunique():,}")
return train, val, test, test_new_items, test_new_users
def engineer_user_features(customers, transactions):
user_stats = transactions.groupby("customer_id").agg(
txn_count=("t_dat", "count"),
avg_price=("price", "mean"),
std_price=("price", "std"),
unique_articles=("article_id", "nunique"),
days_active=("t_dat", lambda x: (x.max() - x.min()).days),
).reset_index()
user_stats["std_price"] = user_stats["std_price"].fillna(0)
users = customers[["customer_id", "age", "club_member_status", "fashion_news_frequency"]].copy()
users["age"] = users["age"].fillna(users["age"].median())
users["club_member_status"] = users["club_member_status"].map(
{"ACTIVE": 1, "PRE-CREATE": 0.5, "LEFT CLUB": 0}
).fillna(0)
users["fashion_news_frequency"] = users["fashion_news_frequency"].map(
{"Regularly": 1, "Monthly": 0.5, "NONE": 0}
).fillna(0)
users = users.merge(user_stats, on="customer_id", how="left")
users = users.fillna(0)
feature_cols = ["age", "club_member_status", "fashion_news_frequency",
"txn_count", "avg_price", "std_price", "unique_articles", "days_active"]
for col in feature_cols:
col_max = users[col].max()
if col_max > 0:
users[col] = users[col] / col_max
return users, feature_cols
def engineer_item_features(articles):
items = articles[["article_id", "product_type_name", "colour_group_name",
"department_name", "section_name", "garment_group_name",
"prod_name", "detail_desc"]].copy()
items["detail_desc"] = items["detail_desc"].fillna("")
items["prod_name"] = items["prod_name"].fillna("")
return items
def build_user_interaction_history(transactions, max_history=20):
history = defaultdict(list)
sorted_txn = transactions.sort_values("t_dat", ascending=False)
for _, row in sorted_txn.iterrows():
uid = row["customer_id"]
if len(history[uid]) < max_history:
history[uid].append(row["article_id"])
return dict(history)
def prepare_data():
articles, customers, transactions = load_raw_data()
print("\nSampling and filtering...")
transactions = sample_recent_weeks(transactions, config.NUM_WEEKS)
transactions = filter_active_users(transactions, config.MIN_USER_TRANSACTIONS)
transactions = assign_week_numbers(transactions)
active_article_ids = set(transactions["article_id"].unique())
active_customer_ids = set(transactions["customer_id"].unique())
articles = articles[articles["article_id"].isin(active_article_ids)]
customers = customers[customers["customer_id"].isin(active_customer_ids)]
print("\nCreating splits...")
train, val, test, test_new_items, test_new_users = create_splits(transactions)
print("\nEngineering features...")
users_df, user_feature_cols = engineer_user_features(customers, train)
items_df = engineer_item_features(articles)
print("\nBuilding interaction history...")
user_history = build_user_interaction_history(train)
return {
"articles": articles,
"customers": customers,
"train": train,
"val": val,
"test": test,
"test_new_items": test_new_items,
"test_new_users": test_new_users,
"users_df": users_df,
"user_feature_cols": user_feature_cols,
"items_df": items_df,
"user_history": user_history,
}
if __name__ == "__main__":
data = prepare_data()
print("\nData pipeline complete!")
print(f" Users: {len(data['users_df']):,}")
print(f" Items: {len(data['items_df']):,}")Your data pipeline is locked in. Every module from here on will call prepare_data() to get clean, split, feature-engineered data. Next up, you will pre-compute the multi-modal item embeddings that power the retrieval model's understanding of what each fashion article looks like and what it is.
Pre-Compute Multi-Modal Item Embeddings
Your data pipeline is working and you have clean, split datasets ready for modeling. The next challenge is representing each fashion item in a way that captures both its visual appearance and its textual description.
In this step, you will pre-compute multi-modal embeddings for every article in the dataset. A ResNet-50 model extracts visual features from product images, and Sentence-BERT encodes textual metadata into dense vectors. Fusing these two signals gives the Item Tower a rich representation of what each item looks like and what it is described as.
In this step, get ready to:
- Generate 2048-dim visual embeddings for every article using ResNet-50.
- Generate 384-dim text embeddings using the all-MiniLM-L6-v2 Sentence-BERT model.
- Fuse visual and text embeddings into 2432-dim multi-modal vectors for each item.
Generate visual embeddings with ResNet-50
Fashion is inherently visual. A product photo captures silhouette, texture, and color in ways that text alone cannot. ResNet-50 is a pre-trained image classification model that produces a 2048-dimensional feature vector for each image, which becomes the visual half of your item representation.
- Create a new file called feature_engineering.py in your hm_recommender folder and add the imports and visual embedding extraction functions by pasting this code:
import torch
import numpy as np
from torchvision import models, transforms
from torchvision.models import ResNet50_Weights
from sentence_transformers import SentenceTransformer
from PIL import Image
import os
from tqdm import tqdm
import config
def get_image_transform():
return transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]),
])
def extract_visual_embeddings(article_ids, images_dir, batch_size=32):
print("Extracting visual embeddings with ResNet-50...")
device = config.DEVICE
# Load pre-trained ResNet-50 and remove the final classification layer
weights = ResNet50_Weights.DEFAULT
resnet = models.resnet50(weights=weights)
resnet = torch.nn.Sequential(*list(resnet.children())[:-1])
resnet = resnet.to(device)
resnet.eval()
transform = get_image_transform()
embeddings = {}
missing_count = 0
article_list = list(article_ids)
for i in tqdm(range(0, len(article_list), batch_size), desc="Visual embeddings"):
batch_ids = article_list[i:i + batch_size]
batch_images = []
batch_valid_ids = []
for aid in batch_ids:
# H&M images are organized in subdirectories by first 3 digits
subdir = aid[:3]
img_path = os.path.join(images_dir, subdir, f"{aid}.jpg")
if os.path.exists(img_path):
try:
img = Image.open(img_path).convert("RGB")
img_tensor = transform(img)
batch_images.append(img_tensor)
batch_valid_ids.append(aid)
except Exception:
missing_count += 1
else:
missing_count += 1
if batch_images:
batch_tensor = torch.stack(batch_images).to(device)
with torch.no_grad():
features = resnet(batch_tensor).squeeze(-1).squeeze(-1)
features = features.cpu().numpy()
for j, aid in enumerate(batch_valid_ids):
embeddings[aid] = features[j]
# Use mean embedding as fallback for articles without images
if embeddings:
mean_emb = np.mean(list(embeddings.values()), axis=0)
for aid in article_list:
if aid not in embeddings:
embeddings[aid] = mean_emb
print(f" Generated {len(embeddings)} visual embeddings ({missing_count} used fallback)")
return embeddingsWhat does this code do?
get_image_transform() defines the standard ImageNet preprocessing pipeline. Every image is resized to 224x224 pixels, converted to a tensor, and normalized with ImageNet mean/std values.
extract_visual_embeddings() loads a pre-trained ResNet-50, strips its final classification layer (leaving a 2048-dim feature extractor), and processes article images in batches on your MPS device. The H&M images are stored in subdirectories named by the first 3 digits of the article ID. Any article without an image gets the mean embedding as a fallback, so every item ends up with a vector.
- Save feature_engineering.py.
You will not see any output yet. The next chunks add the remaining embedding functions before you run the full pipeline.
Why strip ResNet-50's final layer?
The final layer of ResNet-50 is a classification head trained on ImageNet's 1000 categories. That layer outputs class probabilities, which are useless for fashion items. By removing it, you get the 2048-dim feature vector from the layer before it. These features capture general visual patterns like shape, texture, and color that transfer well to fashion.
Generate text embeddings with Sentence-BERT
Images capture what an item looks like, but text metadata captures functional context. The product type, department, color, and description contain information about fit, material, and intended use that pixels alone cannot convey.
- In feature_engineering.py, add the text embedding function below extract_visual_embeddings() by pasting this code:
def extract_text_embeddings(items_df, batch_size=256):
print("Extracting text embeddings with Sentence-BERT...")
model = SentenceTransformer(config.SBERT_MODEL_NAME)
# Concatenate all textual metadata into one descriptive string per article
texts = []
for _, row in items_df.iterrows():
parts = [
str(row.get("prod_name", "")),
str(row.get("product_type_name", "")),
str(row.get("colour_group_name", "")),
str(row.get("department_name", "")),
str(row.get("section_name", "")),
str(row.get("garment_group_name", "")),
str(row.get("detail_desc", "")),
]
text = " ".join([p for p in parts if p and p != "nan"])
texts.append(text)
# Encode all texts into 384-dim vectors
embeddings_array = model.encode(texts, batch_size=batch_size, show_progress_bar=True)
text_embeddings = {}
for i, aid in enumerate(items_df["article_id"].values):
text_embeddings[aid] = embeddings_array[i]
print(f" Generated {len(text_embeddings)} text embeddings")
return text_embeddingsWhat does this code do?
For each article, the function joins seven metadata fields (product name, product type, colour, department, section, garment group, and detail description) into a single string. This gives Sentence-BERT a complete picture of the item in natural language.
The all-MiniLM-L6-v2 model encodes each string into a 384-dimensional vector. These vectors place semantically similar items close together in embedding space. A "navy slim-fit chino" will land near other casual trousers, even if the exact words differ.
- Save feature_engineering.py.
Still no output yet. The next chunk adds the fusion logic and the orchestrator that runs everything together.
Fuse visual and text embeddings into a unified item representation
Each item now has two vectors: a 2048-dim visual vector and a 384-dim text vector. Concatenating them produces a 2432-dim multi-modal vector that captures both appearance and description. Later, the Item Tower will learn a projection from this 2432-dim space down to the 64-dim shared embedding space.
- In feature_engineering.py, add the fusion utilities, the orchestrator function, and the main block below extract_text_embeddings() by pasting this code:
def fuse_embeddings(visual_emb, text_emb, article_ids):
print("Fusing visual + text embeddings...")
fused = {}
for aid in article_ids:
v = visual_emb.get(aid, np.zeros(config.VISUAL_DIM))
t = text_emb.get(aid, np.zeros(config.TEXT_DIM))
fused[aid] = np.concatenate([v, t])
print(f" Fused {len(fused)} multi-modal embeddings (dim={config.MULTIMODAL_DIM})")
return fused
def save_embeddings(embeddings_dict, filename):
path = os.path.join(config.EMBEDDINGS_DIR, filename)
ids = list(embeddings_dict.keys())
vectors = np.array([embeddings_dict[k] for k in ids])
np.savez(path, ids=np.array(ids), vectors=vectors)
print(f" Saved to {path}")
def load_embeddings(filename):
path = os.path.join(config.EMBEDDINGS_DIR, filename)
data = np.load(path, allow_pickle=True)
return dict(zip(data["ids"], data["vectors"]))
def compute_all_embeddings(items_df, article_ids):
visual_path = os.path.join(config.EMBEDDINGS_DIR, "visual_embeddings.npz")
text_path = os.path.join(config.EMBEDDINGS_DIR, "text_embeddings.npz")
fused_path = os.path.join(config.EMBEDDINGS_DIR, "fused_embeddings.npz")
# Check for cached visual embeddings
if os.path.exists(visual_path):
print("Loading cached visual embeddings...")
visual_emb = load_embeddings("visual_embeddings.npz")
else:
if os.path.exists(config.IMAGES_DIR):
visual_emb = extract_visual_embeddings(article_ids, config.IMAGES_DIR)
else:
print("No images directory found. Using random visual embeddings as placeholder.")
visual_emb = {aid: np.random.randn(config.VISUAL_DIM).astype(np.float32)
for aid in article_ids}
save_embeddings(visual_emb, "visual_embeddings.npz")
# Check for cached text embeddings
if os.path.exists(text_path):
print("Loading cached text embeddings...")
text_emb = load_embeddings("text_embeddings.npz")
else:
text_emb = extract_text_embeddings(items_df)
save_embeddings(text_emb, "text_embeddings.npz")
# Check for cached fused embeddings
if os.path.exists(fused_path):
print("Loading cached fused embeddings...")
fused_emb = load_embeddings("fused_embeddings.npz")
else:
fused_emb = fuse_embeddings(visual_emb, text_emb, article_ids)
save_embeddings(fused_emb, "fused_embeddings.npz")
return fused_emb
if __name__ == "__main__":
from data_pipeline import prepare_data
data = prepare_data()
article_ids = data["items_df"]["article_id"].values
fused = compute_all_embeddings(data["items_df"], article_ids)
print(f"\nDone! {len(fused)} fused embeddings ready.")What does this code do?
fuse_embeddings() concatenates each article's 2048-dim visual vector with its 384-dim text vector, producing a 2432-dim multi-modal vector. If either embedding is missing for an article, a zero vector is used as a fallback.
save_embeddings() and load_embeddings() handle persistence using NumPy's .npz format, storing article IDs alongside their vectors.
compute_all_embeddings() is the orchestrator. It checks for cached .npz files before computing anything. If your images directory does not exist, it generates random visual placeholders so the pipeline still works end to end. This caching logic means you only pay the computation cost once.
- Save feature_engineering.py.
- Run the embedding pipeline by running this command:
python feature_engineering.py- You should see progress bars for visual embeddings (if images exist) and text embeddings, followed by a final line like Done! XXXX fused embeddings ready. where XXXX matches the number of active articles in your sample.
What if I don't have the images directory?
The H&M dataset's images are a separate 28GB download. If you skipped downloading them, the code detects the missing data/images directory and generates random visual embeddings as placeholders. Your pipeline will still work end to end. The text embeddings carry the majority of the semantic signal, so results will still be meaningful.
- Confirm the embedding files were saved by running:
ls embeddings/- You should see visual_embeddings.npz, text_embeddings.npz, and fused_embeddings.npz listed in the output.
Script failing or running out of memory?
- If you see an OutOfMemoryError during visual embedding extraction, reduce the batch_size parameter in extract_visual_embeddings() from 32 to 16 or 8.
- If the Sentence-BERT download stalls, check your internet connection. The all-MiniLM-L6-v2 model is about 80MB and downloads on first use.
- If you re-run the script, the caching logic skips any embedding step that already has a saved .npz file. To regenerate from scratch, delete the files inside the embeddings/ folder first.
- Still stuck? Help me debug an error when running feature_engineering.py for my H&M recommender project.
✔️ Awesome, I've got everything!
Double check you have saved feature_engineering.py and that all three .npz files exist in your embeddings/ directory.
ⓧ I'd like to double check the full code
Here is the complete feature_engineering.py file. Make sure yours matches exactly.
import torch
import numpy as np
from torchvision import models, transforms
from torchvision.models import ResNet50_Weights
from sentence_transformers import SentenceTransformer
from PIL import Image
import os
from tqdm import tqdm
import config
def get_image_transform():
return transforms.Compose([
transforms.Resize((224, 224)),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225]),
])
def extract_visual_embeddings(article_ids, images_dir, batch_size=32):
print("Extracting visual embeddings with ResNet-50...")
device = config.DEVICE
weights = ResNet50_Weights.DEFAULT
resnet = models.resnet50(weights=weights)
resnet = torch.nn.Sequential(*list(resnet.children())[:-1])
resnet = resnet.to(device)
resnet.eval()
transform = get_image_transform()
embeddings = {}
missing_count = 0
article_list = list(article_ids)
for i in tqdm(range(0, len(article_list), batch_size), desc="Visual embeddings"):
batch_ids = article_list[i:i + batch_size]
batch_images = []
batch_valid_ids = []
for aid in batch_ids:
subdir = aid[:3]
img_path = os.path.join(images_dir, subdir, f"{aid}.jpg")
if os.path.exists(img_path):
try:
img = Image.open(img_path).convert("RGB")
img_tensor = transform(img)
batch_images.append(img_tensor)
batch_valid_ids.append(aid)
except Exception:
missing_count += 1
else:
missing_count += 1
if batch_images:
batch_tensor = torch.stack(batch_images).to(device)
with torch.no_grad():
features = resnet(batch_tensor).squeeze(-1).squeeze(-1)
features = features.cpu().numpy()
for j, aid in enumerate(batch_valid_ids):
embeddings[aid] = features[j]
if embeddings:
mean_emb = np.mean(list(embeddings.values()), axis=0)
for aid in article_list:
if aid not in embeddings:
embeddings[aid] = mean_emb
print(f" Generated {len(embeddings)} visual embeddings ({missing_count} used fallback)")
return embeddings
def extract_text_embeddings(items_df, batch_size=256):
print("Extracting text embeddings with Sentence-BERT...")
model = SentenceTransformer(config.SBERT_MODEL_NAME)
texts = []
for _, row in items_df.iterrows():
parts = [
str(row.get("prod_name", "")),
str(row.get("product_type_name", "")),
str(row.get("colour_group_name", "")),
str(row.get("department_name", "")),
str(row.get("section_name", "")),
str(row.get("garment_group_name", "")),
str(row.get("detail_desc", "")),
]
text = " ".join([p for p in parts if p and p != "nan"])
texts.append(text)
embeddings_array = model.encode(texts, batch_size=batch_size, show_progress_bar=True)
text_embeddings = {}
for i, aid in enumerate(items_df["article_id"].values):
text_embeddings[aid] = embeddings_array[i]
print(f" Generated {len(text_embeddings)} text embeddings")
return text_embeddings
def fuse_embeddings(visual_emb, text_emb, article_ids):
print("Fusing visual + text embeddings...")
fused = {}
for aid in article_ids:
v = visual_emb.get(aid, np.zeros(config.VISUAL_DIM))
t = text_emb.get(aid, np.zeros(config.TEXT_DIM))
fused[aid] = np.concatenate([v, t])
print(f" Fused {len(fused)} multi-modal embeddings (dim={config.MULTIMODAL_DIM})")
return fused
def save_embeddings(embeddings_dict, filename):
path = os.path.join(config.EMBEDDINGS_DIR, filename)
ids = list(embeddings_dict.keys())
vectors = np.array([embeddings_dict[k] for k in ids])
np.savez(path, ids=np.array(ids), vectors=vectors)
print(f" Saved to {path}")
def load_embeddings(filename):
path = os.path.join(config.EMBEDDINGS_DIR, filename)
data = np.load(path, allow_pickle=True)
return dict(zip(data["ids"], data["vectors"]))
def compute_all_embeddings(items_df, article_ids):
visual_path = os.path.join(config.EMBEDDINGS_DIR, "visual_embeddings.npz")
text_path = os.path.join(config.EMBEDDINGS_DIR, "text_embeddings.npz")
fused_path = os.path.join(config.EMBEDDINGS_DIR, "fused_embeddings.npz")
if os.path.exists(visual_path):
print("Loading cached visual embeddings...")
visual_emb = load_embeddings("visual_embeddings.npz")
else:
if os.path.exists(config.IMAGES_DIR):
visual_emb = extract_visual_embeddings(article_ids, config.IMAGES_DIR)
else:
print("No images directory found. Using random visual embeddings as placeholder.")
visual_emb = {aid: np.random.randn(config.VISUAL_DIM).astype(np.float32)
for aid in article_ids}
save_embeddings(visual_emb, "visual_embeddings.npz")
if os.path.exists(text_path):
print("Loading cached text embeddings...")
text_emb = load_embeddings("text_embeddings.npz")
else:
text_emb = extract_text_embeddings(items_df)
save_embeddings(text_emb, "text_embeddings.npz")
if os.path.exists(fused_path):
print("Loading cached fused embeddings...")
fused_emb = load_embeddings("fused_embeddings.npz")
else:
fused_emb = fuse_embeddings(visual_emb, text_emb, article_ids)
save_embeddings(fused_emb, "fused_embeddings.npz")
return fused_emb
if __name__ == "__main__":
from data_pipeline import prepare_data
data = prepare_data()
article_ids = data["items_df"]["article_id"].values
fused = compute_all_embeddings(data["items_df"], article_ids)
print(f"\nDone! {len(fused)} fused embeddings ready.")Your items now have rich multi-modal representations stored on disk. Next up, you will use an LLM to generate commonsense knowledge triples that capture why people buy each item, adding a third layer of understanding to your recommendation engine.
Generate COSMO-Inspired Commonsense Knowledge with Phi-4-mini
Your multi-modal embeddings capture what each fashion item looks like and how it is described. But they miss something critical: why someone would buy it.
Amazon's COSMO system (SIGMOD 2024) addresses this gap by using an LLM to generate commonsense knowledge triples that capture purchase intent. In this step, you will build a similar pipeline using Phi-4-mini to produce intent knowledge for every article in your dataset.
In this step, get ready to:
- Design a structured prompt that generates commonsense triples across 4 COSMO relation types.
- Generate and filter knowledge triples for all articles using Phi-4-mini.
- Encode the triples into 384-dim intent embeddings with Sentence-BERT.
Set up the LLM knowledge generation pipeline
Amazon's original COSMO system used OPT-175B on clusters of A100 GPUs. You will adapt the same idea for local execution using Phi-4-mini (3.8B parameters), which fits comfortably on your Mac at float16 precision.
What is COSMO?
COSMO (Common Sense for Merchandising and Organization) is a system Amazon published at SIGMOD 2024. It builds a commonsense knowledge graph that encodes implicit purchase motivations. For example, "pregnant" implies a need for "slip-resistant shoes."
Amazon reported a 60% improvement in search relevance and 0.7% sales uplift on 10% of US traffic. Your adapted version uses 4 fashion-specific relation types instead of COSMO's full 15.
- Create a new file called cosmo_knowledge.py in your hm_recommender folder (the same level as config.py).
- Add the imports and COSMO prompt template by copying this code:
import json
import os
import re
import numpy as np
from tqdm import tqdm
from transformers import AutoModelForCausalLM, AutoTokenizer
from sentence_transformers import SentenceTransformer
import torch
import config
COSMO_PROMPT_TEMPLATE = """You are an e-commerce knowledge expert. Given a fashion product, generate commonsense knowledge triples that explain WHY a customer would buy this item.
Product: {prod_name}
Type: {product_type}
Color: {color}
Department: {department}
Description: {description}
Generate exactly 4 knowledge triples in this JSON format:
[
{{"relation": "used_for_function", "value": "<practical function>"}},
{{"relation": "used_for_event", "value": "<occasion or event>"}},
{{"relation": "used_for_audience", "value": "<target buyer persona>"}},
{{"relation": "used_for_season", "value": "<season or weather>"}}
]
Return ONLY the JSON array, nothing else."""What does this code do?
The imports bring in the Hugging Face transformers library for loading Phi-4-mini, plus sentence_transformers for encoding triples into embeddings later.
The prompt template asks the LLM to generate exactly 4 triples, one per COSMO relation type: used_for_function (practical purpose), used_for_event (occasion), used_for_audience (buyer persona), and used_for_season (weather/time). The double curly braces {{}} are Python format-string escapes that produce literal {} in the final prompt sent to the model.
- Save cosmo_knowledge.py.
- Confirm your editor shows no syntax errors. The file is not runnable yet.
Next, you need two functions: one to load the LLM into memory, and one to call it for each article.
- Add the load_llm() function below the prompt template:
def load_llm():
print(f"Loading LLM: {config.LLM_MODEL_NAME}...")
tokenizer = AutoTokenizer.from_pretrained(config.LLM_MODEL_NAME, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
config.LLM_MODEL_NAME,
torch_dtype=torch.float16,
device_map="auto",
trust_remote_code=True,
)
model.eval()
print(" LLM loaded successfully.")
return model, tokenizerWhat does this function do?
- AutoTokenizer.from_pretrained downloads and loads the tokenizer for Phi-4-mini (microsoft/Phi-4-mini-instruct).
- AutoModelForCausalLM.from_pretrained loads the model weights in float16 precision (~7.6 GB RAM). The device_map="auto" setting places the model on whatever hardware is available (MPS on Apple Silicon, CUDA on NVIDIA, or CPU).
- model.eval() disables training-only behaviors like dropout so inference is deterministic.
You won't see a change yet. The next function uses this to generate triples.
- Add the generate_triples_for_item() function below load_llm():
def generate_triples_for_item(model, tokenizer, item_row):
prompt = COSMO_PROMPT_TEMPLATE.format(
prod_name=item_row.get("prod_name", "Unknown"),
product_type=item_row.get("product_type_name", "Unknown"),
color=item_row.get("colour_group_name", "Unknown"),
department=item_row.get("department_name", "Unknown"),
description=item_row.get("detail_desc", "No description"),
)
messages = [{"role": "user", "content": prompt}]
inputs = tokenizer.apply_chat_template(messages, return_tensors="pt", return_dict=True)
inputs = {k: v.to(model.device) for k, v in inputs.items()}
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=config.LLM_MAX_NEW_TOKENS,
temperature=config.LLM_TEMPERATURE,
do_sample=True,
pad_token_id=tokenizer.eos_token_id,
)
response = tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)
# Parse JSON from response
try:
json_match = re.search(r'\[.*\]', response, re.DOTALL)
if json_match:
triples = json.loads(json_match.group())
return triples
except (json.JSONDecodeError, AttributeError):
pass
# Fallback: generate generic triples from metadata
return [
{"relation": "used_for_function", "value": f"wearing as {item_row.get('product_type_name', 'clothing')}"},
{"relation": "used_for_event", "value": "everyday wear"},
{"relation": "used_for_audience", "value": f"{item_row.get('section_name', 'general')} shoppers"},
{"relation": "used_for_season", "value": "all seasons"},
]What does this function do?
- It fills the prompt template with the article's metadata (name, type, color, department, description), then sends it to Phi-4-mini using the chat template format.
- torch.no_grad() disables gradient tracking to save memory during inference.
- The response is parsed with a regex to extract the JSON array. If parsing fails (the LLM occasionally produces malformed JSON), a fallback generates generic triples from the item's existing metadata.
- Save cosmo_knowledge.py.
- Confirm no syntax errors appear. The functions are not callable on their own yet.
Seeing an import error for transformers?
- Make sure you installed the transformers library in your virtual environment. Run pip install transformers if it is missing.
- Confirm your virtual environment is activated. Your terminal prompt should show the environment name.
Why can't Python find my transformers import?
Generate knowledge triples for all articles
Raw LLM outputs can be noisy. Some triples will just restate the product name or produce vague answers like "anyone" for the audience. You need a quality filter before saving the results.
- Add the filter_low_quality_triples() function below generate_triples_for_item():
def filter_low_quality_triples(triples, item_row):
filtered = []
prod_name_lower = str(item_row.get("prod_name", "")).lower()
product_type_lower = str(item_row.get("product_type_name", "")).lower()
for triple in triples:
value = triple.get("value", "").lower().strip()
# Filter trivially derived from product name
if value == prod_name_lower or value == product_type_lower:
continue
# Filter too generic
if value in ["clothing", "wearing clothes", "general use", "anyone", ""]:
continue
filtered.append(triple)
return filtered if filtered else triples # keep originals if all filteredWhat does this function do?
This is a simplified version of COSMO's multi-stage quality filtering. It removes triples where the value exactly matches the product name or type (trivially derived), and triples that are too generic like "clothing" or "anyone."
If all triples get filtered out, the function keeps the originals as a safety net. Amazon's full system additionally uses DeBERTa classifiers and human annotation, but keyword filtering works well for this project's scope.
You won't see a change yet. The next function ties everything together into a batch pipeline.
- Add the generate_all_knowledge() function below filter_low_quality_triples():
def generate_all_knowledge(items_df, max_items=None):
if os.path.exists(config.COSMO_TRIPLES_PATH):
print("Loading cached COSMO knowledge triples...")
with open(config.COSMO_TRIPLES_PATH, "r") as f:
return json.load(f)
model, tokenizer = load_llm()
all_knowledge = {}
items_to_process = items_df.head(max_items) if max_items else items_df
for _, row in tqdm(items_to_process.iterrows(), total=len(items_to_process),
desc="Generating COSMO knowledge"):
aid = row["article_id"]
triples = generate_triples_for_item(model, tokenizer, row)
triples = filter_low_quality_triples(triples, row)
all_knowledge[aid] = triples
# Save
with open(config.COSMO_TRIPLES_PATH, "w") as f:
json.dump(all_knowledge, f, indent=2)
print(f" Generated knowledge for {len(all_knowledge)} articles")
print(f" Saved to {config.COSMO_TRIPLES_PATH}")
# Free GPU memory
del model
del tokenizer
if torch.backends.mps.is_available():
torch.mps.empty_cache()
return all_knowledgeWhat does this function do?
- The caching check at the top skips the entire LLM pipeline if knowledge/cosmo_triples.json already exists. This means you can re-run the script without waiting 30-60 minutes again.
- The max_items parameter lets you process a subset of articles for testing. When set to None, it processes all active articles.
- After generation is complete, the function explicitly deletes the model and clears the MPS cache. Phi-4-mini uses ~7.6 GB at float16, and freeing that memory is essential before the rest of the pipeline runs.
- Save cosmo_knowledge.py.
Test with a small batch first
Before processing all articles, you can test the LLM pipeline on a handful of items to verify the output format. In the __main__ block (which you will add shortly), you could temporarily call generate_all_knowledge(data["items_df"], max_items=5) to process only 5 articles. Once you confirm the triples look reasonable, delete the cosmo_triples.json file and re-run without the limit.
Running out of memory loading Phi-4-mini?
- Close any other memory-heavy applications (browsers with many tabs, other Python processes, Docker containers).
- The model needs approximately 7.6 GB at float16. On a 16 GB Mac, the data pipeline and embeddings should be freed from memory first. The caching in this script handles that since the LLM is loaded fresh each run.
Help me fix memory issues loading Phi-4-mini on my Mac.
Encode knowledge triples into intent embeddings
The JSON triples are human-readable, but the ranking model needs numerical vectors. You will concatenate each article's triples into a single text string and encode it with Sentence-BERT to produce 384-dim "intent embeddings."
Why encode triples with Sentence-BERT instead of using them directly?
The ranking model in Step 8 computes cosine similarity between a user's intent profile and a candidate item's intent. Dense vector similarity is orders of magnitude faster than string matching across thousands of candidates.
Encoding also captures semantic similarity. "casual friday" and "relaxed office wear" would have similar embeddings even though the strings are different.
- Add the encode_intent_embeddings() function below generate_all_knowledge():
def encode_intent_embeddings(knowledge_dict):
intent_path = os.path.join(config.EMBEDDINGS_DIR, "intent_embeddings.npz")
if os.path.exists(intent_path):
print("Loading cached intent embeddings...")
data = np.load(intent_path, allow_pickle=True)
return dict(zip(data["ids"], data["vectors"]))
print("Encoding COSMO triples into intent embeddings...")
sbert = SentenceTransformer(config.SBERT_MODEL_NAME)
texts = []
article_ids = []
for aid, triples in knowledge_dict.items():
triple_texts = [f"{t['relation']}: {t['value']}" for t in triples]
combined = ". ".join(triple_texts)
texts.append(combined)
article_ids.append(aid)
embeddings = sbert.encode(texts, batch_size=256, show_progress_bar=True)
intent_emb = {}
for i, aid in enumerate(article_ids):
intent_emb[aid] = embeddings[i]
# Save
ids_arr = np.array(article_ids)
vecs_arr = np.array([intent_emb[aid] for aid in article_ids])
np.savez(intent_path, ids=ids_arr, vectors=vecs_arr)
print(f" Generated {len(intent_emb)} intent embeddings (dim={config.INTENT_DIM})")
return intent_embWhat does this function do?
- Like the other embedding functions, it checks for a cached file first (embeddings/intent_embeddings.npz) and skips recomputation if it exists.
- For each article, it joins all triple texts into a single string like "used_for_function: casual comfort. used_for_event: weekend brunch. used_for_audience: young professionals. used_for_season: spring". This combined string is encoded by all-MiniLM-L6-v2 into a single 384-dim vector.
- The result is a dictionary mapping each article ID to a 384-dim intent embedding that captures why someone buys the item, complementing the visual/text embeddings that capture what the item is.
Finally, add the main entry point that runs the full COSMO pipeline end to end.
- Add the __main__ block at the bottom of cosmo_knowledge.py:
if __name__ == "__main__":
from data_pipeline import prepare_data
data = prepare_data()
knowledge = generate_all_knowledge(data["items_df"])
intent_emb = encode_intent_embeddings(knowledge)
print(f"\nCOSMO pipeline complete! {len(intent_emb)} intent embeddings ready.")What does this block do?
This is the script's entry point. It loads the data pipeline output, runs knowledge generation for all articles (or loads from cache), and then encodes the triples into intent embeddings. The two steps are sequential because the encoding depends on the generated triples.
- Save cosmo_knowledge.py.
- Run the COSMO pipeline by running this command:
python cosmo_knowledge.pyHow long will this take?
The LLM knowledge generation processes one article at a time. Depending on how many active articles your sampled dataset has, this takes approximately 30-60 minutes on an M4 Mac. You will see a progress bar tracking the generation.
The Sentence-BERT encoding step afterward is much faster (a few minutes at most). Once both files are cached, future runs skip the generation entirely.
- Verify the output by checking for both generated files by running:
ls knowledge/cosmo_triples.json embeddings/intent_embeddings.npzYou should see both files listed. The terminal output from the script should end with a line like COSMO pipeline complete! XXXX intent embeddings ready. where XXXX matches the number of active articles in your dataset.
Seeing JSON parse errors in the output?
- This is expected for some articles. The fallback logic in generate_triples_for_item() handles failed parses by generating metadata-based triples instead. As long as the script completes and the output files exist, the pipeline worked correctly.
- If the script crashes entirely (does not produce the JSON file), check that Phi-4-mini downloaded successfully. The first run downloads approximately 7.6 GB of model weights from Hugging Face.
Help me debug my COSMO knowledge generation pipeline.
✔️ Awesome, I've got everything!
Double check you have saved cosmo_knowledge.py and that both knowledge/cosmo_triples.json and embeddings/intent_embeddings.npz exist on disk.
ⓧ I'd like to double check the full code
Here is the complete cosmo_knowledge.py file. Compare it against yours and fix any differences.
import json
import os
import re
import numpy as np
from tqdm import tqdm
from transformers import AutoModelForCausalLM, AutoTokenizer
from sentence_transformers import SentenceTransformer
import torch
import config
COSMO_PROMPT_TEMPLATE = """You are an e-commerce knowledge expert. Given a fashion product, generate commonsense knowledge triples that explain WHY a customer would buy this item.
Product: {prod_name}
Type: {product_type}
Color: {color}
Department: {department}
Description: {description}
Generate exactly 4 knowledge triples in this JSON format:
[
{{"relation": "used_for_function", "value": "<practical function>"}},
{{"relation": "used_for_event", "value": "<occasion or event>"}},
{{"relation": "used_for_audience", "value": "<target buyer persona>"}},
{{"relation": "used_for_season", "value": "<season or weather>"}}
]
Return ONLY the JSON array, nothing else."""
def load_llm():
print(f"Loading LLM: {config.LLM_MODEL_NAME}...")
tokenizer = AutoTokenizer.from_pretrained(config.LLM_MODEL_NAME, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
config.LLM_MODEL_NAME,
torch_dtype=torch.float16,
device_map="auto",
trust_remote_code=True,
)
model.eval()
print(" LLM loaded successfully.")
return model, tokenizer
def generate_triples_for_item(model, tokenizer, item_row):
prompt = COSMO_PROMPT_TEMPLATE.format(
prod_name=item_row.get("prod_name", "Unknown"),
product_type=item_row.get("product_type_name", "Unknown"),
color=item_row.get("colour_group_name", "Unknown"),
department=item_row.get("department_name", "Unknown"),
description=item_row.get("detail_desc", "No description"),
)
messages = [{"role": "user", "content": prompt}]
inputs = tokenizer.apply_chat_template(messages, return_tensors="pt", return_dict=True)
inputs = {k: v.to(model.device) for k, v in inputs.items()}
with torch.no_grad():
outputs = model.generate(
**inputs,
max_new_tokens=config.LLM_MAX_NEW_TOKENS,
temperature=config.LLM_TEMPERATURE,
do_sample=True,
pad_token_id=tokenizer.eos_token_id,
)
response = tokenizer.decode(outputs[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)
# Parse JSON from response
try:
json_match = re.search(r'\[.*\]', response, re.DOTALL)
if json_match:
triples = json.loads(json_match.group())
return triples
except (json.JSONDecodeError, AttributeError):
pass
# Fallback: generate generic triples from metadata
return [
{"relation": "used_for_function", "value": f"wearing as {item_row.get('product_type_name', 'clothing')}"},
{"relation": "used_for_event", "value": "everyday wear"},
{"relation": "used_for_audience", "value": f"{item_row.get('section_name', 'general')} shoppers"},
{"relation": "used_for_season", "value": "all seasons"},
]
def filter_low_quality_triples(triples, item_row):
filtered = []
prod_name_lower = str(item_row.get("prod_name", "")).lower()
product_type_lower = str(item_row.get("product_type_name", "")).lower()
for triple in triples:
value = triple.get("value", "").lower().strip()
# Filter trivially derived from product name
if value == prod_name_lower or value == product_type_lower:
continue
# Filter too generic
if value in ["clothing", "wearing clothes", "general use", "anyone", ""]:
continue
filtered.append(triple)
return filtered if filtered else triples # keep originals if all filtered
def generate_all_knowledge(items_df, max_items=None):
if os.path.exists(config.COSMO_TRIPLES_PATH):
print("Loading cached COSMO knowledge triples...")
with open(config.COSMO_TRIPLES_PATH, "r") as f:
return json.load(f)
model, tokenizer = load_llm()
all_knowledge = {}
items_to_process = items_df.head(max_items) if max_items else items_df
for _, row in tqdm(items_to_process.iterrows(), total=len(items_to_process),
desc="Generating COSMO knowledge"):
aid = row["article_id"]
triples = generate_triples_for_item(model, tokenizer, row)
triples = filter_low_quality_triples(triples, row)
all_knowledge[aid] = triples
# Save
with open(config.COSMO_TRIPLES_PATH, "w") as f:
json.dump(all_knowledge, f, indent=2)
print(f" Generated knowledge for {len(all_knowledge)} articles")
print(f" Saved to {config.COSMO_TRIPLES_PATH}")
# Free GPU memory
del model
del tokenizer
if torch.backends.mps.is_available():
torch.mps.empty_cache()
return all_knowledge
def encode_intent_embeddings(knowledge_dict):
intent_path = os.path.join(config.EMBEDDINGS_DIR, "intent_embeddings.npz")
if os.path.exists(intent_path):
print("Loading cached intent embeddings...")
data = np.load(intent_path, allow_pickle=True)
return dict(zip(data["ids"], data["vectors"]))
print("Encoding COSMO triples into intent embeddings...")
sbert = SentenceTransformer(config.SBERT_MODEL_NAME)
texts = []
article_ids = []
for aid, triples in knowledge_dict.items():
triple_texts = [f"{t['relation']}: {t['value']}" for t in triples]
combined = ". ".join(triple_texts)
texts.append(combined)
article_ids.append(aid)
embeddings = sbert.encode(texts, batch_size=256, show_progress_bar=True)
intent_emb = {}
for i, aid in enumerate(article_ids):
intent_emb[aid] = embeddings[i]
# Save
ids_arr = np.array(article_ids)
vecs_arr = np.array([intent_emb[aid] for aid in article_ids])
np.savez(intent_path, ids=ids_arr, vectors=vecs_arr)
print(f" Generated {len(intent_emb)} intent embeddings (dim={config.INTENT_DIM})")
return intent_emb
if __name__ == "__main__":
from data_pipeline import prepare_data
data = prepare_data()
knowledge = generate_all_knowledge(data["items_df"])
intent_emb = encode_intent_embeddings(knowledge)
print(f"\nCOSMO pipeline complete! {len(intent_emb)} intent embeddings ready.")Your items now have three layers of representation: visual features capturing appearance, text features capturing product attributes, and intent embeddings capturing purchase motivation. Next up, you will build the two-tower retrieval model that puts these embeddings to work.
Build and Train the Two-Tower Retrieval Model
Your multi-modal embeddings and COSMO intent knowledge are ready. Now you need a model that can learn which items each user is likely to buy next.
The two-tower architecture solves this by encoding users and items into the same 64-dimensional embedding space. A User Tower processes demographics and purchase history. An Item Tower processes visual and text features. Items that land close to a user in this space are strong recommendation candidates.
In this step, get ready to:
- Implement a two-tower model with DropoutNet cold-start handling and in-batch negative sampling.
- Build the training pipeline with MPS acceleration and cosine annealing.
- Train for 20 epochs and extract 64-dim trained embeddings for all users and items.
Implement the two-tower model architecture
The two-tower approach is used at YouTube, Pinterest, and Instagram Explore. Each tower independently produces an embedding, and the dot product between them measures relevance. Training the towers end-to-end means the model learns what makes a user-item pair a good match.
- Create a new file called models.py in your hm_recommender folder.
- Add the imports and the UserTower class by copying this code:
import torch
import torch.nn as nn
import torch.nn.functional as F
import config
class UserTower(nn.Module):
def __init__(self, user_feature_dim, history_dim, embedding_dim, hidden_dim, dropout_rate, dropoutnet_p):
super().__init__()
self.dropoutnet_p = dropoutnet_p
# Process user demographics (age, club status, etc.)
self.user_feat_net = nn.Sequential(
nn.Linear(user_feature_dim, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(dropout_rate),
)
# Process mean-pooled embeddings of previously purchased items
self.history_net = nn.Sequential(
nn.Linear(history_dim, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(dropout_rate),
)
# Merge both signals into a single user embedding
self.combine_net = nn.Sequential(
nn.Linear(hidden_dim * 2, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(dropout_rate),
nn.Linear(hidden_dim, embedding_dim),
)
def forward(self, user_features, history_embedding):
user_out = self.user_feat_net(user_features)
# DropoutNet: randomly zero out history during training
# Forces the model to recommend using demographics alone
if self.training and self.dropoutnet_p > 0:
mask = (torch.rand(history_embedding.size(0), 1, device=history_embedding.device)
> self.dropoutnet_p).float()
history_embedding = history_embedding * mask
history_out = self.history_net(history_embedding)
combined = torch.cat([user_out, history_out], dim=1)
embedding = self.combine_net(combined)
return embeddingWhat does this code do?
The UserTower encodes a user into a 64-dimensional vector. It has two input branches: one for demographic features and one for purchase history (mean-pooled item embeddings).
The DropoutNet mechanism is the key cold-start trick. During training, the history embedding is randomly zeroed out with probability 0.5. This forces the model to learn useful recommendations from demographics alone, so new users without any history still get quality results.
You won't see output yet. The next chunk adds the Item Tower and the combined model.
- Below the UserTower class, add the ItemTower and TwoTowerModel classes by copying this code:
class ItemTower(nn.Module):
def __init__(self, multimodal_dim, embedding_dim, hidden_dim, dropout_rate):
super().__init__()
# 3-layer feedforward: 2432-dim multimodal input -> 64-dim embedding
self.net = nn.Sequential(
nn.Linear(multimodal_dim, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(dropout_rate),
nn.Linear(hidden_dim, hidden_dim // 2),
nn.BatchNorm1d(hidden_dim // 2),
nn.ReLU(),
nn.Dropout(dropout_rate),
nn.Linear(hidden_dim // 2, embedding_dim),
)
def forward(self, multimodal_features):
return self.net(multimodal_features)
class TwoTowerModel(nn.Module):
def __init__(self, user_feature_dim=config.USER_FEATURE_DIM,
history_dim=config.MULTIMODAL_DIM,
multimodal_dim=config.MULTIMODAL_DIM,
embedding_dim=config.EMBEDDING_DIM,
hidden_dim=config.HIDDEN_DIM,
dropout_rate=config.DROPOUT_RATE,
dropoutnet_p=config.DROPOUTNET_P):
super().__init__()
self.user_tower = UserTower(user_feature_dim, history_dim, embedding_dim,
hidden_dim, dropout_rate, dropoutnet_p)
self.item_tower = ItemTower(multimodal_dim, embedding_dim, hidden_dim, dropout_rate)
# Learnable temperature for softmax scaling
self.temperature = nn.Parameter(torch.tensor(0.07))
def forward(self, user_features, history_embedding, item_features):
user_emb = self.user_tower(user_features, history_embedding)
item_emb = self.item_tower(item_features)
# L2 normalize so dot product = cosine similarity
user_emb = F.normalize(user_emb, p=2, dim=1)
item_emb = F.normalize(item_emb, p=2, dim=1)
return user_emb, item_emb
def compute_loss(self, user_emb, item_emb):
# In-batch negatives: every other item in the batch is a negative
logits = torch.matmul(user_emb, item_emb.T) / self.temperature.abs()
labels = torch.arange(logits.size(0), device=logits.device)
loss = F.cross_entropy(logits, labels)
return loss
def get_user_embedding(self, user_features, history_embedding):
self.eval()
with torch.no_grad():
emb = self.user_tower(user_features, history_embedding)
emb = F.normalize(emb, p=2, dim=1)
return emb
def get_item_embedding(self, item_features):
self.eval()
with torch.no_grad():
emb = self.item_tower(item_features)
emb = F.normalize(emb, p=2, dim=1)
return embWhat does this code do?
- ItemTower compresses the 2432-dim fused embedding (ResNet-50 visual + Sentence-BERT text) down to 64 dimensions through three linear layers. Since it only uses intrinsic item properties (pixels and words), brand-new items with zero purchase history are immediately positioned in the embedding space.
- TwoTowerModel wires both towers together. The forward() method returns L2-normalized embeddings so dot product equals cosine similarity.
- compute_loss() uses in-batch negative sampling. Within each batch, every other item serves as a negative for each user. This avoids explicit negative mining and naturally provides hard negatives when the batch size is large.
- temperature is a learnable parameter (initialized to 0.07) that scales the logits before softmax. The model learns the optimal sharpness for distinguishing positive from negative pairs.
- Save models.py.
- Verify the model loads correctly by running this command in your terminal:
python -c "from models import TwoTowerModel; print('Model loaded successfully')"You should see Model loaded successfully printed after the device line.
Seeing an import error?
- Make sure you are running the command from inside the hm_recommender folder where config.py and models.py both live.
- Check for indentation errors in your models.py. Python requires consistent indentation (4 spaces) for class methods.
- Why can't Python find my models module?
Why in-batch negatives instead of explicit negative sampling?
Explicit negative sampling requires pre-computing or randomly selecting items the user did not interact with. This is slow and often produces "easy" negatives the model already distinguishes well.
In-batch negatives reuse all other items already loaded in the batch as negatives for each user. With a batch size of 512, each positive pair gets 511 free negatives. This technique was introduced in the YouTube DNN paper and is the standard at Google, Meta, and Pinterest.
✔️ Awesome, I've got everything!
Double check you've saved models.py before moving on.
ⓧ I'd like to double check the full code
import torch
import torch.nn as nn
import torch.nn.functional as F
import config
class UserTower(nn.Module):
def __init__(self, user_feature_dim, history_dim, embedding_dim, hidden_dim, dropout_rate, dropoutnet_p):
super().__init__()
self.dropoutnet_p = dropoutnet_p
self.user_feat_net = nn.Sequential(
nn.Linear(user_feature_dim, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(dropout_rate),
)
self.history_net = nn.Sequential(
nn.Linear(history_dim, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(dropout_rate),
)
self.combine_net = nn.Sequential(
nn.Linear(hidden_dim * 2, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(dropout_rate),
nn.Linear(hidden_dim, embedding_dim),
)
def forward(self, user_features, history_embedding):
user_out = self.user_feat_net(user_features)
if self.training and self.dropoutnet_p > 0:
mask = (torch.rand(history_embedding.size(0), 1, device=history_embedding.device)
> self.dropoutnet_p).float()
history_embedding = history_embedding * mask
history_out = self.history_net(history_embedding)
combined = torch.cat([user_out, history_out], dim=1)
embedding = self.combine_net(combined)
return embedding
class ItemTower(nn.Module):
def __init__(self, multimodal_dim, embedding_dim, hidden_dim, dropout_rate):
super().__init__()
self.net = nn.Sequential(
nn.Linear(multimodal_dim, hidden_dim),
nn.BatchNorm1d(hidden_dim),
nn.ReLU(),
nn.Dropout(dropout_rate),
nn.Linear(hidden_dim, hidden_dim // 2),
nn.BatchNorm1d(hidden_dim // 2),
nn.ReLU(),
nn.Dropout(dropout_rate),
nn.Linear(hidden_dim // 2, embedding_dim),
)
def forward(self, multimodal_features):
return self.net(multimodal_features)
class TwoTowerModel(nn.Module):
def __init__(self, user_feature_dim=config.USER_FEATURE_DIM,
history_dim=config.MULTIMODAL_DIM,
multimodal_dim=config.MULTIMODAL_DIM,
embedding_dim=config.EMBEDDING_DIM,
hidden_dim=config.HIDDEN_DIM,
dropout_rate=config.DROPOUT_RATE,
dropoutnet_p=config.DROPOUTNET_P):
super().__init__()
self.user_tower = UserTower(user_feature_dim, history_dim, embedding_dim,
hidden_dim, dropout_rate, dropoutnet_p)
self.item_tower = ItemTower(multimodal_dim, embedding_dim, hidden_dim, dropout_rate)
self.temperature = nn.Parameter(torch.tensor(0.07))
def forward(self, user_features, history_embedding, item_features):
user_emb = self.user_tower(user_features, history_embedding)
item_emb = self.item_tower(item_features)
user_emb = F.normalize(user_emb, p=2, dim=1)
item_emb = F.normalize(item_emb, p=2, dim=1)
return user_emb, item_emb
def compute_loss(self, user_emb, item_emb):
logits = torch.matmul(user_emb, item_emb.T) / self.temperature.abs()
labels = torch.arange(logits.size(0), device=logits.device)
loss = F.cross_entropy(logits, labels)
return loss
def get_user_embedding(self, user_features, history_embedding):
self.eval()
with torch.no_grad():
emb = self.user_tower(user_features, history_embedding)
emb = F.normalize(emb, p=2, dim=1)
return emb
def get_item_embedding(self, item_features):
self.eval()
with torch.no_grad():
emb = self.item_tower(item_features)
emb = F.normalize(emb, p=2, dim=1)
return embCreate the training pipeline and evaluation utilities
The model architecture is defined. Now you need the machinery to feed data through it, run the training loop, and measure how well it performs.
- Create a new file called train_retrieval.py in your hm_recommender folder.
- Add the imports, the InteractionDataset class, and the two helper functions by copying this code:
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import numpy as np
from tqdm import tqdm
import os
import config
from models import TwoTowerModel
class InteractionDataset(Dataset):
"""Wraps (user, item) pairs into tensors for the DataLoader."""
def __init__(self, interactions_df, user_features_dict, item_embeddings_dict,
user_history_embeddings, user_feature_cols):
self.pairs = interactions_df[["customer_id", "article_id"]].values
self.user_features_dict = user_features_dict
self.item_embeddings_dict = item_embeddings_dict
self.user_history_embeddings = user_history_embeddings
self.user_feature_cols = user_feature_cols
def __len__(self):
return len(self.pairs)
def __getitem__(self, idx):
uid, aid = self.pairs[idx]
# Look up pre-computed features; fall back to zeros if missing
user_feat = self.user_features_dict.get(uid, np.zeros(len(self.user_feature_cols)))
item_feat = self.item_embeddings_dict.get(aid, np.zeros(config.MULTIMODAL_DIM))
history_feat = self.user_history_embeddings.get(uid, np.zeros(config.MULTIMODAL_DIM))
return (
torch.tensor(user_feat, dtype=torch.float32),
torch.tensor(history_feat, dtype=torch.float32),
torch.tensor(item_feat, dtype=torch.float32),
)
def build_user_features_dict(users_df, feature_cols):
"""Convert the users DataFrame into a dict for O(1) lookup."""
result = {}
for _, row in users_df.iterrows():
result[row["customer_id"]] = row[feature_cols].values.astype(np.float32)
return result
def build_user_history_embeddings(user_history, item_embeddings_dict):
"""Mean-pool each user's purchased item embeddings into one vector."""
result = {}
for uid, item_ids in user_history.items():
embs = [item_embeddings_dict[aid] for aid in item_ids if aid in item_embeddings_dict]
if embs:
result[uid] = np.mean(embs, axis=0).astype(np.float32)
else:
result[uid] = np.zeros(config.MULTIMODAL_DIM, dtype=np.float32)
return resultWhat does this code do?
- InteractionDataset is a PyTorch Dataset that wraps (user, item) transaction pairs. For each pair, it looks up the user's demographic features, the item's fused multimodal embedding, and the user's mean-pooled interaction history.
- build_user_features_dict() converts the users DataFrame into a dictionary for fast lookup during training.
- build_user_history_embeddings() averages the fused embeddings of each user's previously purchased items into a single 2432-dim vector. This is the interaction history that the User Tower processes (and that DropoutNet randomly zeroes out).
You won't see output yet. The next chunk adds the training loop.
- Below the helper functions, add the train_model() function by copying this code:
def train_model(data, fused_embeddings):
print("\n=== Training Two-Tower Retrieval Model ===")
device = config.DEVICE
# Build lookup dicts from the prepared data
user_feat_dict = build_user_features_dict(data["users_df"], data["user_feature_cols"])
history_emb_dict = build_user_history_embeddings(data["user_history"], fused_embeddings)
# Create train and validation datasets
train_dataset = InteractionDataset(
data["train"], user_feat_dict, fused_embeddings,
history_emb_dict, data["user_feature_cols"]
)
val_dataset = InteractionDataset(
data["val"], user_feat_dict, fused_embeddings,
history_emb_dict, data["user_feature_cols"]
)
train_loader = DataLoader(train_dataset, batch_size=config.BATCH_SIZE, shuffle=True,
num_workers=0, drop_last=True)
val_loader = DataLoader(val_dataset, batch_size=config.BATCH_SIZE, shuffle=False,
num_workers=0)
# Initialize model with the correct user feature dimension
model = TwoTowerModel(
user_feature_dim=len(data["user_feature_cols"]),
history_dim=config.MULTIMODAL_DIM,
).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=config.LEARNING_RATE,
weight_decay=config.WEIGHT_DECAY)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=config.NUM_EPOCHS)
best_val_loss = float("inf")
for epoch in range(config.NUM_EPOCHS):
model.train()
total_loss = 0
num_batches = 0
pbar = tqdm(train_loader, desc=f"Epoch {epoch+1}/{config.NUM_EPOCHS}")
for user_feat, history_feat, item_feat in pbar:
user_feat = user_feat.to(device)
history_feat = history_feat.to(device)
item_feat = item_feat.to(device)
# Forward pass: get embeddings, compute in-batch negative loss
user_emb, item_emb = model(user_feat, history_feat, item_feat)
loss = model.compute_loss(user_emb, item_emb)
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), config.GRAD_CLIP)
optimizer.step()
total_loss += loss.item()
num_batches += 1
pbar.set_postfix({"loss": f"{loss.item():.4f}"})
scheduler.step()
avg_loss = total_loss / max(num_batches, 1)
# Validation pass
model.eval()
val_loss = 0
val_batches = 0
with torch.no_grad():
for user_feat, history_feat, item_feat in val_loader:
user_feat = user_feat.to(device)
history_feat = history_feat.to(device)
item_feat = item_feat.to(device)
user_emb, item_emb = model(user_feat, history_feat, item_feat)
loss = model.compute_loss(user_emb, item_emb)
val_loss += loss.item()
val_batches += 1
avg_val_loss = val_loss / max(val_batches, 1)
print(f" Epoch {epoch+1}: train_loss={avg_loss:.4f} val_loss={avg_val_loss:.4f}")
# Save checkpoint if this is the best validation loss so far
if avg_val_loss < best_val_loss:
best_val_loss = avg_val_loss
torch.save(model.state_dict(), os.path.join(config.MODELS_DIR, "two_tower_best.pt"))
print(f" Saved best model (val_loss={best_val_loss:.4f})")
# Reload the best checkpoint
model.load_state_dict(torch.load(os.path.join(config.MODELS_DIR, "two_tower_best.pt"),
map_location=device))
print("Training complete! Best model loaded.")
return modelWhat does this code do?
- The training loop runs for 20 epochs on your MPS device (Apple Silicon GPU). Each epoch iterates over all (user, item) interaction pairs in batches of 512.
- Adam optimizer with learning rate 1e-3 and weight decay 1e-5 handles parameter updates. CosineAnnealingLR gradually reduces the learning rate following a cosine curve, helping the model converge smoothly.
- Gradient clipping at 1.0 prevents exploding gradients. After each epoch, the model is evaluated on the validation set. The best checkpoint (lowest validation loss) is saved to models/two_tower_best.pt.
You won't see output yet. The next chunk adds the embedding extraction and the script entry point.
- Below the train_model() function, add the extract_all_embeddings() function and the __main__ block by copying this code:
def extract_all_embeddings(model, data, fused_embeddings):
print("\nExtracting final embeddings...")
device = config.DEVICE
model.eval()
# Pass all items through the trained Item Tower
all_item_ids = list(fused_embeddings.keys())
item_features = np.array([fused_embeddings[aid] for aid in all_item_ids], dtype=np.float32)
item_tensor = torch.tensor(item_features).to(device)
item_embs = []
with torch.no_grad():
for i in range(0, len(item_tensor), config.BATCH_SIZE):
batch = item_tensor[i:i + config.BATCH_SIZE]
emb = model.item_tower(batch)
emb = nn.functional.normalize(emb, p=2, dim=1)
item_embs.append(emb.cpu().numpy())
item_embeddings = np.concatenate(item_embs, axis=0)
np.savez(os.path.join(config.EMBEDDINGS_DIR, "trained_item_embeddings.npz"),
ids=np.array(all_item_ids), vectors=item_embeddings)
# Pass all users through the trained User Tower (with full history, no dropout)
user_feat_dict = build_user_features_dict(data["users_df"], data["user_feature_cols"])
history_emb_dict = build_user_history_embeddings(data["user_history"], fused_embeddings)
all_user_ids = list(user_feat_dict.keys())
user_features = np.array([user_feat_dict[uid] for uid in all_user_ids], dtype=np.float32)
history_features = np.array([history_emb_dict.get(uid, np.zeros(config.MULTIMODAL_DIM, dtype=np.float32))
for uid in all_user_ids], dtype=np.float32)
user_embs = []
with torch.no_grad():
for i in range(0, len(user_features), config.BATCH_SIZE):
uf = torch.tensor(user_features[i:i + config.BATCH_SIZE]).to(device)
hf = torch.tensor(history_features[i:i + config.BATCH_SIZE]).to(device)
emb = model.user_tower(uf, hf)
emb = nn.functional.normalize(emb, p=2, dim=1)
user_embs.append(emb.cpu().numpy())
user_embeddings = np.concatenate(user_embs, axis=0)
np.savez(os.path.join(config.EMBEDDINGS_DIR, "trained_user_embeddings.npz"),
ids=np.array(all_user_ids), vectors=user_embeddings)
print(f" Saved {len(all_item_ids)} item and {len(all_user_ids)} user embeddings")
return dict(zip(all_item_ids, item_embeddings)), dict(zip(all_user_ids, user_embeddings))
if __name__ == "__main__":
from data_pipeline import prepare_data
from feature_engineering import compute_all_embeddings
data = prepare_data()
fused = compute_all_embeddings(data["items_df"], data["items_df"]["article_id"].values)
model = train_model(data, fused)
item_embs, user_embs = extract_all_embeddings(model, data, fused)What does this code do?
- extract_all_embeddings() passes every item through the trained Item Tower and every user through the trained User Tower to produce final 64-dim embeddings. During extraction, the model is in eval() mode, so DropoutNet is disabled and users get their full interaction history.
- The __main__ block wires the full pipeline together: load data, compute fused embeddings (cached from the earlier step), train the model, then extract embeddings. This is what runs when you execute the script.
- Save train_retrieval.py.
✔️ Awesome, I've got everything!
Double check you've saved train_retrieval.py before moving on.
ⓧ I'd like to double check the full code
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import numpy as np
from tqdm import tqdm
import os
import config
from models import TwoTowerModel
class InteractionDataset(Dataset):
def __init__(self, interactions_df, user_features_dict, item_embeddings_dict,
user_history_embeddings, user_feature_cols):
self.pairs = interactions_df[["customer_id", "article_id"]].values
self.user_features_dict = user_features_dict
self.item_embeddings_dict = item_embeddings_dict
self.user_history_embeddings = user_history_embeddings
self.user_feature_cols = user_feature_cols
def __len__(self):
return len(self.pairs)
def __getitem__(self, idx):
uid, aid = self.pairs[idx]
user_feat = self.user_features_dict.get(uid, np.zeros(len(self.user_feature_cols)))
item_feat = self.item_embeddings_dict.get(aid, np.zeros(config.MULTIMODAL_DIM))
history_feat = self.user_history_embeddings.get(uid, np.zeros(config.MULTIMODAL_DIM))
return (
torch.tensor(user_feat, dtype=torch.float32),
torch.tensor(history_feat, dtype=torch.float32),
torch.tensor(item_feat, dtype=torch.float32),
)
def build_user_features_dict(users_df, feature_cols):
result = {}
for _, row in users_df.iterrows():
result[row["customer_id"]] = row[feature_cols].values.astype(np.float32)
return result
def build_user_history_embeddings(user_history, item_embeddings_dict):
result = {}
for uid, item_ids in user_history.items():
embs = [item_embeddings_dict[aid] for aid in item_ids if aid in item_embeddings_dict]
if embs:
result[uid] = np.mean(embs, axis=0).astype(np.float32)
else:
result[uid] = np.zeros(config.MULTIMODAL_DIM, dtype=np.float32)
return result
def train_model(data, fused_embeddings):
print("\n=== Training Two-Tower Retrieval Model ===")
device = config.DEVICE
user_feat_dict = build_user_features_dict(data["users_df"], data["user_feature_cols"])
history_emb_dict = build_user_history_embeddings(data["user_history"], fused_embeddings)
train_dataset = InteractionDataset(
data["train"], user_feat_dict, fused_embeddings,
history_emb_dict, data["user_feature_cols"]
)
val_dataset = InteractionDataset(
data["val"], user_feat_dict, fused_embeddings,
history_emb_dict, data["user_feature_cols"]
)
train_loader = DataLoader(train_dataset, batch_size=config.BATCH_SIZE, shuffle=True,
num_workers=0, drop_last=True)
val_loader = DataLoader(val_dataset, batch_size=config.BATCH_SIZE, shuffle=False,
num_workers=0)
model = TwoTowerModel(
user_feature_dim=len(data["user_feature_cols"]),
history_dim=config.MULTIMODAL_DIM,
).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=config.LEARNING_RATE,
weight_decay=config.WEIGHT_DECAY)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=config.NUM_EPOCHS)
best_val_loss = float("inf")
for epoch in range(config.NUM_EPOCHS):
model.train()
total_loss = 0
num_batches = 0
pbar = tqdm(train_loader, desc=f"Epoch {epoch+1}/{config.NUM_EPOCHS}")
for user_feat, history_feat, item_feat in pbar:
user_feat = user_feat.to(device)
history_feat = history_feat.to(device)
item_feat = item_feat.to(device)
user_emb, item_emb = model(user_feat, history_feat, item_feat)
loss = model.compute_loss(user_emb, item_emb)
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), config.GRAD_CLIP)
optimizer.step()
total_loss += loss.item()
num_batches += 1
pbar.set_postfix({"loss": f"{loss.item():.4f}"})
scheduler.step()
avg_loss = total_loss / max(num_batches, 1)
model.eval()
val_loss = 0
val_batches = 0
with torch.no_grad():
for user_feat, history_feat, item_feat in val_loader:
user_feat = user_feat.to(device)
history_feat = history_feat.to(device)
item_feat = item_feat.to(device)
user_emb, item_emb = model(user_feat, history_feat, item_feat)
loss = model.compute_loss(user_emb, item_emb)
val_loss += loss.item()
val_batches += 1
avg_val_loss = val_loss / max(val_batches, 1)
print(f" Epoch {epoch+1}: train_loss={avg_loss:.4f} val_loss={avg_val_loss:.4f}")
if avg_val_loss < best_val_loss:
best_val_loss = avg_val_loss
torch.save(model.state_dict(), os.path.join(config.MODELS_DIR, "two_tower_best.pt"))
print(f" Saved best model (val_loss={best_val_loss:.4f})")
model.load_state_dict(torch.load(os.path.join(config.MODELS_DIR, "two_tower_best.pt"),
map_location=device))
print("Training complete! Best model loaded.")
return model
def extract_all_embeddings(model, data, fused_embeddings):
print("\nExtracting final embeddings...")
device = config.DEVICE
model.eval()
all_item_ids = list(fused_embeddings.keys())
item_features = np.array([fused_embeddings[aid] for aid in all_item_ids], dtype=np.float32)
item_tensor = torch.tensor(item_features).to(device)
item_embs = []
with torch.no_grad():
for i in range(0, len(item_tensor), config.BATCH_SIZE):
batch = item_tensor[i:i + config.BATCH_SIZE]
emb = model.item_tower(batch)
emb = nn.functional.normalize(emb, p=2, dim=1)
item_embs.append(emb.cpu().numpy())
item_embeddings = np.concatenate(item_embs, axis=0)
np.savez(os.path.join(config.EMBEDDINGS_DIR, "trained_item_embeddings.npz"),
ids=np.array(all_item_ids), vectors=item_embeddings)
user_feat_dict = build_user_features_dict(data["users_df"], data["user_feature_cols"])
history_emb_dict = build_user_history_embeddings(data["user_history"], fused_embeddings)
all_user_ids = list(user_feat_dict.keys())
user_features = np.array([user_feat_dict[uid] for uid in all_user_ids], dtype=np.float32)
history_features = np.array([history_emb_dict.get(uid, np.zeros(config.MULTIMODAL_DIM, dtype=np.float32))
for uid in all_user_ids], dtype=np.float32)
user_embs = []
with torch.no_grad():
for i in range(0, len(user_features), config.BATCH_SIZE):
uf = torch.tensor(user_features[i:i + config.BATCH_SIZE]).to(device)
hf = torch.tensor(history_features[i:i + config.BATCH_SIZE]).to(device)
emb = model.user_tower(uf, hf)
emb = nn.functional.normalize(emb, p=2, dim=1)
user_embs.append(emb.cpu().numpy())
user_embeddings = np.concatenate(user_embs, axis=0)
np.savez(os.path.join(config.EMBEDDINGS_DIR, "trained_user_embeddings.npz"),
ids=np.array(all_user_ids), vectors=user_embeddings)
print(f" Saved {len(all_item_ids)} item and {len(all_user_ids)} user embeddings")
return dict(zip(all_item_ids, item_embeddings)), dict(zip(all_user_ids, user_embeddings))
if __name__ == "__main__":
from data_pipeline import prepare_data
from feature_engineering import compute_all_embeddings
data = prepare_data()
fused = compute_all_embeddings(data["items_df"], data["items_df"]["article_id"].values)
model = train_model(data, fused)
item_embs, user_embs = extract_all_embeddings(model, data, fused)Now create the evaluation utilities that you will use to measure retrieval quality in the next step.
- Create a new file called evaluate.py in your hm_recommender folder.
- Add the evaluation metric functions by copying this code:
import numpy as np
from collections import defaultdict
def mean_average_precision_at_k(actual, predicted, k=12):
"""MAP@K: measures ranking quality of the top-K predictions."""
scores = []
for user_id in actual:
if user_id not in predicted:
continue
actual_items = set(actual[user_id])
pred_items = predicted[user_id][:k]
if not actual_items:
continue
score = 0.0
hits = 0
for i, item in enumerate(pred_items):
if item in actual_items:
hits += 1
score += hits / (i + 1)
scores.append(score / min(len(actual_items), k))
return np.mean(scores) if scores else 0.0
def recall_at_k(actual, predicted, k=50):
"""Recall@K: fraction of actual purchases found in the top-K."""
recalls = []
for user_id in actual:
if user_id not in predicted:
continue
actual_items = set(actual[user_id])
pred_items = set(predicted[user_id][:k])
if not actual_items:
continue
recalls.append(len(actual_items & pred_items) / len(actual_items))
return np.mean(recalls) if recalls else 0.0
def ndcg_at_k(actual, predicted, k=12):
"""NDCG@K: normalized discounted cumulative gain at K."""
scores = []
for user_id in actual:
if user_id not in predicted:
continue
actual_items = set(actual[user_id])
pred_items = predicted[user_id][:k]
if not actual_items:
continue
dcg = 0.0
for i, item in enumerate(pred_items):
if item in actual_items:
dcg += 1.0 / np.log2(i + 2)
ideal_hits = min(len(actual_items), k)
idcg = sum(1.0 / np.log2(i + 2) for i in range(ideal_hits))
scores.append(dcg / idcg if idcg > 0 else 0.0)
return np.mean(scores) if scores else 0.0
def build_ground_truth(test_df):
"""Convert a test DataFrame into a dict of user -> list of purchased article IDs."""
ground_truth = defaultdict(list)
for _, row in test_df.iterrows():
ground_truth[row["customer_id"]].append(row["article_id"])
return dict(ground_truth)
def evaluate_predictions(actual, predicted, label="Overall"):
"""Run all metrics and print a summary."""
map12 = mean_average_precision_at_k(actual, predicted, k=12)
r10 = recall_at_k(actual, predicted, k=10)
r50 = recall_at_k(actual, predicted, k=50)
ndcg12 = ndcg_at_k(actual, predicted, k=12)
print(f"\n=== {label} Metrics ===")
print(f" MAP@12: {map12:.4f}")
print(f" Recall@10: {r10:.4f}")
print(f" Recall@50: {r50:.4f}")
print(f" NDCG@12: {ndcg12:.4f}")
return {"MAP@12": map12, "Recall@10": r10, "Recall@50": r50, "NDCG@12": ndcg12}What do these metrics measure?
- MAP@12 (Mean Average Precision at 12) is the official H&M Kaggle competition metric. It rewards putting correct items earlier in the ranked list.
- Recall@K measures what fraction of items the user actually bought appear in your top-K predictions. Higher K catches more items but lowers the bar.
- NDCG@12 (Normalized Discounted Cumulative Gain) penalizes relevant items that appear lower in the list using a logarithmic discount. It captures ranking quality more granularly than MAP.
- Save evaluate.py.
- Verify the module loads by running this command:
python -c "from evaluate import evaluate_predictions; print('Evaluation utils ready')"You should see Evaluation utils ready printed.
Seeing a ModuleNotFoundError?
- Make sure evaluate.py is saved in the same hm_recommender folder as your other Python files.
- Confirm your terminal is running from inside hm_recommender, not its parent folder.
- Help me fix my evaluate.py import error.
✔️ Awesome, I've got everything!
Double check you've saved evaluate.py before moving on.
ⓧ I'd like to double check the full code
import numpy as np
from collections import defaultdict
def mean_average_precision_at_k(actual, predicted, k=12):
scores = []
for user_id in actual:
if user_id not in predicted:
continue
actual_items = set(actual[user_id])
pred_items = predicted[user_id][:k]
if not actual_items:
continue
score = 0.0
hits = 0
for i, item in enumerate(pred_items):
if item in actual_items:
hits += 1
score += hits / (i + 1)
scores.append(score / min(len(actual_items), k))
return np.mean(scores) if scores else 0.0
def recall_at_k(actual, predicted, k=50):
recalls = []
for user_id in actual:
if user_id not in predicted:
continue
actual_items = set(actual[user_id])
pred_items = set(predicted[user_id][:k])
if not actual_items:
continue
recalls.append(len(actual_items & pred_items) / len(actual_items))
return np.mean(recalls) if recalls else 0.0
def ndcg_at_k(actual, predicted, k=12):
scores = []
for user_id in actual:
if user_id not in predicted:
continue
actual_items = set(actual[user_id])
pred_items = predicted[user_id][:k]
if not actual_items:
continue
dcg = 0.0
for i, item in enumerate(pred_items):
if item in actual_items:
dcg += 1.0 / np.log2(i + 2)
ideal_hits = min(len(actual_items), k)
idcg = sum(1.0 / np.log2(i + 2) for i in range(ideal_hits))
scores.append(dcg / idcg if idcg > 0 else 0.0)
return np.mean(scores) if scores else 0.0
def build_ground_truth(test_df):
ground_truth = defaultdict(list)
for _, row in test_df.iterrows():
ground_truth[row["customer_id"]].append(row["article_id"])
return dict(ground_truth)
def evaluate_predictions(actual, predicted, label="Overall"):
map12 = mean_average_precision_at_k(actual, predicted, k=12)
r10 = recall_at_k(actual, predicted, k=10)
r50 = recall_at_k(actual, predicted, k=50)
ndcg12 = ndcg_at_k(actual, predicted, k=12)
print(f"\n=== {label} Metrics ===")
print(f" MAP@12: {map12:.4f}")
print(f" Recall@10: {r10:.4f}")
print(f" Recall@50: {r50:.4f}")
print(f" NDCG@12: {ndcg12:.4f}")
return {"MAP@12": map12, "Recall@10": r10, "Recall@50": r50, "NDCG@12": ndcg12}Train the model and extract embeddings
All three files are ready. Time to run the training pipeline and produce the trained embeddings that power the retrieval stage.
How long does training take?
On an M4 Mac with MPS acceleration, training 20 epochs typically takes 10-30 minutes depending on the number of active users and articles in your sampled data. The script first re-runs the data pipeline and loads cached embeddings, then begins the training loop.
- Start training by running this command:
python train_retrieval.pyYou should see progress bars for each epoch with a running loss value. After each epoch, the script prints the train and validation loss.
What should I see?
The training output shows epoch-by-epoch progress. The training loss should decrease over time. When the validation loss improves, you see a "Saved best model" message. After 20 epochs, the script automatically extracts 64-dim embeddings for all users and items.
Expect output like:
=== Training Two-Tower Retrieval Model ===
Epoch 1/20: 100%|████| 42/42 [00:05<00:00]
Epoch 1: train_loss=6.2148 val_loss=5.8921
Saved best model (val_loss=5.8921)
...
Epoch 20: train_loss=4.1203 val_loss=4.3567
Training complete! Best model loaded.
Extracting final embeddings...
Saved 8234 item and 12456 user embeddings- After training completes, verify the output files exist by running:
ls models/ embeddings/trained_*You should see models/two_tower_best.pt, embeddings/trained_item_embeddings.npz, and embeddings/trained_user_embeddings.npz in the listing.
Training crashed or stalled?
- If you see an MPS-related error, open config.py and temporarily change the device detection to force CPU: set DEVICE = torch.device("cpu"). Training will be slower but reliable.
- If you see an out-of-memory error, reduce BATCH_SIZE in config.py from 512 to 256 and run the script again.
- If the loss is nan from the start, double check that your fused embeddings were generated correctly. Rerun feature_engineering.py to regenerate them.
- My two-tower training is crashing. Help me debug.
Your two-tower model is trained and every user and item now has a compact 64-dimensional embedding. Next up, you will build a FAISS index for fast approximate nearest neighbor retrieval and measure how well the system finds relevant items.
Build the FAISS Index and Evaluate Retrieval
Your two-tower model has produced 64-dimensional embeddings for every user and every item. The retrieval problem is now a nearest neighbor search: given a user embedding, find the items whose embeddings are closest in the shared space.
Scanning every item embedding linearly would be too slow in production. FAISS (Facebook AI Similarity Search) solves this with an optimized index structure that returns the top-K most similar items in milliseconds. In this step, you will build that index, then validate retrieval quality on the test set and on two cold-start holdout sets.
In this step, get ready to:
- Build a FAISS approximate nearest neighbor index from the trained item embeddings.
- Evaluate retrieval quality (Recall@10, Recall@50, NDCG@12) on the test set.
- Evaluate cold-start performance for new items and new users.
Build the FAISS approximate nearest neighbor index
FAISS offers several index types optimized for different scales. For your sampled dataset (tens of thousands of items), IndexFlatIP performs exact inner product search. It is fast enough at this scale and avoids the complexity of approximate index tuning.
Why inner product instead of Euclidean distance?
Your two-tower model L2-normalizes its output embeddings. When vectors are unit-length, maximizing inner product (dot product) is mathematically equivalent to minimizing Euclidean distance. Inner product is slightly faster to compute, so it is the standard choice for normalized embeddings.
- Create a new file called faiss_index.py in your hm_recommender folder.
- Add the imports, the FAISSIndex class definition, and the build method by copying this code:
import faiss
import numpy as np
import os
import config
class FAISSIndex:
def __init__(self, embedding_dim=config.EMBEDDING_DIM):
# Create an exact inner product index for the given dimension
self.index = faiss.IndexFlatIP(embedding_dim)
self.item_ids = []
def build(self, item_ids, item_embeddings):
# Store the mapping from index position to article ID
self.item_ids = list(item_ids)
# Stack all item embeddings into a single matrix
vectors = np.array([item_embeddings[aid] for aid in self.item_ids], dtype=np.float32)
# L2-normalize so inner product equals cosine similarity
faiss.normalize_L2(vectors)
self.index.add(vectors)
print(f"FAISS index built with {self.index.ntotal} items")What does this code do?
The FAISSIndex class wraps a FAISS IndexFlatIP object. The __init__ method creates an empty index with the configured embedding dimension (64).
The build method takes a dictionary of article IDs to embedding vectors, stacks them into a matrix, L2-normalizes them, and adds them to the index. The item_ids list maps each index position back to the original article ID so you can translate FAISS results into real recommendations.
- Next, add the search and batch_search methods below the build method inside the FAISSIndex class:
def search(self, query_embedding, top_k=50):
# Reshape single query to 2D and normalize
query = np.array(query_embedding, dtype=np.float32).reshape(1, -1)
faiss.normalize_L2(query)
distances, indices = self.index.search(query, top_k)
# Map FAISS indices back to article IDs
results = []
for i, idx in enumerate(indices[0]):
if idx < len(self.item_ids):
results.append((self.item_ids[idx], float(distances[0][i])))
return results
def batch_search(self, query_embeddings, top_k=50):
# Process multiple queries at once for efficiency
queries = np.array(query_embeddings, dtype=np.float32)
faiss.normalize_L2(queries)
distances, indices = self.index.search(queries, top_k)
all_results = []
for i in range(len(queries)):
results = []
for j, idx in enumerate(indices[i]):
if idx < len(self.item_ids):
results.append((self.item_ids[idx], float(distances[i][j])))
all_results.append(results)
return all_resultsWhat does this code do?
The search method takes a single user embedding, normalizes it, and queries the index for the top-K most similar items. It returns a list of (article_id, similarity_score) tuples.
The batch_search method does the same for multiple users in a single call. FAISS is optimized for batch queries, so this is much faster than looping over individual searches when evaluating the full test set.
- Add the persistence methods and the convenience function below batch_search, still inside the class. Then add the standalone build_and_save_index function outside the class at the bottom of the file:
def save(self, path=None):
if path is None:
path = config.FAISS_INDEX_PATH
# Write the FAISS index binary to disk
faiss.write_index(self.index, path)
# Save the article ID mapping alongside the index
ids_path = path.replace(".bin", "_ids.npy")
np.save(ids_path, np.array(self.item_ids))
print(f"FAISS index saved to {path}")
def load(self, path=None):
if path is None:
path = config.FAISS_INDEX_PATH
# Read the FAISS index binary from disk
self.index = faiss.read_index(path)
ids_path = path.replace(".bin", "_ids.npy")
self.item_ids = list(np.load(ids_path, allow_pickle=True))
print(f"FAISS index loaded: {self.index.ntotal} items")
def build_and_save_index(item_embeddings):
# Convenience function to build and persist the index in one call
index = FAISSIndex()
index.build(list(item_embeddings.keys()), item_embeddings)
index.save()
return indexWhat does this code do?
The save method writes two files: the FAISS binary index (faiss_index.bin) and a NumPy array of article IDs (faiss_index_ids.npy). The load method reads both back to restore the full index.
The standalone build_and_save_index function wraps the build-then-save workflow into a single call. Other scripts (like the evaluation and ranking stages) will use this to create the index in one line.
- Save faiss_index.py.
✔️ Awesome, I've got everything!
Great. Double check you have saved the file before continuing.
ⓧ I'd like to double check the full code
import faiss
import numpy as np
import os
import config
class FAISSIndex:
def __init__(self, embedding_dim=config.EMBEDDING_DIM):
self.index = faiss.IndexFlatIP(embedding_dim)
self.item_ids = []
def build(self, item_ids, item_embeddings):
self.item_ids = list(item_ids)
vectors = np.array([item_embeddings[aid] for aid in self.item_ids], dtype=np.float32)
faiss.normalize_L2(vectors)
self.index.add(vectors)
print(f"FAISS index built with {self.index.ntotal} items")
def search(self, query_embedding, top_k=50):
query = np.array(query_embedding, dtype=np.float32).reshape(1, -1)
faiss.normalize_L2(query)
distances, indices = self.index.search(query, top_k)
results = []
for i, idx in enumerate(indices[0]):
if idx < len(self.item_ids):
results.append((self.item_ids[idx], float(distances[0][i])))
return results
def batch_search(self, query_embeddings, top_k=50):
queries = np.array(query_embeddings, dtype=np.float32)
faiss.normalize_L2(queries)
distances, indices = self.index.search(queries, top_k)
all_results = []
for i in range(len(queries)):
results = []
for j, idx in enumerate(indices[i]):
if idx < len(self.item_ids):
results.append((self.item_ids[idx], float(distances[i][j])))
all_results.append(results)
return all_results
def save(self, path=None):
if path is None:
path = config.FAISS_INDEX_PATH
faiss.write_index(self.index, path)
ids_path = path.replace(".bin", "_ids.npy")
np.save(ids_path, np.array(self.item_ids))
print(f"FAISS index saved to {path}")
def load(self, path=None):
if path is None:
path = config.FAISS_INDEX_PATH
self.index = faiss.read_index(path)
ids_path = path.replace(".bin", "_ids.npy")
self.item_ids = list(np.load(ids_path, allow_pickle=True))
print(f"FAISS index loaded: {self.index.ntotal} items")
def build_and_save_index(item_embeddings):
index = FAISSIndex()
index.build(list(item_embeddings.keys()), item_embeddings)
index.save()
return indexNow build and save the FAISS index from your trained item embeddings.
- Build the FAISS index by running this command in your terminal from inside the hm_recommender folder:
python -c "
import numpy as np
from faiss_index import build_and_save_index
data = np.load('embeddings/trained_item_embeddings.npz', allow_pickle=True)
item_embs = dict(zip(data['ids'], data['vectors']))
print(f'Loaded {len(item_embs)} item embeddings')
index = build_and_save_index(item_embs)
print('Done!')
"- You should see output confirming the index was built and saved. The number of items matches the number of active articles in your sampled dataset.
- Verify the index files exist by running:
ls artifacts/- You should see faiss_index.bin and faiss_index_ids.npy in the output.
Not seeing the index files?
Make sure you are running the command from inside the hm_recommender folder, not a parent directory. The artifacts/ directory is created relative to where you run the script.
If you see an ImportError for faiss, confirm that faiss-cpu is installed in your virtual environment by running pip list | grep faiss.
Still stuck? Help me debug why my FAISS index is not building or saving correctly.
Evaluate retrieval quality on the test set
Building the index is only half the story. You need to measure how well the retrieval stage surfaces items that users actually purchased in the held-out test week.
You already have an evaluate.py module with metric functions from a previous step. Now you will use those functions together with the FAISS index to compute Recall@10, Recall@50, and NDCG@12 on the test set.
What do these metrics mean?
- Recall@K measures what fraction of the user's actual purchases appear in the top-K retrieved candidates. Higher is better.
- NDCG@12 (Normalized Discounted Cumulative Gain) rewards relevant items that appear earlier in the ranked list. A perfect score means every purchased item is at the very top.
- MAP@12 (Mean Average Precision) is the primary competition metric from the H&M Kaggle challenge. It combines precision and recall at cutoff 12.
- Run the retrieval evaluation on your test set by running this command:
python -c "
import numpy as np
from collections import defaultdict
from data_pipeline import prepare_data
from faiss_index import FAISSIndex
from evaluate import build_ground_truth, evaluate_predictions
# Load data and embeddings
data = prepare_data()
user_data = np.load('embeddings/trained_user_embeddings.npz', allow_pickle=True)
user_embs = dict(zip(user_data['ids'], user_data['vectors']))
# Load the FAISS index
faiss_idx = FAISSIndex()
faiss_idx.load()
# Build ground truth from test set purchases
ground_truth = build_ground_truth(data['test'])
# Retrieve top-50 candidates for each test user
predictions = {}
for uid in ground_truth:
if uid in user_embs:
results = faiss_idx.search(user_embs[uid], top_k=50)
predictions[uid] = [item_id for item_id, score in results]
print(f'Evaluated {len(predictions)} users')
metrics = evaluate_predictions(ground_truth, predictions, label='Retrieval (Overall)')
"What does this script do?
- It loads the test set ground truth (the items each user actually purchased in week 5).
- For each test user, it queries the FAISS index with the user's trained embedding to retrieve the top-50 candidate items.
- It passes the predictions and ground truth to evaluate_predictions, which computes and prints MAP@12, Recall@10, Recall@50, and NDCG@12.
- You should see a metrics table printed to the terminal with values for MAP@12, Recall@10, Recall@50, and NDCG@12. The exact numbers depend on your data sample and training run.
What should my metrics look like?
Retrieval metrics for the two-tower model on a 5-week sample will be modest. Recall@50 in the range of 0.05 to 0.15 is reasonable at this stage. The retrieval model's job is not to be perfect on its own. It just needs to capture a good pool of candidates that the ranking model (coming next) can re-score with richer cross-features.
Seeing all zeros for metrics?
Check that your test set is not empty. If the data sampling filtered too aggressively, there may be no test-week transactions. Run print(len(data['test'])) to verify.
Also confirm that the user IDs in your trained embeddings match the user IDs in the test set. A mismatch means no predictions are generated.
Help me debug why my retrieval evaluation metrics are all zeros.
Evaluate cold-start performance
The whole point of the DropoutNet training heuristic was to handle cold-start scenarios. Now you will measure whether it actually works.
You have two cold-start holdout sets from the data pipeline: new items (articles that only appear in the test week) and new users (customers that only appear in the test week). You will evaluate each separately to quantify the cold-start gap.
- Run the cold-start evaluation by running this command:
python -c "
import numpy as np
import torch
from collections import defaultdict
from data_pipeline import prepare_data
from feature_engineering import load_embeddings
from faiss_index import FAISSIndex
from evaluate import build_ground_truth, evaluate_predictions, recall_at_k
from models import TwoTowerModel
from train_retrieval import build_user_features_dict, build_user_history_embeddings
import config
# Load data and artifacts
data = prepare_data()
faiss_idx = FAISSIndex()
faiss_idx.load()
user_data = np.load('embeddings/trained_user_embeddings.npz', allow_pickle=True)
user_embs = dict(zip(user_data['ids'], user_data['vectors']))
# === New-Item Cold-Start ===
# Check how many new items appear in top-50 retrievals
new_item_truth = build_ground_truth(data['test_new_items'])
new_item_preds = {}
for uid in new_item_truth:
if uid in user_embs:
results = faiss_idx.search(user_embs[uid], top_k=50)
new_item_preds[uid] = [item_id for item_id, score in results]
new_item_metrics = evaluate_predictions(new_item_truth, new_item_preds, label='New-Item Cold-Start')
# === New-User Cold-Start ===
# Re-run user tower with zeroed interaction history (demographics only)
fused_emb = load_embeddings('fused_embeddings.npz')
user_feat_dict = build_user_features_dict(data['users_df'], data['user_feature_cols'])
model = TwoTowerModel(user_feature_dim=len(data['user_feature_cols']), history_dim=config.MULTIMODAL_DIM)
model.load_state_dict(torch.load('models/two_tower_best.pt', map_location='cpu'))
model.eval()
new_user_truth = build_ground_truth(data['test_new_users'])
new_user_preds = {}
for uid in new_user_truth:
if uid in user_feat_dict:
uf = torch.tensor(user_feat_dict[uid], dtype=torch.float32).unsqueeze(0)
# Zero out history to simulate brand-new user
zero_hist = torch.zeros(1, config.MULTIMODAL_DIM)
emb = model.get_user_embedding(uf, zero_hist)
results = faiss_idx.search(emb.numpy()[0], top_k=50)
new_user_preds[uid] = [item_id for item_id, score in results]
new_user_metrics = evaluate_predictions(new_user_truth, new_user_preds, label='New-User Cold-Start (No History)')
# Compare with full-history baseline for context
full_history_preds = {}
for uid in new_user_truth:
if uid in user_embs:
results = faiss_idx.search(user_embs[uid], top_k=50)
full_history_preds[uid] = [item_id for item_id, score in results]
full_history_metrics = evaluate_predictions(new_user_truth, full_history_preds, label='New-User Baseline (Full History)')
print('\nCold-start evaluation complete!')
"What does this script do?
The script runs two separate cold-start evaluations.
- New-item cold-start checks whether articles with zero training history still appear in relevant users' top-50 retrievals. Because the Item Tower encodes visual and text features (not interaction history), new items get a meaningful position in the embedding space from day one.
- New-user cold-start re-runs the User Tower with zeroed interaction history, simulating a brand-new customer who has only demographic features. This is exactly the scenario DropoutNet prepared the model for during training. The script then compares these results against a hypothetical full-history baseline to quantify the gap.
- You should see three sets of metrics printed: new-item cold-start, new-user cold-start (no history), and new-user baseline (full history).
How do I interpret these cold-start results?
The gap between the new-user cold-start metrics and the full-history baseline shows how much retrieval quality degrades without purchase data. A small gap means DropoutNet successfully taught the model to rely on demographic features when history is unavailable.
For new items, any non-zero Recall means the visual and text embeddings are doing their job. New items are positioned meaningfully in the shared space without needing any interaction signal.
Seeing very few or zero new users/items in the cold-start sets?
The cold-start holdout sets depend on the data sampling window. If the 5-week sample is small, there may be very few articles or customers appearing only in the test week. This is expected with smaller samples.
If the new-user set is empty, it means all test-week customers also appeared in the training weeks. The cold-start metrics will show zeros because there are no new users to evaluate. This does not indicate a bug.
Help me understand why my cold-start holdout sets are empty or very small.
Your retrieval pipeline is now validated with real metrics. You know exactly how well the two-tower model surfaces relevant candidates, and you have cold-start numbers that prove the DropoutNet heuristic works. Next, you will build the ranking model that re-scores these candidates with richer features, including the COSMO intent embeddings.
Build the Ranking Model with MORS and COSMO Features
Your FAISS index retrieves 50 candidates per user based on embedding similarity. The retrieval stage prioritizes broad coverage, pulling in anything that looks roughly relevant.
Now you need a ranking model to precision-score those candidates using richer signals. In this step, you'll build a LightGBM ranking model that considers cross-features like category preferences, color matching, and COSMO intent alignment. You'll also implement Multi-Objective Ranking (MORS) to balance style relevance with eco-friendliness.
In this step, get ready to:
- Engineer ranking features including COSMO intent similarity.
- Implement the Multi-Objective Ranking Score (MORS) and the full reranking pipeline.
- Train two LightGBM rankers, run a COSMO ablation study, and evaluate the full pipeline end-to-end.
Engineer ranking features including COSMO intent features
The retrieval stage returns candidates ranked by a single signal: embedding dot-product similarity. The ranking stage enriches each candidate with cross-features that capture the relationship between a specific user and a specific item.
- Create a new file called train_ranker.py in your hm_recommender folder (alongside your other Python files).
- Add the imports and the eco-score helper function by pasting this code:
import numpy as np
import pandas as pd
import lightgbm as lgb
from sklearn.metrics.pairwise import cosine_similarity as sklearn_cosine
from collections import defaultdict
import os
import config
from evaluate import build_ground_truth
# Compute an eco-friendliness score from product metadata
def compute_eco_score(detail_desc, department_name):
text = f"{detail_desc} {department_name}".lower()
# Count how many eco keywords appear in the text
keyword_count = sum(1 for kw in config.ECO_KEYWORDS if kw in text)
# Normalize to [0, 1] with a cap at 3 keyword matches
return min(keyword_count / 3.0, 1.0)What does this code do?
The imports bring in LightGBM for the ranking model, scikit-learn's cosine similarity for COSMO intent matching, and your existing build_ground_truth function from evaluate.py.
The compute_eco_score() function scans a product's description and department name for eco-friendly keywords (organic, recycled, sustainable, etc.) defined in config.ECO_KEYWORDS. The count is normalized to a 0-1 range, capped at 3 matches. This score feeds into the MORS formula later.
- Save train_ranker.py.
Now you need the main feature engineering function. For each user-item pair, this function computes cross-features that capture how well a candidate item matches a specific user's preferences. The COSMO intent similarity feature is the key addition here.
- In train_ranker.py, add the engineer_ranking_features() function below compute_eco_score():
def engineer_ranking_features(user_id, candidate_items, retrieval_scores,
user_features_dict, items_df, train_df,
user_feature_cols, intent_embeddings=None,
user_history=None):
items_lookup = items_df.set_index("article_id").to_dict("index")
# Build user purchase history stats for category/color matching
user_txns = train_df[train_df["customer_id"] == user_id] if len(train_df) > 0 else pd.DataFrame()
user_categories = set()
user_colors = set()
if len(user_txns) > 0:
merged_cat = user_txns.merge(items_df[["article_id", "product_type_name"]], on="article_id", how="left")
user_categories = set(merged_cat["product_type_name"].dropna())
merged_col = user_txns.merge(items_df[["article_id", "colour_group_name"]], on="article_id", how="left")
user_colors = set(merged_col["colour_group_name"].dropna())
# Compute user intent profile from COSMO embeddings of purchased items
user_intent_profile = None
if intent_embeddings and user_history and user_id in user_history:
hist_intents = [intent_embeddings[aid] for aid in user_history[user_id]
if aid in intent_embeddings]
if hist_intents:
user_intent_profile = np.mean(hist_intents, axis=0).reshape(1, -1)
features = []
for item_id, score in zip(candidate_items, retrieval_scores):
item_info = items_lookup.get(item_id, {})
feat = {
# Retrieval similarity score from FAISS
"retrieval_score": score,
# Does this item's category match something the user bought before?
"category_match": 1.0 if item_info.get("product_type_name") in user_categories else 0.0,
# Does this item's color match something the user bought before?
"color_match": 1.0 if item_info.get("colour_group_name") in user_colors else 0.0,
# Eco-friendliness score from keyword matching
"eco_score": compute_eco_score(
str(item_info.get("detail_desc", "")),
str(item_info.get("department_name", ""))
),
}
# COSMO intent similarity: cosine between user intent profile and item intent
if user_intent_profile is not None and intent_embeddings and item_id in intent_embeddings:
item_intent = intent_embeddings[item_id].reshape(1, -1)
feat["intent_similarity"] = float(sklearn_cosine(user_intent_profile, item_intent)[0][0])
else:
feat["intent_similarity"] = 0.0
features.append(feat)
return pd.DataFrame(features)What does this code do?
This function builds a feature vector for each (user, candidate item) pair. It computes five features per candidate.
- retrieval_score: the raw FAISS similarity score from the retrieval stage.
- category_match and color_match: binary features indicating whether the candidate's product type or color appears in the user's purchase history.
- eco_score: the eco-friendliness score from the keyword matcher.
- intent_similarity: cosine similarity between the user's intent profile (average of COSMO intent embeddings from purchased items) and the candidate item's intent embedding. This is the COSMO-enhanced feature that captures why a user might want an item.
- Save train_ranker.py.
Seeing an import error?
If Python can't find sklearn, make sure your virtual environment is activated and that you installed scikit-learn in Step 1. Also confirm evaluate.py exists in the same directory as train_ranker.py.
Help me debug an import error in train_ranker.py
Implement MORS and the reranking pipeline
Before writing the training functions, you need two more pieces: the data preparation function that builds training pairs from your validation set, and the core training function that fits a LightGBM model.
- In train_ranker.py, add the prepare_ranking_data() function below engineer_ranking_features():
def prepare_ranking_data(data, faiss_index, user_embeddings, items_df,
user_features_dict, user_feature_cols,
intent_embeddings=None, user_history=None):
print("Preparing ranking training data...")
ground_truth = build_ground_truth(data["val"])
all_features = []
all_labels = []
all_groups = []
# Process up to 2000 users to keep training manageable
for user_id in list(ground_truth.keys())[:2000]:
if user_id not in user_embeddings:
continue
# Retrieve candidates for this user via FAISS
user_emb = user_embeddings[user_id]
candidates = faiss_index.search(user_emb, top_k=config.TOP_K_RETRIEVAL)
candidate_ids = [c[0] for c in candidates]
candidate_scores = [c[1] for c in candidates]
# Label: 1 if user actually purchased this item, 0 otherwise
actual_items = set(ground_truth[user_id])
labels = [1 if cid in actual_items else 0 for cid in candidate_ids]
# Compute ranking features for all candidates
feats = engineer_ranking_features(
user_id, candidate_ids, candidate_scores,
user_features_dict, items_df, data["train"], user_feature_cols,
intent_embeddings=intent_embeddings, user_history=user_history
)
all_features.append(feats)
all_labels.extend(labels)
all_groups.append(len(candidate_ids))
features_df = pd.concat(all_features, ignore_index=True)
return features_df, np.array(all_labels), all_groupsWhat does this code do?
This function builds the training dataset for the ranker. For up to 2,000 users in the validation set, it retrieves 50 FAISS candidates per user, then computes ranking features for each candidate.
Labels are binary: 1 if the user actually purchased the candidate item in the validation period, 0 otherwise. The all_groups list tracks how many candidates belong to each user. LightGBM's lambdarank objective uses these groups to learn pairwise preferences within each user's candidate list.
- Save train_ranker.py.
Next, add the function that trains the LightGBM ranker itself.
- In train_ranker.py, add the train_ranker() function below prepare_ranking_data():
def train_ranker(features_df, labels, groups, model_name="lgbm_ranker.txt"):
print(f"Training LightGBM ranker ({model_name})...")
train_data = lgb.Dataset(features_df, label=labels, group=groups)
# Lambdarank objective learns pairwise item ordering per user group
params = {
"objective": "lambdarank",
"metric": "ndcg",
"ndcg_eval_at": [12],
"learning_rate": 0.1,
"num_leaves": 31,
"min_data_in_leaf": 10,
"verbose": -1,
}
model = lgb.train(params, train_data, num_boost_round=100)
model_path = os.path.join(config.MODELS_DIR, model_name)
model.save_model(model_path)
print(f"Ranker saved to {model_path}")
return modelWhat does this code do?
The lambdarank objective is a learning-to-rank algorithm that optimizes NDCG directly. It learns which items should rank higher for each user by comparing pairs within a group.
The model_name parameter lets you save different models under different filenames. You'll use this to train one model with COSMO features and one without for the ablation study.
- Save train_ranker.py.
Why LightGBM instead of a neural ranker?
LightGBM with lambdarank is the industry standard for recommendation ranking stages. It was used by silver medalists in the H&M Kaggle competition. Gradient-boosted trees handle cross-features naturally, train in seconds, and produce interpretable feature importances.
A neural cross-encoder could capture deeper feature interactions, but it would add training complexity without a proportional quality gain at this dataset scale.
The final two functions implement the MORS formula and tie the full reranking pipeline together. MORS (Multi-Objective Ranking Score) blends the ranker's style-relevance prediction with the eco-friendliness score using a tunable alpha parameter.
- In train_ranker.py, add the apply_mors() and rerank_candidates() functions below train_ranker():
# MORS formula: blend style relevance with eco-friendliness
def apply_mors(style_scores, eco_scores, alpha=config.MORS_ALPHA):
return alpha * style_scores + (1 - alpha) * eco_scores
def rerank_candidates(ranker_model, user_id, candidates, retrieval_scores,
user_features_dict, items_df, train_df, user_feature_cols,
alpha=config.MORS_ALPHA, intent_embeddings=None, user_history=None):
# Compute ranking features for this user's candidates
feats = engineer_ranking_features(
user_id, candidates, retrieval_scores,
user_features_dict, items_df, train_df, user_feature_cols,
intent_embeddings=intent_embeddings, user_history=user_history
)
# Get LightGBM style predictions and eco scores
style_scores = ranker_model.predict(feats)
eco_scores = feats["eco_score"].values
# Combine with MORS
final_scores = apply_mors(style_scores, eco_scores, alpha)
# Sort by final score descending
ranked = sorted(zip(candidates, final_scores), key=lambda x: -x[1])
return [(item_id, score) for item_id, score in ranked]What does this code do?
The apply_mors() function implements the formula: Final = (alpha * S) + ((1 - alpha) * E). With the default alpha of 0.8, the score is 80% style relevance and 20% eco-friendliness.
The rerank_candidates() function runs the full ranking pipeline for one user: compute features, predict with LightGBM, blend with MORS, and return items sorted by final score. This is the function the Gradio app will call at inference time.
- Save train_ranker.py.
Not sure where to place each function?
The function order in your file must be: compute_eco_score, engineer_ranking_features, prepare_ranking_data, train_ranker, apply_mors, rerank_candidates. Functions call each other, so order matters. Use the full-file double check below to compare.
Help me check my train_ranker.py structure
✔️ Awesome, I've got everything!
Double check you've saved train_ranker.py with all six functions.
ⓧ I'd like to double check the full code
Here is the complete train_ranker.py file. Compare yours against this:
import numpy as np
import pandas as pd
import lightgbm as lgb
from sklearn.metrics.pairwise import cosine_similarity as sklearn_cosine
from collections import defaultdict
import os
import config
from evaluate import build_ground_truth
def compute_eco_score(detail_desc, department_name):
text = f"{detail_desc} {department_name}".lower()
keyword_count = sum(1 for kw in config.ECO_KEYWORDS if kw in text)
return min(keyword_count / 3.0, 1.0)
def engineer_ranking_features(user_id, candidate_items, retrieval_scores,
user_features_dict, items_df, train_df,
user_feature_cols, intent_embeddings=None,
user_history=None):
items_lookup = items_df.set_index("article_id").to_dict("index")
# User purchase history stats
user_txns = train_df[train_df["customer_id"] == user_id] if len(train_df) > 0 else pd.DataFrame()
user_categories = set()
user_colors = set()
if len(user_txns) > 0:
merged_cat = user_txns.merge(items_df[["article_id", "product_type_name"]], on="article_id", how="left")
user_categories = set(merged_cat["product_type_name"].dropna())
merged_col = user_txns.merge(items_df[["article_id", "colour_group_name"]], on="article_id", how="left")
user_colors = set(merged_col["colour_group_name"].dropna())
# Compute user intent profile if intent embeddings available
user_intent_profile = None
if intent_embeddings and user_history and user_id in user_history:
hist_intents = [intent_embeddings[aid] for aid in user_history[user_id]
if aid in intent_embeddings]
if hist_intents:
user_intent_profile = np.mean(hist_intents, axis=0).reshape(1, -1)
features = []
for item_id, score in zip(candidate_items, retrieval_scores):
item_info = items_lookup.get(item_id, {})
feat = {
"retrieval_score": score,
"category_match": 1.0 if item_info.get("product_type_name") in user_categories else 0.0,
"color_match": 1.0 if item_info.get("colour_group_name") in user_colors else 0.0,
"eco_score": compute_eco_score(
str(item_info.get("detail_desc", "")),
str(item_info.get("department_name", ""))
),
}
# COSMO intent similarity feature
if user_intent_profile is not None and intent_embeddings and item_id in intent_embeddings:
item_intent = intent_embeddings[item_id].reshape(1, -1)
feat["intent_similarity"] = float(sklearn_cosine(user_intent_profile, item_intent)[0][0])
else:
feat["intent_similarity"] = 0.0
features.append(feat)
return pd.DataFrame(features)
def prepare_ranking_data(data, faiss_index, user_embeddings, items_df,
user_features_dict, user_feature_cols,
intent_embeddings=None, user_history=None):
print("Preparing ranking training data...")
ground_truth = build_ground_truth(data["val"])
all_features = []
all_labels = []
all_groups = []
for user_id in list(ground_truth.keys())[:2000]:
if user_id not in user_embeddings:
continue
user_emb = user_embeddings[user_id]
candidates = faiss_index.search(user_emb, top_k=config.TOP_K_RETRIEVAL)
candidate_ids = [c[0] for c in candidates]
candidate_scores = [c[1] for c in candidates]
actual_items = set(ground_truth[user_id])
labels = [1 if cid in actual_items else 0 for cid in candidate_ids]
feats = engineer_ranking_features(
user_id, candidate_ids, candidate_scores,
user_features_dict, items_df, data["train"], user_feature_cols,
intent_embeddings=intent_embeddings, user_history=user_history
)
all_features.append(feats)
all_labels.extend(labels)
all_groups.append(len(candidate_ids))
features_df = pd.concat(all_features, ignore_index=True)
return features_df, np.array(all_labels), all_groups
def train_ranker(features_df, labels, groups, model_name="lgbm_ranker.txt"):
print(f"Training LightGBM ranker ({model_name})...")
train_data = lgb.Dataset(features_df, label=labels, group=groups)
params = {
"objective": "lambdarank",
"metric": "ndcg",
"ndcg_eval_at": [12],
"learning_rate": 0.1,
"num_leaves": 31,
"min_data_in_leaf": 10,
"verbose": -1,
}
model = lgb.train(params, train_data, num_boost_round=100)
model_path = os.path.join(config.MODELS_DIR, model_name)
model.save_model(model_path)
print(f"Ranker saved to {model_path}")
return model
def apply_mors(style_scores, eco_scores, alpha=config.MORS_ALPHA):
return alpha * style_scores + (1 - alpha) * eco_scores
def rerank_candidates(ranker_model, user_id, candidates, retrieval_scores,
user_features_dict, items_df, train_df, user_feature_cols,
alpha=config.MORS_ALPHA, intent_embeddings=None, user_history=None):
feats = engineer_ranking_features(
user_id, candidates, retrieval_scores,
user_features_dict, items_df, train_df, user_feature_cols,
intent_embeddings=intent_embeddings, user_history=user_history
)
style_scores = ranker_model.predict(feats)
eco_scores = feats["eco_score"].values
final_scores = apply_mors(style_scores, eco_scores, alpha)
ranked = sorted(zip(candidates, final_scores), key=lambda x: -x[1])
return [(item_id, score) for item_id, score in ranked]Train the LightGBM ranker, run ablation, and evaluate end-to-end
With all the ranking functions written, you can now train two models and compare them. The ablation study trains one ranker with the COSMO intent_similarity feature and one without it. This quantifies exactly how much the LLM-generated knowledge improves ranking quality.
- Run the full training, ablation, and evaluation pipeline by running this script in your terminal from the hm_recommender folder:
python -c "
import numpy as np
import pandas as pd
from data_pipeline import prepare_data
from feature_engineering import load_embeddings
from faiss_index import FAISSIndex
from train_ranker import prepare_ranking_data, train_ranker, rerank_candidates, build_user_history_embeddings, apply_mors
from train_retrieval import build_user_features_dict
from evaluate import build_ground_truth, evaluate_predictions
import config
# Load data and artifacts
print('Loading data...')
data = prepare_data()
user_feat_dict = build_user_features_dict(data['users_df'], data['user_feature_cols'])
# Load trained embeddings
user_data = np.load('embeddings/trained_user_embeddings.npz', allow_pickle=True)
user_emb_dict = dict(zip(user_data['ids'], user_data['vectors']))
# Load intent embeddings
intent_data = np.load('embeddings/intent_embeddings.npz', allow_pickle=True)
intent_emb_dict = dict(zip(intent_data['ids'], intent_data['vectors']))
# Load FAISS index
faiss_idx = FAISSIndex()
faiss_idx.load()
# === Train WITH COSMO features ===
print('\n=== Training ranker WITH COSMO features ===')
feats_cosmo, labels_cosmo, groups_cosmo = prepare_ranking_data(
data, faiss_idx, user_emb_dict, data['items_df'],
user_feat_dict, data['user_feature_cols'],
intent_embeddings=intent_emb_dict, user_history=data['user_history']
)
ranker_cosmo = train_ranker(feats_cosmo, labels_cosmo, groups_cosmo, 'lgbm_ranker.txt')
# === Train WITHOUT COSMO features (ablation) ===
print('\n=== Training ranker WITHOUT COSMO features ===')
feats_no_cosmo = feats_cosmo.drop(columns=['intent_similarity'])
ranker_no_cosmo = train_ranker(feats_no_cosmo, labels_cosmo, groups_cosmo, 'lgbm_ranker_no_cosmo.txt')
# === Evaluate end-to-end on test set ===
print('\n=== Evaluating full pipeline ===')
ground_truth = build_ground_truth(data['test'])
all_metrics = []
# Evaluate with COSMO ranker
predictions_cosmo = {}
for uid in list(ground_truth.keys())[:1000]:
if uid not in user_emb_dict:
continue
candidates = faiss_idx.search(user_emb_dict[uid], top_k=config.TOP_K_RETRIEVAL)
cids = [c[0] for c in candidates]
cscores = [c[1] for c in candidates]
reranked = rerank_candidates(
ranker_cosmo, uid, cids, cscores,
user_feat_dict, data['items_df'], data['train'], data['user_feature_cols'],
intent_embeddings=intent_emb_dict, user_history=data['user_history']
)
predictions_cosmo[uid] = [r[0] for r in reranked]
metrics_cosmo = evaluate_predictions(ground_truth, predictions_cosmo, 'With COSMO')
metrics_cosmo['config'] = 'with_cosmo'
all_metrics.append(metrics_cosmo)
# Evaluate without COSMO ranker
predictions_no_cosmo = {}
for uid in list(ground_truth.keys())[:1000]:
if uid not in user_emb_dict:
continue
candidates = faiss_idx.search(user_emb_dict[uid], top_k=config.TOP_K_RETRIEVAL)
cids = [c[0] for c in candidates]
cscores = [c[1] for c in candidates]
reranked = rerank_candidates(
ranker_no_cosmo, uid, cids, cscores,
user_feat_dict, data['items_df'], data['train'], data['user_feature_cols']
)
predictions_no_cosmo[uid] = [r[0] for r in reranked]
metrics_no_cosmo = evaluate_predictions(ground_truth, predictions_no_cosmo, 'Without COSMO')
metrics_no_cosmo['config'] = 'without_cosmo'
all_metrics.append(metrics_no_cosmo)
# Cold-start evaluations
gt_new_items = build_ground_truth(data['test_new_items'])
if gt_new_items:
m = evaluate_predictions(gt_new_items, predictions_cosmo, 'New Items Cold-Start')
m['config'] = 'new_items'
all_metrics.append(m)
gt_new_users = build_ground_truth(data['test_new_users'])
if gt_new_users:
m = evaluate_predictions(gt_new_users, predictions_cosmo, 'New Users Cold-Start')
m['config'] = 'new_users'
all_metrics.append(m)
# Save all metrics to CSV
metrics_df = pd.DataFrame(all_metrics)
metrics_df.to_csv('artifacts/evaluation_metrics.csv', index=False)
print('\nAll metrics saved to artifacts/evaluation_metrics.csv')
print(metrics_df.to_string())
"What does this script do?
This script runs the full ranking pipeline in four stages.
- It trains a LightGBM ranker WITH the COSMO intent_similarity feature, saved as lgbm_ranker.txt.
- It trains a second ranker WITHOUT COSMO features (the ablation baseline), saved as lgbm_ranker_no_cosmo.txt.
- It evaluates both rankers end-to-end (retrieve with FAISS, rerank with LightGBM + MORS) on the test set and the cold-start holdout sets.
- It saves all metrics (MAP@12, Recall@10, Recall@50, NDCG@12) to artifacts/evaluation_metrics.csv.
This takes a few minutes
Preparing ranking data requires FAISS lookups for up to 2,000 users. The LightGBM training itself is fast (seconds), but the feature engineering loop takes a few minutes depending on your dataset size.
- Wait for the script to finish. You should see evaluation metrics printed for each configuration: "With COSMO", "Without COSMO", "New Items Cold-Start", and "New Users Cold-Start".
- Confirm the trained models exist by running:
ls models/You should see lgbm_ranker.txt, lgbm_ranker_no_cosmo.txt, and two_tower_best.pt listed.
- Confirm the evaluation metrics CSV was created by running:
ls artifacts/evaluation_metrics.csvYou should see the file listed. The CSV contains your overall metrics, cold-start metrics, and the COSMO ablation comparison.
Seeing a LightGBM error or empty metrics?
If LightGBM throws an error about empty datasets, your FAISS retrievals may not be matching any validation-set purchases. Make sure you ran retrieval training and FAISS index building in the previous steps. Check that artifacts/faiss_index.bin exists.
If the cold-start metrics show all zeros, that can happen if there are no new items or new users in your sampled test week. The overall and ablation metrics are the most important results.
Help me debug the ranking training or evaluation pipeline
What is the ablation study showing?
The ablation compares two rankers that are identical except for one feature: intent_similarity. If the "With COSMO" metrics are higher than "Without COSMO", it proves the LLM-generated commonsense knowledge adds real value to the ranking.
This is the same experimental methodology Amazon used to validate their COSMO system, which showed a 0.7% sales uplift on 10% of US traffic. Your numbers will be different (smaller dataset, smaller LLM), but the structure of the experiment is research-paper-grade.
Your full two-stage pipeline is now trained and evaluated. Next up, you'll build an interactive Gradio UI that lets users explore recommendations, adjust the MORS alpha slider in real time, and see the COSMO intent explanations for each recommendation.
Build and Deploy the Gradio App to Hugging Face Spaces
Your two-stage pipeline is fully trained and evaluated. You have a FAISS index for fast candidate retrieval, a LightGBM ranker for precision scoring, and COSMO intent embeddings that capture purchase motivation.
Now it is time to make this pipeline interactive and shareable. In this step, you will build a Gradio web app that ties the full retrieval-ranking pipeline together, then deploy it to Hugging Face Spaces so anyone with a URL can see your recommender in action.
In this step, get ready to:
- Create an interactive Gradio app with a recommendations tab and a metrics dashboard.
- Deploy the app and all model artifacts to Hugging Face Spaces.
- Test the live app and get your public URL for your portfolio.
Create the interactive Gradio app
The Gradio app is the frontend for your entire recommendation pipeline. It loads all your pre-computed artifacts at startup, then serves real-time recommendations through a clean web interface with MORS scoring and COSMO intent explanations.
You will build app/app.py in three chunks: artifact loading, the recommendation function, and the Gradio UI definition.
- In app/app.py, add the imports and artifact loading code:
import gradio as gr
import numpy as np
import pandas as pd
import json
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import config
from faiss_index import FAISSIndex
import lightgbm as lgb
from train_ranker import engineer_ranking_features, apply_mors
# Load artifacts
print("Loading artifacts...")
item_data = np.load(os.path.join(config.EMBEDDINGS_DIR, "trained_item_embeddings.npz"),
allow_pickle=True)
item_emb_dict = dict(zip(item_data["ids"], item_data["vectors"]))
user_data = np.load(os.path.join(config.EMBEDDINGS_DIR, "trained_user_embeddings.npz"),
allow_pickle=True)
user_emb_dict = dict(zip(user_data["ids"], user_data["vectors"]))
# Load intent embeddings
intent_path = os.path.join(config.EMBEDDINGS_DIR, "intent_embeddings.npz")
intent_emb_dict = {}
if os.path.exists(intent_path):
intent_data = np.load(intent_path, allow_pickle=True)
intent_emb_dict = dict(zip(intent_data["ids"], intent_data["vectors"]))
# Load COSMO knowledge for display
cosmo_knowledge = {}
if os.path.exists(config.COSMO_TRIPLES_PATH):
with open(config.COSMO_TRIPLES_PATH, "r") as f:
cosmo_knowledge = json.load(f)
articles = pd.read_csv(config.ARTICLES_PATH, dtype={"article_id": str})
items_df = articles[["article_id", "prod_name", "product_type_name",
"colour_group_name", "department_name", "section_name",
"garment_group_name", "detail_desc"]].copy()
items_df["detail_desc"] = items_df["detail_desc"].fillna("")
faiss_idx = FAISSIndex()
faiss_idx.load()
ranker = lgb.Booster(model_file=os.path.join(config.MODELS_DIR, "lgbm_ranker.txt"))
metrics_path = os.path.join(config.ARTIFACTS_DIR, "evaluation_metrics.csv")
metrics_df = pd.read_csv(metrics_path) if os.path.exists(metrics_path) else None
user_ids = list(user_emb_dict.keys())What does this code do?
The sys.path.insert line adds the parent directory to Python's module search path so the app can import config, faiss_index, and train_ranker from the project root.
At startup, the app loads six artifacts into memory: item embeddings, user embeddings, intent embeddings, COSMO knowledge triples, the FAISS index, and the LightGBM ranker. It also loads the evaluation metrics CSV for the dashboard tab. All of these are pre-computed, so the app never runs the LLM or trains anything at serving time.
- Save app/app.py.
You will not see any output yet. The next chunk adds the function that generates recommendations.
- Below the artifact loading code in app/app.py, add the get_recommendations() function:
def get_recommendations(user_id, alpha, num_results):
if user_id not in user_emb_dict:
return "User ID not found."
# Retrieve candidates from FAISS
user_emb = user_emb_dict[user_id]
candidates = faiss_idx.search(user_emb, top_k=config.TOP_K_RETRIEVAL)
candidate_ids = [c[0] for c in candidates]
candidate_scores = [c[1] for c in candidates]
# Compute ranking features including COSMO intent similarity
feats = engineer_ranking_features(
user_id, candidate_ids, candidate_scores,
{}, items_df, pd.DataFrame(), [],
intent_embeddings=intent_emb_dict
)
# Apply LightGBM ranking + MORS scoring
style_scores = ranker.predict(feats)
eco_scores = feats["eco_score"].values
final_scores = apply_mors(style_scores, eco_scores, alpha)
# Sort by final score and take top results
ranked = sorted(zip(candidate_ids, final_scores), key=lambda x: -x[1])
top_items = ranked[:int(num_results)]
# Build the results table with COSMO explanations
results = []
for item_id, score in top_items:
item_info = items_df[items_df["article_id"] == item_id]
if not item_info.empty:
row = item_info.iloc[0]
# Get COSMO intent explanation
intent_text = ""
if item_id in cosmo_knowledge:
triples = cosmo_knowledge[item_id]
intent_text = "; ".join([f"{t['relation']}: {t['value']}" for t in triples[:3]])
results.append({
"Article ID": item_id,
"Product": row.get("prod_name", "Unknown"),
"Type": row.get("product_type_name", ""),
"Color": row.get("colour_group_name", ""),
"Score": f"{score:.3f}",
"Why (COSMO Intent)": intent_text,
})
return pd.DataFrame(results)What does this function do?
This function runs the full two-stage pipeline in real time. First, it retrieves the top 50 candidates from FAISS using the user's embedding. Then it computes ranking features (including COSMO intent similarity) and scores them with the LightGBM ranker.
The apply_mors() call blends the style relevance score with the eco-friendliness score using the alpha parameter from the slider. For each recommended item, the function also pulls the COSMO knowledge triples to explain *why* that item was suggested.
- Save app/app.py.
You still will not see a change yet. The next chunk builds the actual UI that calls this function.
- Below get_recommendations() in app/app.py, add the metrics helper and the Gradio Blocks UI:
def get_metrics_display():
if metrics_df is not None:
return metrics_df.to_html(index=False)
return "No metrics available. Run evaluation first."
# Build the Gradio Blocks interface
with gr.Blocks(theme=gr.themes.Soft(), title="H&M Fashion Recommender") as demo:
gr.Markdown("# H&M Two-Stage Fashion Recommender with LLM-Enhanced Understanding")
gr.Markdown("Production-style recommendation system with two-tower retrieval, "
"COSMO-inspired LLM knowledge (Phi-4-mini), LightGBM ranking, "
"and MORS multi-objective scoring.")
with gr.Tabs():
# Tab 1: Interactive recommendations
with gr.TabItem("Recommendations"):
with gr.Row():
user_input = gr.Dropdown(choices=user_ids[:500], label="Select Customer",
info="Choose a customer ID")
alpha_slider = gr.Slider(0.0, 1.0, value=0.8, step=0.05,
label="MORS Alpha (Style vs Eco)",
info="1.0 = pure style, 0.0 = pure eco")
num_slider = gr.Slider(1, 20, value=12, step=1,
label="Number of Results")
rec_button = gr.Button("Get Recommendations", variant="primary")
rec_output = gr.Dataframe(label="Recommended Items")
rec_button.click(
fn=get_recommendations,
inputs=[user_input, alpha_slider, num_slider],
outputs=rec_output,
)
# Tab 2: Evaluation metrics dashboard
with gr.TabItem("Metrics Dashboard"):
gr.Markdown("## Evaluation Results")
gr.Markdown("Includes overall metrics, cold-start evaluation, and COSMO ablation study.")
metrics_html = gr.HTML(value=get_metrics_display())
gr.Markdown("---")
gr.Markdown("Built with PyTorch, FAISS, LightGBM, Phi-4-mini (COSMO knowledge), and Gradio. "
"Two-tower retrieval + LLM-enhanced ranking architecture.")
if __name__ == "__main__":
demo.launch()What does this UI do?
The gr.Blocks layout creates two tabs. The **Recommendations** tab has a customer dropdown (limited to 500 users for performance), a MORS alpha slider to control the style-vs-sustainability tradeoff, and a results slider. Clicking **Get Recommendations** runs the full pipeline and displays a table with product details and COSMO intent explanations.
The **Metrics Dashboard** tab renders your evaluation CSV as an HTML table. This is what makes the demo recruiter-ready and research-paper-ready.
- Save app/app.py.
- Test the app locally by running this command from your hm_recommender folder:
python app/app.pyYou should see output ending with a local URL like http://127.0.0.1:7860. Open that URL in your browser to see the Gradio interface.
App not loading?
If you see a ModuleNotFoundError for config or faiss_index, make sure you are running the command from the hm_recommender root folder, not from inside the app/ directory.
If the page loads but shows no data in the dropdown, confirm that embeddings/trained_user_embeddings.npz exists. That file is created in the retrieval training step.
Help me debug my Gradio app not loading
✔️ Awesome, I've got everything!
Great. Double check you have saved app/app.py and that it runs locally before moving on to deployment.
ⓧ I'd like to double check the full code
Here is the complete app/app.py file for comparison:
import gradio as gr
import numpy as np
import pandas as pd
import json
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import config
from faiss_index import FAISSIndex
import lightgbm as lgb
from train_ranker import engineer_ranking_features, apply_mors
# Load artifacts
print("Loading artifacts...")
item_data = np.load(os.path.join(config.EMBEDDINGS_DIR, "trained_item_embeddings.npz"),
allow_pickle=True)
item_emb_dict = dict(zip(item_data["ids"], item_data["vectors"]))
user_data = np.load(os.path.join(config.EMBEDDINGS_DIR, "trained_user_embeddings.npz"),
allow_pickle=True)
user_emb_dict = dict(zip(user_data["ids"], user_data["vectors"]))
# Load intent embeddings
intent_path = os.path.join(config.EMBEDDINGS_DIR, "intent_embeddings.npz")
intent_emb_dict = {}
if os.path.exists(intent_path):
intent_data = np.load(intent_path, allow_pickle=True)
intent_emb_dict = dict(zip(intent_data["ids"], intent_data["vectors"]))
# Load COSMO knowledge for display
cosmo_knowledge = {}
if os.path.exists(config.COSMO_TRIPLES_PATH):
with open(config.COSMO_TRIPLES_PATH, "r") as f:
cosmo_knowledge = json.load(f)
articles = pd.read_csv(config.ARTICLES_PATH, dtype={"article_id": str})
items_df = articles[["article_id", "prod_name", "product_type_name",
"colour_group_name", "department_name", "section_name",
"garment_group_name", "detail_desc"]].copy()
items_df["detail_desc"] = items_df["detail_desc"].fillna("")
faiss_idx = FAISSIndex()
faiss_idx.load()
ranker = lgb.Booster(model_file=os.path.join(config.MODELS_DIR, "lgbm_ranker.txt"))
metrics_path = os.path.join(config.ARTIFACTS_DIR, "evaluation_metrics.csv")
metrics_df = pd.read_csv(metrics_path) if os.path.exists(metrics_path) else None
user_ids = list(user_emb_dict.keys())
def get_recommendations(user_id, alpha, num_results):
if user_id not in user_emb_dict:
return "User ID not found."
user_emb = user_emb_dict[user_id]
candidates = faiss_idx.search(user_emb, top_k=config.TOP_K_RETRIEVAL)
candidate_ids = [c[0] for c in candidates]
candidate_scores = [c[1] for c in candidates]
feats = engineer_ranking_features(
user_id, candidate_ids, candidate_scores,
{}, items_df, pd.DataFrame(), [],
intent_embeddings=intent_emb_dict
)
style_scores = ranker.predict(feats)
eco_scores = feats["eco_score"].values
final_scores = apply_mors(style_scores, eco_scores, alpha)
ranked = sorted(zip(candidate_ids, final_scores), key=lambda x: -x[1])
top_items = ranked[:int(num_results)]
results = []
for item_id, score in top_items:
item_info = items_df[items_df["article_id"] == item_id]
if not item_info.empty:
row = item_info.iloc[0]
# Get COSMO intent explanation
intent_text = ""
if item_id in cosmo_knowledge:
triples = cosmo_knowledge[item_id]
intent_text = "; ".join([f"{t['relation']}: {t['value']}" for t in triples[:3]])
results.append({
"Article ID": item_id,
"Product": row.get("prod_name", "Unknown"),
"Type": row.get("product_type_name", ""),
"Color": row.get("colour_group_name", ""),
"Score": f"{score:.3f}",
"Why (COSMO Intent)": intent_text,
})
return pd.DataFrame(results)
def get_metrics_display():
if metrics_df is not None:
return metrics_df.to_html(index=False)
return "No metrics available. Run evaluation first."
with gr.Blocks(theme=gr.themes.Soft(), title="H&M Fashion Recommender") as demo:
gr.Markdown("# H&M Two-Stage Fashion Recommender with LLM-Enhanced Understanding")
gr.Markdown("Production-style recommendation system with two-tower retrieval, "
"COSMO-inspired LLM knowledge (Phi-4-mini), LightGBM ranking, "
"and MORS multi-objective scoring.")
with gr.Tabs():
with gr.TabItem("Recommendations"):
with gr.Row():
user_input = gr.Dropdown(choices=user_ids[:500], label="Select Customer",
info="Choose a customer ID")
alpha_slider = gr.Slider(0.0, 1.0, value=0.8, step=0.05,
label="MORS Alpha (Style vs Eco)",
info="1.0 = pure style, 0.0 = pure eco")
num_slider = gr.Slider(1, 20, value=12, step=1,
label="Number of Results")
rec_button = gr.Button("Get Recommendations", variant="primary")
rec_output = gr.Dataframe(label="Recommended Items")
rec_button.click(
fn=get_recommendations,
inputs=[user_input, alpha_slider, num_slider],
outputs=rec_output,
)
with gr.TabItem("Metrics Dashboard"):
gr.Markdown("## Evaluation Results")
gr.Markdown("Includes overall metrics, cold-start evaluation, and COSMO ablation study.")
metrics_html = gr.HTML(value=get_metrics_display())
gr.Markdown("---")
gr.Markdown("Built with PyTorch, FAISS, LightGBM, Phi-4-mini (COSMO knowledge), and Gradio. "
"Two-tower retrieval + LLM-enhanced ranking architecture.")
if __name__ == "__main__":
demo.launch()Deploy to Hugging Face Spaces
Hugging Face Spaces hosts Gradio apps for free on a CPU tier with 16 GB RAM and 2 vCPUs. Your app only serves pre-computed embeddings and model inference, so the free tier is more than enough.
Why is the LLM not deployed?
Phi-4-mini ran offline during the COSMO knowledge generation step. The knowledge triples it produced are already saved as cosmo_triples.json and encoded into intent embeddings. Only these pre-computed files go to the Space. This keeps the deployment fast and well within the free tier limits.
- Go to huggingface.co/spaces and log in to your Hugging Face account.
- Click **Create new Space**.
- Enter a name for your Space (for example, hm-fashion-recommender).
- Select **Gradio** as the Space SDK.
- Select **CPU basic** (free tier) as the hardware.
- Set the visibility to **Public**.
- Click **Create Space**.
Hugging Face creates a Git repository for your Space. Now you need to clone it and add your project files.
- Clone your new Space repository by running this command (replace the placeholder with your actual username and space name):
cd ~/Desktop
git clone https://huggingface.co/spaces/[[HF_USERNAME="your-huggingface-username"]]/[[SPACE_NAME="hm-fashion-recommender"]]
cd [[SPACE_NAME="hm-fashion-recommender"]]Authentication error when cloning?
If Git asks for credentials, use your Hugging Face username and a User Access Token (not your password). You can generate a token at huggingface.co/settings/tokens. Make sure the token has write permissions.
Help me authenticate with Hugging Face for git clone
- Copy all the necessary project files into the Space repository by running these commands from inside the cloned Space folder:
# Copy application code
cp -r ~/Desktop/hm_recommender/app .
cp ~/Desktop/hm_recommender/config.py .
cp ~/Desktop/hm_recommender/faiss_index.py .
cp ~/Desktop/hm_recommender/train_ranker.py .
cp ~/Desktop/hm_recommender/evaluate.py .
cp ~/Desktop/hm_recommender/requirements.txt .
# Copy model artifacts
cp -r ~/Desktop/hm_recommender/models .
cp -r ~/Desktop/hm_recommender/embeddings .
cp -r ~/Desktop/hm_recommender/artifacts .
cp -r ~/Desktop/hm_recommender/knowledge .
# Copy article metadata (needed for product names)
mkdir -p data
cp ~/Desktop/hm_recommender/data/articles.csv data/What files are being deployed?
You are copying the app code, the Python modules it imports (config.py, faiss_index.py, train_ranker.py, evaluate.py), the trained model files, pre-computed embeddings, FAISS index, COSMO knowledge triples, evaluation metrics, and the article metadata CSV. The data pipeline scripts, training scripts, and images are NOT deployed. Only the artifacts needed for serving are included.
Hugging Face Spaces reads its configuration from a YAML block at the top of the repository's README.md. You need to create this file so the Space knows which SDK and app file to use.
- Create a README.md file in the root of your Space repository with the following content:
---
title: HM Fashion Recommender
emoji: 👗
colorFrom: purple
colorTo: pink
sdk: gradio
sdk_version: "6.13.0"
app_file: app/app.py
pinned: false
---
Two-stage fashion recommendation system with COSMO-inspired LLM knowledge enhancement.What does this configuration do?
The sdk: gradio tells Hugging Face to run this as a Gradio app. The sdk_version pins it to Gradio 6.13.0 to match your local setup. The app_file: app/app.py tells Spaces where to find your main application file, relative to the repository root.
- Push everything to your Space by running these commands:
git add .
git commit -m "Deploy fashion recommender with COSMO knowledge"
git pushThe push may take a few minutes depending on the size of your model artifacts. After the push completes, Hugging Face automatically starts building and deploying your Space.
Push rejected or taking too long?
If the push fails due to large file sizes, install Git LFS first with git lfs install and then track large files with git lfs track "*.npz" "*.bin" "*.pt" "*.txt" before committing. Hugging Face Spaces supports Git LFS for files over 10 MB.
Help me push large model files to my Hugging Face Space
Test the deployed app and get your public URL
- Go to your Space URL at https://huggingface.co/spaces/ followed by your username and space name.
- Click the **Logs** tab to monitor the build process.
How long does the build take?
The first build installs all dependencies from requirements.txt (including PyTorch and other packages). This can take 5 to 10 minutes on the free tier. Subsequent rebuilds are faster because dependencies are cached.
- Once the build completes and the status shows **Running**, click the **App** tab to see your live app.
- Select a customer from the dropdown and click **Get Recommendations**.
- Verify that recommendations appear with product names, colors, MORS scores, and COSMO intent explanations in the **Why (COSMO Intent)** column.
- Try moving the **MORS Alpha** slider to 0.5 and clicking **Get Recommendations** again to see how the eco-friendliness weighting changes the results.
- Switch to the **Metrics Dashboard** tab and confirm you see the evaluation metrics table with MAP@12, Recall@10, Recall@50, and NDCG@12 values.
Verify that response times are under 5 seconds for each recommendation request on the free tier.
- Copy the public URL from your browser's address bar. This is the link you will share in your portfolio and research paper.
Space shows a build error?
Check the **Logs** tab for the specific error message. Common issues include a missing requirements.txt at the repository root or the app_file path in README.md not matching the actual file location.
If you see a memory error during artifact loading, your embedding files may be too large for the 16 GB RAM limit. Try reducing the number of user IDs loaded in the dropdown (change user_ids[:500] to a smaller number like user_ids[:100]).
Help me debug my Hugging Face Space build error
Your full recommendation pipeline is now live on the internet. Next up, you will tackle the Secret Mission and add temporal decay weighting with an A/B test simulation.
Secret mission
Temporal Decay + A/B Test Simulation
Your recommendation system treats all purchases equally, whether they happened yesterday or five weeks ago. In fashion, recency matters. Add temporal decay weighting to prioritize recent behavior and build an A/B test simulation that compares MORS configurations side by side.
Before you go
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs almost entirely locally on your Mac, so there are no ongoing compute costs. The only hosted resource is your Hugging Face Space, which runs on the free tier.
Resources you used:
- Python virtual environment with PyTorch, sentence-transformers, FAISS, LightGBM, Gradio, and Hugging Face transformers.
- H&M Kaggle dataset files (articles.csv, customers.csv, transactions_train.csv, and images directory).
- Project directory hm_recommender/ with all subdirectories (data/, models/, embeddings/, artifacts/, knowledge/, app/).
- Trained model checkpoints (two-tower retrieval model and LightGBM ranker).
- Pre-computed embeddings (visual, text, fused, intent, trained user and item embeddings).
- FAISS index file and COSMO knowledge triples JSON.
- Hugging Face Space (free tier deployment).
- Kaggle account and Hugging Face account.
Keep everything running
No action needed. Choose this if you are still actively building on this project or plan to extend it with the Secret Mission.
- Your Hugging Face Space stays live on the free tier at no cost.
- Your local project files, embeddings, trained models, and FAISS index remain on disk.
- Your Python virtual environment stays ready to activate.
Pause - I'll come back to this later
Free up local memory and disk space while keeping your files safe for later.
- Deactivate your Python virtual environment by running deactivate in your terminal.
- Pause your Hugging Face Space by going to your Space's **Settings** tab and clicking **Pause Space**. This stops the runtime but keeps all files intact.
- Your local project folder, trained models, embeddings, and COSMO knowledge triples remain on disk and can be reused when you return.
Delete - I don't want to use this again
Remove all project resources and start fresh.
- Delete your Hugging Face Space by going to your Space's **Settings** tab, scrolling to the bottom, and clicking **Delete this Space**.
- Delete the entire hm_recommender/ project folder from your Desktop. This removes all code, trained models, embeddings, the FAISS index, and COSMO knowledge triples.
- Delete the H&M dataset files if you no longer need them. These can be re-downloaded from Kaggle at any time.
- Delete the Python virtual environment folder (the venv/ directory inside your project) to free up disk space from installed packages.
- Your Kaggle and Hugging Face accounts are free to keep. No need to delete them.
Mission accomplished
That's a Wrap!
Nice work! You just built a production-grade two-stage fashion recommendation system with LLM-enhanced understanding from the ground up, trained on real H&M data.
You learned how to:
- Built a two-tower retrieval model with multi-modal embeddings (ResNet-50 visual + Sentence-BERT text), DropoutNet cold-start handling, and FAISS approximate nearest neighbor search for fast candidate retrieval.
- Implemented Amazon COSMO-inspired commonsense knowledge generation using Phi-4-mini, producing intent triples that capture why people buy fashion items and encoding them into ranking features.
- Built a LightGBM ranking model with Multi-Objective Ranking Score (MORS) balancing style and sustainability, validated the full pipeline with MAP@12, Recall@K, and NDCG metrics, and deployed an interactive Gradio demo to Hugging Face Spaces with a metrics dashboard and COSMO intent explanations.
- Secret Mission: Added temporal decay weighting for user interaction history and an A/B test simulation comparing MORS alpha configurations.
Ready to quiz yourself?