

Serveml Advanced Features

FastAPI configuration

By default, we do not change the FastAPI configuration, so you end up with a generic configuration.

One way to avoid that is to define a configuration file (for example fastapi.cfg) like this :

[fastapi]
title = WineQualityApi
description = This API helps you determine if the quality of the Wine is good or not
version = 0.1.0

And then to give it to your application :

from serveml.api import ApiBuilder
from serveml.inputs import BasicInput
from serveml.loader import load_mlflow_model
from serveml.predictions import GenericPrediction


# load model
model = load_mlflow_model(
    # MlFlow model path
    'models:/sklearn_model/1',
    # MlFlow Tracking URI
    'http://localhost:5000',
)


# Implement deserializer for input data
class WineComposition(BasicInput):
    alcohol: float
    chlorides: float
    citric_acid: float
    density: float
    fixed_acidity: float
    free_sulfur_dioxide: int
    pH: float
    residual_sugar: float
    sulphates: float
    total_sulfur_dioxide: int
    volatile_acidity: int


# implement application
app = ApiBuilder(GenericPrediction(model), WineComposition, configuration_path='fastapi.cfg').build_api()

Defining your own prediction object

The GenericPrediction object implements the AbstractPrediction class in order to ease the interface between the API and the model prediction.

Under the hood, the GenericPrediction does a little a bit more than that, in order :

• _transforms_input: Transforms JSON into pandas.DataFrame
• _fetch_data: Fetch data from an external source (by default does nothing)
• _combine_fetched_data_with_input: Combine fetched data with pandas.DataFrame (by default only returns pandas.DataFrame)
• _apply_model: Apply model
• _transform_output: Transforms result (either pandas.DataFrame, numpy.ndarray, pandas.Series) into JSON

Retrieving additional data before applying model

Let's say that the data given in input of the API is not enough, and you need additional information in order to make your prediction.

This could be done easily by overriding these two functions in the GenericPrediction class (inherited by the AbstractPrediction class):

import logging

from abc import ABC, abstractmethod

import numpy as np
import pandas as pd

from serveml.inputs import BasicInput
from serveml.utils import pydantic_model_to_pandas, pandas_to_dict


class AbstractPrediction(ABC):
    """
    Abstract class to define methods called during predict
    """

    def __init__(self, model):
        self.model = model

    @abstractmethod
    def _transform_input(self, input):
        """
        Function called right after API call. It is supposed to transform
        <pydantic.BaseModel> object into the input data format needed to apply
        model
        """

    @staticmethod
    def _fetch_data(input: BasicInput):
        """
        Helper function in case we need additional data. In most of the cases,
        can be ignored
        """
        pass

    @staticmethod
    def _combine_fetched_data_with_input(fetched_data, transformed_input):
        return transformed_input

    @abstractmethod
    def _apply_model(self, transformed_input):
        """
        Function called to apply Machine Learning model to predict from the
        transformed input
        """

    @abstractmethod
    def _transform_output(self, output):
        """
        Function called right after applying model to input data. Supposed to
        transform the data that we got after the predict in order to
        """

    def predict(self, input, uuid):
        """
        Main function that will be used by all the childs to apply model.
        Here are the steps made :
            - Transform <pydantic.BaseModel> objet to target input object
            before applying model
            - Apply model
            - Transform output into more suitable format for an API
            - Add an uuid to the request to track request made
        """
        logging.debug(
            "Got input: {} for request_id: {}".format(input.dict(), uuid)
        )
        logging.info("Transforming input for request: {}".format(uuid))
        transformed_input = self._transform_input(input)
        logging.debug(
            "Input transformed to this: {}".format(transformed_input)
        )
        logging.info("Fetching data for request: {}".format(uuid))
        fetched_data = self._fetch_data(input)
        logging.debug("Fetched data: {}".format(fetched_data))
        logging.info(
            "Combining fetched data and input for request {}".format(uuid)
        )
        combined_data = self._combine_fetched_data_with_input(
            fetched_data, transformed_input
        )
        logging.debug("Combined data: {}".format(combined_data))
        logging.info("Applying input for request {}".format(uuid))
        output = self._apply_model(combined_data)
        logging.debug("Prediction output: {}".format(output))
        logging.info("Transforming output for request {}".format(uuid))
        transformed_output = self._transform_output(output)
        logging.debug("Transformed output: {}".format(transformed_output))
        transformed_output["request_id"] = uuid
        return transformed_output


class GenericPrediction(AbstractPrediction):
    """
    Implementation of <serveml.ml.model.AbstractModel> for scikit-learn
    """

    def _transform_input(self, input) -> pd.DataFrame:
        """
        Transforms <pydantic.BaseModel> object to <pandas.DataFrame>
        """
        return pydantic_model_to_pandas(input)

    def _apply_model(self, transformed_input: pd.DataFrame):
        """
        Applies the sklearn model to the <pandas.DataFrame>.
        Returns either one of these:
            - <pandas.DataFrame>
            - <pandas.Series>
            - <numpy.ndarray>
        """
        return self.model.predict(transformed_input)

    def _transform_output(self, output) -> dict:
        """
        Transforms output given by <serveml.ml.sklearn._apply_model> to
        prepare sending result with API.
        """
        if isinstance(output, np.ndarray):
            result = output.tolist()
        elif isinstance(output, pd.DataFrame):
            result = pandas_to_dict(output)
        elif isinstance(output, pd.Series):
            result = output.to_dict()
        else:
            result = None
        return {"result": result}

Defining outputs for /predict and /feedback

By default, we do not define output models for both /predict and /feedback endpoints.

But you can specify them if you wish to. Here is how :

from typing import List

from serveml.api import ApiBuilder
from serveml.inputs import BasicInput
from serveml.loader import load_mlflow_model
from serveml.outputs import BasicFeedbackOutput, BasicPredictOutput
from serveml.predictions import GenericPrediction


# load model
model = load_mlflow_model(
    # MlFlow model path
    'models:/sklearn_model/1',
    # MlFlow Tracking URI
    'http://localhost:5000',
)


# Implement deserializer for input data
class WineComposition(BasicInput):
    alcohol: float
    chlorides: float
    citric_acid: float
    density: float
    fixed_acidity: float
    free_sulfur_dioxide: int
    pH: float
    residual_sugar: float
    sulphates: float
    total_sulfur_dioxide: int
    volatile_acidity: int


# Implement deserializer for /predict output data
class WineQuality(BasicPredictOutput):
    result: List[float]


# Implement deserializer for /predict output data
class WineFeedbackOutput(BasicFeedbackOutput):
    status: bool


# implement application
app = ApiBuilder(
    GenericPrediction(model),
    WineComposition,
    predict_output_class=WineQuality,
    feedback_output_class=WineFeedbackOutput
).build_api()