streamlit for QCRI/PropagandaTechniquesAnalysis-en-BERT

app.py

@@ -0,0 +1,29 @@
+import streamlit as st
+import torch
+from transformers import BertTokenizerFast
+
+from model import BertForTokenAndSequenceJointClassification
+
+@st.cache(allow_output_mutation=True)
+def load_model():
+    tokenizer = BertTokenizerFast.from_pretrained('bert-base-cased')
+    model = BertForTokenAndSequenceJointClassification.from_pretrained(
+            "QCRI/PropagandaTechniquesAnalysis-en-BERT",
+             revision="v0.1.0")
+    return tokenizer, model
+    
+tokenizer, model = load_model()
+
+input = st.text_area('Input', """\
+In some instances, it can be highly dangerous to use a medicine for the prevention or treatment of COVID-19 that has not been approved by or has not received emergency use authorization from the FDA.
+""")
+
+inputs = tokenizer.encode_plus(input, return_tensors="pt")
+outputs = model(**inputs)
+sequence_class_index = torch.argmax(outputs.sequence_logits, dim=-1)
+sequence_class = model.sequence_tags[sequence_class_index[0]]
+token_class_index = torch.argmax(outputs.token_logits, dim=-1)
+tokens = tokenizer.convert_ids_to_tokens(inputs.input_ids[0][1:-1])
+tags = [model.token_tags[i] for i in token_class_index[0].tolist()[1:-1]]
+
+st.table(list(zip(tokens, tags)))

model.py

@@ -0,0 +1,128 @@
+__author__ = "Yifan Zhang (yzhang@hbku.edu.qa)"
+__copyright__ = "Copyright (C) 2021, Qatar Computing Research Institute, HBKU, Doha"
+
+
+from dataclasses import dataclass
+from typing import Optional, Tuple
+import torch
+from torch import nn
+from torch.nn.functional import sigmoid
+from transformers import BertPreTrainedModel, BertModel
+from transformers.file_utils import ModelOutput
+
+
+TOKEN_TAGS = (
+    "<PAD>", "O", 
+    "Name_Calling,Labeling", "Repetition", "Slogans", "Appeal_to_fear-prejudice", "Doubt",
+    "Exaggeration,Minimisation", "Flag-Waving", "Loaded_Language",
+    "Reductio_ad_hitlerum", "Bandwagon",
+    "Causal_Oversimplification", "Obfuscation,Intentional_Vagueness,Confusion", "Appeal_to_Authority", "Black-and-White_Fallacy",
+    "Thought-terminating_Cliches", "Red_Herring", "Straw_Men", "Whataboutism"
+)
+
+
+SEQUENCE_TAGS = ("Non-prop", "Prop")
+
+
+@dataclass
+class TokenAndSequenceJointClassifierOutput(ModelOutput):
+    loss: Optional[torch.FloatTensor] = None
+    token_logits: torch.FloatTensor = None
+    sequence_logits: torch.FloatTensor = None
+    hidden_states: Optional[Tuple[torch.FloatTensor]] = None
+    attentions: Optional[Tuple[torch.FloatTensor]] = None
+
+
+class BertForTokenAndSequenceJointClassification(BertPreTrainedModel):
+
+    def __init__(self, config):
+        super().__init__(config)
+        self.num_token_labels = 20
+        self.num_sequence_labels = 2
+
+        self.token_tags = TOKEN_TAGS
+        self.sequence_tags = SEQUENCE_TAGS 
+
+        self.alpha = 0.9
+
+        self.bert = BertModel(config)
+        self.dropout = nn.Dropout(config.hidden_dropout_prob)
+        self.classifier = nn.ModuleList([
+            nn.Linear(config.hidden_size, self.num_token_labels),
+            nn.Linear(config.hidden_size, self.num_sequence_labels),
+        ])
+        self.masking_gate = nn.Linear(2, 1)
+
+        self.init_weights()
+        self.merge_classifier_1 = nn.Linear(self.num_token_labels + self.num_sequence_labels, self.num_token_labels)
+
+    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=True,
+    ):
+        return_dict = return_dict if return_dict is not None else self.config.use_return_dict
+
+        outputs = self.bert(
+            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=output_hidden_states,
+        )
+
+        sequence_output = outputs[0]
+        pooler_output = outputs[1]
+
+        sequence_output = self.dropout(sequence_output)
+        token_logits = self.classifier[0](sequence_output)
+
+        pooler_output = self.dropout(pooler_output)
+        sequence_logits = self.classifier[1](pooler_output)
+
+        gate = torch.sigmoid(self.masking_gate(sequence_logits))
+
+        gates = gate.unsqueeze(1).repeat(1, token_logits.size()[1], token_logits.size()[2])
+
+        weighted_token_logits = torch.mul(gates, token_logits)
+
+        logits = [weighted_token_logits, sequence_logits]
+
+        loss = None
+        if labels is not None:
+            criterion = nn.CrossEntropyLoss(ignore_index=0)
+            binary_criterion = nn.BCEWithLogitsLoss(pos_weight=torch.Tensor([3932/14263]).cuda())
+            loss_fct = CrossEntropyLoss()
+            weighted_token_logits = weighted_token_logits.view(-1, weighted_token_logits.shape[-1])
+            sequence_logits = sequence_logits.view(-1, sequence_logits.shape[-1])
+
+            token_loss = criterion(weighted_token_logits, labels)
+            sequence_label = torch.LongTensor([1] if any([label > 0 for label in labels]) else [0])
+            sequence_loss = binary_criterion(sequence_logits, sequence_label)
+
+            loss = self.alpha*loss[0] + (1-self.alpha)*loss[1]
+
+        if not return_dict:
+            output = (logits,) + outputs[2:]
+            return ((loss,) + output) if loss is not None else output
+
+        return TokenAndSequenceJointClassifierOutput(
+            loss=loss,
+            token_logits=weighted_token_logits,
+            sequence_logits=sequence_logits,
+            hidden_states=outputs.hidden_states,
+            attentions=outputs.attentions,
+        )
+
+

requirements.txt

@@ -0,0 +1,3 @@
+streamlit
+transformers
+torch