KServe Python Serving Runtime API¶

class ModelServer:
    def __init__(
        self,
        http_port: int = args.http_port,
        grpc_port: int = args.grpc_port,
        workers: int = args.workers,
        max_threads: int = args.max_threads,
        max_asyncio_workers: int = args.max_asyncio_workers,
        registered_models: Optional[ModelRepository] = None,
        enable_grpc: bool = args.enable_grpc,
        enable_docs_url: bool = args.enable_docs_url,
        enable_latency_logging: bool = args.enable_latency_logging,
        access_log_format: str = args.access_log_format,
    ):
        """KServe ModelServer Constructor

        Args:
            http_port: HTTP port. Default: ``8080``.
            grpc_port: GRPC port. Default: ``8081``.
            workers: Number of uvicorn workers. Default: ``1``.
            max_threads: Max number of gRPC processing threads. Default: ``4``
            max_asyncio_workers: Max number of AsyncIO threads. Default: ``None``
            registered_models: A optional Model repository with registered models.
            enable_grpc: Whether to turn on grpc server. Default: ``True``
            enable_docs_url: Whether to turn on ``/docs`` Swagger UI. Default: ``False``.
            enable_latency_logging: Whether to log latency metric. Default: ``True``.
            access_log_format: Format to set for the access log (provided by asgi-logger). Default: ``None``.
                               it allows to override only the `uvicorn.access`'s format configuration with a richer
                               set of fields (output hardcoded to `stdout`). This limitation is currently due to the
                               ASGI specs that don't describe how access logging should be implemented in detail
                               (please refer to this Uvicorn
                               [github issue](https://github.com/encode/uvicorn/issues/527) for more info).
        """
        self.registered_models = (
            ModelRepository() if registered_models is None else registered_models
        )
        self.http_port = http_port
        self.grpc_port = grpc_port
        self.workers = workers
        self.max_threads = max_threads
        self.max_asyncio_workers = max_asyncio_workers
        self.enable_grpc = enable_grpc
        self.enable_docs_url = enable_docs_url
        self.enable_latency_logging = enable_latency_logging
        self.dataplane = DataPlane(model_registry=self.registered_models)
        self.model_repository_extension = ModelRepositoryExtension(
            model_registry=self.registered_models
        )
        self._grpc_server = None
        self._rest_server = None
        if self.enable_grpc:
            self._grpc_server = GRPCServer(
                grpc_port,
                self.dataplane,
                self.model_repository_extension,
                kwargs=vars(args),
            )
        if args.configure_logging:
            # If the logger does not have any handlers, then the logger is not configured.
            # For backward compatibility, we configure the logger here.
            if len(logger.handlers) == 0:
                logging.configure_logging(args.log_config_file)
        self.access_log_format = access_log_format
        self._custom_exception_handler = None

    async def _serve_rest(self):
        logger.info(f"Starting uvicorn with {self.workers} workers")
        loop = asyncio.get_event_loop()
        if sys.platform not in ["win32", "win64"]:
            sig_list = [signal.SIGINT, signal.SIGTERM, signal.SIGQUIT]
        else:
            sig_list = [signal.SIGINT, signal.SIGTERM]

        for sig in sig_list:
            loop.add_signal_handler(
                sig, lambda s=sig: asyncio.create_task(self.stop(sig=s))
            )
        if self._custom_exception_handler is None:
            loop.set_exception_handler(self.default_exception_handler)
        else:
            loop.set_exception_handler(self._custom_exception_handler)
        self._rest_server = UvicornServer(
            app,
            self.http_port,
            self.dataplane,
            self.model_repository_extension,
            # By setting log_config to None we tell Uvicorn not to configure logging as it is already
            # configured by kserve.
            log_config=None,
            access_log_format=self.access_log_format,
            workers=self.workers,
        )
        await self._rest_server.run()

    def start(self, models: List[BaseKServeModel]) -> None:
        """Start the model server with a set of registered models.

        Args:
            models: a list of models to register to the model server.
        """
        if isinstance(models, list):
            at_least_one_model_ready = False
            for model in models:
                if isinstance(model, BaseKServeModel):
                    if model.ready:
                        at_least_one_model_ready = True
                        self.register_model(model)
                        # pass whether to log request latency into the model
                        model.enable_latency_logging = self.enable_latency_logging
                    model.start()
                else:
                    raise RuntimeError("Model type should be 'BaseKServeModel'")
            if not at_least_one_model_ready and models:
                raise NoModelReady(models)
        else:
            raise RuntimeError("Unknown model collection type")

        if self.max_asyncio_workers is None:
            # formula as suggest in https://bugs.python.org/issue35279
            self.max_asyncio_workers = min(32, utils.cpu_count() + 4)
        logger.info(f"Setting max asyncio worker threads as {self.max_asyncio_workers}")
        asyncio.get_event_loop().set_default_executor(
            concurrent.futures.ThreadPoolExecutor(max_workers=self.max_asyncio_workers)
        )

        async def servers_task():
            servers = [self._serve_rest()]
            if self.enable_grpc:
                servers.append(self._grpc_server.start(self.max_threads))
            await asyncio.gather(*servers)

        asyncio.run(servers_task())

    async def stop(self, sig: Optional[int] = None):
        """Stop the instances of REST and gRPC model servers.

        Args:
            sig: The signal to stop the server. Default: ``None``.
        """
        logger.info("Stopping the model server")
        if self._rest_server:
            logger.info("Stopping the rest server")
            await self._rest_server.stop()
        if self._grpc_server:
            logger.info("Stopping the grpc server")
            await self._grpc_server.stop(sig)
        for model_name in list(self.registered_models.get_models().keys()):
            self.registered_models.unload(model_name)

    def register_exception_handler(
        self,
        handler: Callable[[asyncio.events.AbstractEventLoop, Dict[str, Any]], None],
    ):
        """Add a custom handler as the event loop exception handler.

        If a handler is not provided, the default exception handler will be set.

        handler should be a callable object, it should have a signature matching '(loop, context)', where 'loop'
        will be a reference to the active event loop, 'context' will be a dict object (see `call_exception_handler()`
        documentation for details about context).
        """
        self._custom_exception_handler = handler

    def default_exception_handler(
        self, loop: asyncio.events.AbstractEventLoop, context: Dict[str, Any]
    ):
        """Default exception handler for event loop.

        This is called when an exception occurs and no exception handler is set.
        This can be called by a custom exception handler that wants to defer to the default handler behavior.
        """
        if "exception" in context:
            logger.error(f"Caught exception: {context.get('exception')}")
        logger.error(f"message: { context.get('message')}")
        loop.default_exception_handler(context)

    def register_model(self, model: BaseKServeModel):
        """Register a model to the model server.

        Args:
            model: The model object.
        """
        if not model.name:
            raise Exception("Failed to register model, model.name must be provided.")
        self.registered_models.update(model)
        logger.info("Registering model: %s", model.name)

