Loading your learning…
Loading your learning…
A single history may mix sportswear and kitchen products. First summarize relevant items within each known category; then decide which category summaries matter for the candidate. items [B, N, D] and target [B, D] contain feature vectors. category_ids [B, N] assigns one category 0…G−1 to each item; −1 marks padding. Output interest [B, D] is a feature vector, not a click probability. B is the batch size, N the padded history length, D the vector width, and G the number of categories. Category labels are supplied by the data; the network learns relevance, not the category assignments.
import torchfrom torch import nn class Relevance(nn.Module): def __init__(self, dim): super().__init__() self.net = nn.Sequential(nn.Linear(4 * dim, 16), nn.ReLU(), nn.Linear(16, 1)) def forward(self, items, target): query = target[:, None, :].expand_as(items) features = torch.cat([items, query, items - query, items * query], dim=-1) return self.net(features).squeeze(-1).sigmoid() class HierarchicalInterest(nn.Module): def __init__(self, dim, categories): super().__init__() self.categories = categories self.item_score = Relevance(dim) self.category_score = Relevance(dim) def forward(self, items, target, category_ids): # items [B, N, D], target [B, D], category_ids [B, N]; -1 means padding. groups = torch.arange(self.categories, device=items.device) membership = category_ids[:, None, :] == groups[None, :, None] # [B, G, N] item_weights = self.item_score(items, target) # [B, N] group_weights = item_weights[:, None, :] * membership # [B, G, N] mass = group_weights.sum(dim=-1, keepdim=True) # [B, G, 1] numerator = (group_weights[..., None] * items[:, None, :, :]).sum(dim=2) summaries = numerator / mass.clamp_min(torch.finfo(items.dtype).tiny) category_weights = self.category_score(summaries, target) # [B, G] category_weights = category_weights.masked_fill(mass.squeeze(-1) == 0, 0) interest = (category_weights[..., None] * summaries).sum(dim=1) # [B, D] return interest, group_weights, category_weights if __name__ == "__main__": torch.manual_seed(0) items = torch.tensor([[[1., 0.], [3., 0.], [0., 2.]], [[0., 1.], [0., 0.], [0., 0.]]]) # [2, 3, 2] target = torch.tensor([[1., 0.5], [0.5, 1.]]) categories = torch.tensor([[0, 0, 1], [1, -1, -1]]) pooling = HierarchicalInterest(dim=2, categories=2) interest, item_weights, category_weights = pooling(items, target, categories) print("Interest:", interest.shape) # [2, 2] print("Weights by group:", item_weights.shape) # [2, 2, 3] print("Category weights:", category_weights.detach()) # [2, 2]Both scorers compare a vector with the target using the vector, target, signed difference, and elementwise product. Their input is [B, rows, 4D]; the last Linear returns one scalar per row. Sigmoid bounds scores between 0 and 1. Separate parameters let item relevance and category relevance be learned differently. Positive item weights make weighted averaging meaningful. These gates are relevance weights, not click probabilities, and they do not sum to one.
from torch import nn class Relevance(nn.Module): def __init__(self, dim): super().__init__() self.net = nn.Sequential(nn.Linear(4 * dim, 16), nn.ReLU(), nn.Linear(16, 1)) def forward(self, items, target): query = target[:, None, :].expand_as(items) features = torch.cat([items, query, items - query, items * query], dim=-1) return self.net(features).squeeze(-1).sigmoid() class HierarchicalInterest(nn.Module): def __init__(self, dim, categories): super().__init__() self.categories = categories self.item_score = Relevance(dim) self.category_score = Relevance(dim) def forward(self, items, target, category_ids):membership [B, G, N] is true where an item belongs to a category. With categories [0, 0, 1], group 0 selects the first two items and group 1 selects the third. Padding −1 matches no group. Multiply membership by item_weights [B, 1, N] to get one masked weight per group and history position.
def forward(self, items, target, category_ids): # items [B, N, D], target [B, D], category_ids [B, N]; -1 means padding. groups = torch.arange(self.categories, device=items.device) membership = category_ids[:, None, :] == groups[None, :, None] # [B, G, N] item_weights = self.item_score(items, target) # [B, N] group_weights = item_weights[:, None, :] * membership # [B, G, N] mass = group_weights.sum(dim=-1, keepdim=True) # [B, G, 1] numerator = (group_weights[..., None] * items[:, None, :, :]).sum(dim=2)