import math from typing import List, Optional, Tuple, Union import torch.nn.functional as F import torch import torch.utils.checkpoint from torch import nn from torch.nn import CrossEntropyLoss from dataclasses import dataclass from transformers.modeling_outputs import ( BaseModelOutputWithPastAndCrossAttentions, BaseModelOutputWithPoolingAndCrossAttentions, MaskedLMOutput, ModelOutput, ) from transformers.modeling_utils import ( PreTrainedModel, find_pruneable_heads_and_indices, prune_linear_layer, ) from transformers.utils import logging from .configuration_proprime import ProPrimeConfig from torch.nn.functional import scaled_dot_product_attention logger = logging.get_logger(__name__) def consine_based_loss(x1, x2): cos = nn.CosineSimilarity(dim=0, eps=1e-6) x1 = x1 - x1.mean() x2 = x2 - x2.mean() return 1 - cos(x1, x2).mean() PROPRIME_PRETRAINED_MODEL_ARCHIVE_LIST = [ "AI4protein/ProPrime_650M", ] def rotate_half(x): return torch.cat((-x[..., x.shape[-1] // 2 :], x[..., : x.shape[-1] // 2]), dim=-1) def apply_rotary_pos_emb(x, cos, sin): cos = cos[:, :, : x.shape[-2], :] sin = sin[:, :, : x.shape[-2], :] return (x * cos) + (rotate_half(x) * sin) def gelu(x): return x * 0.5 * (1.0 + torch.erf(x / math.sqrt(2.0))) class RotaryEmbedding(torch.nn.Module): def __init__(self, dim: int): super().__init__() # Generate and save the inverse frequency buffer (non trainable) inv_freq = 1.0 / ( 10000 ** (torch.arange(0, dim, 2, dtype=torch.int64).float() / dim) ) inv_freq = inv_freq self.register_buffer("inv_freq", inv_freq) self._seq_len_cached = None self._cos_cached = None self._sin_cached = None def _update_cos_sin_tables(self, x, seq_dimension=2): seq_len = x.shape[seq_dimension] # Reset the tables if the sequence length has changed, # or if we're on a new device (possibly due to tracing for instance) if seq_len != self._seq_len_cached or self._cos_cached.device != x.device: self._seq_len_cached = seq_len t = torch.arange(x.shape[seq_dimension], device=x.device).type_as( self.inv_freq ) freqs = torch.outer(t, self.inv_freq) emb = torch.cat((freqs, freqs), dim=-1).to(x.device) self._cos_cached = emb.cos()[None, None, :, :] self._sin_cached = emb.sin()[None, None, :, :] return self._cos_cached, self._sin_cached def forward( self, q: torch.Tensor, k: torch.Tensor ) -> Tuple[torch.Tensor, torch.Tensor]: self._cos_cached, self._sin_cached = self._update_cos_sin_tables( k, seq_dimension=-2 ) return ( apply_rotary_pos_emb(q, self._cos_cached, self._sin_cached), apply_rotary_pos_emb(k, self._cos_cached, self._sin_cached), ) class ProPrimeEmbeddings(nn.Module): def __init__(self, config): super().__init__() self.word_embeddings = nn.Embedding( config.vocab_size, config.hidden_size, padding_idx=config.pad_token_id ) if config.emb_layer_norm_before: self.layer_norm = nn.LayerNorm( config.hidden_size, eps=config.layer_norm_eps ) else: self.layer_norm = None self.dropout = nn.Dropout(config.hidden_dropout_prob) self.position_embedding_type = getattr( config, "position_embedding_type", "absolute" ) self.register_buffer( "position_ids", torch.arange(config.max_position_embeddings).expand((1, -1)), persistent=False, ) self.padding_idx = config.pad_token_id if self.position_embedding_type == "absolute": self.position_embeddings = nn.Embedding( config.max_position_embeddings, config.hidden_size, padding_idx=self.padding_idx, ) self.token_dropout = config.token_dropout self.mask_token_id = config.mask_token_id def forward( self, input_ids=None, attention_mask=None, position_ids=None, inputs_embeds=None, past_key_values_length=0, ): if position_ids is None: if input_ids is not None: position_ids = create_position_ids_from_input_ids( input_ids, self.padding_idx, past_key_values_length ) else: position_ids = self.create_position_ids_from_inputs_embeds( inputs_embeds ) if inputs_embeds is None: inputs_embeds = self.word_embeddings(input_ids) embeddings = inputs_embeds if self.token_dropout: embeddings = embeddings.masked_fill( (input_ids == self.mask_token_id).unsqueeze(-1), 0.0 ) mask_ratio_train = 0.15 * 0.8 src_lengths = attention_mask.sum(-1) mask_ratio_observed = (input_ids == self.mask_token_id).sum( -1 ).float() / src_lengths embeddings = ( embeddings * (1 - mask_ratio_train) / (1 - mask_ratio_observed)[:, None, None] ).to(embeddings.dtype) if self.position_embedding_type == "absolute": position_embeddings = self.position_embeddings(position_ids) embeddings = embeddings + position_embeddings if self.layer_norm is not None: embeddings = self.layer_norm(embeddings) if attention_mask is not None: embeddings = (embeddings * attention_mask.unsqueeze(-1)).to( embeddings.dtype ) # Matt: I think this line was copied incorrectly from BERT, disabling it for now. # embeddings = self.dropout(embeddings) return embeddings def create_position_ids_from_inputs_embeds(self, inputs_embeds): input_shape = inputs_embeds.size()[:-1] sequence_length = input_shape[1] position_ids = torch.arange( self.padding_idx + 1, sequence_length + self.padding_idx + 1, dtype=torch.long, device=inputs_embeds.device, ) return position_ids.unsqueeze(0).expand(input_shape) class ProPrimeSelfAttention(nn.Module): def __init__(self, config, position_embedding_type=None): super().__init__() if config.hidden_size % config.num_attention_heads != 0 and not hasattr( config, "embedding_size" ): raise ValueError( f"The hidden size ({config.hidden_size}) is not a multiple of the number of attention " f"heads ({config.num_attention_heads})" ) self.num_attention_heads = config.num_attention_heads self.attention_head_size = int(config.hidden_size / config.num_attention_heads) self.all_head_size = self.num_attention_heads * self.attention_head_size self.query = nn.Linear(config.hidden_size, self.all_head_size) self.key = nn.Linear(config.hidden_size, self.all_head_size) self.value = nn.Linear(config.hidden_size, self.all_head_size) self.dropout = nn.Dropout(config.attention_probs_dropout_prob) self.position_embedding_type = position_embedding_type or getattr( config, "position_embedding_type", "absolute" ) self.rotary_embeddings = None if ( self.position_embedding_type == "relative_key" or self.position_embedding_type == "relative_key_query" ): self.max_position_embeddings = config.max_position_embeddings self.distance_embedding = nn.Embedding( 2 * config.max_position_embeddings - 1, self.attention_head_size ) elif self.position_embedding_type == "rotary": self.rotary_embeddings = RotaryEmbedding(dim=self.attention_head_size) self.flash_attention = config.flash_attention self.is_decoder = config.is_decoder self.config = config def transpose_for_scores(self, x: torch.Tensor) -> torch.Tensor: new_x_shape = x.size()[:-1] + ( self.num_attention_heads, self.attention_head_size, ) x = x.view(new_x_shape) return x.permute(0, 2, 1, 3) def forward( self, hidden_states: torch.Tensor, attention_mask: Optional[torch.FloatTensor] = None, head_mask: Optional[torch.FloatTensor] = None, encoder_hidden_states: Optional[torch.FloatTensor] = None, encoder_attention_mask: Optional[torch.FloatTensor] = None, past_key_value: Optional[Tuple[Tuple[torch.FloatTensor]]] = None, output_attentions: Optional[bool] = False, ) -> Tuple[torch.Tensor]: mixed_query_layer = self.query(hidden_states) # If this is instantiated as a cross-attention module, the keys # and values come from an encoder; the attention mask needs to be # such that the encoder's padding tokens are not attended to. is_cross_attention = encoder_hidden_states is not None if is_cross_attention and past_key_value is not None: # reuse k,v, cross_attentions key_layer = past_key_value[0] value_layer = past_key_value[1] attention_mask = encoder_attention_mask elif is_cross_attention: key_layer = self.transpose_for_scores(self.key(encoder_hidden_states)) value_layer = self.transpose_for_scores(self.value(encoder_hidden_states)) attention_mask = encoder_attention_mask elif past_key_value is not None: key_layer = self.transpose_for_scores(self.key(hidden_states)) value_layer = self.transpose_for_scores(self.value(hidden_states)) key_layer = torch.cat([past_key_value[0], key_layer], dim=2) value_layer = torch.cat([past_key_value[1], value_layer], dim=2) else: key_layer = self.transpose_for_scores(self.key(hidden_states)) value_layer = self.transpose_for_scores(self.value(hidden_states)) query_layer = self.transpose_for_scores(mixed_query_layer) query_layer = query_layer * self.attention_head_size**-0.5 if self.is_decoder: past_key_value = (key_layer, value_layer) if self.position_embedding_type == "rotary": query_layer, key_layer = self.rotary_embeddings(query_layer, key_layer) if not self.flash_attention: # Take the dot product between "query" and "key" to get the raw attention scores. attention_scores = torch.matmul(query_layer, key_layer.transpose(-1, -2)) if ( self.position_embedding_type == "relative_key" or self.position_embedding_type == "relative_key_query" ): seq_length = hidden_states.size()[1] position_ids_l = torch.arange( seq_length, dtype=torch.long, device=hidden_states.device ).view(-1, 1) position_ids_r = torch.arange( seq_length, dtype=torch.long, device=hidden_states.device ).view(1, -1) distance = position_ids_l - position_ids_r positional_embedding = self.distance_embedding( distance + self.max_position_embeddings - 1 ) positional_embedding = positional_embedding.to( dtype=query_layer.dtype ) # fp16 compatibility if self.position_embedding_type == "relative_key": relative_position_scores = torch.einsum( "bhld,lrd->bhlr", query_layer, positional_embedding ) attention_scores = attention_scores + relative_position_scores elif self.position_embedding_type == "relative_key_query": relative_position_scores_query = torch.einsum( "bhld,lrd->bhlr", query_layer, positional_embedding ) relative_position_scores_key = torch.einsum( "bhrd,lrd->bhlr", key_layer, positional_embedding ) attention_scores = ( attention_scores + relative_position_scores_query + relative_position_scores_key ) if attention_mask is not None: attention_scores = attention_scores + attention_mask # Normalize the attention scores to probabilities. attention_probs = nn.functional.softmax(attention_scores, dim=-1) # This is actually dropping out entire tokens to attend to, which might # seem a bit unusual, but is taken from the original Transformer paper. attention_probs = self.dropout(attention_probs) # Mask heads if we want to if head_mask is not None: attention_probs = attention_probs * head_mask context_layer = torch.matmul(attention_probs, value_layer) else: if self.training: context_layer = scaled_dot_product_attention( query_layer, key_layer, value_layer, attn_mask=attention_mask, dropout_p=self.config.attention_probs_dropout_prob, scale=1, # we have query_layer = query_layer * self.attention_head_size**-0.5 ) else: context_layer = scaled_dot_product_attention( query_layer, key_layer, value_layer, attn_mask=attention_mask, scale=1, # we have query_layer = query_layer * self.attention_head_size**-0.5 ) context_layer = context_layer.permute(0, 2, 1, 3).contiguous() new_context_layer_shape = context_layer.size()[:-2] + (self.all_head_size,) context_layer = context_layer.view(new_context_layer_shape) outputs = ( (context_layer, attention_probs) if output_attentions else (context_layer,) ) if self.is_decoder: outputs = outputs + (past_key_value,) return outputs class ProPrimeSelfOutput(nn.Module): def __init__(self, config): super().__init__() self.dense = nn.Linear(config.hidden_size, config.hidden_size) self.dropout = nn.Dropout(config.hidden_dropout_prob) def forward(self, hidden_states, input_tensor): hidden_states = self.dense(hidden_states) hidden_states = self.dropout(hidden_states) hidden_states = hidden_states + input_tensor return hidden_states class ProPrimeAttention(nn.Module): def __init__(self, config): super().__init__() self.self = ProPrimeSelfAttention(config) self.output = ProPrimeSelfOutput(config) self.pruned_heads = set() self.LayerNorm = nn.LayerNorm(config.hidden_size, eps=config.layer_norm_eps) def prune_heads(self, heads): if len(heads) == 0: return heads, index = find_pruneable_heads_and_indices( heads, self.self.num_attention_heads, self.self.attention_head_size, self.pruned_heads, ) # Prune linear layers self.self.query = prune_linear_layer(self.self.query, index) self.self.key = prune_linear_layer(self.self.key, index) self.self.value = prune_linear_layer(self.self.value, index) self.output.dense = prune_linear_layer(self.output.dense, index, dim=1) # Update hyper params and store pruned heads self.self.num_attention_heads = self.self.num_attention_heads - len(heads) self.self.all_head_size = ( self.self.attention_head_size * self.self.num_attention_heads ) self.pruned_heads = self.pruned_heads.union(heads) def forward( self, hidden_states, attention_mask=None, head_mask=None, encoder_hidden_states=None, encoder_attention_mask=None, past_key_value=None, output_attentions=False, ): hidden_states_ln = self.LayerNorm(hidden_states) self_outputs = self.self( hidden_states_ln, attention_mask, head_mask, encoder_hidden_states, encoder_attention_mask, past_key_value, output_attentions, ) attention_output = self.output(self_outputs[0], hidden_states) outputs = (attention_output,) + self_outputs[ 1: ] # add attentions if we output them return outputs class ProPrimeIntermediate(nn.Module): def __init__(self, config): super().__init__() self.dense = nn.Linear(config.hidden_size, config.intermediate_size) def forward(self, hidden_states: torch.Tensor) -> torch.Tensor: hidden_states = self.dense(hidden_states) hidden_states = gelu(hidden_states) return hidden_states class ProPrimeOutput(nn.Module): def __init__(self, config): super().__init__() self.dense = nn.Linear(config.intermediate_size, config.hidden_size) self.dropout = nn.Dropout(config.hidden_dropout_prob) def forward(self, hidden_states, input_tensor): hidden_states = self.dense(hidden_states) hidden_states = self.dropout(hidden_states) hidden_states = hidden_states + input_tensor return hidden_states class ProPrimeLayer(nn.Module): def __init__(self, config): super().__init__() self.chunk_size_feed_forward = config.chunk_size_feed_forward self.seq_len_dim = 1 self.attention = ProPrimeAttention(config) self.is_decoder = config.is_decoder self.add_cross_attention = config.add_cross_attention if self.add_cross_attention: if not self.is_decoder: raise RuntimeError( f"{self} should be used as a decoder model if cross attention is added" ) self.crossattention = ProPrimeAttention(config) self.intermediate = ProPrimeIntermediate(config) self.output = ProPrimeOutput(config) self.LayerNorm = nn.LayerNorm(config.hidden_size, eps=config.layer_norm_eps) def forward( self, hidden_states, attention_mask=None, head_mask=None, encoder_hidden_states=None, encoder_attention_mask=None, past_key_value=None, output_attentions=False, ): # decoder uni-directional self-attention cached key/values tuple is at positions 1,2 self_attn_past_key_value = ( past_key_value[:2] if past_key_value is not None else None ) self_attention_outputs = self.attention( hidden_states, attention_mask, head_mask, output_attentions=output_attentions, past_key_value=self_attn_past_key_value, ) attention_output = self_attention_outputs[0] # if decoder, the last output is tuple of self-attn cache if self.is_decoder: outputs = self_attention_outputs[1:-1] present_key_value = self_attention_outputs[-1] else: outputs = self_attention_outputs[ 1: ] # add self attentions if we output attention weights cross_attn_present_key_value = None if self.is_decoder and encoder_hidden_states is not None: if not hasattr(self, "crossattention"): raise AttributeError( f"If `encoder_hidden_states` are passed, {self} has to be instantiated" " with cross-attention layers by setting `config.add_cross_attention=True`" ) # cross_attn cached key/values tuple is at positions 3,4 of past_key_value tuple cross_attn_past_key_value = ( past_key_value[-2:] if past_key_value is not None else None ) cross_attention_outputs = self.crossattention( attention_output, attention_mask, head_mask, encoder_hidden_states, encoder_attention_mask, cross_attn_past_key_value, output_attentions, ) attention_output = cross_attention_outputs[0] outputs = ( outputs + cross_attention_outputs[1:-1] ) # add cross attentions if we output attention weights # add cross-attn cache to positions 3,4 of present_key_value tuple cross_attn_present_key_value = cross_attention_outputs[-1] present_key_value = present_key_value + cross_attn_present_key_value layer_output = self.feed_forward_chunk(attention_output) outputs = (layer_output,) + outputs # if decoder, return the attn key/values as the last output if self.is_decoder: outputs = outputs + (present_key_value,) return outputs def feed_forward_chunk(self, attention_output): attention_output_ln = self.LayerNorm(attention_output) intermediate_output = self.intermediate(attention_output_ln) layer_output = self.output(intermediate_output, attention_output) return layer_output class ProPrimeEncoder(nn.Module): def __init__(self, config): super().__init__() self.config = config self.layer = nn.ModuleList( [ProPrimeLayer(config) for _ in range(config.num_hidden_layers)] ) self.emb_layer_norm_after = nn.LayerNorm( config.hidden_size, eps=config.layer_norm_eps ) self.gradient_checkpointing = False def forward( self, hidden_states, attention_mask=None, head_mask=None, encoder_hidden_states=None, encoder_attention_mask=None, past_key_values=None, use_cache=None, output_attentions=False, output_hidden_states=False, return_dict=True, ): if self.gradient_checkpointing and self.training: if use_cache: logger.warning_once( "`use_cache=True` is incompatible with `config.gradient_checkpointing=True`. Setting " "`use_cache=False`..." ) use_cache = False all_hidden_states = () if output_hidden_states else None all_self_attentions = () if output_attentions else None all_cross_attentions = ( () if output_attentions and self.config.add_cross_attention else None ) next_decoder_cache = () if use_cache else None for i, layer_module in enumerate(self.layer): if output_hidden_states: all_hidden_states = all_hidden_states + (hidden_states,) layer_head_mask = head_mask[i] if head_mask is not None else None past_key_value = past_key_values[i] if past_key_values is not None else None if self.gradient_checkpointing and self.training: layer_outputs = self._gradient_checkpointing_func( layer_module.__call__, hidden_states, attention_mask, layer_head_mask, encoder_hidden_states, encoder_attention_mask, past_key_value, output_attentions, ) else: layer_outputs = layer_module( hidden_states, attention_mask, layer_head_mask, encoder_hidden_states, encoder_attention_mask, past_key_value, output_attentions, ) hidden_states = layer_outputs[0] if use_cache: next_decoder_cache = next_decoder_cache + (layer_outputs[-1],) if output_attentions: all_self_attentions = all_self_attentions + (layer_outputs[1],) if self.config.add_cross_attention: all_cross_attentions = all_cross_attentions + (layer_outputs[2],) if self.emb_layer_norm_after: hidden_states = self.emb_layer_norm_after(hidden_states) if output_hidden_states: all_hidden_states = all_hidden_states + (hidden_states,) if not return_dict: return tuple( v for v in [ hidden_states, next_decoder_cache, all_hidden_states, all_self_attentions, all_cross_attentions, ] if v is not None ) return BaseModelOutputWithPastAndCrossAttentions( last_hidden_state=hidden_states, past_key_values=next_decoder_cache, hidden_states=all_hidden_states, attentions=all_self_attentions, cross_attentions=all_cross_attentions, ) class ProPrimePreTrainedModel(PreTrainedModel): """ An abstract class to handle weights initialization and a simple interface for downloading and loading pretrained models. """ config_class = ProPrimeConfig base_model_prefix = "proprime" supports_gradient_checkpointing = True _no_split_modules = [ "ProPrimeLayer", "ProPrimeEmbeddings", ] # Copied from transformers.models.bert.modeling_bert.BertPreTrainedModel._init_weights def _init_weights(self, module): """Initialize the weights""" if isinstance(module, nn.Linear): # Slightly different from the TF version which uses truncated_normal for initialization # cf https://github.com/pytorch/pytorch/pull/5617 module.weight.data.normal_(mean=0.0, std=self.config.initializer_range) if module.bias is not None: module.bias.data.zero_() elif isinstance(module, nn.Embedding): module.weight.data.normal_(mean=0.0, std=self.config.initializer_range) if module.padding_idx is not None: module.weight.data[module.padding_idx].zero_() elif isinstance(module, nn.LayerNorm): module.bias.data.zero_() module.weight.data.fill_(1.0) class ProPrimeModel(ProPrimePreTrainedModel): base_model_prefix = "proprime" def __init__(self, config, add_pooling_layer=True): super().__init__(config) self.config = config self.embeddings = ProPrimeEmbeddings(config) self.encoder = ProPrimeEncoder(config) self.post_init() def get_input_embeddings(self): return self.embeddings.word_embeddings def set_input_embeddings(self, value): self.embeddings.word_embeddings = value def _prune_heads(self, heads_to_prune): """ Prunes heads of the model. heads_to_prune: dict of {layer_num: list of heads to prune in this layer} See base class PreTrainedModel """ for layer, heads in heads_to_prune.items(): self.encoder.layer[layer].attention.prune_heads(heads) def forward( self, input_ids: Optional[torch.Tensor] = None, attention_mask: Optional[torch.Tensor] = None, position_ids: Optional[torch.Tensor] = None, head_mask: Optional[torch.Tensor] = None, inputs_embeds: Optional[torch.Tensor] = None, encoder_hidden_states: Optional[torch.Tensor] = None, encoder_attention_mask: Optional[torch.Tensor] = None, past_key_values: Optional[List[torch.FloatTensor]] = None, use_cache: Optional[bool] = None, output_attentions: Optional[bool] = None, output_hidden_states: Optional[bool] = None, return_dict: Optional[bool] = None, ) -> Union[Tuple[torch.Tensor], BaseModelOutputWithPoolingAndCrossAttentions]: output_attentions = ( output_attentions if output_attentions is not None else self.config.output_attentions ) output_hidden_states = ( output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states ) return_dict = ( return_dict if return_dict is not None else self.config.use_return_dict ) if self.config.is_decoder: use_cache = use_cache if use_cache is not None else self.config.use_cache else: use_cache = False if input_ids is not None and inputs_embeds is not None: raise ValueError( "You cannot specify both input_ids and inputs_embeds at the same time" ) elif input_ids is not None: self.warn_if_padding_and_no_attention_mask(input_ids, attention_mask) input_shape = input_ids.size() elif inputs_embeds is not None: input_shape = inputs_embeds.size()[:-1] else: raise ValueError("You have to specify either input_ids or inputs_embeds") batch_size, seq_length = input_shape device = input_ids.device if input_ids is not None else inputs_embeds.device # past_key_values_length past_key_values_length = ( past_key_values[0][0].shape[2] if past_key_values is not None else 0 ) if attention_mask is None: attention_mask = torch.ones( ((batch_size, seq_length + past_key_values_length)), device=device ) extended_attention_mask: torch.Tensor = self.get_extended_attention_mask( attention_mask, input_shape ) if self.config.is_decoder and encoder_hidden_states is not None: encoder_batch_size, encoder_sequence_length, _ = ( encoder_hidden_states.size() ) encoder_hidden_shape = (encoder_batch_size, encoder_sequence_length) if encoder_attention_mask is None: encoder_attention_mask = torch.ones(encoder_hidden_shape, device=device) encoder_extended_attention_mask = self.invert_attention_mask( encoder_attention_mask ) else: encoder_extended_attention_mask = None head_mask = self.get_head_mask(head_mask, self.config.num_hidden_layers) embedding_output = self.embeddings( input_ids=input_ids, position_ids=position_ids, attention_mask=attention_mask, inputs_embeds=inputs_embeds, past_key_values_length=past_key_values_length, ) encoder_outputs = self.encoder( embedding_output, attention_mask=extended_attention_mask, head_mask=head_mask, encoder_hidden_states=encoder_hidden_states, encoder_attention_mask=encoder_extended_attention_mask, past_key_values=past_key_values, use_cache=use_cache, output_attentions=output_attentions, output_hidden_states=output_hidden_states, return_dict=return_dict, ) sequence_output = encoder_outputs[0] return BaseModelOutputWithPoolingAndCrossAttentions( last_hidden_state=sequence_output, past_key_values=encoder_outputs.past_key_values, hidden_states=encoder_outputs.hidden_states, attentions=encoder_outputs.attentions, cross_attentions=encoder_outputs.cross_attentions, ) class ProPrimeForMaskedLM(ProPrimePreTrainedModel): _tied_weights_keys = ["lm_head.decoder.weight"] def __init__(self, config): super().__init__(config) if config.is_decoder: logger.warning( "If you want to use `ProPrimeForMaskedLM` make sure `config.is_decoder=False` for " "bi-directional self-attention." ) self.pro_prime = ProPrimeModel(config, add_pooling_layer=False) self.lm_head = ProPrimeLMHead(config) self.init_weights() def get_input_embeddings(self): return self.pro_prime.embeddings.word_embeddings def get_output_embeddings(self): return self.lm_head.decoder def set_output_embeddings(self, new_embeddings): self.lm_head.decoder = new_embeddings def forward( self, input_ids: Optional[torch.LongTensor] = None, attention_mask: Optional[torch.Tensor] = None, position_ids: Optional[torch.LongTensor] = None, head_mask: Optional[torch.Tensor] = None, inputs_embeds: Optional[torch.FloatTensor] = None, encoder_hidden_states: Optional[torch.FloatTensor] = None, encoder_attention_mask: Optional[torch.Tensor] = None, labels: Optional[torch.LongTensor] = None, output_attentions: Optional[bool] = None, output_hidden_states: Optional[bool] = None, return_dict: Optional[bool] = None, ) -> Union[Tuple, MaskedLMOutput]: return_dict = ( return_dict if return_dict is not None else self.config.use_return_dict ) outputs = self.pro_prime( input_ids, attention_mask=attention_mask, position_ids=position_ids, head_mask=head_mask, inputs_embeds=inputs_embeds, encoder_hidden_states=encoder_hidden_states, encoder_attention_mask=encoder_attention_mask, output_attentions=output_attentions, output_hidden_states=output_hidden_states, return_dict=return_dict, ) sequence_output = outputs[0] prediction_scores = self.lm_head(sequence_output) masked_lm_loss = None if labels is not None: loss_fct = CrossEntropyLoss() labels = labels.to(prediction_scores.device) masked_lm_loss = loss_fct( prediction_scores.view(-1, self.config.vocab_size), labels.view(-1) ) if not return_dict: output = (prediction_scores,) + outputs[2:] return ( ((masked_lm_loss,) + output) if masked_lm_loss is not None else output ) return MaskedLMOutput( loss=masked_lm_loss, logits=prediction_scores, hidden_states=outputs.hidden_states, attentions=outputs.attentions, ) class ProPrimeLMHead(nn.Module): def __init__(self, config): super().__init__() self.dense = nn.Linear(config.hidden_size, config.hidden_size) self.layer_norm = nn.LayerNorm(config.hidden_size, eps=config.layer_norm_eps) self.decoder = nn.Linear(config.hidden_size, config.vocab_size, bias=False) self.bias = nn.Parameter(torch.zeros(config.vocab_size)) def forward(self, features, **kwargs): x = self.dense(features) x = gelu(x) x = self.layer_norm(x) # project back to size of vocabulary with bias x = self.decoder(x) + self.bias return x class ProPrimeStructureHead(nn.Module): def __init__(self, config): super().__init__() self.dense = nn.Linear(config.hidden_size, config.hidden_size, bias=False) self.layer_norm = nn.LayerNorm(config.hidden_size, eps=config.layer_norm_eps) self.decoder = nn.Linear(config.hidden_size, config.structure_vocab_size, bias=False) self.bias = nn.Parameter(torch.zeros(config.structure_vocab_size)) def forward(self, features, **kwargs): x = self.dense(features) x = gelu(x) x = self.layer_norm(x) # project back to size of vocabulary with bias x = self.decoder(x) + self.bias return x def create_position_ids_from_input_ids( input_ids, padding_idx, past_key_values_length=0 ): """ Replace non-padding symbols with their position numbers. Position numbers begin at padding_idx+1. Padding symbols are ignored. This is modified from fairseq's `utils.make_positions`. Args: x: torch.Tensor x: Returns: torch.Tensor """ # The series of casts and type-conversions here are carefully balanced to both work with ONNX export and XLA. mask = input_ids.ne(padding_idx).int() incremental_indices = ( torch.cumsum(mask, dim=1).type_as(mask) + past_key_values_length ) * mask return incremental_indices.long() + padding_idx # POOLING_HEAD class MaskedConv1d(nn.Conv1d): """A masked 1-dimensional convolution layer. Takes the same arguments as torch.nn.Conv1D, except that the padding is set automatically. Shape: Input: (N, L, in_channels) input_mask: (N, L, 1), optional Output: (N, L, out_channels) """ def __init__( self, in_channels: int, out_channels: int, kernel_size: int, stride: int = 1, dilation: int = 1, groups: int = 1, bias: bool = True, ): """ :param in_channels: input channels :param out_channels: output channels :param kernel_size: the kernel width :param stride: filter shift :param dilation: dilation factor :param groups: perform depth-wise convolutions :param bias: adds learnable bias to output """ padding = dilation * (kernel_size - 1) // 2 super().__init__( in_channels, out_channels, kernel_size, stride=stride, dilation=dilation, groups=groups, bias=bias, padding=padding, ) def forward(self, x, input_mask=None): if input_mask is not None: x = x * input_mask return super().forward(x.transpose(1, 2)).transpose(1, 2) class Attention1d(nn.Module): def __init__(self, config): super().__init__() self.layer = MaskedConv1d(config.hidden_size, 1, 1) self.out = nn.Linear(config.hidden_size, config.hidden_size) def forward(self, x, input_mask=None): batch_szie = x.shape[0] attn = self.layer(x) attn = attn.view(batch_szie, -1) if input_mask is not None: attn = attn.masked_fill_( ~input_mask.view(batch_szie, -1).bool(), float("-inf") ) attn = F.softmax(attn, dim=-1).view(batch_szie, -1, 1) out = (attn * x).sum(dim=1) out = self.out(out) return out class FFN1d(nn.Module): def __init__(self, config): super().__init__() self.fc1 = nn.Linear(config.hidden_size, config.intermediate_size) self.fc2 = nn.Linear(config.intermediate_size, config.hidden_size) self.act = nn.GELU() def forward(self, x): x = self.fc1(x) x = self.act(x) x = self.fc2(x) return x class Attention1dPooling(nn.Module): """Outputs of the model with the attention1d""" def __init__( self, config ): # [batch x sequence(751) x embedding (1280)] --> [batch x embedding] --> [batch x 1] super(Attention1dPooling, self).__init__() self.attention1d = Attention1d(config) self.ffn = FFN1d(config) # self.norm1 = nn.BatchNorm1d(config.hidden_size) # self.norm2 = nn.BatchNorm1d(config.hidden_size) self.dropout1 = nn.Dropout(config.hidden_dropout_prob) self.dropout2 = nn.Dropout(config.hidden_dropout_prob) def forward(self, x, input_mask): attn_out = self.attention1d(x, input_mask=input_mask.unsqueeze(-1)) x = self.dropout1(attn_out) # x = self.norm1(x) ffn_out = self.ffn(x) x = x + self.dropout2(ffn_out) # x = self.norm2(x) return x @dataclass class MaskedLMOutput(ModelOutput): loss: Optional[torch.FloatTensor] = None mlm_loss: Optional[torch.FloatTensor] = None value_loss: Optional[torch.FloatTensor] = None predicted_values: Optional[torch.FloatTensor] = None logits: torch.FloatTensor = None sequence_hidden_states: Optional[Tuple[torch.FloatTensor, ...]] = None hidden_states: Optional[Tuple[torch.FloatTensor, ...]] = None attentions: Optional[Tuple[torch.FloatTensor, ...]] = None class ProPrimeMV(ProPrimePreTrainedModel): _tied_weights_keys = ["lm_head.decoder.weight"] def __init__(self, config): super().__init__(config) self.pro_prime = ProPrimeModel(config, add_pooling_layer=False) self.lm_head = ProPrimeLMHead(config) self.sequence_pooling = Attention1dPooling(config) self.value_projection = nn.Sequential( nn.Linear(config.hidden_size, config.hidden_size), nn.Tanh(), nn.Linear(config.hidden_size, 1), ) self.init_weights() def get_input_embeddings(self): return self.pro_prime.embeddings.word_embeddings def get_output_embeddings(self): return self.lm_head.decoder def set_output_embeddings(self, new_embeddings): self.lm_head.decoder = new_embeddings def forward( self, input_ids: Optional[torch.LongTensor] = None, attention_mask: Optional[torch.Tensor] = None, position_ids: Optional[torch.LongTensor] = None, head_mask: Optional[torch.Tensor] = None, inputs_embeds: Optional[torch.FloatTensor] = None, encoder_hidden_states: Optional[torch.FloatTensor] = None, encoder_attention_mask: Optional[torch.Tensor] = None, labels: Optional[torch.LongTensor] = None, values: Optional[torch.FloatTensor] = None, output_attentions: Optional[bool] = None, output_hidden_states: Optional[bool] = None, return_dict: Optional[bool] = None, ) -> Union[Tuple, MaskedLMOutput]: return_dict = ( return_dict if return_dict is not None else self.config.use_return_dict ) outputs = self.pro_prime( input_ids, attention_mask=attention_mask, position_ids=position_ids, head_mask=head_mask, inputs_embeds=inputs_embeds, encoder_hidden_states=encoder_hidden_states, encoder_attention_mask=encoder_attention_mask, output_attentions=output_attentions, output_hidden_states=output_hidden_states, return_dict=return_dict, ) sequence_output = outputs[0] prediction_scores = self.lm_head(sequence_output) masked_lm_loss = None if labels is not None: loss_fct = CrossEntropyLoss() labels = labels.to(prediction_scores.device) masked_lm_loss = loss_fct( prediction_scores.view(-1, self.config.vocab_size), labels.view(-1) ) if not return_dict: output = (prediction_scores,) + outputs[2:] return ( ((masked_lm_loss,) + output) if masked_lm_loss is not None else output ) if values is not None: sequence_states = self.sequence_pooling(sequence_output, attention_mask) predicted_values = self.value_projection(sequence_states) values = values.to(predicted_values.dtype) values = values.reshape(-1, 1) value_loss = nn.MSELoss()(predicted_values, values) loss = masked_lm_loss + 0.01 * value_loss else: sequence_states = self.sequence_pooling(sequence_output, attention_mask) predicted_values = self.value_projection(sequence_states) value_loss = None loss = masked_lm_loss return MaskedLMOutput( loss=loss, mlm_loss=masked_lm_loss, value_loss=value_loss, logits=prediction_scores, predicted_values=predicted_values.reshape(-1), hidden_states=outputs.hidden_states, sequence_hidden_states=sequence_states, attentions=outputs.attentions, ) @dataclass class PretrainedOutput(ModelOutput): loss: Optional[torch.FloatTensor] = None mlm_loss: Optional[torch.FloatTensor] = None structure_loss: Optional[torch.FloatTensor] = None corr_loss: Optional[torch.FloatTensor] = None value_loss: Optional[torch.FloatTensor] = None predicted_values: Optional[torch.FloatTensor] = None logits: torch.FloatTensor = None structure_logits: torch.FloatTensor = None sequence_hidden_states: Optional[Tuple[torch.FloatTensor, ...]] = None hidden_states: Optional[Tuple[torch.FloatTensor, ...]] = None attentions: Optional[Tuple[torch.FloatTensor, ...]] = None class ProPrimeForPretraining(ProPrimePreTrainedModel): _tied_weights_keys = ["lm_head.decoder.weight"] base_model_prefix = "proprime" def __init__(self, config): super().__init__(config) self.pro_prime = ProPrimeModel(config, add_pooling_layer=False) self.lm_head = ProPrimeLMHead(config) self.structure_head = ProPrimeStructureHead(config) self.sequence_pooling = Attention1dPooling(config) self.value_projection = nn.Sequential( nn.Linear(config.hidden_size, config.hidden_size), nn.Tanh(), nn.Linear(config.hidden_size, 1), ) self.init_weights() def get_input_embeddings(self): return self.pro_prime.embeddings.word_embeddings def get_output_embeddings(self): return self.lm_head.decoder def set_output_embeddings(self, new_embeddings): self.lm_head.decoder = new_embeddings def forward( self, input_ids: Optional[torch.LongTensor] = None, attention_mask: Optional[torch.Tensor] = None, position_ids: Optional[torch.LongTensor] = None, head_mask: Optional[torch.Tensor] = None, inputs_embeds: Optional[torch.FloatTensor] = None, encoder_hidden_states: Optional[torch.FloatTensor] = None, encoder_attention_mask: Optional[torch.Tensor] = None, labels: Optional[torch.LongTensor] = None, structure_labels: Optional[torch.LongTensor] = None, values: Optional[torch.FloatTensor] = None, mutant_input_ids: Optional[torch.LongTensor] = None, # Corr mutant_index: Optional[torch.LongTensor] = None, # Corr mutant_type: Optional[torch.LongTensor] = None, # Corr wild_type: Optional[torch.LongTensor] = None, # Corr output_attentions: Optional[bool] = None, output_hidden_states: Optional[bool] = None, return_dict: Optional[bool] = None, ) -> Union[Tuple, MaskedLMOutput]: return_dict = ( return_dict if return_dict is not None else self.config.use_return_dict ) outputs = self.pro_prime( input_ids, attention_mask=attention_mask, position_ids=position_ids, head_mask=head_mask, inputs_embeds=inputs_embeds, encoder_hidden_states=encoder_hidden_states, encoder_attention_mask=encoder_attention_mask, output_attentions=output_attentions, output_hidden_states=output_hidden_states, return_dict=return_dict, ) sequence_output = outputs[0] mlm_scores = self.lm_head(sequence_output) structure_scores = self.structure_head(sequence_output) sequence_states = self.sequence_pooling(sequence_output, attention_mask) predicted_values = self.value_projection(sequence_states) loss = 0 if mutant_input_ids is not None: with torch.no_grad(): mutant_outputs = self.pro_prime( mutant_input_ids, attention_mask=attention_mask, position_ids=position_ids, head_mask=head_mask, inputs_embeds=inputs_embeds, encoder_hidden_states=encoder_hidden_states, encoder_attention_mask=encoder_attention_mask, output_attentions=output_attentions, output_hidden_states=output_hidden_states, return_dict=return_dict, ) mutant_sequence_output = mutant_outputs[0] mutant_sequence_states = self.sequence_pooling(mutant_sequence_output, attention_mask) mutant_predicted_values = self.value_projection(mutant_sequence_states) values_diff = mutant_predicted_values - predicted_values logits = mlm_scores.log_softmax(dim=-1) mt_probs = logits[torch.arange(logits.size(0)), mutant_index, mutant_type] wt_probs = logits[torch.arange(logits.size(0)), mutant_index, wild_type] mutant_effects = mt_probs - wt_probs corr_loss = consine_based_loss(values_diff.squeeze(), mutant_effects.squeeze()) loss += corr_loss else: corr_loss = None if labels is not None: loss_fct = CrossEntropyLoss() labels = labels.to(mlm_scores.device) mlm_loss = loss_fct( mlm_scores.view(-1, self.config.vocab_size), labels.view(-1) ) loss += mlm_loss else: mlm_loss = None if structure_labels is not None: loss_fct = CrossEntropyLoss() structure_labels = structure_labels.to(structure_scores.device) structure_loss = loss_fct( structure_scores.view(-1, self.config.structure_vocab_size), structure_labels.view(-1) ) loss += structure_loss else: structure_loss = None if values is not None: loss_fct = nn.MSELoss() values = values.to(predicted_values.dtype) values = values.reshape(-1, 1) value_loss = nn.MSELoss()(predicted_values, values) loss += 0.01 * value_loss else: value_loss = None return PretrainedOutput( loss=loss, mlm_loss=mlm_loss, structure_loss=structure_loss, value_loss=value_loss, corr_loss=corr_loss, logits=mlm_scores, structure_logits=structure_scores, predicted_values=predicted_values, hidden_states=outputs.hidden_states, sequence_hidden_states=sequence_states, attentions=outputs.attentions, ) ProPrimeForMaskedLM.register_for_auto_class("AutoModelForMaskedLM") ProPrimeForPretraining.register_for_auto_class("AutoModel")