handler.py

from typing import Dict, List, Any, Optional, Tuple, Union
from dataclasses import dataclass
import torch
from torch import nn
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss
import numpy as np
import transformers
from transformers import AutoTokenizer, BertTokenizer
from transformers import Pipeline, pipeline
from transformers.pipelines import PIPELINE_REGISTRY
from transformers import models
from transformers.modeling_outputs import SequenceClassifierOutput
from transformers.models.bert.configuration_bert import BertConfig
from transformers.models.bert.modeling_bert import (
    BertPreTrainedModel,
    BERT_INPUTS_DOCSTRING,
    _TOKENIZER_FOR_DOC,
    _CHECKPOINT_FOR_DOC,
    BERT_START_DOCSTRING,
    _CONFIG_FOR_DOC,
    _SEQ_CLASS_EXPECTED_OUTPUT,
    _SEQ_CLASS_EXPECTED_LOSS,
    BertModel,
)

from transformers.file_utils import (
    add_code_sample_docstrings,
    add_start_docstrings_to_model_forward,
    add_start_docstrings
)

@add_start_docstrings(
    """
    Bert Model transformer with a sequence classification/regression head on top (a linear layer on top of the pooled
    output) e.g. for GLUE tasks.
    """,
    BERT_START_DOCSTRING,
)
class BertForSequenceClassification(BertPreTrainedModel):
  def __init__(self, config, **kwargs):
    super().__init__(transformers.PretrainedConfig())
    #task_labels_map={"binary_classification": 2, "label_classification": 5}
    self.tasks = kwargs.get("tasks_map", {})
    self.config = config

    self.bert = BertModel(config)
    classifier_dropout = (
        config.classifier_dropout
        if config.classifier_dropout is not None
        else config.hidden_dropout_prob
    )
    self.dropout = nn.Dropout(classifier_dropout)
    ## add task specific output heads
    self.classifier1 = nn.Linear(
        config.hidden_size, self.tasks[0].num_labels
    )
    self.classifier2 = nn.Linear(
        config.hidden_size, self.tasks[1].num_labels
    )

    self.init_weights()

  @add_start_docstrings_to_model_forward(
  BERT_INPUTS_DOCSTRING.format("batch_size, sequence_length")
  )
  @add_code_sample_docstrings(
      processor_class=_TOKENIZER_FOR_DOC,
      checkpoint=_CHECKPOINT_FOR_DOC,
      output_type=SequenceClassifierOutput,
      config_class=_CONFIG_FOR_DOC,
      expected_output=_SEQ_CLASS_EXPECTED_OUTPUT,
      expected_loss=_SEQ_CLASS_EXPECTED_LOSS,
  )
  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,
    task_ids=None,
) -> Union[Tuple[torch.Tensor], SequenceClassifierOutput]:
    r"""
    labels (:obj:`torch.LongTensor` of shape :obj:`(batch_size,)`, `optional`):
        Labels for computing the sequence classification/regression loss. Indices should be in :obj:`[0, ...,
        config.num_labels - 1]`. If :obj:`config.num_labels == 1` a regression loss is computed (Mean-Square loss),
        If :obj:`config.num_labels > 1` a classification loss is computed (Cross-Entropy).
    """
    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,
        return_dict=return_dict,
    )

    pooled_output = outputs[1]

    pooled_output = self.dropout(pooled_output)
  
    unique_task_ids_list = torch.unique(task_ids).tolist()
    loss_list = []
    logits = None
    for unique_task_id in unique_task_ids_list:

      loss = None
      task_id_filter = task_ids == unique_task_id 

      if unique_task_id == 0:
        logits = self.classifier1(pooled_output[task_id_filter])
      elif unique_task_id == 1:
        logits = self.classifier2(pooled_output[task_id_filter])

      
      if labels is not None: 
        loss_fct = CrossEntropyLoss()
        loss = loss_fct(logits.view(-1, self.tasks[unique_task_id].num_labels), labels[task_id_filter].view(-1))
        loss_list.append(loss)
    
    # logits are only used for eval. and in case of eval the batch is not multi task
    # For training only the loss is used

    if loss_list:
      loss = torch.stack(loss_list).mean()
    if not return_dict:
      output = (logits,) + outputs[2:]
      return ((loss,) + output) if loss is not None else output
    
    return SequenceClassifierOutput(
            loss=loss,
            logits=logits,
            hidden_states=outputs.hidden_states,
            attentions=outputs.attentions,
        )

