from transformers import GPT2Config, AutoTokenizer, GPT2Config from transformers import PretrainedConfig, PreTrainedModel import transformers from typing import Optional, Tuple, Callable import torch import torch.nn as nn from transformers.modeling_utils import PreTrainedModel, PretrainedConfig from .utils import CABlock, _GPT2LMHeadModel from .configuration_prot2text import Prot2TextConfig import os import numpy as np from transformers.generation.configuration_utils import GenerationConfig from transformers.generation.logits_process import LogitsProcessorList from transformers.generation.stopping_criteria import StoppingCriteriaList from .pdb2graph import PDB2Graph, download_alphafold_structure from .graphs import * from .utils_dataset import * try: from graphein.protein.config import ProteinGraphConfig, DSSPConfig from graphein.protein.features.nodes.amino_acid import amino_acid_one_hot, meiler_embedding, expasy_protein_scale, hydrogen_bond_acceptor, hydrogen_bond_donor from graphein.protein.features.nodes.dssp import phi, psi, asa, rsa, secondary_structure from graphein.protein.edges.distance import (add_peptide_bonds, add_hydrogen_bond_interactions, add_distance_threshold, ) except ImportError: raise Exception('You need to install graphein from source in addition to DSSP to use this model please refer to https://github.com/a-r-j/graphein and https://ssbio.readthedocs.io/en/latest/instructions/dssp.html') try: from torch_geometric.nn import RGCNConv, global_mean_pool except ImportError: raise Exception('You need to install torch geometric and its dependecies to use this model please refer to https://pytorch-geometric.readthedocs.io/en/latest/install/installation.html') class EncoderRGCN(PreTrainedModel): ''' This class implement the RGCN encoder to encode the protein structure ''' def __init__(self, input_dim, hidden_dim=512, n_layers=6, emb_dim=512, dropout=0.2, num_relation=7, prot2text_version='1.0'): super(EncoderRGCN, self).__init__(PretrainedConfig(name='RGCN')) self.n_layers = n_layers self.output_dim = emb_dim self.prot2text_version = prot2text_version self.fc0 = nn.Linear(input_dim, hidden_dim) self.batchnorm_final = nn.BatchNorm1d(hidden_dim) self.batch_norms = nn.ModuleList() self.batch_norms.append(nn.BatchNorm1d(hidden_dim)) lst = list() lst.append(RGCNConv(hidden_dim, hidden_dim, num_relations=num_relation)) for i in range(n_layers-1): lst.append(RGCNConv(hidden_dim,hidden_dim, num_relations=num_relation)) self.conv = nn.ModuleList(lst) self.fc1 = nn.Linear(hidden_dim, hidden_dim) self.fc2 = nn.Linear(hidden_dim, self.output_dim) self.dropout = nn.Dropout(p=dropout) self.relu = nn.LeakyReLU() self.batchnorm = nn.BatchNorm1d(hidden_dim) self.main_input_name = 'nothing' def forward(self, x:Optional[torch.FloatTensor] = None, edge_index:Optional[torch.LongTensor] = None, edge_type:Optional[torch.LongTensor] = None, batch:Optional[torch.LongTensor] = None, **kargs): #construct pyg edge index shape (2, num_edges) from edge_list x = self.relu(self.fc0(x)) for i in range(self.n_layers): x = self.conv[i](x, edge_index, edge_type) out = global_mean_pool(x, batch) out = self.relu(self.fc1(out)) out = self.relu(self.fc2(out)) return out.unsqueeze(1) class Prot2TextModel(PreTrainedModel): config_class = Prot2TextConfig _keys_to_ignore_on_load_missing = [r"transformer"] base_model_prefix = "decoder" def __init__(self, config): super().__init__(config) self.gpt_config = GPT2Config.from_dict(config.gpt_config) # if we are using RGCN to encode the protein's structure, define the RGCN encoder if config.rgcn: self.encoder = EncoderRGCN(input_dim=config.rgcn_input_dim, hidden_dim=self.gpt_config.n_embd, n_layers=config.rgcn_n_layers, emb_dim=self.gpt_config.n_embd, prot2text_version=self.config.prot2text_version) # define the GPT2 decoder self.decoder = _GPT2LMHeadModel(self.gpt_config) # if using ESM to encode protein's sequence, define the ESM layer, the Projection layer and the fusion layer if config.esm: self.esm_config = PretrainedConfig.from_dict(config.esm_config) self.esm = transformers.EsmModel(self.esm_config) self.to_embedding = nn.Linear(self.esm_config.hidden_size, self.gpt_config.n_embd) if config.cross_esm_graph and config.rgcn: self.h = nn.ModuleList([CABlock(self.gpt_config, layer_idx=i) for i in range(4)]) self.ln_f = nn.LayerNorm(self.gpt_config.n_embd, eps=self.gpt_config.layer_norm_epsilon) self.config = config def get_encoder(self): return self.encoder def get_decoder(self): return self.decoder def get_input_embeddings(self): if hasattr(self, "transformer"): return self.transformer.wte return self.decoder.transformer.wte def warm_up(self, gpt_model=None, esm_model=None): if esm_model is not None: self.esm = transformers.EsmModel.from_pretrained(esm_model) if gpt_model is not None: self.decoder = _GPT2LMHeadModel.from_pretrained(gpt_model, add_cross_attention=True, use_cache=False) self.decoder.resize_token_embeddings(self.gpt_config.vocab_size) self.decoder.config = self.gpt_config def forward(self, encoder_input_ids: Optional[torch.LongTensor] = None, edge_index: Optional[torch.LongTensor] = None, batch: Optional[torch.LongTensor] = None, x: Optional[torch.FloatTensor] = None, edge_type: Optional[torch.LongTensor] = None, decoder_input_ids: Optional[torch.LongTensor] = None, past_key_values: Optional[Tuple[Tuple[torch.Tensor]]] = None, past_key_values_graph_esm: Optional[Tuple[Tuple[torch.Tensor]]] = None, decoder_attention_mask: Optional[torch.FloatTensor] = None, attention_mask: Optional[torch.FloatTensor] = None, token_type_ids: Optional[torch.LongTensor] = None, position_ids: Optional[torch.LongTensor] = None, head_mask: Optional[torch.FloatTensor] = None, inputs_embeds: Optional[torch.FloatTensor] = None, encoder_hidden_states: Optional[torch.Tensor] = None, encoder_attention_mask: Optional[torch.FloatTensor] = None, labels: Optional[torch.LongTensor] = None, use_cache: Optional[bool] = None, output_attentions: Optional[bool] = None, output_hidden_states: Optional[bool] = None, return_dict: Optional[bool] = None, get_graph_emb: Optional[bool] = False, **delete_args, ): use_cache = use_cache if use_cache is not None else self.gpt_config.use_cache return_dict = return_dict if return_dict is not None else self.gpt_config.use_return_dict if decoder_input_ids is not None and len(decoder_input_ids.size()) == 3: decoder_input_ids = decoder_input_ids.squeeze(0) if x is not None and self.config.rgcn: graph_emb = self.encoder(x, edge_index, edge_type, batch) graph_mask = None if self.config.esm: if self.config.prot2text_version=='1.0': if encoder_input_ids.size()[1] != 1021: raise ValueError("For this version of the model you need to PAD/Truncate the amino acid sequence for the ESM model to 1021") esm_emb = self.esm(input_ids=encoder_input_ids, attention_mask=attention_mask, return_dict=return_dict).last_hidden_state esm_emb = self.to_embedding(esm_emb) if not self.config.cross_esm_graph and self.config.rgcn: graph_emb = torch.cat((graph_emb, esm_emb), dim=1) t_add = torch.ones((attention_mask.size(0), 1)).to(attention_mask.get_device()) attention_mask = torch.cat((t_add, attention_mask), dim=1) elif self.config.cross_esm_graph and self.config.rgcn: if past_key_values_graph_esm is None: past_length = 0 past_key_values_graph_esm = tuple([None] * len(self.h)) else: past_length = past_key_values_graph_esm[0][0].size(-2) output_shape = esm_emb.size() all_self_attentions = () if output_attentions else None all_cross_attentions = () if output_attentions and self.gpt_config.add_cross_attention else None all_hidden_states = () if output_hidden_states else None for i, (block, layer_past) in enumerate(zip(self.h, past_key_values_graph_esm)): outputs = block( esm_emb, layer_past=layer_past, attention_mask=attention_mask, encoder_hidden_states=graph_emb, encoder_attention_mask=graph_mask, use_cache=use_cache, output_attentions=False, ) esm_emb = outputs[0] esm_emb = self.ln_f(esm_emb) esm_emb = esm_emb.view(output_shape) graph_emb = esm_emb else: graph_emb = esm_emb else: attention_mask = None if self.config.prot2text_version=='1.0': attention_mask = None if get_graph_emb: return graph_emb transformer_outputs = self.decoder(input_ids=decoder_input_ids, past_key_values=past_key_values, attention_mask=decoder_attention_mask, token_type_ids=token_type_ids, position_ids=position_ids, head_mask=head_mask, inputs_embeds=inputs_embeds, encoder_hidden_states=graph_emb, encoder_attention_mask=attention_mask, labels=labels, use_cache=use_cache, output_attentions=output_attentions, output_hidden_states=output_hidden_states, return_dict=return_dict, ) return transformer_outputs @torch.no_grad() def generate_protein_description(self, protein_pdbID=None, protein_sequence=None, edge_index: Optional[torch.LongTensor] = None, x: Optional[torch.FloatTensor] = None, edge_type: Optional[torch.LongTensor] = None, tokenizer=None, device='cpu' ): if self.config.esm and not self.config.rgcn and protein_sequence==None: raise ValueError( "The model you are trying to use is based only on protein sequence, please provide an amino-acid protein_sequence" ) if self.config.rgcn and protein_pdbID==None and (x==None or edge_index==None or edge_type==None): raise ValueError( "The model you are trying to use is based on protein structure, please provide a AlphaFold ID (you must have to have internet connection using protein_pdbID, or provide the triplet inputs: x (node features), edge_index and edge_type" ) if self.config.esm: esmtokenizer = AutoTokenizer.from_pretrained(self.config.esm_model_name) if protein_pdbID==None and protein_sequence==None: raise ValueError( "you need to provide either a protein AlphaFold Id or an amino-acid sequence" ) if protein_pdbID!=None: config = {"node_metadata_functions": [amino_acid_one_hot, expasy_protein_scale, meiler_embedding, hydrogen_bond_acceptor, hydrogen_bond_donor ], "edge_construction_functions": [add_peptide_bonds, add_hydrogen_bond_interactions, partial(add_distance_threshold, long_interaction_threshold=3, threshold=10.),], "graph_metadata_functions":[asa,phi, psi, secondary_structure, rsa], "dssp_config": DSSPConfig()} config = ProteinGraphConfig(**config) PATH_TO_DATA = f"~/.tmp/pdb/pdb" OUTPUT_FOLDER = f"~/.tmp/pdb/raw" save_dir = f"~/.tmp/pdb/" isExist = os.path.exists(PATH_TO_DATA) if not isExist: os.makedirs(PATH_TO_DATA) isExist = os.path.exists(OUTPUT_FOLDER) if not isExist: os.makedirs(OUTPUT_FOLDER) isExist = os.path.exists(save_dir+'processed') if not isExist: os.makedirs(save_dir+'processed') structure_filename = download_alphafold_structure(uniprot_id=protein_pdbID, out_dir=PATH_TO_DATA) if structure_filename is None: raise ValueError("Error! the ID does not exist in AlphaFoldDB or you do not have internet connection") graph_filename = structure_filename.split('/') graph_filename[-2] = 'raw' graph_filename[-1] = graph_filename[-1].replace('.pdb', '.pt') graph_filename = '/'.join(graph_filename) process_filename = structure_filename.split('/') process_filename[-2] = 'processed' process_filename[-1] = process_filename[-1].replace('.pdb', '.pt') process_filename = '/'.join(process_filename) try: gpdb = PDB2Graph(root = PATH_TO_DATA, output_folder = OUTPUT_FOLDER, config=config, n_processors=1).create_pyg_graph(structure_filename) seq = esmtokenizer(gpdb.sequence, add_special_tokens=True, truncation=True, max_length=1021, padding='max_length',return_tensors="pt") # torch.save(gpdb, graph_filename) gpdb.edge_type = [np.array(gpdb.edge_type.transpose(0,1))] gpdb.encoder_input_ids = seq['input_ids'] gpdb.attention_mask = seq['attention_mask'] torch.save(gpdb, process_filename) except: os.remove(structure_filename) raise ValueError('creating graphs did not work, probably the pdb file of alphaFold is damaged') self.eval() inputs = gpdb inputs = inputs.to_dict() inputs['edge_type'] = torch.cat([torch.tensor(inputs['edge_type'][i]) for i in range(len(inputs['edge_type']))], dim=0) inputs['edge_type'] = torch.argmax(inputs['edge_type'], dim=1) for key in ['num_nodes', 'node_id', 'name', 'sequence', 'distance_matrix', 'distance', 'coordinates']: inputs.pop(key) inputs['decoder_input_ids'] = inputs['encoder_input_ids'][:,0:1].clone() inputs['decoder_input_ids'][:,0] = tokenizer.bos_token_id inputs["decoder_attention_mask"] = torch.ones(inputs['decoder_input_ids'].shape[0], 1) self.to(device) inputs = {k: v.to(device=device, non_blocking=True) if hasattr(v, 'to') else v for k, v in inputs.items()} encoder_state = dict() encoder_state['hidden_states'] = self(**inputs, get_graph_emb=True, output_attentions=True) encoder_state['attentions'] = inputs['attention_mask'] for key in ['edge_index', 'edge_type', 'x', 'encoder_input_ids']: inputs.pop(key) tok_ids = self.decoder.generate(input_ids=inputs['decoder_input_ids'], encoder_outputs=encoder_state, use_cache=True, output_attentions=False, output_scores=False, return_dict_in_generate=True, encoder_attention_mask=inputs['attention_mask'], length_penalty=1.0, no_repeat_ngram_size=None, early_stopping=False, num_beams=1) generated = tokenizer.batch_decode(tok_ids.get('sequences'), skip_special_tokens=True) os.remove(structure_filename) os.remove(graph_filename) os.remove(process_filename) return generated[0].replace('<|stop_token|>', '').replace('<|graph_token|>', '') else: seq = esmtokenizer([protein_sequence], add_special_tokens=True, truncation=True, max_length=1021, padding='max_length', return_tensors="pt") inputs={} inputs['encoder_input_ids'] = seq['input_ids'] inputs['attention_mask'] = seq['attention_mask'] inputs['decoder_input_ids'] = inputs['encoder_input_ids'][:,0:1].clone() inputs['decoder_input_ids'][:,0] = tokenizer.bos_token_id self.to(device) inputs = {k: v.to(device=device, non_blocking=True) if hasattr(v, 'to') else v for k, v in inputs.items()} encoder_state = dict() encoder_state['hidden_states'] = self(**inputs, get_graph_emb=True, output_attentions=True) generated = tokenizer.batch_decode(self.decoder.generate(input_ids=inputs['decoder_input_ids'], encoder_outputs=encoder_state, use_cache=True), skip_special_tokens=True) return generated[0].replace('<|stop_token|>', '').replace('<|graph_token|>', '') @torch.no_grad() def generate(self, inputs: Optional[torch.Tensor] = None, generation_config: Optional[GenerationConfig] = None, logits_processor: Optional[LogitsProcessorList] = None, stopping_criteria: Optional[StoppingCriteriaList] = None, prefix_allowed_tokens_fn: Optional[Callable[[int, torch.Tensor], List[int]]] = None, synced_gpus: Optional[bool] = None, assistant_model: Optional["PreTrainedModel"] = None, streamer: Optional["BaseStreamer"] = None, **kwargs, ): encoder_state = self(**kwargs, get_graph_emb=True) input_ids = kwargs['decoder_input_ids'] attention_mask = kwargs['decoder_attention_mask'] kwargs['encoder_attention_mask'] = kwargs['attention_mask'] if not self.config.cross_esm_graph and self.config.rgcn and self.config.esm: t_add = torch.ones((kwargs['encoder_attention_mask'].size(0), 1)).to(kwargs['encoder_attention_mask'].get_device()) kwargs['encoder_attention_mask'] = torch.cat((t_add, kwargs['encoder_attention_mask']), dim=1) for key in ['edge_index', 'edge_type', 'x', 'encoder_input_ids', 'decoder_input_ids', 'decoder_attention_mask', 'batch', 'attention_mask', 'max_length', '_num_nodes', 'node_id', 'name', 'sequence', 'distance_matrix', 'distance', 'coordinates', 'ptr', 'num_nodes',]: if key in kwargs.keys(): kwargs.pop(key) return self.decoder.generate(input_ids=input_ids, generation_config=generation_config, logits_processor=logits_processor, stopping_criteria=stopping_criteria, prefix_allowed_tokens_fn=prefix_allowed_tokens_fn, synced_gpus=synced_gpus, assistant_model=assistant_model, streamer=streamer, encoder_outputs={'hidden_states': encoder_state, 'attentions':0}, **kwargs )