class BaseKServeModel(ABC):
    """
    A base class to inherit all of the kserve models from.

    This class implements the expectations of model repository and model server.
    """

    def __init__(self, name: str):
        """
        Adds the required attributes

        Args:
            name: The name of the model.
        """
        self.name = name
        self.ready = False

    async def healthy(self) -> bool:
        """
        Check the health of this model. By default returns `self.ready`.

        Returns:
            True if healthy, false otherwise
        """
        return self.ready

    def load(self) -> bool:
        """Load handler can be overridden to load the model from storage.
        The `self.ready` should be set to True after the model is loaded. The flag is used for model health check.

        Returns:
            bool: True if model is ready, False otherwise
        """
        self.ready = True
        return self.ready

    def start(self):
        """Start handler can be overridden to perform model setup"""
        self.ready = True

    def stop(self):
        """Stop handler can be overridden to perform model teardown"""
        self.ready = False

class InferenceModel(BaseKServeModel):
    """
    Abstract class representing a model that supports standard inference and prediction.
    """

    @abstractmethod
    def __call__(
        self,
        body: Union[Dict, CloudEvent, InferRequest],
        headers: Optional[Dict[str, str]] = None,
        verb: InferenceVerb = InferenceVerb.PREDICT,
    ) -> InferReturnType:
        pass

    def get_input_types(self) -> List[Dict]:
        # Override this function to return appropriate input format expected by your model.
        # Refer https://kserve.github.io/website/0.9/modelserving/inference_api/#model-metadata-response-json-object

        # Eg.
        # return [{ "name": "", "datatype": "INT32", "shape": [1,5], }]
        return []

    def get_output_types(self) -> List[Dict]:
        # Override this function to return appropriate output format returned by your model.
        # Refer https://kserve.github.io/website/0.9/modelserving/inference_api/#model-metadata-response-json-object

        # Eg.
        # return [{ "name": "", "datatype": "INT32", "shape": [1,5], }]
        return []

