""" Training script for the Two-Tower Pinterest retrieval model. Run: python scripts/train.py [--config config.yaml] Key design choices logged here (interview talking points): - In-batch negatives: scales O(B^2) without extra data - Learnable temperature: adapts sharpness during training - Temporal val split: prevents future leakage - Early stopping on Recall@10 """ import sys import os import yaml import argparse import numpy as np import torch import torch.optim as optim from pathlib import Path from tqdm import tqdm from loguru import logger # ── allow running from project root ────────────────────────────────────────── sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from data.generate_data import main as generate_data from pipeline.dataset import build_dataloaders from models.two_tower import build_model, InfoNCELoss, HardNegativeMiner from evaluation.metrics import compute_all_metrics from inference.faiss_index import build_faiss_index, search_index def parse_args(): p = argparse.ArgumentParser() p.add_argument("--config", default="config.yaml") p.add_argument("--regenerate-data", action="store_true", help="Force re-generate synthetic dataset") return p.parse_args() def encode_all_items(model, item_features: np.ndarray, batch_size: int, device) -> np.ndarray: """Encode the entire item corpus into embeddings.""" model.eval() all_embs = [] t = torch.from_numpy(item_features).to(device) with torch.no_grad(): for start in range(0, len(t), batch_size): batch = t[start: start + batch_size] emb = model.encode_items(batch) all_embs.append(emb.cpu().numpy()) return np.vstack(all_embs) def evaluate(model, loader, item_features, cfg, device) -> dict: """ Evaluate retrieval quality using brute-force exact search (val/test). At scale this would use the FAISS index. """ model.eval() all_user_embs, all_pos_pin_ids = [], [] with torch.no_grad(): for user_feats, item_feats, weights in loader: user_feats = user_feats.to(device) u_emb = model.encode_users(user_feats) all_user_embs.append(u_emb.cpu().numpy()) user_embs = np.vstack(all_user_embs) # (N_val, D) # Gather ground-truth pin_ids from val/test loader dataset dataset = loader.dataset pos_pin_ids = dataset.pin_ids # (N_val,) # Encode full item corpus item_embs = encode_all_items( model, item_features, cfg["training"]["batch_size"], device ) # (num_items, D) # Exact inner product search (embeddings already L2-normalized → cosine) scores = user_embs @ item_embs.T # (N_val, num_items) k_max = max(cfg["evaluation"]["k_values"]) top_k_idx = np.argsort(-scores, axis=1)[:, :k_max] # (N_val, k_max) return compute_all_metrics(top_k_idx, pos_pin_ids, cfg["evaluation"]["k_values"]) def train_epoch(model, loader, optimizer, criterion, miner, device, epoch): model.train() total_loss = 0.0 total_hard_negs = 0 for user_feats, item_feats, weights in tqdm(loader, desc=f"Epoch {epoch}", leave=False): user_feats = user_feats.to(device) item_feats = item_feats.to(device) weights = weights.to(device) u_emb, i_emb = model(user_feats, item_feats) # Log hard negative statistics _, n_hard = miner.mine(u_emb.detach(), i_emb.detach()) total_hard_negs += n_hard loss = criterion(u_emb, i_emb, model.temperature, weights) optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_( model.parameters(), max_norm=1.0 ) optimizer.step() total_loss += loss.item() return total_loss / len(loader), total_hard_negs def main(): args = parse_args() with open(args.config) as f: cfg = yaml.safe_load(f) # ── Setup ──────────────────────────────────────────────────────────────── Path(cfg["paths"]["model_dir"]).mkdir(parents=True, exist_ok=True) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") logger.info(f"Device: {device}") # ── Data ───────────────────────────────────────────────────────────────── data_dir = Path(cfg["paths"]["data_dir"]) if args.regenerate_data or not (data_dir / "raw/users.parquet").exists(): logger.info("Generating synthetic dataset...") generate_data(args.config) train_loader, val_loader, test_loader, meta = build_dataloaders(cfg) logger.info( f"Data loaded | train: {len(train_loader.dataset)} " f"| val: {len(val_loader.dataset)} " f"| test: {len(test_loader.dataset)}" ) # ── Model + Optimizer ──────────────────────────────────────────────────── model = build_model(cfg, meta["user_feat_dim"], meta["item_feat_dim"]).to(device) logger.info(f"Model parameters: {model.num_parameters():,}") optimizer = optim.AdamW( model.parameters(), lr=cfg["training"]["learning_rate"], weight_decay=cfg["training"]["weight_decay"], ) scheduler = optim.lr_scheduler.CosineAnnealingLR( optimizer, T_max=cfg["training"]["epochs"], eta_min=1e-5 ) criterion = InfoNCELoss() miner = HardNegativeMiner(cfg["training"]["hard_negative_ratio"]) # ── Training Loop ──────────────────────────────────────────────────────── best_recall = 0.0 patience_counter = 0 history = {"train_loss": [], "val_recall@10": [], "val_ndcg@10": []} for epoch in range(1, cfg["training"]["epochs"] + 1): train_loss, n_hard = train_epoch( model, train_loader, optimizer, criterion, miner, device, epoch ) scheduler.step() val_metrics = evaluate(model, val_loader, meta["item_features"], cfg, device) history["train_loss"].append(train_loss) history["val_recall@10"].append(val_metrics.get("recall@10", 0)) history["val_ndcg@10"].append(val_metrics.get("ndcg@10", 0)) logger.info( f"Epoch {epoch:3d} | loss: {train_loss:.4f} | " f"Recall@10: {val_metrics.get('recall@10', 0):.4f} | " f"NDCG@10: {val_metrics.get('ndcg@10', 0):.4f} | " f"temp: {model.temperature.item():.4f} | " f"hard_negs: {n_hard}" ) # ── Early Stopping ──────────────────────────────────────────────── recall10 = val_metrics.get("recall@10", 0) if recall10 > best_recall: best_recall = recall10 patience_counter = 0 ckpt = Path(cfg["paths"]["model_dir"]) / "best_model.pt" torch.save({ "epoch": epoch, "model_state": model.state_dict(), "optimizer_state": optimizer.state_dict(), "val_metrics": val_metrics, "cfg": cfg, "meta": { "user_feat_dim": meta["user_feat_dim"], "item_feat_dim": meta["item_feat_dim"], }, }, ckpt) logger.info(f" ✓ New best Recall@10: {best_recall:.4f} → saved") else: patience_counter += 1 if patience_counter >= cfg["training"]["patience"]: logger.info(f"Early stopping at epoch {epoch}") break # ── Final Evaluation ───────────────────────────────────────────────────── logger.info("Loading best model for test evaluation...") ckpt = torch.load(Path(cfg["paths"]["model_dir"]) / "best_model.pt", map_location=device) model.load_state_dict(ckpt["model_state"]) test_metrics = evaluate(model, test_loader, meta["item_features"], cfg, device) logger.info("=" * 60) logger.info("TEST RESULTS:") for k, v in test_metrics.items(): logger.info(f" {k}: {v:.4f}") logger.info("=" * 60) # ── Build FAISS Index ───────────────────────────────────────────────────── logger.info("Building FAISS index...") item_embs = encode_all_items( model, meta["item_features"], cfg["training"]["batch_size"], device ) index = build_faiss_index(item_embs, cfg) import faiss faiss.write_index(index, cfg["paths"]["index_path"]) np.save(cfg["paths"]["embeddings_path"], item_embs) logger.info(f"FAISS index saved → {cfg['paths']['index_path']}") # Save training history import json hist_path = Path(cfg["paths"]["model_dir"]) / "training_history.json" with open(hist_path, "w") as f: json.dump(history, f, indent=2) logger.info("✅ Training complete.") if __name__ == "__main__": main()