Spaces:
Runtime error
Runtime error
Upload model.py with huggingface_hub
Browse files
model.py
ADDED
|
@@ -0,0 +1,67 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
"""Two-tower neural recommender.
|
| 2 |
+
|
| 3 |
+
User tower: user_id ──► Embedding ──► MLP ──► L2-normalized user_vec
|
| 4 |
+
Track tower: (track_id, artist_id) ──► Embeddings ──► MLP ──► L2-normalized track_vec
|
| 5 |
+
|
| 6 |
+
Score = dot(user_vec, track_vec). Trained with InfoNCE / in-batch negatives.
|
| 7 |
+
"""
|
| 8 |
+
|
| 9 |
+
from __future__ import annotations
|
| 10 |
+
|
| 11 |
+
import torch
|
| 12 |
+
import torch.nn as nn
|
| 13 |
+
import torch.nn.functional as F
|
| 14 |
+
|
| 15 |
+
|
| 16 |
+
def mlp(in_dim: int, hidden: int, out_dim: int, dropout: float = 0.1) -> nn.Sequential:
|
| 17 |
+
return nn.Sequential(
|
| 18 |
+
nn.Linear(in_dim, hidden),
|
| 19 |
+
nn.GELU(),
|
| 20 |
+
nn.Dropout(dropout),
|
| 21 |
+
nn.Linear(hidden, out_dim),
|
| 22 |
+
)
|
| 23 |
+
|
| 24 |
+
|
| 25 |
+
class UserTower(nn.Module):
|
| 26 |
+
def __init__(self, n_users: int, embed_dim: int = 64, out_dim: int = 128):
|
| 27 |
+
super().__init__()
|
| 28 |
+
self.user_emb = nn.Embedding(n_users, embed_dim)
|
| 29 |
+
self.proj = mlp(embed_dim, 256, out_dim)
|
| 30 |
+
|
| 31 |
+
def forward(self, user_idx: torch.Tensor) -> torch.Tensor:
|
| 32 |
+
x = self.user_emb(user_idx)
|
| 33 |
+
return F.normalize(self.proj(x), dim=-1)
|
| 34 |
+
|
| 35 |
+
|
| 36 |
+
class TrackTower(nn.Module):
|
| 37 |
+
def __init__(self, n_tracks: int, n_artists: int, embed_dim: int = 64, out_dim: int = 128):
|
| 38 |
+
super().__init__()
|
| 39 |
+
self.track_emb = nn.Embedding(n_tracks, embed_dim)
|
| 40 |
+
self.artist_emb = nn.Embedding(n_artists, embed_dim)
|
| 41 |
+
self.proj = mlp(embed_dim * 2, 256, out_dim)
|
| 42 |
+
|
| 43 |
+
def forward(self, track_idx: torch.Tensor, artist_idx: torch.Tensor) -> torch.Tensor:
|
| 44 |
+
x = torch.cat([self.track_emb(track_idx), self.artist_emb(artist_idx)], dim=-1)
|
| 45 |
+
return F.normalize(self.proj(x), dim=-1)
|
| 46 |
+
|
| 47 |
+
|
| 48 |
+
class TwoTower(nn.Module):
|
| 49 |
+
def __init__(self, n_users: int, n_tracks: int, n_artists: int, out_dim: int = 128):
|
| 50 |
+
super().__init__()
|
| 51 |
+
self.user_tower = UserTower(n_users, out_dim=out_dim)
|
| 52 |
+
self.track_tower = TrackTower(n_tracks, n_artists, out_dim=out_dim)
|
| 53 |
+
self.log_temp = nn.Parameter(torch.tensor(0.0)) # learnable temperature, like CLIP
|
| 54 |
+
|
| 55 |
+
def forward(
|
| 56 |
+
self, user_idx: torch.Tensor, track_idx: torch.Tensor, artist_idx: torch.Tensor
|
| 57 |
+
) -> torch.Tensor:
|
| 58 |
+
u = self.user_tower(user_idx) # (B, D)
|
| 59 |
+
t = self.track_tower(track_idx, artist_idx) # (B, D)
|
| 60 |
+
logits = (u @ t.T) * self.log_temp.exp() # (B, B)
|
| 61 |
+
return logits
|
| 62 |
+
|
| 63 |
+
|
| 64 |
+
def info_nce_loss(logits: torch.Tensor) -> torch.Tensor:
|
| 65 |
+
"""Symmetric InfoNCE — diagonal is the positive pair, off-diagonal are in-batch negatives."""
|
| 66 |
+
targets = torch.arange(logits.size(0), device=logits.device)
|
| 67 |
+
return 0.5 * (F.cross_entropy(logits, targets) + F.cross_entropy(logits.T, targets))
|