# Two Tower Recommender Systems

> Use when building two-tower recommendation models.

- Skill: `loopyluci/two-tower-recommender-systems` (Agent Skill)
- Install (CLI): `npx skillmds@latest add loopyluci/two-tower-recommender-systems`
- Raw SKILL.md: https://api.skillmd.com/api/skills/loopyluci/two-tower-recommender-systems/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: MIT
- Author: LoopyLuci (https://skillmd.com/u/loopyluci)
- Updated: 2026-09-08
- Page: https://skillmd.com/skills/loopyluci/two-tower-recommender-systems

---


# Two-Tower Recommender Systems

Building two-tower (dual-encoder) recommendation models — from candidate generation through approximate nearest neighbor retrieval, training optimization, and serving.

## When to Use

- Large-scale candidate retrieval (millions of items)
- YouTube-style recommendation architecture
- Building embedding-based retrieval systems
- Scaling recommendations to large item catalogs

## Two-Tower Architecture

```python
import torch, torch.nn as nn

class TwoTowerRecommender(nn.Module):
    """Dual-encoder for retrieval: user tower + item tower."""
    def __init__(self, n_users: int, n_items: int, dim: int = 64):
        super().__init__()
        self.user_embed = nn.Embedding(n_users, dim, padding_idx=0)
        self.item_embed = nn.Embedding(n_items, dim, padding_idx=0)
        
        self.user_tower = nn.Sequential(
            nn.Linear(dim, dim*2), nn.ReLU(), nn.Linear(dim*2, dim))
        self.item_tower = nn.Sequential(
            nn.Linear(dim, dim*2), nn.ReLU(), nn.Linear(dim*2, dim))
    
    def forward(self, user_ids, item_ids):
        user_vecs = self.user_tower(self.user_embed(user_ids))
        item_vecs = self.item_tower(self.item_embed(item_ids))
        return (user_vecs @ item_vecs.T)  # similarity matrix
    
    def encode_users(self, user_ids):
        return self.user_tower(self.user_embed(user_ids))
    
    def encode_items(self, item_ids):
        return self.item_tower(self.item_embed(item_ids))


# Sampled softmax loss for efficient training
def sampled_softmax_loss(user_logits, item_logits, labels, n_negatives=100):
    # In-batch negatives
    logits = user_logits @ item_logits.T  # (batch, batch)
    labels = torch.arange(len(user_logits))
    return F.cross_entropy(logits / 0.05, labels)
```

## Verification Checklist

- [ ] User and item towers defined with feature embeddings
- [ ] Training with in-batch negatives or sampled softmax
- [ ] Item embeddings indexed for ANN retrieval (FAISS)
- [ ] Candidate retrieval recall benchmarked (recall@k)
- [ ] Serving pipeline (user query → ANN → ranker → top-k)
- [ ] Freshness: new items indexed regularly
- [ ] A/B test framework for recommendation quality