# -*- coding: utf-8 -*-
# @Time : 2020/10/6
# @Author : Yingqian Min
# @Email : eliver_min@foxmail.com
r"""
ConvNCF
################################################
Reference:
Xiangnan He et al. "Outer Product-based Neural Collaborative Filtering." in IJCAI 2018.
Reference code:
https://github.com/duxy-me/ConvNCF
"""
import torch
import torch.nn as nn
import copy
from recbole.model.abstract_recommender import GeneralRecommender
from recbole.model.layers import MLPLayers, CNNLayers
from recbole.model.general_recommender.bpr import BPR
from recbole.utils import InputType
[docs]class ConvNCFBPRLoss(nn.Module):
"""ConvNCFBPRLoss, based on Bayesian Personalized Ranking,
Shape:
- Pos_score: (N)
- Neg_score: (N), same shape as the Pos_score
- Output: scalar.
Examples::
>>> loss = ConvNCFBPRLoss()
>>> pos_score = torch.randn(3, requires_grad=True)
>>> neg_score = torch.randn(3, requires_grad=True)
>>> output = loss(pos_score, neg_score)
>>> output.backward()
"""
def __init__(self):
super(ConvNCFBPRLoss, self).__init__()
[docs] def forward(self, pos_score, neg_score):
distance = pos_score - neg_score
loss = torch.sum(torch.log((1 + torch.exp(-distance))))
return loss
[docs]class ConvNCF(GeneralRecommender):
r"""ConvNCF is a a new neural network framework for collaborative filtering based on NCF.
It uses an outer product operation above the embedding layer,
which results in a semantic-rich interaction map that encodes pairwise correlations between embedding dimensions.
We carefully design the data interface and use sparse tensor to train and test efficiently.
We implement the model following the original author with a pairwise training mode.
"""
input_type = InputType.PAIRWISE
def __init__(self, config, dataset):
super(ConvNCF, self).__init__(config, dataset)
# load dataset info
self.LABEL = config["LABEL_FIELD"]
# load parameters info
self.embedding_size = config["embedding_size"]
self.cnn_channels = config["cnn_channels"]
self.cnn_kernels = config["cnn_kernels"]
self.cnn_strides = config["cnn_strides"]
self.dropout_prob = config["dropout_prob"]
self.regs = config["reg_weights"]
self.train_method = config["train_method"]
self.pre_model_path = config["pre_model_path"]
# define layers and loss
assert self.train_method in ["after_pretrain", "no_pretrain"]
if self.train_method == "after_pretrain":
assert self.pre_model_path != ""
pretrain_state = torch.load(self.pre_model_path)["state_dict"]
bpr = BPR(config=config, dataset=dataset)
bpr.load_state_dict(pretrain_state)
self.user_embedding = copy.deepcopy(bpr.user_embedding)
self.item_embedding = copy.deepcopy(bpr.item_embedding)
else:
self.user_embedding = nn.Embedding(self.n_users, self.embedding_size)
self.item_embedding = nn.Embedding(self.n_items, self.embedding_size)
self.cnn_layers = CNNLayers(
self.cnn_channels, self.cnn_kernels, self.cnn_strides, activation="relu"
)
self.predict_layers = MLPLayers(
[self.cnn_channels[-1], 1], self.dropout_prob, activation="none"
)
self.loss = ConvNCFBPRLoss()
[docs] def forward(self, user, item):
user_e = self.user_embedding(user)
item_e = self.item_embedding(item)
interaction_map = torch.bmm(user_e.unsqueeze(2), item_e.unsqueeze(1))
interaction_map = interaction_map.unsqueeze(1)
cnn_output = self.cnn_layers(interaction_map)
cnn_output = cnn_output.sum(axis=(2, 3))
prediction = self.predict_layers(cnn_output)
prediction = prediction.squeeze(-1)
return prediction
[docs] def reg_loss(self):
r"""Calculate the L2 normalization loss of model parameters.
Including embedding matrices and weight matrices of model.
Returns:
loss(torch.FloatTensor): The L2 Loss tensor. shape of [1,]
"""
reg_1, reg_2 = self.regs[:2]
loss_1 = reg_1 * self.user_embedding.weight.norm(2)
loss_2 = reg_1 * self.item_embedding.weight.norm(2)
loss_3 = 0
for name, parm in self.cnn_layers.named_parameters():
if name.endswith("weight"):
loss_3 = loss_3 + reg_2 * parm.norm(2)
for name, parm in self.predict_layers.named_parameters():
if name.endswith("weight"):
loss_3 = loss_3 + reg_2 * parm.norm(2)
return loss_1 + loss_2 + loss_3
[docs] def calculate_loss(self, interaction):
user = interaction[self.USER_ID]
pos_item = interaction[self.ITEM_ID]
neg_item = interaction[self.NEG_ITEM_ID]
pos_item_score = self.forward(user, pos_item)
neg_item_score = self.forward(user, neg_item)
loss = self.loss(pos_item_score, neg_item_score)
opt_loss = loss + self.reg_loss()
return opt_loss
[docs] def predict(self, interaction):
user = interaction[self.USER_ID]
item = interaction[self.ITEM_ID]
return self.forward(user, item)