1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150
| import torch import torch.nn as nn
class LHUCLayer(nn.Module): """ LHUC (Learning Hidden Unit Contributions) Layer 核心功能:根据用户个性化特征,学习对共享隐层单元的缩放权重。 """ def __init__(self, hidden_dim, user_dim, scaling_func='2sigmoid'): """ :param hidden_dim: 共享隐层的维度(需要被缩放的维度) :param user_dim: 用户个性化特征的维度(Gate的输入维度) :param scaling_func: 缩放函数的类型,'2sigmoid' 是推荐系统中常用的(范围 0~2), 'exp' 或 'sigmoid' 也是可选方案。 """ super(LHUCLayer, self).__init__() self.scaling_func = scaling_func self.gate = nn.Linear(user_dim, hidden_dim) nn.init.zeros_(self.gate.bias) nn.init.xavier_uniform_(self.gate.weight)
def forward(self, hidden_input, user_feature): """ :param hidden_input: (Batch, Hidden_Dim) 共享网络的隐层输出 :param user_feature: (Batch, User_Dim) 用户个性化特征 (如 User ID Embedding) """ gate_out = self.gate(user_feature) if self.scaling_func == '2sigmoid': scale = 2.0 * torch.sigmoid(gate_out) elif self.scaling_func == 'sigmoid': scale = torch.sigmoid(gate_out) elif self.scaling_func == 'exp': scale = torch.exp(gate_out) else: scale = gate_out
return hidden_input * scale
class SimplePPNet(nn.Module): """ 一个集成 LHUC 的简单推荐模型 (类似 PPNet 结构) 结构:Embedding -> MLP (with LHUC) -> Output """ def __init__(self, feature_dim, user_num, user_emb_dim, hidden_units=[64, 32]): super(SimplePPNet, self).__init__() self.user_embedding = nn.Embedding(user_num, user_emb_dim) self.fcs = nn.ModuleList() self.lhuc_layers = nn.ModuleList() input_dim = feature_dim for unit in hidden_units: self.fcs.append(nn.Linear(input_dim, unit)) self.lhuc_layers.append(LHUCLayer(unit, user_emb_dim)) input_dim = unit self.activation = nn.ReLU() self.final = nn.Linear(input_dim, 1) self.sigmoid = nn.Sigmoid()
def forward(self, dense_features, user_ids): user_emb = self.user_embedding(user_ids) curr_out = dense_features for fc, lhuc in zip(self.fcs, self.lhuc_layers): curr_out = fc(curr_out) curr_out = self.activation(curr_out) curr_out = lhuc(curr_out, user_emb) logits = self.final(curr_out) return self.sigmoid(logits)
def test_lhuc_implementation(): print("=== 开始测试 LHUC 实现 ===") BATCH_SIZE = 4 FEATURE_DIM = 10 USER_NUM = 100 USER_EMB_DIM = 8 HIDDEN_UNITS = [16, 8] model = SimplePPNet(FEATURE_DIM, USER_NUM, USER_EMB_DIM, HIDDEN_UNITS) print(f"模型结构:\n{model}") dummy_features = torch.randn(BATCH_SIZE, FEATURE_DIM) dummy_user_ids = torch.randint(0, USER_NUM, (BATCH_SIZE,)) print(f"\n输入数据形状: Feature {dummy_features.shape}, UserID {dummy_user_ids.shape}") try: output = model(dummy_features, dummy_user_ids) print(f"前向传播输出形状: {output.shape}") print(f"输出值示例: {output.detach().view(-1).numpy()}") except Exception as e: print(f"前向传播失败: {e}") return
try: loss = output.mean() loss.backward() print("反向传播成功,梯度正常计算。") lhuc_grad = model.lhuc_layers[0].gate.weight.grad if lhuc_grad is not None: print(f"LHUC Gate 梯度范数: {lhuc_grad.norm().item():.4f}") else: print("警告: LHUC 层没有梯度!") except Exception as e: print(f"反向传播失败: {e}")
if __name__ == "__main__": test_lhuc_implementation()
|