@dataclass
class Task:
  id: int
  name: str
  type: str
  num_labels: int

def softmax(_outputs):
  maxes = np.max(_outputs, axis=-1, keepdims=True)
  shifted_exp = np.exp(_outputs - maxes)
  return shifted_exp / shifted_exp.sum(axis=-1, keepdims=True)

class BiBert_MultiTaskPipeline(Pipeline):


  def _sanitize_parameters(self, **kwargs):

    preprocess_kwargs = {}
    if "task_id" in kwargs:
      preprocess_kwargs["task_id"] = kwargs["task_id"]
    
    forward_kwargs = {}
    if "task_id" in kwargs:
      forward_kwargs["task_id"] = kwargs["task_id"]

    postprocess_kwargs = {}
    if "top_k" in kwargs:
      postprocess_kwargs["top_k"] = kwargs["top_k"]
      postprocess_kwargs["_legacy"] = False
    return preprocess_kwargs, forward_kwargs, postprocess_kwargs

    

  def preprocess(self, inputs, task_id):
    return_tensors = self.framework
    feature = self.tokenizer(inputs, padding = True, return_tensors=return_tensors).to(self.device)
    task_ids = np.full(shape=1,fill_value=task_id, dtype=int)
    feature["task_ids"] = torch.IntTensor(task_ids)
    return feature
  
  def _forward(self, model_inputs, task_id):
    return self.model(**model_inputs)

  def postprocess(self, model_outputs, top_k=1, _legacy=True):
    outputs = model_outputs["logits"][0]
    outputs = outputs.numpy()
    scores = softmax(outputs)

    if top_k == 1 and _legacy:
      return {"label": self.model.config.id2label[scores.argmax().item()], "score": scores.max().item()}

    dict_scores = [
        {"label": self.model.config.id2label[i], "score": score.item()} for i, score in enumerate(scores)
    ]
    if not _legacy:
      dict_scores.sort(key=lambda x: x["score"], reverse=True)
      if top_k is not None:
          dict_scores = dict_scores[:top_k]
    return dict_scores


class EndpointHandler():
  def __init__(self, path=""):
    # Preload all the elements you are going to need at inference.
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    tokenizer = AutoTokenizer.from_pretrained(path)

    PIPELINE_REGISTRY.register_pipeline("bibert-multitask-classification", pipeline_class=BiBert_MultiTaskPipeline, pt_model=BertForSequenceClassification)
    tasks = [
        Task(id=0, name='label_classification', type='seq_classification', num_labels=5),
        Task(id=1, name='binary_classification', type='seq_classification', num_labels=2)
        ]


    model = BertForSequenceClassification.from_pretrained(path, tasks_map=tasks).to(device)
    
    self.classifier_s = pipeline("bibert-multitask-classification", model = model, task_id="0", tokenizer=tokenizer, device = device)
    self.classifier_p = pipeline("bibert-multitask-classification", model = model, task_id="1", tokenizer=tokenizer, device = device)

  def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
    """
    data args:
        inputs (:obj: `str` | `PIL.Image` | `np.array`)
        kwargs
    Return:
        A :obj:`list` | `dict`: will be serialized and returned
    """
    inputs = data.pop("inputs", data)
    lang = data.pop("lang", None)
    
    prediction_p = self.classifier_p(inputs) 
    label = prediction_p[0]['label']
    score = prediction_p[0]['score']
    
    if label == '0' and score >= 0.75:
      label = 2
      return {"label":label, "score": score}
    else:
      prediction_s = self.classifier_s(inputs)
      label = prediction_s[0]['label']
      score = prediction_s[0]['score']
    return prediction_s