feat: DF solution

autotests/helm/values.yaml

@@ -25,8 +25,8 @@ global:
     activeDeadlineSeconds: 3600 # 1h
 
   env:
-    PARTICIPANT_NAME: <REPLACE_WITH_USERNAME>
-    api_host: http://inca-smc-mlops-challenge-solution.default.svc.cluster.local/<REPLACE_WITH_ENDPOINT>
+    PARTICIPANT_NAME: Dmitrii Fedotov
+    api_host: http://inca-smc-mlops-challenge-solution.default.svc.cluster.local/api/v1/process/
 
     # K6, do not edit!
     K6_PROMETHEUS_RW_SERVER_URL: http://kube-prometheus-stack-prometheus.monitoring.svc.cluster.local:9090/api/v1/write

solution/Dockerfile

@@ -0,0 +1,12 @@
+FROM pytorch/pytorch:2.0.0-cuda11.7-cudnn8-runtime
+
+SHELL ["/bin/bash", "-o", "pipefail", "-c"]
+# Install dependencies
+COPY requirements.txt .
+RUN source /opt/conda/etc/profile.d/conda.sh && conda activate base && \
+    pip install --no-cache-dir -r requirements.txt
+
+COPY src /opt/app
+ENV PYTHONPATH=/opt/app:$PYTHONPATH
+WORKDIR /opt/app
+ENTRYPOINT ["./start.sh"]

solution/helm/envs/df.yaml

@@ -0,0 +1,13 @@
+global:
+#  env:
+#    EXAMPLE: "example"
+  resources:
+    limits:
+      nvidia.com/gpu: 1
+  pod:
+    ports:
+      - name: http
+        containerPort: x8000
+        protocol: TCP
+  service:
+    targetPort: 8080

solution/requirements.txt

@@ -0,0 +1,7 @@
+fastapi==0.95.2
+pydantic==1.10.7
+torch==2.0.1
+uvicorn==0.22.0
+tweetnlp==0.4.3
+orjson==3.8.12
+gunicorn==20.1.0

solution/src/api/v1/health_check.py

@@ -0,0 +1,18 @@
+from http import HTTPStatus
+
+from fastapi import APIRouter
+from starlette.responses import JSONResponse
+
+router = APIRouter()
+
+
+@router.get(
+    path="/",
+    summary="Create Notification",
+    description="Create notification for user id",
+)
+async def health_check():
+    return JSONResponse(
+        status_code=HTTPStatus.OK,
+        content={"message": "Health Check"},
+    )

solution/src/api/v1/process.py

@@ -0,0 +1,41 @@
+from http import HTTPStatus
+
+from fastapi import APIRouter, Body, Depends, HTTPException
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+
+from services.get_result import MLModelProcess, get_model_service
+
+router = APIRouter()
+empty_response=[]
+
+@router.post(
+    path="/",
+    summary="Get Result from Models",
+    description="Get Result from Models",
+)
+async def process_data(
+    request: Request,
+    model_service: MLModelProcess = Depends(
+        get_model_service,
+    ),
+):
+    payload = await request.body()
+    payload = payload.decode("utf-8")
+    #cache empty payload
+    if payload == "":
+        global empty_response
+        if len(empty_response)>0:
+            return empty_response
+        else:
+            try:
+                response=await model_service.process_data(payload)
+                empty_response=response
+                return response
+            except Exception as e:
+                raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))
+
+    try:
+        return await model_service.process_data(payload)
+    except Exception as e:
+        raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR, detail=str(e))

solution/src/core/config.py

@@ -0,0 +1,40 @@
+import torch
+from pydantic import BaseSettings
+
+from models.model_process import ModelProcess
+
+
+class Settings(BaseSettings):
+    project_name: str = "models_api"
+    device: int = 0 if torch.cuda.is_available() else -1
+    MODELS: list[ModelProcess] = [
+        {
+            "model_name": "cardiffnlp",
+            "model_path": "cardiffnlp/twitter-xlm-roberta-base-sentiment",
+            "output_functions": ["label_upper"],
+        },
+        {
+            "model_name": "jy46604790",
+            "model_path": "jy46604790/Fake-News-Bert-Detect",
+        },
+        {
+            "model_name": "ivanlau",
+            "model_path": "ivanlau/language-detection-fine-tuned-on-xlm-roberta-base",
+        },
+        {
+            "model_name": "svalabs",
+            "model_path": "svalabs/twitter-xlm-roberta-crypto-spam",
+        },
+        {
+            "model_name": "EIStakovskii",
+            "model_path": "EIStakovskii/xlm_roberta_base_multilingual_toxicity_classifier_plus",
+        },
+    ]
+
+    class Config:
+        case_sensitive = False
+        env_file = "models_api.env"
+        env_file_encoding = "utf-8"
+
+
+settings = Settings()

