SCMG/models/Transformer/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):
        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)))
        att = att.masked_fill(self.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 CrossAttention(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.n_head = config[varables.NUM_HEADS]
        self.single_head_dim = config[varables.DIM_ATTENTION] // self.n_head
        self.attention_features = config[varables.DIM_ATTENTION]
        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]))

    def forward(self, x_encoder,x_decoder, layer_past=None):
        B_encoder, T_encoder, C_encoder = x_encoder.size()
        B_decoder, T_decoder, C_decoder = x_decoder.size()
        k = self.key(  x_encoder).view(B_encoder, T_encoder, self.n_head,self.single_head_dim).transpose(1, 2)
        q = self.query(x_decoder).view(B_encoder, T_decoder, self.n_head,self.single_head_dim).transpose(1, 2)
        v = self.value(x_encoder).view(B_encoder, T_encoder, self.n_head,self.single_head_dim).transpose(1, 2)
        att = (q @ k.transpose(-2, -1)) * (1.0 / math.sqrt(k.size(-1)))
        att = att.masked_fill(self.mask[:,:,:T_decoder,:T_encoder] == 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_encoder, T_decoder, self.attention_features)
        y = self.dropout_residue(self.projection(y))
        return y




class EncoderBlock(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):
         # = y_input
        x = x + self.attn(self.ln1(x))
        x = x + self.mlp(self.ln2(x))
        return x

class DecoderBlock(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.masked_attn = CausalSelfAttention(config)
        self.cross_attn = CrossAttention(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_encoder,x):
         # = y_input
        x = x + self.masked_attn(self.ln1(x))
        x = x + self.cross_attn(x_encoder,self.ln1(x))
        x = x + self.mlp(self.ln2(x))
        return x

class Model(nn.Module):
    def __init__(self, config):
        super().__init__()
        self.tok_emb = nn.Embedding(len(config["vocab_encoder"]), 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.encoder_blocks = nn.ModuleList([EncoderBlock(config) for _ in range(config[varables.NUM_LAYERS])])
        self.decoder_blocks = nn.ModuleList([DecoderBlock(config) for _ in range(config[varables.NUM_LAYERS])])
        # self.blocks = nn.Sequential(*[DecoderBlock(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], len(config["vocab_encoder"]), 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):
        def collate(results):
            x_in = [a[0] for a in results]
            y_in = [a[1] for a in results]
            boundary = -1
            max_len_x = max([len(a) for a in x_in])
            max_len_y = max([len(a) for a in y_in])
            x = torch.tensor([(a+[vocab[varables.TOKEN_PAD]]*(max_len_x-len(a))) for a in x_in],dtype=torch.long)
            y = torch.tensor([(a+[vocab[varables.TOKEN_PAD]]*(max_len_y-len(a))) for a in y_in],dtype=torch.long)
            return x,y,boundary
        return collate

    def forward(self, x_in, y_in, y_out=None,boundary=None):
        x_in = self.drop(self.tok_emb(x_in) + self.pos_emb[:, :x_in.size()[1], :])
        y_in = self.drop(self.tok_emb(y_in) + self.pos_emb[:, :y_in.size()[1], :])        
        #
        for encoder_block in self.encoder_blocks:
            x_in = encoder_block(x_in)
        x_in = self.ln_f(x_in)
        for decoder_block in self.decoder_blocks:
            y_in = decoder_block(x_in,y_in)
        y_in = self.ln_f(y_in)        
        logits = self.head(y_in)
        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

# mark test