class Model(InferenceModel):
    def __init__(self, name: str, predictor_config: Optional[PredictorConfig] = None):
        """KServe Model Public Interface

        Model is intended to be subclassed to implement the model handlers.

        Args:
            name: The name of the model.
            predictor_config: The configurations for http call to the predictor.
        """
        super().__init__(name)

        # The predictor config member fields are kept for backwards compatibility as they could be set outside
        self.protocol = (
            predictor_config.predictor_protocol
            if predictor_config
            else PredictorProtocol.REST_V1.value
        )
        self.predictor_host = (
            predictor_config.predictor_host if predictor_config else None
        )
        # The default timeout matches what is set in generated Istio virtual service resources.
        # We generally don't want things to time out at the request level here,
        # timeouts should be handled elsewhere in the system.
        self.timeout = (
            predictor_config.predictor_request_timeout_seconds
            if predictor_config
            else 600
        )
        self.use_ssl = predictor_config.predictor_use_ssl if predictor_config else False
        self.explainer_host = None
        self._http_client_instance = None
        self._grpc_client_stub = None
        self.enable_latency_logging = False

    async def __call__(
        self,
        body: Union[Dict, CloudEvent, InferRequest],
        headers: Optional[Dict[str, str]] = None,
        verb: InferenceVerb = InferenceVerb.PREDICT,
    ) -> InferReturnType:
        """Method to call predictor or explainer with the given input.

        Args:
            body: Request body.
            verb: The inference verb for predict/generate/explain
            headers: Request headers.

        Returns:
            Response output from preprocess -> predict/generate/explain -> postprocess
        """
        request_id = headers.get("x-request-id", "N.A.") if headers else "N.A."

        # latency vars
        preprocess_ms = 0
        explain_ms = 0
        predict_ms = 0
        postprocess_ms = 0
        prom_labels = get_labels(self.name)

        with PRE_HIST_TIME.labels(**prom_labels).time():
            start = time.time()
            payload = (
                await self.preprocess(body, headers)
                if inspect.iscoroutinefunction(self.preprocess)
                else self.preprocess(body, headers)
            )
            preprocess_ms = get_latency_ms(start, time.time())
        payload = self.validate(payload)
        if verb == InferenceVerb.EXPLAIN:
            with EXPLAIN_HIST_TIME.labels(**prom_labels).time():
                start = time.time()
                response = (
                    (await self.explain(payload, headers))
                    if inspect.iscoroutinefunction(self.explain)
                    else self.explain(payload, headers)
                )
                explain_ms = get_latency_ms(start, time.time())
        elif verb == InferenceVerb.PREDICT:
            with PREDICT_HIST_TIME.labels(**prom_labels).time():
                start = time.time()
                response = (
                    (await self.predict(payload, headers))
                    if inspect.iscoroutinefunction(self.predict)
                    else self.predict(payload, headers)
                )
                predict_ms = get_latency_ms(start, time.time())
        else:
            raise NotImplementedError

        with POST_HIST_TIME.labels(**prom_labels).time():
            start = time.time()
            response = (
                await self.postprocess(response, headers)
                if inspect.iscoroutinefunction(self.postprocess)
                else self.postprocess(response, headers)
            )
            postprocess_ms = get_latency_ms(start, time.time())

        if self.enable_latency_logging is True:
            trace_logger.info(
                f"requestId: {request_id}, preprocess_ms: {preprocess_ms}, "
                f"explain_ms: {explain_ms}, predict_ms: {predict_ms}, "
                f"postprocess_ms: {postprocess_ms}"
            )

        return response

    @property
    def _http_client(self) -> InferenceRESTClient:
        if self._http_client_instance is None and self.predictor_host:
            config = RESTConfig(protocol=self.protocol, timeout=self.timeout, retries=3)
            self._http_client_instance = InferenceRESTClient(config=config)
        return self._http_client_instance

    @property
    def _grpc_client(self) -> InferenceGRPCClient:
        if self._grpc_client_stub is None and self.predictor_host:
            self._grpc_client_stub = InferenceGRPCClient(
                url=self.predictor_host, use_ssl=self.use_ssl, timeout=self.timeout
            )
        return self._grpc_client_stub

    def validate(self, payload):
        if isinstance(payload, ModelInferRequest):
            return payload
        if isinstance(payload, InferRequest):
            return payload
        # TODO: validate the request if self.get_input_types() defines the input types.
        if self.protocol == PredictorProtocol.REST_V2.value:
            if "inputs" in payload and not isinstance(payload["inputs"], list):
                raise InvalidInput('Expected "inputs" to be a list')
        elif self.protocol == PredictorProtocol.REST_V1.value:
            if (
                isinstance(payload, Dict)
                and "instances" in payload
                and not isinstance(payload["instances"], list)
            ):
                raise InvalidInput('Expected "instances" to be a list')
        return payload

    def load(self) -> bool:
        """Load handler can be overridden to load the model from storage.
        The `self.ready` should be set to True after the model is loaded. The flag is used for model health check.

        Returns:
            bool: True if model is ready, False otherwise
        """
        self.ready = True
        return self.ready

    async def preprocess(
        self, payload: Union[Dict, InferRequest], headers: Dict[str, str] = None
    ) -> Union[Dict, InferRequest]:
        """`preprocess` handler can be overridden for data or feature transformation.
        The model decodes the request body to `Dict` for v1 endpoints and `InferRequest` for v2 endpoints.

        Args:
            payload: Payload of the request.
            headers: Request headers.

        Returns:
            A Dict or InferRequest in KServe Model Transformer mode which is transmitted on the wire to predictor.
            Tensors in KServe Predictor mode which is passed to predict handler for performing the inference.
        """

        return payload

    async def postprocess(
        self, result: Union[Dict, InferResponse], headers: Dict[str, str] = None
    ) -> Union[Dict, InferResponse]:
        """The `postprocess` handler can be overridden for inference result or response transformation.
        The predictor sends back the inference result in `Dict` for v1 endpoints and `InferResponse` for v2 endpoints.

        Args:
            result: The inference result passed from `predict` handler or the HTTP response from predictor.
            headers: Request headers.

        Returns:
            A Dict or InferResponse after post-process to return back to the client.
        """
        return result

    async def _http_predict(
        self, payload: Union[Dict, InferRequest], headers: Dict[str, str] = None
    ) -> Union[Dict, InferResponse]:
        # Adjusting headers. Inject content type if not exist.
        # Also, removing host, as the header is the one passed to transformer and contains transformer's host
        predict_headers = {"Content-Type": "application/json"}
        if headers is not None:
            if "x-request-id" in headers:
                predict_headers["x-request-id"] = headers["x-request-id"]
            if "x-b3-traceid" in headers:
                predict_headers["x-b3-traceid"] = headers["x-b3-traceid"]

        protocol = "https" if self.use_ssl else "http"
        predict_base_url = PREDICTOR_BASE_URL_FORMAT.format(
            protocol, self.predictor_host
        )
        response = await self._http_client.infer(
            predict_base_url,
            model_name=self.name,
            data=payload,
            headers=predict_headers,
        )
        return response

    async def _grpc_predict(
        self,
        payload: Union[ModelInferRequest, InferRequest],
        headers: Dict[str, str] = None,
    ) -> InferResponse:
        if isinstance(payload, ModelInferRequest):
            payload = InferRequest.from_grpc(payload)
        async_result = await self._grpc_client.infer(
            infer_request=payload,
            headers=(
                ("request_type", "grpc_v2"),
                ("response_type", "grpc_v2"),
                ("x-request-id", headers.get("x-request-id", "")),
            ),
        )
        return async_result

    async def predict(
        self,
        payload: Union[Dict, InferRequest, ModelInferRequest],
        headers: Dict[str, str] = None,
    ) -> Union[Dict, InferResponse, AsyncIterator[Any]]:
        """The `predict` handler can be overridden for performing the inference.
            By default, the predict handler makes call to predictor for the inference step.

        Args:
            payload: Model inputs passed from `preprocess` handler.
            headers: Request headers.

        Returns:
            Inference result or a Response from the predictor.

        Raises:
            HTTPStatusError when getting back an error response from the predictor.
        """
        if not self.predictor_host:
            raise NotImplementedError("Could not find predictor_host.")
        if self.protocol == PredictorProtocol.GRPC_V2.value:
            return await self._grpc_predict(payload, headers)
        else:
            return await self._http_predict(payload, headers)

    async def explain(self, payload: Dict, headers: Dict[str, str] = None) -> Dict:
        """`explain` handler can be overridden to implement the model explanation.
        The default implementation makes call to the explainer if ``explainer_host`` is specified.

        Args:
            payload: Explainer model inputs passed from preprocess handler.
            headers: Request headers.

        Returns:
            An Explanation for the inference result.

        Raises:
            HTTPStatusError when getting back an error response from the explainer.
        """
        if self.explainer_host is None:
            raise NotImplementedError("Could not find explainer_host.")

        explain_headers = {"content-type": "application/json"}
        if headers is not None:
            if "content-type" in headers:
                explain_headers["content-type"] = headers["content-type"]
            if "x-request-id" in headers:
                explain_headers["x-request-id"] = headers["x-request-id"]
            if "x-b3-traceid" in headers:
                explain_headers["x-b3-traceid"] = headers["x-b3-traceid"]

        protocol = "https" if self.use_ssl else "http"
        # Currently explainer only supports the kserve v1 endpoints
        explain_base_url = EXPLAINER_BASE_URL_FORMAT.format(
            protocol, self.explainer_host
        )
        response = await self._http_client.explain(
            explain_base_url,
            model_name=self.name,
            data=payload,
            headers=explain_headers,
        )
        return response

