Extend MLServer by implementing the MLModel class. The core lifecycle involves:
load(): An asynchronous method used to locate model data (using get_model_uri), parse/process files, and instantiate the model. Set self.ready = True when loading is complete.predict(): An asynchronous method that receives an InferenceRequest payload, validates it, and returns an InferenceResponse containing the model's outputs.
Use mlserver.errors.InferenceError to signal errors during request validation.
from typing import List
from mlserver import MLModel, types
from mlserver.errors import InferenceError
from mlserver.utils import get_model_uri
WELLKNOWN_MODEL_FILENAMES = ["model.json", "model.dat"]
class CustomMLModel(MLModel):
async def load(self) -> bool:
# get URI to model data
model_uri = await get_model_uri(self._settings, wellknown_filenames=WELLKNOWN_MODEL_FILENAMES)
# parse/process file and instantiate the model
self._load_model_from_file(model_uri)
# set ready to signal that model is loaded
self.ready = True
return self.ready
async def predict(self, payload: types.InferenceRequest) -> types.InferenceResponse:
payload = self._check_request(payload)
return types.InferenceResponse(
model_name=self.name,
model_version=self.version,
outputs=self._predict_outputs(payload),
)
def _load_model_from_file(self, file_uri):
# TODO: load model from file and instantiate class data
return
def _check_request(self, payload: types.InferenceRequest) -> types.InferenceRequest:
# TODO: validate request and raise InferenceError on error
return payload
def _predict_outputs(self, payload: types.InferenceRequest) -> List[types.ResponseOutput]:
inputs = payload.inputs
# TODO: transform inputs and run prediction logic
outputs = []
return outputs
