meta_train.py

import torch
from torch import nn
from collections import namedtuple
import math
from utils import DEVICE
from pytorch_transformers.modeling_bert import (
    BertEncoder,
    BertPreTrainedModel,
    BertConfig,
)


class GeLU(nn.Module):
    """Implementation of the gelu activation function.
    For information: OpenAI GPT's gelu is slightly different (and gives slightly different results):
    0.5 * x * (1 + torch.tanh(math.sqrt(2 / math.pi) * (x + 0.044715 * torch.pow(x, 3))))
    Also see https://arxiv.org/abs/1606.08415
    """

    def __init__(self):
        super().__init__()

    def forward(self, x):
        return x * 0.5 * (1.0 + torch.erf(x / math.sqrt(2.0)))


class BertLayerNorm(nn.Module):
    def __init__(self, hidden_size, eps=1e-12):
        """Construct a layernorm module in the TF style (epsilon inside the square root)."""
        super(BertLayerNorm, self).__init__()
        self.weight = nn.Parameter(torch.ones(hidden_size))
        self.bias = nn.Parameter(torch.zeros(hidden_size))
        self.variance_epsilon = eps

    def forward(self, x):
        u = x.mean(-1, keepdim=True)
        s = (x - u).pow(2).mean(-1, keepdim=True)
        x = (x - u) / torch.sqrt(s + self.variance_epsilon)
        return self.weight * x + self.bias


class mlp_meta(nn.Module):
    def __init__(self, config):
        super().__init__()
        self.mlp = nn.Sequential(
            nn.Linear(config.hid_dim, config.hid_dim),
            GeLU(),
            BertLayerNorm(config.hid_dim, eps=1e-12),
            nn.Dropout(config.dropout),
        )

    def forward(self, x):
        return self.mlp(x)


class Bert_Transformer_Layer(BertPreTrainedModel):
    def __init__(self, fusion_config):
        super().__init__(BertConfig(**fusion_config))
        bertconfig_fusion = BertConfig(**fusion_config)
        self.encoder = BertEncoder(bertconfig_fusion)
        self.init_weights()

    def forward(self, input, mask=None):
        """
        input:(bs, 4, dim)
        """
        batch, feats, dim = input.size()
        if mask is not None:
            mask_ = torch.ones(size=(batch, feats), device=mask.device)
            mask_[:, 1:] = mask
            mask_ = torch.bmm(
                mask_.view(batch, 1, -1).transpose(1, 2), mask_.view(batch, 1, -1)
            )
            mask_ = mask_.unsqueeze(1)

        else:
            mask = torch.Tensor([1.0]).to(input.device)
            mask_ = mask.repeat(batch, 1, feats, feats)

        extend_mask = (1 - mask_) * -10000
        assert not extend_mask.requires_grad
        head_mask = [None] * self.config.num_hidden_layers

        enc_output = self.encoder(input, extend_mask, head_mask=head_mask)
        output = enc_output[0]
        all_attention = enc_output[1]

        return output, all_attention


class mmdPreModel(nn.Module):
    def __init__(
        self,
        config,
        num_mlp=0,
        transformer_flag=False,
        num_hidden_layers=1,
        mlp_flag=True,
    ):
        super(mmdPreModel, self).__init__()
        self.num_mlp = num_mlp
        self.transformer_flag = transformer_flag
        self.mlp_flag = mlp_flag
        token_num = config.token_num
        self.mlp = nn.Sequential(
            nn.Linear(config.in_dim, config.hid_dim),
            GeLU(),
            BertLayerNorm(config.hid_dim, eps=1e-12),
            nn.Dropout(config.dropout),
            # nn.Linear(config.hid_dim, config.out_dim),
        )
        self.fusion_config = {
            "hidden_size": config.in_dim,
            "num_hidden_layers": num_hidden_layers,
            "num_attention_heads": 4,
            "output_attentions": True,
        }
        if self.num_mlp > 0:
            self.mlp2 = nn.ModuleList([mlp_meta(config) for _ in range(self.num_mlp)])
        if self.transformer_flag:
            self.transformer = Bert_Transformer_Layer(self.fusion_config)
        self.feature = nn.Linear(config.hid_dim * token_num, config.out_dim)

    def forward(self, features):
        """
        input: [batch, token_num, hidden_size], output: [batch, token_num * config.out_dim]
        """

        if self.transformer_flag:
            features, _ = self.transformer(features)
        if self.mlp_flag:
            features = self.mlp(features)

        if self.num_mlp > 0:
            # features = self.mlp2(features)
            for _ in range(1):
                for mlp in self.mlp2:
                    features = mlp(features)

        features = self.feature(features.view(features.shape[0], -1))
        return features  # features.view(features.shape[0], -1)


class NetLoader:
    def __init__(self):
        token_num, hidden_size = 100, 768
        Config = namedtuple(
            "Config", ["in_dim", "hid_dim", "dropout", "out_dim", "token_num"]
        )
        config = Config(
            in_dim=hidden_size,
            token_num=token_num,
            hid_dim=512,
            dropout=0.2,
            out_dim=300,
        )
        self.config = config
        self.net = mmdPreModel(
            config=config, num_mlp=0, transformer_flag=True, num_hidden_layers=1
        )
        checkpoint_filename = "./net.pt"
        checkpoint = torch.load(checkpoint_filename, map_location=DEVICE)
        self.net.load_state_dict(checkpoint["net"])
        self.sigma, self.sigma0_u, self.ep = (
            checkpoint["sigma"],
            checkpoint["sigma0_u"],
            checkpoint["ep"],
        )
        self.net = self.net.to(DEVICE)
        self.sigma, self.sigma0_u, self.ep = (
            self.sigma.to(DEVICE),
            self.sigma0_u.to(DEVICE),
            self.ep.to(DEVICE),
        )


net = NetLoader()