Coverage for hopwise/model/sequential_recommender/fdsa.py: 85%
111 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-09-30 13:25 +0000
1# @Time : 2020/9/18 11:27
2# @Author : Hui Wang
3# @Email : hui.wang@ruc.edu.cn
5r"""FDSA
6################################################
8Reference:
9 Tingting Zhang et al. "Feature-level Deeper Self-Attention Network for Sequential Recommendation."
10 In IJCAI 2019
12"""
14import torch
15from torch import nn
17from hopwise.model.abstract_recommender import SequentialRecommender
18from hopwise.model.layers import (
19 FeatureSeqEmbLayer,
20 TransformerEncoder,
21 VanillaAttention,
22)
23from hopwise.model.loss import BPRLoss
26class FDSA(SequentialRecommender):
27 r"""FDSA is similar with the GRU4RecF implemented in hopwise, which uses two different Transformer encoders to
28 encode items and features respectively and concatenates the two subparts' outputs as the final output.
30 """
32 def __init__(self, config, dataset):
33 super().__init__(config, dataset)
35 # load parameters info
36 self.n_layers = config["n_layers"]
37 self.n_heads = config["n_heads"]
38 self.hidden_size = config["hidden_size"] # same as embedding_size
39 self.inner_size = config["inner_size"] # the dimensionality in feed-forward layer
40 self.hidden_dropout_prob = config["hidden_dropout_prob"]
41 self.attn_dropout_prob = config["attn_dropout_prob"]
42 self.hidden_act = config["hidden_act"]
43 self.layer_norm_eps = config["layer_norm_eps"]
45 self.selected_features = config["selected_features"]
46 self.pooling_mode = config["pooling_mode"]
47 self.device = config["device"]
48 self.num_feature_field = len(config["selected_features"])
50 self.initializer_range = config["initializer_range"]
51 self.loss_type = config["loss_type"]
53 # define layers and loss
54 self.item_embedding = nn.Embedding(self.n_items, self.hidden_size, padding_idx=0)
55 self.position_embedding = nn.Embedding(self.max_seq_length, self.hidden_size)
57 self.feature_embed_layer = FeatureSeqEmbLayer(
58 dataset,
59 self.hidden_size,
60 self.selected_features,
61 self.pooling_mode,
62 self.device,
63 )
65 self.item_trm_encoder = TransformerEncoder(
66 n_layers=self.n_layers,
67 n_heads=self.n_heads,
68 hidden_size=self.hidden_size,
69 inner_size=self.inner_size,
70 hidden_dropout_prob=self.hidden_dropout_prob,
71 attn_dropout_prob=self.attn_dropout_prob,
72 hidden_act=self.hidden_act,
73 layer_norm_eps=self.layer_norm_eps,
74 )
76 self.feature_att_layer = VanillaAttention(self.hidden_size, self.hidden_size)
77 # For simplicity, we use same architecture for item_trm and feature_trm
78 self.feature_trm_encoder = TransformerEncoder(
79 n_layers=self.n_layers,
80 n_heads=self.n_heads,
81 hidden_size=self.hidden_size,
82 inner_size=self.inner_size,
83 hidden_dropout_prob=self.hidden_dropout_prob,
84 attn_dropout_prob=self.attn_dropout_prob,
85 hidden_act=self.hidden_act,
86 layer_norm_eps=self.layer_norm_eps,
87 )
89 self.LayerNorm = nn.LayerNorm(self.hidden_size, eps=self.layer_norm_eps)
90 self.dropout = nn.Dropout(self.hidden_dropout_prob)
91 self.concat_layer = nn.Linear(self.hidden_size * 2, self.hidden_size)
92 if self.loss_type == "BPR":
93 self.loss_fct = BPRLoss()
94 elif self.loss_type == "CE":
95 self.loss_fct = nn.CrossEntropyLoss()
96 else:
97 raise NotImplementedError("Make sure 'loss_type' in ['BPR', 'CE']!")
99 # parameters initialization
100 self.apply(self._init_weights)
101 self.other_parameter_name = ["feature_embed_layer"]
103 def _init_weights(self, module):
104 """Initialize the weights"""
105 if isinstance(module, (nn.Linear, nn.Embedding)):
106 # Slightly different from the TF version which uses truncated_normal for initialization
107 # cf https://github.com/pytorch/pytorch/pull/5617
108 module.weight.data.normal_(mean=0.0, std=self.initializer_range)
109 elif isinstance(module, nn.LayerNorm):
110 module.bias.data.zero_()
111 module.weight.data.fill_(1.0)
112 if isinstance(module, nn.Linear) and module.bias is not None:
113 module.bias.data.zero_()
115 def forward(self, item_seq, item_seq_len):
116 item_emb = self.item_embedding(item_seq)
118 position_ids = torch.arange(item_seq.size(1), dtype=torch.long, device=item_seq.device)
119 position_ids = position_ids.unsqueeze(0).expand_as(item_seq)
120 position_embedding = self.position_embedding(position_ids)
122 # get item_trm_input
123 # item position add position embedding
124 item_emb = item_emb + position_embedding
125 item_emb = self.LayerNorm(item_emb)
126 item_trm_input = self.dropout(item_emb)
128 sparse_embedding, dense_embedding = self.feature_embed_layer(None, item_seq)
129 sparse_embedding = sparse_embedding["item"]
130 dense_embedding = dense_embedding["item"]
132 # concat the sparse embedding and float embedding
133 feature_table = []
134 if sparse_embedding is not None:
135 feature_table.append(sparse_embedding)
136 if dense_embedding is not None:
137 feature_table.append(dense_embedding)
139 # [batch len num_features hidden_size]
140 feature_table = torch.cat(feature_table, dim=-2)
142 # feature_emb [batch len hidden]
143 # weight [batch len num_features]
144 # if only one feature, the weight would be 1.0
145 feature_emb, attn_weight = self.feature_att_layer(feature_table)
146 # feature position add position embedding
147 feature_emb = feature_emb + position_embedding
148 feature_emb = self.LayerNorm(feature_emb)
149 feature_trm_input = self.dropout(feature_emb)
151 extended_attention_mask = self.get_attention_mask(item_seq)
153 item_trm_output = self.item_trm_encoder(
154 item_trm_input, extended_attention_mask, output_all_encoded_layers=True
155 )
156 item_output = item_trm_output[-1]
158 feature_trm_output = self.feature_trm_encoder(
159 feature_trm_input, extended_attention_mask, output_all_encoded_layers=True
160 ) # [B Len H]
161 feature_output = feature_trm_output[-1]
163 item_output = self.gather_indexes(item_output, item_seq_len - 1) # [B H]
164 feature_output = self.gather_indexes(feature_output, item_seq_len - 1) # [B H]
166 output_concat = torch.cat((item_output, feature_output), -1) # [B 2*H]
167 output = self.concat_layer(output_concat)
168 output = self.LayerNorm(output)
169 seq_output = self.dropout(output)
170 return seq_output # [B H]
172 def calculate_loss(self, interaction):
173 item_seq = interaction[self.ITEM_SEQ]
174 item_seq_len = interaction[self.ITEM_SEQ_LEN]
175 seq_output = self.forward(item_seq, item_seq_len)
176 pos_items = interaction[self.POS_ITEM_ID]
177 if self.loss_type == "BPR":
178 neg_items = interaction[self.NEG_ITEM_ID]
179 pos_items_emb = self.item_embedding(pos_items)
180 neg_items_emb = self.item_embedding(neg_items)
181 pos_score = torch.sum(seq_output * pos_items_emb, dim=-1) # [B]
182 neg_score = torch.sum(seq_output * neg_items_emb, dim=-1) # [B]
183 loss = self.loss_fct(pos_score, neg_score)
184 return loss
185 else: # self.loss_type = 'CE'
186 test_item_emb = self.item_embedding.weight
187 logits = torch.matmul(seq_output, test_item_emb.transpose(0, 1))
188 loss = self.loss_fct(logits, pos_items)
189 return loss
191 def predict(self, interaction):
192 item_seq = interaction[self.ITEM_SEQ]
193 item_seq_len = interaction[self.ITEM_SEQ_LEN]
194 test_item = interaction[self.ITEM_ID]
195 seq_output = self.forward(item_seq, item_seq_len)
196 test_item_emb = self.item_embedding(test_item)
197 scores = torch.mul(seq_output, test_item_emb).sum(dim=1) # [B]
198 return scores
200 def full_sort_predict(self, interaction):
201 item_seq = interaction[self.ITEM_SEQ]
202 item_seq_len = interaction[self.ITEM_SEQ_LEN]
203 seq_output = self.forward(item_seq, item_seq_len)
204 test_items_emb = self.item_embedding.weight
205 scores = torch.matmul(seq_output, test_items_emb.transpose(0, 1)) # [B, n_items]
206 return scores