SCMG/models/UTFMG/model.py

import math
import logging

import torch
import torch.nn as nn
from torch.nn import functional as F

logger = logging.getLogger(__name__)
from SCMG.config import varables

# class ModelConfig():
#     rate_dropout_embedding = 0.1
#     rate_dropout_residue = 0.1
#     rate_dropout_attention = 0.1
#     block_size=125
#     def __init__(self, size_vocab, **kwargs):
#         self.size_vocab = size_vocab
#         for k,v in kwargs.items():
#             setattr(self, k, v)

class CausalSelfAttention(nn.Module):
    def __init__(self, config):
        super().__init__()
        assert config[varables.DIM_ATTENTION] % config[varables.NUM_HEADS] == 0
        self.key = nn.Linear(config[varables.DIM_EMBEDDING], config[varables.DIM_ATTENTION])
        self.query = nn.Linear(config[varables.DIM_EMBEDDING], config[varables.DIM_ATTENTION])
        self.value = nn.Linear(config[varables.DIM_EMBEDDING], config[varables.DIM_ATTENTION])
        self.dropout_attention = nn.Dropout(config[varables.RATE_DROPOUT])
        self.dropout_residue = nn.Dropout(config[varables.RATE_DROPOUT])
        self.projection = nn.Linear(config[varables.DIM_ATTENTION], config[varables.DIM_EMBEDDING])
        self.register_buffer("mask", torch.tril(torch.ones(config[varables.SIZE_BLOCK], config[varables.SIZE_BLOCK]))
                                     .view(1, 1, config[varables.SIZE_BLOCK], config[varables.SIZE_BLOCK]))
        self.n_head = config[varables.NUM_HEADS]
        self.single_head_dim = config[varables.DIM_ATTENTION] // self.n_head
        self.attention_features = config[varables.DIM_ATTENTION]

    def forward(self, x, layer_past=None,boundary=None):
        B, T, C = x.size()
        k =   self.key(x).view(B, T, self.n_head,self.single_head_dim).transpose(1, 2)
        q = self.query(x).view(B, T, self.n_head,self.single_head_dim).transpose(1, 2)
        v = self.value(x).view(B, T, self.n_head,self.single_head_dim).transpose(1, 2)
        att = (q @ k.transpose(-2, -1)) * (1.0 / math.sqrt(k.size(-1)))
        if boundary is None:
            att = att.masked_fill(self.mask[:,:,:T,:T] == 0, float('-inf'))
        else:
            mask = torch.tril(torch.ones(config[varables.SIZE_BLOCK], config[varables.SIZE_BLOCK]))\
                            .view(1, 1, config[varables.SIZE_BLOCK], config[varables.SIZE_BLOCK])\
                            .repeat(B,1,1,1)
            for i in range(len(boundary)):
                mask[i,0,:boundary[i],::boundary[i]] = 1
            att = att.masked_fill(mask[:,:,:T,:T] == 0, float('-inf'))
        att = F.softmax(att, dim=-1)
        att = self.dropout_attention(att)
        y = att @ v
        y = y.transpose(1, 2).contiguous().view(B, T, self.attention_features)
        y = self.dropout_residue(self.projection(y))
        return y

class Block(nn.Module):
    def __init__(self, config):
        super().__init__()
        self.ln1 = nn.LayerNorm(config[varables.DIM_EMBEDDING])
        self.ln2 = nn.LayerNorm(config[varables.DIM_EMBEDDING])
        self.attn = CausalSelfAttention(config)
        self.mlp = nn.Sequential(
            nn.Linear(config[varables.DIM_EMBEDDING], config[varables.DIM_FEEDFORWARD]),
            nn.GELU(),
            nn.Linear(config[varables.DIM_FEEDFORWARD], config[varables.DIM_EMBEDDING]),
            nn.Dropout(config[varables.RATE_DROPOUT]),
        )

    def forward(self, x, boundary):
         # = y_input
        x = x + self.attn(self.ln1(x),boundary)
        x = x + self.mlp(self.ln2(x))
        return x

class Model(nn.Module):
    def __init__(self, config):
        super().__init__()
        self.tok_emb = nn.Embedding(config[varables.SIZE_VOCAB], config[varables.DIM_EMBEDDING])
        self.pos_emb = nn.Parameter(torch.zeros(1, config[varables.SIZE_BLOCK], config[varables.DIM_EMBEDDING]))
        self.drop = nn.Dropout(config[varables.RATE_DROPOUT])
        self.blocks = nn.ModuleList([Block(config) for _ in range(config[varables.NUM_LAYERS])])
        # self.blocks = nn.Sequential(*[Block(config) for _ in range(config[varables.NUM_LAYERS])])
        self.ln_f = nn.LayerNorm(config[varables.DIM_EMBEDDING])
        self.head = nn.Linear(config[varables.DIM_EMBEDDING], config[varables.SIZE_VOCAB], bias=False)
        self.block_size = config[varables.SIZE_BLOCK]
        self.apply(self._init_weights)
        logger.info("number of parameters: %e", sum(p.numel() for p in self.parameters()))
        self.optimizer = None

    def get_block_size(self):
        return self.block_size

    def _init_weights(self, module):
        if isinstance(module, (nn.Linear, nn.Embedding)):
            module.weight.data.normal_(mean=0.0, std=0.02)
            if isinstance(module, nn.Linear) and module.bias is not None:
                module.bias.data.zero_()
        elif isinstance(module, nn.LayerNorm):
            module.bias.data.zero_()
            module.weight.data.fill_(1.0)
    def init_optimizers(self,train_config):
        optimizer = torch.optim.Adam(self.parameters(), lr=train_config[varables.RATE_LEARNING])
        return optimizer
    def init_scheduler(self,train_config):
        scheduler = torch.optim.lr_scheduler.StepLR(self.optimizer, step_size=train_config[varables.SIZE_STEP], gamma=train_config[varables.GAMMA])
        return scheduler
    def get_collate_fn(self, vocab,vocab2):
        def collate(results):
            x_in = None
            y_in = [a[0] + [vocab[varables.TOKEN_SEP]] + a[1] for a in results]
            boundary = [a[2] for a in results]
            max_len = max([len(a) for a in y_in])
            y = torch.tensor([(a+[vocab[varables.TOKEN_PAD]]*(max_len-len(a))) for a in y_in],dtype=torch.long)
            return x_in,y,boundary
        return collate

    def forward(self, x_in, y_in, y_out=None,boundary=None):
        b, t = y_in.size()
        assert t <= self.block_size
        token_embeddings = self.tok_emb(y_in)
        position_embeddings = self.pos_emb[:, :t, :]
        x = self.drop(token_embeddings + position_embeddings)
        for block in self.blocks:
            x = block(x,boundary)
        x = self.ln_f(x)
        logits = self.head(x)
        loss = None
        if y_out is not None:
            loss = F.cross_entropy(logits.view(-1, logits.size(-1)), y_out.view(-1))
        return logits, loss