|
from typing import Dict, List, Any |
|
from transformers import pipeline |
|
|
|
class EndpointHandler(): |
|
def __init__(self, path=""): |
|
self.pipeline = pipeline("text-classification", model=path) |
|
|
|
def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]: |
|
""" |
|
data args: |
|
inputs (:obj: `str`) |
|
date (:obj: `str`) |
|
Return: |
|
A :obj:`list` | `dict`: will be serialized and returned |
|
""" |
|
|
|
inputs = data.pop("inputs", data) |
|
|
|
|
|
prediction = self.pipeline(inputs) |
|
|
|
|
|
label_mapping = { |
|
'LABEL_0': 'credit_card', |
|
'LABEL_1': 'credit_reporting', |
|
'LABEL_2': 'debt_collection', |
|
'LABEL_3': 'mortgages_and_loans', |
|
'LABEL_4': 'retail_banking' |
|
} |
|
|
|
|
|
mapped_output = [{'label': label_mapping.get(item['label'], item['label']), 'score': item['score']} for item in |
|
prediction] |
|
|
|
return mapped_output |
|
|