app and model

app.py

@@ -0,0 +1,56 @@
+import gradio as gr
+
+import numpy as np
+import torch
+from transformers import AutoTokenizer, Pipeline
+
+from inference_utils import prepare_stance_texts
+from models import StanceEncoderModel
+
+CLASS_DICT = {0: 'FAVOR', 1: 'AGAINST', 2: 'NEITHER'}
+
+params = {'lang': 'pl',
+          'masked_lm_prompt': 4,}
+
+
+class StancePipeline(Pipeline):
+
+    def _sanitize_parameters(self, **pipeline_parameters):
+        return pipeline_parameters, {}, {}
+
+    def preprocess(self, input):
+        prompt_text, prompt_target = prepare_stance_texts([input['text'],], [input['target'],], params, self.tokenizer)
+        inputs = self.tokenizer(prompt_text, prompt_target, return_tensors="pt", padding=True, truncation='only_first')
+        return {'input_ids': inputs['input_ids'], 'attention_mask': inputs['attention_mask'],
+                'sequence_ids': torch.tensor((np.array(inputs.sequence_ids()) == 1).astype(int)).unsqueeze(0)}
+
+    def _forward(self, model_inputs):
+        outputs = self.model(**model_inputs)
+        return outputs
+
+    def postprocess(self, model_outputs):
+        probas = model_outputs["logits"].softmax(-1)
+        score = probas.max(-1)[0].item()
+        return {'stance': CLASS_DICT[probas.argmax(-1).item()], 'score': score}
+
+
+pipeline = StancePipeline(model=StanceEncoderModel.from_pretrained('clarin-knext/stance-pl-1'),
+                          tokenizer=AutoTokenizer.from_pretrained('clarin-knext/stance-pl-1'),
+                          batch_size=1)
+
+
+def predict(text, target):
+    predictions = pipeline({'text': text, 'target': target})
+    return predictions['stance'], predictions['score']
+
+
+gradio_app = gr.Interface(
+    predict,
+    inputs=[gr.TextArea(label="Text", placeholder="text"),
+            gr.Textbox(label="Target", placeholder="stance target")],
+    outputs=[gr.Label(label="Stance"), gr.Label(label="Score")],
+    title="Polish stance detection",
+)
+
+if __name__ == "__main__":
+    gradio_app.launch()

inference_utils.py

@@ -0,0 +1,11 @@
+from typing import List, Dict
+import pandas as pd
+
+from preprocessing import format_masked_lm_prompt
+
+
+def prepare_stance_texts(texts: List[str], targets: List[str], params: Dict[str, any], tokenizer):
+    texts_df = pd.DataFrame({'text': texts, 'target': targets})
+    prompt_text, prompt_target = format_masked_lm_prompt(texts_df, tokenizer=tokenizer,
+                                                         prompt_type=params['masked_lm_prompt'], lang='pl',)
+    return prompt_text, prompt_target

models.py

