Coverage for hopwise/model/sequential_recommender/npe.py: 88%

1# @Time : 2020/11/22 14:56 

2# @Author : Shao Weiqi 

3# @Reviewer : Lin Kun 

4# @Email : shaoweiqi@ruc.edu.cn 

5 

6r"""NPE 

7################################################ 

8 

9Reference: 

10 ThaiBinh Nguyen, et al. "NPE: Neural Personalized Embedding for Collaborative Filtering" in IJCAI 2018. 

11 

12Reference code: 

13 https://github.com/wubinzzu/NeuRec 

14 

15""" 

16 

17import torch 

18from torch import nn 

19from torch.nn.init import xavier_normal_ 

20 

21from hopwise.model.abstract_recommender import SequentialRecommender 

22from hopwise.model.loss import BPRLoss 

23 

24 

25class NPE(SequentialRecommender): 

26 r"""models a user’s click to an item in two terms: the personal preference of the user for the item, 

27 and the relationships between this item and other items clicked by the user 

28 

29 """ 

30 

31 def __init__(self, config, dataset): 

32 super().__init__(config, dataset) 

33 

34 # load the dataset information 

35 self.n_user = dataset.num(self.USER_ID) 

36 self.device = config["device"] 

37 

38 # load the parameters information 

39 self.embedding_size = config["embedding_size"] 

40 self.dropout_prob = config["dropout_prob"] 

41 

42 # define layers and loss type 

43 self.user_embedding = nn.Embedding(self.n_user, self.embedding_size) 

44 self.item_embedding = nn.Embedding(self.n_items, self.embedding_size) 

45 self.embedding_seq_item = nn.Embedding(self.n_items, self.embedding_size, padding_idx=0) 

46 self.relu = nn.ReLU() 

47 self.dropout = nn.Dropout(self.dropout_prob) 

48 

49 self.loss_type = config["loss_type"] 

50 if self.loss_type == "BPR": 

51 self.loss_fct = BPRLoss() 

52 elif self.loss_type == "CE": 

53 self.loss_fct = nn.CrossEntropyLoss() 

54 else: 

55 raise NotImplementedError("Make sure 'loss_type' in ['BPR', 'CE']!") 

56 

57 # init the parameters of the module 

58 self.apply(self._init_weights) 

59 

60 def _init_weights(self, module): 

61 if isinstance(module, nn.Embedding): 

62 xavier_normal_(module.weight.data) 

63 

64 def forward(self, seq_item, user): 

65 user_embedding = self.dropout(self.relu(self.user_embedding(user))) 

66 # batch_size * embedding_size 

67 seq_item_embedding = self.item_embedding(seq_item).sum(dim=1) 

68 seq_item_embedding = self.dropout(self.relu(seq_item_embedding)) 

69 # batch_size * embedding_size 

70 

71 return user_embedding + seq_item_embedding 

72 

73 def calculate_loss(self, interaction): 

74 seq_item = interaction[self.ITEM_SEQ] 

75 user = interaction[self.USER_ID] 

76 seq_output = self.forward(seq_item, user) 

77 pos_items = interaction[self.POS_ITEM_ID] 

78 pos_items_embs = self.item_embedding(pos_items) 

79 if self.loss_type == "BPR": 

80 neg_items = interaction[self.NEG_ITEM_ID] 

81 neg_items_emb = self.relu(self.item_embedding(neg_items)) 

82 pos_items_emb = self.relu(pos_items_embs) 

83 pos_score = torch.sum(seq_output * pos_items_emb, dim=-1) 

84 neg_score = torch.sum(seq_output * neg_items_emb, dim=-1) 

85 loss = self.loss_fct(pos_score, neg_score) 

86 return loss 

87 else: # self.loss_type = 'CE' 

88 test_item_emb = self.relu(self.item_embedding.weight) 

89 logits = torch.matmul(seq_output, test_item_emb.transpose(0, 1)) 

90 loss = self.loss_fct(logits, pos_items) 

91 return loss 

92 

93 def predict(self, interaction): 

94 item_seq = interaction[self.ITEM_SEQ] 

95 test_item = interaction[self.ITEM_ID] 

96 user = interaction[self.USER_ID] 

97 seq_output = self.forward(item_seq, user) 

98 test_item_emb = self.relu(self.item_embedding(test_item)) 

99 scores = torch.mul(seq_output, test_item_emb).sum(dim=1) 

100 return scores 

101 

102 def full_sort_predict(self, interaction): 

103 item_seq = interaction[self.ITEM_SEQ] 

104 user = interaction[self.USER_ID] 

105 seq_output = self.forward(item_seq, user) 

106 test_items_emb = self.relu(self.item_embedding.weight) 

107 scores = torch.matmul(seq_output, test_items_emb.transpose(0, 1)) 

108 return scores