solution/src/helpers/output_functions.py

@@ -0,0 +1,3 @@
+def label_upper(output: dict):
+    output["label"] = output["label"].upper()
+    return output

solution/src/main.py

@@ -0,0 +1,53 @@
+import uvicorn
+from fastapi import FastAPI
+from fastapi.responses import ORJSONResponse
+from transformers import pipeline
+
+from api.v1 import health_check, process
+from core.config import settings
+from ml_models import init_models as ml_models
+from models.model_process import ModelPipeline
+
+app = FastAPI(
+    title=settings.project_name,
+    docs_url="/api/openapi",
+    openapi_url="/api/openapi.json",
+    default_response_class=ORJSONResponse,
+)
+
+
+@app.on_event("startup")
+async def startup():
+    # load models
+    if ml_models.models is None:
+        ml_models.models = []
+    for model in settings.MODELS:
+        clf = pipeline(
+            "text-classification",
+            model=model.model_path,
+            tokenizer=model.model_path,
+            device=settings.device,
+        )
+        ml_models.models.append(ModelPipeline(pipeline=clf, model_process=model))
+    app.include_router(
+        health_check.router, prefix="/api/v1/health_check", tags=["Health Check"]
+    )
+
+
+@app.on_event("shutdown")
+async def shutdown():
+    ...
+
+
+app.include_router(
+    process.router,
+    prefix="/api/v1/process",
+    tags=["Process Model API"],
+)
+
+if __name__ == "__main__":
+    uvicorn.run(
+        "main:app",
+        host="0.0.0.0",
+        port=8080,
+    )

solution/src/ml_models/init_models.py

@@ -0,0 +1,7 @@
+from models.model_process import ModelPipeline
+
+models: list[ModelPipeline] = None
+
+
+async def load_models() -> list[ModelPipeline]:
+    return models

solution/src/models/model_process.py

@@ -0,0 +1,18 @@
+from typing import Optional
+
+from pydantic import BaseModel
+from transformers import TextClassificationPipeline
+
+
+class ModelProcess(BaseModel):
+    model_name: str
+    model_path: str
+    output_functions: Optional[list[str]] = None
+
+
+class ModelPipeline(BaseModel):
+    model_process: ModelProcess
+    pipeline: TextClassificationPipeline
+
+    class Config:
+        arbitrary_types_allowed = True

solution/src/models/models_request.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+
+
+class ModelRequest(BaseModel):
+    text: str
+
+    class Config:
+        schema_extra = {
+            "example": {
+                "text": "Tis is how true happiness looks like",
+            }
+        }

solution/src/services/base_service.py

@@ -0,0 +1,8 @@
+from abc import ABC
+
+from models.model_process import ModelPipeline
+
+
+class BaseService(ABC):
+    def __init__(self, models: list[ModelPipeline]):
+        self.models = models

solution/src/services/get_result.py

@@ -0,0 +1,37 @@
+import logging
+
+from fastapi import Depends
+
+from helpers.output_functions import label_upper
+from ml_models.init_models import load_models
+from models.model_process import ModelPipeline
+from services.base_service import BaseService
+
+logger = logging.getLogger()
+
+
+class MLModelProcess(BaseService):
+    async def process_data(
+        self,
+        payload: str,
+    ):
+        result = {}
+        for model in self.models:
+            model_result = model.pipeline(payload)
+            if len(model_result) > 0:
+                model_result = model_result[0]
+            else:
+                raise Exception("Model result is empty for model: " + model.model_process.model_name)
+            if model.model_process.output_functions is not None:
+                for output_function in model.model_process.output_functions:
+                    func = globals()[output_function]
+                    model_result = func(model_result)
+            result[model.model_process.model_name] = model_result
+
+        return result
+
+
+def get_model_service(
+    loaded_models: list[ModelPipeline] = Depends(load_models),
+) -> MLModelProcess:
+    return MLModelProcess(models=loaded_models)

solution/src/start.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+set -eo pipefail
+
+#Activate conda
+source /opt/conda/etc/profile.d/conda.sh && conda activate base
+
+gunicorn main:app --workers=2 --bind=:8080 --worker-class=uvicorn.workers.UvicornWorker --timeout 300