Upload folder using huggingface_hub

config.json

@@ -0,0 +1,41 @@
+{
+  "_name_or_path": "cardiffnlp/twitter-roberta-base-sentiment",
+  "architectures": [
+    "RobertaForCL"
+  ],
+  "auto_map": {
+    "AutoModel": "modeling_roberta_cl.RobertaForCL"
+  },
+  "attention_probs_dropout_prob": 0.1,
+  "bos_token_id": 0,
+  "classifier_dropout": null,
+  "eos_token_id": 2,
+  "gradient_checkpointing": false,
+  "hidden_act": "gelu",
+  "hidden_dropout_prob": 0.1,
+  "hidden_size": 768,
+  "id2label": {
+    "0": "LABEL_0",
+    "1": "LABEL_1",
+    "2": "LABEL_2"
+  },
+  "initializer_range": 0.02,
+  "intermediate_size": 3072,
+  "label2id": {
+    "LABEL_0": 0,
+    "LABEL_1": 1,
+    "LABEL_2": 2
+  },
+  "layer_norm_eps": 1e-05,
+  "max_position_embeddings": 514,
+  "model_type": "roberta",
+  "num_attention_heads": 12,
+  "num_hidden_layers": 12,
+  "pad_token_id": 1,
+  "position_embedding_type": "absolute",
+  "torch_dtype": "float32",
+  "transformers_version": "4.48.1",
+  "type_vocab_size": 1,
+  "use_cache": true,
+  "vocab_size": 50265
+}

model.safetensors

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:06626dbe31b7e4b4ebb273081631608f988c5c8d7345b90aff0190d04f2c4de5
+size 503080724

modeling_roberta_cl.py