@@ -0,0 +1,85 @@
+import logging
+from typing import Optional, Tuple
+from typing import Union
+
+import torch
+from torch.nn import CrossEntropyLoss
+from transformers import PreTrainedModel, BertForMaskedLM, BertConfig
+from transformers.modeling_outputs import SequenceClassifierOutput
+
+
+class StanceEncoderModel(PreTrainedModel):
+
+    config_class = BertConfig
+    logger = logging.getLogger("StanceEncoderModel")
+
+    def __init__(self, config):
+        super().__init__(config)
+        task_specific_params = config.task_specific_params
+        self.num_labels = task_specific_params.get('num_labels', 3)
+        self.mask_token_id = task_specific_params['mask_token_id']
+        self.verbalizer_token_ids = task_specific_params['verbalizer_token_ids']
+        self.clf_hidden_dim = task_specific_params.get('clf_hidden_dim', 300)
+        self.clf_drop_prob = task_specific_params.get('clf_drop_prob', 0.2)
+        self.clf_gelu_head = task_specific_params.get('clf_gelu_head', False)
+        self.masked_lm = task_specific_params.get('masked_lm', True)
+        self.masked_lm_n_tokens = task_specific_params.get('masked_lm_tokens', 1)
+        self.masked_lm_verbalizer = task_specific_params.get('masked_lm_verbalizer', False)
+
+        base_model = BertForMaskedLM(config)
+        self.base_enc_model = base_model.bert
+        self.lm_head = base_model.cls
+
+        hidden_size_multiplier = 1
+
+        if not self.masked_lm_verbalizer:
+            if self.clf_gelu_head:
+                self.logger.info('using 2 layer gelu classifier head')
+                self.classifier = torch.nn.Sequential(
+                    torch.nn.Linear(self.config.hidden_size * hidden_size_multiplier, self.clf_hidden_dim),
+                    torch.nn.Dropout(self.clf_drop_prob),
+                    torch.nn.GELU(),
+                    torch.nn.Linear(self.clf_hidden_dim, self.num_labels)
+                )
+            else:
+                raise ValueError('classification type head not specified')
+
+    def forward(
+            self,
+            input_ids: Optional[torch.Tensor] = None,
+            attention_mask: Optional[torch.Tensor] = None,
+            token_type_ids: Optional[torch.Tensor] = None,
+            position_ids: Optional[torch.Tensor] = None,
+            head_mask: Optional[torch.Tensor] = None,
+            inputs_embeds: Optional[torch.Tensor] = None,
+            labels: Optional[torch.Tensor] = None,
+            output_attentions: Optional[bool] = None,
+            output_hidden_states: Optional[bool] = None,
+            return_dict: Optional[bool] = None,
+            sequence_ids: Optional[torch.Tensor] = None,
+    ) -> Union[Tuple[torch.Tensor], SequenceClassifierOutput]:
+
+        outputs = self.base_enc_model(
+            input_ids=input_ids,
+            attention_mask=attention_mask
+        )
+
+        masked_token_filter = input_ids == self.mask_token_id
+        masked_repr = outputs.last_hidden_state[masked_token_filter].reshape(len(input_ids), -1)
+
+        if self.masked_lm_verbalizer:
+            logits = self.lm_head(masked_repr)[:, self.verbalizer_token_ids]
+        else:
+            logits = self.classifier(masked_repr)
+
+        loss = None
+        if labels is not None:
+            loss_fct = CrossEntropyLoss()
+            loss = loss_fct(logits.view(-1, self.num_labels), labels.view(-1))
+
+        return SequenceClassifierOutput(
+            loss=loss,
+            logits=logits,
+            hidden_states=outputs.hidden_states,
+            attentions=outputs.attentions,
+        )

preprocessing.py

@@ -0,0 +1,22 @@
+import logging
+
+logger = logging.getLogger('stancedatasets')
+
+
+def format_masked_lm_prompt(stance_df, tokenizer, prompt_type, lang='en', masked_lm_tokens=1):
+    masks_str = "".join([tokenizer.mask_token] * masked_lm_tokens)
+    if lang == 'pl':
+        if prompt_type == 1:
+            return list(stance_df['text']), \
+                list('Moja postawa w kierunku ' + stance_df['target'] + ' jest: ' + masks_str + '.')
+        elif prompt_type == 2:
+            return list(stance_df['text']), \
+                list('Moja postawa w kierunku ' + stance_df['target'] + ' jest ' + masks_str + '.')
+        elif prompt_type == 3:
+            return list(stance_df['text']), \
+                list('Więc moja postawa w kierunku ' + stance_df['target'] + ' jest: ' + masks_str + '.')
+        elif prompt_type == 4:
+            return list(stance_df['text']), \
+                list('Więc moja postawa w kierunku ' + stance_df['target'] + ' jest ' + masks_str + '.')
+
+    raise ValueError(f'unknown prompt_type: {prompt_type} for language {lang}')