nanelimon-organization / tddi-model-service Goto Github PK

from wsgi import BERTModelMicroService

bert_serivce = BERTModelMicroService()

@model_router.get("/example_dump_model")
async def get_label_score(texts: List[str]):
    preprocess_url = "https://cryptic-oasis-68424.herokuapp.com/bulk_preprocess?turkish_char=true"
   
    for text in texts:
        preprocess_response = requests.post(
            preprocess_url, json={"text": text})
        processed_text = preprocess_response.json()["result"]

        results = bert_serivce.predict(processed_text)

    return {"success": True,
            "payload": results}

from fastapi import FastAPI
from api.controllers.model_controller import model_router
from fastapi.middleware.cors import CORSMiddleware
from transformers import BertTokenizer, TFBertForSequenceClassification
from transformers import TextClassificationPipeline

BERT_MODEL_PATH = "api/static/model/bigscience_t0_model"
BERT_TOKENIZER_PATH = "api/static/model/bigscience_t0_tokenizer"


class BERTModelMicroService:
    def __init__(self):
        """
        Initializes a new instance of the BERTModelMicroService class.
        """
        self.app = FastAPI(
            title="BERT Model Micro Service",
            version="0.1.0",
            description="This API analyzes Turkish text using BERT, a natural language processing technology. "
                        "It helps Telco and OTT brands to monitor and analyze Turkish text data to identify patterns in customer feedback "
                        "or detect inappropriate language, and improve their customer experience and reputation management."
        )
        self.make_middleware()
        self.bert_model = TFBertForSequenceClassification.from_pretrained(BERT_MODEL_PATH, from_pt=True) 
        self.bert_tokenizer = BertTokenizer.from_pretrained(BERT_TOKENIZER_PATH, do_lower_case=True)
        self.pipeline = TextClassificationPipeline(model=self.bert_model, tokenizer=self.bert_tokenizer)

    def predict(self, processed_text):
        results = [f"{processed_text[index]} - {i['label']}" for index, i in enumerate(self.pipeline(processed_text))]
        return results

    def make_middleware(self):
        """
        Adds middleware to the application to enable cross-origin resource sharing.
        """
        self.app.add_middleware(
            CORSMiddleware,
            allow_origins=["*"],
            allow_credentials=True,
            allow_methods=["*"],
            allow_headers=["*"]
        )

    def init_routes(self):
        """
        Initialize routes for the application.

        Parameters
        ----------
        app : FastAPI
            The FastAPI instance to attach the routes to.

        Returns
        -------
        None
        """
        @self.app.get("/healthcheck")
        async def healthcheck():
            return {"success": True}

        self.app.include_router(model_router, prefix="/api/v1")

BERTModelMicroService = BERTModelMicroService()
app = BERTModelMicroService.app