class PredictorConfig:
    def __init__(
        self,
        predictor_host: str,
        predictor_protocol: str = PredictorProtocol.REST_V1.value,
        predictor_use_ssl: bool = False,
        predictor_request_timeout_seconds: int = 600,
    ):
        """The configuration for the http call to the predictor

        Args:
            predictor_host: The host name of the predictor
            predictor_protocol: The inference protocol used for predictor http call
            predictor_use_ssl: Enable using ssl for http connection to the predictor
            predictor_request_timeout_seconds: The request timeout seconds for the predictor http call
        """
        self.predictor_host = predictor_host
        self.predictor_protocol = predictor_protocol
        self.predictor_use_ssl = predictor_use_ssl
        self.predictor_request_timeout_seconds = predictor_request_timeout_seconds

class InferInput:
    _name: str
    _shape: List[int]
    _datatype: str
    _parameters: Dict

    def __init__(
        self,
        name: str,
        shape: List[int],
        datatype: str,
        data: Union[List, np.ndarray, InferTensorContents] = None,
        parameters: Optional[Union[Dict, MessageMap[str, InferParameter]]] = None,
    ):
        """An object of InferInput class is used to describe the input tensor of an inference request.

        Args:
            name: The name of the inference input whose data will be described by this object.
            shape : The shape of the associated inference input.
            datatype : The data type of the associated inference input.
            data : The data of the inference input.
                   When data is not set, raw_data is used for gRPC to transmit with numpy array bytes
                   by using `set_data_from_numpy`.
            parameters : The additional inference parameters.
        """

        self._name = name
        self._shape = shape
        self._datatype = datatype.upper()
        self._parameters = parameters
        self._data = data
        self._raw_data = None

    @property
    def name(self) -> str:
        """Get the name of inference input associated with this object.

        Returns:
            The name of the inference input
        """
        return self._name

    @property
    def datatype(self) -> str:
        """Get the datatype of inference input associated with this object.

        Returns:
            The datatype of the inference input.
        """
        return self._datatype

    @property
    def data(self) -> Union[List, np.ndarray, InferTensorContents]:
        """Get the data of the inference input associated with this object.

        Returns:
            The data of the inference input.
        """
        return self._data

    @data.setter
    def data(self, data: List):
        """Set the data of the inference input associated with this object.

        Args:
             data: data of the inference input.
        """
        self._data = data

    @property
    def shape(self) -> List[int]:
        """Get the shape of inference input associated with this object.

        Returns:
            The shape of the inference input.
        """
        return self._shape

    @property
    def parameters(self) -> Union[Dict, MessageMap[str, InferParameter], None]:
        """Get the parameters of the inference input associated with this object.

        Returns:
            The additional inference parameters
        """
        return self._parameters

    @parameters.setter
    def parameters(
        self, params: Optional[Union[Dict, MessageMap[str, InferParameter]]]
    ):
        """Set the parameters of the inference input associated with this object.

        Args:
             params: parameters of the inference input
        """
        self._parameters = params

    @shape.setter
    def shape(self, shape: List[int]):
        """Set the shape of inference input.

        Args:
            shape : The shape of the associated inference input.
        """
        self._shape = shape

    def as_string(self) -> List[List[str]]:
        """
        Decodes the inference input data as a list of strings.

        Returns:
            List[List[str]]: The decoded data as a list of strings.

        Raises:
            InvalidInput: If the datatype of the inference input is not 'BYTES'.
        """
        if self.datatype == "BYTES":
            return [s.decode("utf-8") for li in self._data for s in li]
        else:
            raise InvalidInput(f"invalid datatype {self.datatype} in the input")

    def as_numpy(self) -> np.ndarray:
        """Decode the inference input data as numpy array.

        Returns:
            A numpy array of the inference input data
        Raises:
            InvalidInput: If the datatype of the inference input is not recognized.
        """
        dtype = to_np_dtype(self.datatype)
        if dtype is None:
            raise InvalidInput(f"invalid datatype {dtype} in the input")
        if self._raw_data is not None:
            if self.datatype == "BYTES":
                # String results contain a 4-byte string length
                # followed by the actual string characters. Hence,
                # need to decode the raw bytes to convert into
                # array elements.
                np_array = deserialize_bytes_tensor(self._raw_data)
            else:
                np_array = np.frombuffer(self._raw_data, dtype=dtype)
            return np_array.reshape(self._shape)
        else:
            np_array = np.array(self._data, dtype=dtype)
            return np_array.reshape(self._shape)

    def set_data_from_numpy(self, input_tensor: np.ndarray, binary_data: bool = True):
        """Set the tensor data from the specified numpy array for input associated with this object.

        Args:
            input_tensor : The tensor data in numpy array format.
            binary_data : Indicates whether to set data for the input in binary format
                          or explicit tensor within JSON. The default value is True,
                          which means the data will be delivered as binary data with gRPC or in the
                          HTTP body after the JSON object for REST.

        Raises:
            InferenceError if failed to set data for the tensor.
        """
        if not isinstance(input_tensor, (np.ndarray,)):
            raise InferenceError("input_tensor must be a numpy array")

        dtype = from_np_dtype(input_tensor.dtype)
        if self._datatype != dtype:
            raise InferenceError(
                "got unexpected datatype {} from numpy array, expected {}".format(
                    dtype, self._datatype
                )
            )
        valid_shape = True
        if len(self._shape) != len(input_tensor.shape):
            valid_shape = False
        else:
            for i in range(len(self._shape)):
                if self._shape[i] != input_tensor.shape[i]:
                    valid_shape = False
        if not valid_shape:
            raise InferenceError(
                "got unexpected numpy array shape [{}], expected [{}]".format(
                    str(input_tensor.shape)[1:-1], str(self._shape)[1:-1]
                )
            )

        if not binary_data:
            if self._parameters:
                self._parameters.pop("binary_data_size", None)
            self._raw_data = None
            if self._datatype == "BYTES":
                self._data = []
                try:
                    if input_tensor.size > 0:
                        for obj in np.nditer(
                            input_tensor, flags=["refs_ok"], order="C"
                        ):
                            # We need to convert the object to string using utf-8,
                            # if we want to use the binary_data=False. JSON requires
                            # the input to be a UTF-8 string.
                            if input_tensor.dtype == np.object_:
                                if type(obj.item()) == bytes:
                                    self._data.append(str(obj.item(), encoding="utf-8"))
                                else:
                                    self._data.append(str(obj.item()))
                            else:
                                self._data.append(str(obj.item(), encoding="utf-8"))
                except UnicodeDecodeError:
                    raise InferenceError(
                        f'Failed to encode "{obj.item()}" using UTF-8. Please use binary_data=True, if'
                        " you want to pass a byte array."
                    )
            else:
                self._data = [val.item() for val in input_tensor.flatten()]
        else:
            self._data = None
            if self._datatype == "BYTES":
                serialized_output = serialize_byte_tensor(input_tensor)
                if serialized_output.size > 0:
                    self._raw_data = serialized_output.item()
                else:
                    self._raw_data = b""
            else:
                self._raw_data = input_tensor.tobytes()
            if self._parameters is None:
                self._parameters = {"binary_data_size": len(self._raw_data)}
            else:
                self._parameters["binary_data_size"] = len(self._raw_data)

    def __eq__(self, other):
        if not isinstance(other, InferInput):
            return False
        if self.name != other.name:
            return False
        if self.shape != other.shape:
            return False
        if self.datatype != other.datatype:
            return False
        if self.parameters != other.parameters:
            return False
        if self.data != other.data:
            return False
        if self._raw_data != other._raw_data:
            return False
        return True

    @parameters.setter
    def parameters(self, value):
        self._parameters = value

    def to_dict(self) -> dict:
        return {
            "name": self.name,
            "shape": self.shape,
            "datatype": self.datatype,
            "data": self.data,
            "parameters": self.parameters,
        }

    def __repr__(self) -> str:
        return (
            f'"name": "{self.name}",'
            f'"shape": {self.shape},'
            f'"datatype": "{self.datatype}",'
            f'"data": {self.data},'
            f'"parameters": {self.parameters}'
        )

    def __str__(self) -> str:
        return self.__repr__()