@@ -0,0 +1,395 @@
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import torch.distributed as dist
+from torch import Tensor
+
+import transformers
+from transformers import RobertaTokenizer
+from transformers.models.roberta.modeling_roberta import RobertaForSequenceClassification, RobertaClassificationHead, RobertaLMHead
+from transformers.activations import gelu
+from transformers.file_utils import (
+    add_code_sample_docstrings,
+    add_start_docstrings,
+    add_start_docstrings_to_model_forward,
+    replace_return_docstrings,
+)
+from transformers.modeling_outputs import SequenceClassifierOutput, BaseModelOutputWithPoolingAndCrossAttentions
+
+class MLPLayer(nn.Module):
+    """
+    Head for getting sentence representations over RoBERTa/BERT's CLS representation.
+    """
+
+    def __init__(self, config):
+        super().__init__()
+        self.dense = nn.Linear(config.hidden_size, config.hidden_size)
+        self.activation = nn.Tanh()
+
+    def forward(self, features, **kwargs):
+        x = self.dense(features)
+        x = self.activation(x)
+
+        return x
+
+class ResidualBlock(nn.Module):
+    def __init__(self, dim):
+        super(ResidualBlock, self).__init__()
+        self.fc = nn.Linear(dim, dim)
+        self.relu = nn.ReLU()
+
+    def forward(self, x):
+        out = self.fc(x)
+        out = self.relu(out)
+        out = out + x 
+        return out
+
+class SemanticModel(nn.Module):
+    def __init__(self, num_layers=2, input_dim=768, hidden_dim=512, output_dim=384):
+        super(SemanticModel, self).__init__()
+        
+        self.layers = nn.ModuleList()
+        
+        self.layers.append(nn.Linear(input_dim, hidden_dim))
+        
+        for _ in range(num_layers):
+            self.layers.append(ResidualBlock(hidden_dim))
+
+        self.layers.append(nn.Linear(hidden_dim, output_dim))
+
+    def forward(self, x):
+        for i in range(len(self.layers)):
+            x = self.layers[i](x)
+        
+        return x
+
+class Similarity(nn.Module):
+    """
+    Dot product or cosine similarity
+    """
+
+    def __init__(self, temp):
+        super().__init__()
+        self.temp = temp
+        self.cos = nn.CosineSimilarity(dim=-1)
+
+    def forward(self, x, y):
+        return self.cos(x, y) / self.temp
+
+
+class RobertaClassificationHeadForEmbedding(RobertaClassificationHead):
+    """Head for sentence-level classification tasks."""
+
+    def __init__(self, config):
+        super().__init__(config)
+        self.dense = nn.Linear(config.hidden_size, config.hidden_size)
+        classifier_dropout = (
+            config.classifier_dropout if config.classifier_dropout is not None else config.hidden_dropout_prob
+        )
+        self.dropout = nn.Dropout(classifier_dropout)
+        self.out_proj = nn.Linear(config.hidden_size, config.num_labels)
+
+    def forward(self, features, **kwargs):
+        x = features[:, 0, :]  # take <s> token (equiv. to [CLS])
+        x = self.dropout(x)
+        x = self.dense(x)
+        # x = torch.tanh(x)
+        # x = self.dropout(x)
+        # x = self.out_proj(x)
+        return x
+    
+def cl_init(cls, config):
+    """
+    Contrastive learning class init function.
+    """
+    cls.sim = Similarity(temp=cls.model_args.temp)
+    cls.init_weights()
+
+def remove_diagonal_elements(input_tensor):
+    """
+    Removes the diagonal elements from a square matrix (bs, bs) 
+    and returns a new matrix of size (bs, bs-1).
+    """
+    if input_tensor.size(0) != input_tensor.size(1):
+        raise ValueError("Input tensor must be square (bs, bs).")
+    
+    bs = input_tensor.size(0)
+    mask = ~torch.eye(bs, dtype=torch.bool, device=input_tensor.device)  # Mask for non-diagonal elements
+    output_tensor = input_tensor[mask].view(bs, bs - 1)  # Reshape into (bs, bs-1)
+    return output_tensor
+
+def cl_forward(cls,
+    input_ids=None,
+    attention_mask=None,
+    token_type_ids=None,
+    position_ids=None,
+    head_mask=None,
+    inputs_embeds=None,
+    labels=None,
+    output_attentions=None,
+    output_hidden_states=None,
+    return_dict=None,
+    mlm_input_ids=None,
+    mlm_labels=None,
+    latter_sentiment_spoof_mask=None,
+):
+    return_dict = return_dict if return_dict is not None else cls.config.use_return_dict
+    batch_size = input_ids.size(0)
+    # Number of sentences in one instance
+    # original + cls.model_args.num_paraphrased + cls.model_args.num_negative
+    num_sent = input_ids.size(1)
+
+    mlm_outputs = None
+    # Flatten input for encoding
+    input_ids = input_ids.view((-1, input_ids.size(-1))) # (bs * num_sent, len)
+    attention_mask = attention_mask.view((-1, attention_mask.size(-1))) # (bs * num_sent len)
+    if token_type_ids is not None:
+        token_type_ids = token_type_ids.view((-1, token_type_ids.size(-1))) # (bs * num_sent, len)
+    
+    # Get raw embeddings
+    outputs = cls.roberta(
+        input_ids,
+        attention_mask=attention_mask,
+        token_type_ids=token_type_ids,
+        position_ids=position_ids,
+        head_mask=head_mask,
+        inputs_embeds=inputs_embeds,
+        output_attentions=output_attentions,
+        output_hidden_states=False,
+        return_dict=True,
+    )
+
+    # MLM auxiliary objective
+    if mlm_input_ids is not None:
+        mlm_input_ids = mlm_input_ids.view((-1, mlm_input_ids.size(-1)))
+        mlm_outputs = cls.roberta(
+            mlm_input_ids,
+            attention_mask=attention_mask,
+            token_type_ids=token_type_ids,
+            position_ids=position_ids,
+            head_mask=head_mask,
+            inputs_embeds=inputs_embeds,
+            output_attentions=output_attentions,
+            output_hidden_states=False,
+            return_dict=True,
+        )
+
+    # Pooling
+    sequence_output = outputs[0]  # (bs*num_sent, seq_len, hidden)
+    pooler_output = cls.classifier(sequence_output)  # (bs*num_sent, hidden)
+    pooler_output = pooler_output.view((batch_size, num_sent, pooler_output.size(-1))) # (bs, num_sent, hidden)
+    
+    # Mapping
+    pooler_output = cls.map(pooler_output)  # (bs, num_sent, hidden_states)
+        
+    # Separate representation
+    original = pooler_output[:, 0]
+    paraphrase_list = [pooler_output[:, i] for i in range(1, cls.model_args.num_paraphrased + 1)]
+    if cls.model_args.num_negative == 0:
+        negative_list = []
+    else:
+        negative_list = [pooler_output[:, i] for i in range(cls.model_args.num_paraphrased + 1, cls.model_args.num_paraphrased + cls.model_args.num_negative + 1)]
+
+    # Gather all embeddings if using distributed training
+    if dist.is_initialized() and cls.training:
+        raise NotImplementedError
+    
+    # get sign value before calculating similarity
+    original = torch.tanh(original * 1000)
+    paraphrase_list = [torch.tanh(p * 1000) for p in paraphrase_list]
+    negative_list = [torch.tanh(n * 1000) for n in negative_list]
+    spoofing_cnames = cls.model_args.spoofing_cnames
+    negative_dict = {}
+    for cname, n in zip(spoofing_cnames, negative_list):
+        negative_dict[cname] = n
+
+    # Calculate triplet loss
+    loss_triplet = 0
+    for i in range(batch_size):
+        for j in range(cls.model_args.num_paraphrased):
+            for cname in spoofing_cnames:
+                if cname == 'latter_sentiment_spoof_0' and latter_sentiment_spoof_mask[i] == 0:
+                    continue
+                ori = original[i]
+                pos = paraphrase_list[j][i]
+                neg = negative_dict[cname][i]
+                loss_triplet += F.relu(cls.sim(ori, neg) * cls.model_args.temp  - cls.sim(ori, pos) * cls.model_args.temp  + cls.model_args.margin)
+    loss_triplet /= (batch_size * cls.model_args.num_paraphrased * len(spoofing_cnames))
+
+    # Calculate loss for MLM
+    if mlm_outputs is not None and mlm_labels is not None:
+        raise NotImplementedError
+        # mlm_labels = mlm_labels.view(-1, mlm_labels.size(-1))
+        # prediction_scores = cls.lm_head(mlm_outputs.last_hidden_state)
+        # masked_lm_loss = loss_fct(prediction_scores.view(-1, cls.config.vocab_size), mlm_labels.view(-1))
+        # loss_cl = loss_cl + cls.model_args.mlm_weight * masked_lm_loss
+    
+    # Calculate loss for uniform perturbation and unbiased token preference
+    def sign_loss(x):
+        row = torch.abs(torch.mean(torch.mean(x, dim=0)))
+        col = torch.abs(torch.mean(torch.mean(x, dim=1)))
+        return (row + col)/2
+
+    loss_gr = sign_loss(original)
+
+    # calculate loss_3: similarity between original and paraphrased text
+    loss_3_list = [cls.sim(original, p).unsqueeze(1) for p in paraphrase_list]  # [(bs, 1)] * num_paraphrased
+    loss_3_tensor = torch.cat(loss_3_list, dim=1)  # (bs, num_paraphrased)
+    loss_3 = loss_3_tensor.mean() * cls.model_args.temp
+
+    # calculate loss_sent: similarity between original and sentiment spoofed text
+    negative_sample_loss = {}
+    for cname in spoofing_cnames:
+        negatives = negative_dict[cname]
+        originals = original.clone()
+        if cname == 'latter_sentiment_spoof_0':
+            negatives = negatives[latter_sentiment_spoof_mask == 1]
+            originals = originals[latter_sentiment_spoof_mask == 1]
+        one_negative_loss = cls.sim(originals, negatives).mean() * cls.model_args.temp
+        negative_sample_loss[cname] = one_negative_loss
+
+    # calculate loss_5: similarity between original and other original text
+    ori_ori_cos = cls.sim(original.unsqueeze(1), original.unsqueeze(0))  # (bs, bs)
+    ori_ori_cos_removed = remove_diagonal_elements(ori_ori_cos)  # (bs, bs-1)
+    loss_5 = ori_ori_cos_removed.mean() * cls.model_args.temp
+
+    loss = loss_gr + loss_triplet
+
+    result = {
+        'loss': loss,
+        'loss_gr': loss_gr,
+        'sim_paraphrase': loss_3,
+        'sim_other': loss_5,
+        'hidden_states': outputs.hidden_states,
+        'attentions': outputs.attentions,
+    }
+
+    for cname, l in negative_sample_loss.items():
+        key = f"sim_{cname.replace('_spoof_0', '')}"
+        result[key] = l
+
+    result['loss_tl'] = loss_triplet
+
+    if not return_dict:
+        raise NotImplementedError
+        # output = (cos_sim,) + outputs[2:]
+        # return ((loss,) + output) if loss is not None else output
+    return result
+
+
+def sentemb_forward(
+    cls,
+    input_ids=None,
+    attention_mask=None,
+    token_type_ids=None,
+    position_ids=None,
+    head_mask=None,
+    inputs_embeds=None,
+    labels=None,
+    output_attentions=None,
+    output_hidden_states=None,
+    return_dict=None,
+):
+
+    return_dict = return_dict if return_dict is not None else cls.config.use_return_dict
+
+    outputs = cls.roberta(
+        input_ids,
+        attention_mask=attention_mask,
+        token_type_ids=token_type_ids,
+        position_ids=position_ids,
+        head_mask=head_mask,
+        inputs_embeds=inputs_embeds,
+        output_attentions=output_attentions,
+        output_hidden_states=False,
+        return_dict=True,
+    )
+    sequence_output = outputs[0]
+    pooler_output = cls.classifier(sequence_output)
+
+    # Mapping
+    mapping_output = cls.map(pooler_output)
+    pooler_output = mapping_output
+        
+
+    if not return_dict:
+        return (outputs[0], pooler_output) + outputs[2:]
+
+    return BaseModelOutputWithPoolingAndCrossAttentions(
+        pooler_output=pooler_output,
+        last_hidden_state=outputs.last_hidden_state,
+        hidden_states=outputs.hidden_states,
+    )
+
+
+class RobertaForCL(RobertaForSequenceClassification):
+    _keys_to_ignore_on_load_missing = [r"position_ids"]
+
+    def __init__(self, config, *model_args, **model_kargs):
+        super().__init__(config)
+        self.model_args = model_kargs.get("model_args", None)
+
+        self.classifier = RobertaClassificationHeadForEmbedding(config)
+
+        if self.model_args and getattr(self.model_args, "do_mlm", False):
+            self.lm_head = RobertaLMHead(config)
+            cl_init(self, config)
+
+        self.map = SemanticModel(input_dim=768)
+
+        # Initialize weights and apply final processing
+        self.post_init()
+
+    def initialize_mlp_weights(self, pretrained_model_state_dict):
+        """
+        Initialize MLP weights using the pretrained classifier's weights.
+        """
+        self.mlp.dense.weight.data = pretrained_model_state_dict.classifier.dense.weight.data.clone()
+        self.mlp.dense.bias.data = pretrained_model_state_dict.classifier.dense.bias.data.clone()
+
+    def forward(self,
+        input_ids=None,
+        attention_mask=None,
+        token_type_ids=None,
+        position_ids=None,
+        head_mask=None,
+        inputs_embeds=None,
+        labels=None,
+        output_attentions=None,
+        output_hidden_states=None,
+        return_dict=None,
+        sent_emb=False,
+        mlm_input_ids=None,
+        mlm_labels=None,
+        latter_sentiment_spoof_mask=None,
+    ):
+        if sent_emb:
+            return sentemb_forward(self,
+                input_ids=input_ids,
+                attention_mask=attention_mask,
+                token_type_ids=token_type_ids,
+                position_ids=position_ids,
+                head_mask=head_mask,
+                inputs_embeds=inputs_embeds,
+                labels=labels,
+                output_attentions=output_attentions,
+                output_hidden_states=output_hidden_states,
+                return_dict=return_dict,
+            )
+        else:
+            return cl_forward(self,
+                input_ids=input_ids,
+                attention_mask=attention_mask,
+                token_type_ids=token_type_ids,
+                position_ids=position_ids,
+                head_mask=head_mask,
+                inputs_embeds=inputs_embeds,
+                labels=labels,
+                output_attentions=output_attentions,
+                output_hidden_states=output_hidden_states,
+                return_dict=return_dict,
+                mlm_input_ids=mlm_input_ids,
+                mlm_labels=mlm_labels,
+                latter_sentiment_spoof_mask=latter_sentiment_spoof_mask,
+            )
+