| 102 | |
| 103 | # NeuProcessEncoder |
| 104 | class NeuProcessEncoder(nn.Module): |
| 105 | def __init__(self, input_size=64, hidden_size=64, output_size=64, dropout_prob=0.4, device=None): |
| 106 | super(NeuProcessEncoder, self).__init__() |
| 107 | self.device = device |
| 108 | |
| 109 | # Encoder for item embeddings |
| 110 | layers = [nn.Linear(input_size, hidden_size), |
| 111 | torch.nn.Dropout(dropout_prob), |
| 112 | nn.ReLU(inplace=True), |
| 113 | nn.Linear(hidden_size, output_size)] |
| 114 | self.input_to_hidden = nn.Sequential(*layers) |
| 115 | |
| 116 | # Encoder for latent vector z |
| 117 | self.z1_dim = input_size # 64 |
| 118 | self.z2_dim = hidden_size # 64 |
| 119 | self.z_dim = output_size # 64 |
| 120 | self.z_to_hidden = nn.Linear(self.z1_dim, self.z2_dim) |
| 121 | self.hidden_to_mu = nn.Linear(self.z2_dim, self.z_dim) |
| 122 | self.hidden_to_logsigma = nn.Linear(self.z2_dim, self.z_dim) |
| 123 | |
| 124 | def emb_encode(self, input_tensor): |
| 125 | hidden = self.input_to_hidden(input_tensor) |
| 126 | |
| 127 | return hidden |
| 128 | |
| 129 | def aggregate(self, input_tensor): |
| 130 | return torch.mean(input_tensor, dim=-2) |
| 131 | |
| 132 | def z_encode(self, input_tensor): |
| 133 | hidden = torch.relu(self.z_to_hidden(input_tensor)) |
| 134 | mu = self.hidden_to_mu(hidden) |
| 135 | log_sigma = self.hidden_to_logsigma(hidden) |
| 136 | std = torch.exp(0.5 * log_sigma) |
| 137 | eps = torch.randn_like(std) |
| 138 | z = eps.mul(std).add_(mu) |
| 139 | return z, mu, log_sigma |
| 140 | |
| 141 | def encoder(self, input_tensor): |
| 142 | z_ = self.emb_encode(input_tensor) |
| 143 | z = self.aggregate(z_) |
| 144 | self.z, mu, log_sigma = self.z_encode(z) |
| 145 | return self.z, mu, log_sigma |
| 146 | |
| 147 | def forward(self, input_tensor): |
| 148 | self.z, _, _ = self.encoder(input_tensor) |
| 149 | return self.z |
| 150 | |
| 151 | |
| 152 | class MemoryUnit(nn.Module): |
nothing calls this directly
no outgoing calls
no test coverage detected