import torch.nn as nn import copy, math import torch import numpy as np import torch.nn.functional as F from transformers import AutoModelForMaskedLM, AutoConfig from bertmodel import make_bert, make_bert_without_emb from utils import ContraLoss def load_pretrained_model(): model_checkpoint = "Rostlab/prot_bert" config = AutoConfig.from_pretrained(model_checkpoint) model = AutoModelForMaskedLM.from_config(config) return model class ConoEncoder(nn.Module): def __init__(self, encoder): super(ConoEncoder, self).__init__() self.encoder = encoder self.trainable_encoder = make_bert_without_emb() for param in self.encoder.parameters(): param.requires_grad = False def forward(self, x, mask): # x:(128,54) mask:(128,54) feat = self.encoder(x, attention_mask=mask) # (128,54,128) feat = list(feat.values())[0] # (128,54,128) feat = self.trainable_encoder(feat, mask) # (128,54,128) return feat class MSABlock(nn.Module): def __init__(self, in_dim, out_dim, vocab_size): super(MSABlock, self).__init__() self.embedding = nn.Embedding(vocab_size, in_dim) self.mlp = nn.Sequential( nn.Linear(in_dim, out_dim), nn.LeakyReLU(), nn.Linear(out_dim, out_dim) ) self.init() def init(self): for layer in self.mlp.children(): if isinstance(layer, nn.Linear): nn.init.xavier_uniform_(layer.weight) # nn.init.xavier_uniform_(self.embedding.weight) def forward(self, x): # x: (128,3,54) x = self.embedding(x) # x: (128,3,54,128) x = self.mlp(x) # x: (128,3,54,128) return x class ConoModel(nn.Module): def __init__(self, encoder, msa_block, decoder): super(ConoModel, self).__init__() self.encoder = encoder self.msa_block = msa_block self.feature_combine = nn.Conv2d(in_channels=4, out_channels=1, kernel_size=1) self.decoder = decoder def forward(self, input_ids, msa, attn_idx=None): encoder_output = self.encoder.forward(input_ids, attn_idx) # (128,54,128) msa_output = self.msa_block(msa) # (128,3,54,128) # msa_output = torch.mean(msa_output, dim=1) encoder_output = encoder_output.view(input_ids.shape[0], 54, -1).unsqueeze(1) # (128,1,54,128) output = torch.cat([encoder_output*5, msa_output], dim=1) # (128,4,54,128) output = self.feature_combine(output) # (128,1,54,128) output = output.squeeze(1) # (128,54,128) logits = self.decoder(output) # (128,54,85) return logits class ContraModel(nn.Module): def __init__(self, cono_encoder): super(ContraModel, self).__init__() self.contra_loss = ContraLoss() self.encoder1 = cono_encoder self.encoder2 = make_bert(404, 6, 128) # contrastive decoder self.lstm = nn.LSTM(16, 16, batch_first=True) self.contra_decoder = nn.Sequential( nn.Linear(128, 64), nn.LeakyReLU(), nn.Linear(64, 32), nn.LeakyReLU(), nn.Linear(32, 16), nn.LeakyReLU(), nn.Dropout(0.1), ) # classifier self.pre_classifer = nn.LSTM(128, 64, batch_first=True) self.classifer = nn.Sequential( nn.Linear(128, 32), nn.LeakyReLU(), nn.Linear(32, 6), nn.Softmax(dim=-1) ) self.init() def init(self): for layer in self.contra_decoder.children(): if isinstance(layer, nn.Linear): nn.init.xavier_uniform_(layer.weight) for layer in self.classifer.children(): if isinstance(layer, nn.Linear): nn.init.xavier_uniform_(layer.weight) for layer in self.pre_classifer.children(): if isinstance(layer, nn.Linear): nn.init.xavier_uniform_(layer.weight) for layer in self.lstm.children(): if isinstance(layer, nn.Linear): nn.init.xavier_uniform_(layer.weight) def compute_class_loss(self, feat1, feat2, labels): _, cls_feat1= self.pre_classifer(feat1) _, cls_feat2 = self.pre_classifer(feat2) cls_feat1 = torch.cat([cls_feat1[0], cls_feat1[1]], dim=-1).squeeze(0) cls_feat2 = torch.cat([cls_feat2[0], cls_feat2[1]], dim=-1).squeeze(0) cls1_dis = self.classifer(cls_feat1) cls2_dis = self.classifer(cls_feat2) cls1_loss = F.cross_entropy(cls1_dis, labels.to('cuda:0')) cls2_loss = F.cross_entropy(cls2_dis, labels.to('cuda:0')) return cls1_loss, cls2_loss def compute_contrastive_loss(self, feat1, feat2): contra_feat1 = self.contra_decoder(feat1) contra_feat2 = self.contra_decoder(feat2) _, feat1 = self.lstm(contra_feat1) _, feat2 = self.lstm(contra_feat2) feat1 = torch.cat([feat1[0], feat1[1]], dim=-1).squeeze(0) feat2 = torch.cat([feat2[0], feat2[1]], dim=-1).squeeze(0) ctr_loss = self.contra_loss(feat1, feat2) return ctr_loss def forward(self, x1, x2, labels=None): loss = dict() idx1, attn1 = x1 idx2, attn2 = x2 feat1 = self.encoder1(idx1.to('cuda:0'), attn1.to('cuda:0')) feat2 = self.encoder2(idx2.to('cuda:0'), attn2.to('cuda:0')) cls1_loss, cls2_loss = self.compute_class_loss(feat1, feat2, labels) ctr_loss = self.compute_contrastive_loss(feat1, feat2) loss['cls1_loss'] = cls1_loss loss['cls2_loss'] = cls2_loss loss['ctr_loss'] = ctr_loss return loss