Skip to content

KServe Python Serving Runtime API

ModelServer

Source code in kserve/model_server.py
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
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: ModelRepository = ModelRepository(),
                 enable_grpc: bool = args.enable_grpc,
                 enable_docs_url: bool = args.enable_docs_url,
                 enable_latency_logging: bool = args.enable_latency_logging,
                 configure_logging: bool = args.configure_logging,
                 log_config: Optional[Union[Dict, str]] = args.log_config_file,
                 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: 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``.
            configure_logging: Whether to configure KServe and Uvicorn logging. Default: ``True``.
            log_config: File path or dict containing log config. Default: ``None``.
            access_log_format: Format to set for the access log (provided by asgi-logger). Default: ``None``
        """
        self.registered_models = 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=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)

        # Logs can be passed as a path to a file or a dictConfig.
        # We rely on Uvicorn to configure the loggers for us.
        if configure_logging:
            self.log_config = log_config if log_config is not None else KSERVE_LOG_CONFIG
        else:
            # By setting log_config to None we tell Uvicorn not to configure logging
            self.log_config = None

        self.access_log_format = access_log_format
        self._custom_exception_handler = None

    def start(self, models: Union[List[Model], Dict[str, Deployment]]) -> 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):
            for model in models:
                if isinstance(model, Model):
                    self.register_model(model)
                    # pass whether to log request latency into the model
                    model.enable_latency_logging = self.enable_latency_logging
                else:
                    raise RuntimeError("Model type should be 'Model'")
        elif isinstance(models, dict):
            if all([isinstance(v, Deployment) for v in models.values()]):
                # TODO: make this port number a variable
                rayserve.start(detached=True, http_options={"host": "0.0.0.0", "port": 9071})
                for key in models:
                    models[key].deploy()
                    handle = models[key].get_handle()
                    self.register_model_handle(key, handle)
            else:
                raise RuntimeError("Model type should be RayServe Deployment")
        else:
            raise RuntimeError("Unknown model collection types")

        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 serve():
            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)
            if self.workers == 1:
                self._rest_server = UvicornServer(self.http_port, [],
                                                  self.dataplane, self.model_repository_extension,
                                                  self.enable_docs_url,
                                                  log_config=self.log_config,
                                                  access_log_format=self.access_log_format)
                await self._rest_server.run()
            else:
                # Since py38 MacOS/Windows defaults to use spawn for starting multiprocessing.
                # https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods
                # Spawn does not work with FastAPI/uvicorn in multiprocessing mode, use fork for multiprocessing
                # https://github.com/tiangolo/fastapi/issues/1586
                serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                serversocket.bind(('0.0.0.0', self.http_port))
                serversocket.listen(5)
                multiprocessing.set_start_method('fork')
                self._rest_server = UvicornServer(self.http_port, [serversocket],
                                                  self.dataplane, self.model_repository_extension,
                                                  self.enable_docs_url, log_config=self.log_config,
                                                  access_log_format=self.access_log_format)
                for _ in range(self.workers):
                    p = Process(target=self._rest_server.run_sync)
                    p.start()

        async def servers_task():
            servers = [serve()]
            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)

    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.
        By default, this will shut down the server gracefully.

        This can be called by a custom exception handler that wants to defer to the default handler behavior.
        """
        # gracefully shutdown the server
        loop.run_until_complete(self.stop())
        loop.default_exception_handler(context)

    def register_model_handle(self, name: str, model_handle: DeploymentHandle):
        """Register a model handle to the model server.

        Args:
            name: The name of the model handle.
            model_handle: The model handle object.
        """
        self.registered_models.update_handle(name, model_handle)
        logger.info("Registering model handle: %s", name)

    def register_model(self, model: Model):
        """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)

__init__(http_port=args.http_port, grpc_port=args.grpc_port, workers=args.workers, max_threads=args.max_threads, max_asyncio_workers=args.max_asyncio_workers, registered_models=ModelRepository(), enable_grpc=args.enable_grpc, enable_docs_url=args.enable_docs_url, enable_latency_logging=args.enable_latency_logging, configure_logging=args.configure_logging, log_config=args.log_config_file, access_log_format=args.access_log_format)

KServe ModelServer Constructor

Parameters:

Name Type Description Default
http_port int

HTTP port. Default: 8080.

http_port
grpc_port int

GRPC port. Default: 8081.

grpc_port
workers int

Number of uvicorn workers. Default: 1.

workers
max_threads int

Max number of gRPC processing threads. Default: 4

max_threads
max_asyncio_workers int

Max number of AsyncIO threads. Default: None

max_asyncio_workers
registered_models ModelRepository

Model repository with registered models.

ModelRepository()
enable_grpc bool

Whether to turn on grpc server. Default: True

enable_grpc
enable_docs_url bool

Whether to turn on /docs Swagger UI. Default: False.

enable_docs_url
enable_latency_logging bool

Whether to log latency metric. Default: True.

enable_latency_logging
configure_logging bool

Whether to configure KServe and Uvicorn logging. Default: True.

configure_logging
log_config Optional[Union[Dict, str]]

File path or dict containing log config. Default: None.

log_config_file
access_log_format str

Format to set for the access log (provided by asgi-logger). Default: None

access_log_format
Source code in kserve/model_server.py
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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: ModelRepository = ModelRepository(),
             enable_grpc: bool = args.enable_grpc,
             enable_docs_url: bool = args.enable_docs_url,
             enable_latency_logging: bool = args.enable_latency_logging,
             configure_logging: bool = args.configure_logging,
             log_config: Optional[Union[Dict, str]] = args.log_config_file,
             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: 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``.
        configure_logging: Whether to configure KServe and Uvicorn logging. Default: ``True``.
        log_config: File path or dict containing log config. Default: ``None``.
        access_log_format: Format to set for the access log (provided by asgi-logger). Default: ``None``
    """
    self.registered_models = 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=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)

    # Logs can be passed as a path to a file or a dictConfig.
    # We rely on Uvicorn to configure the loggers for us.
    if configure_logging:
        self.log_config = log_config if log_config is not None else KSERVE_LOG_CONFIG
    else:
        # By setting log_config to None we tell Uvicorn not to configure logging
        self.log_config = None

    self.access_log_format = access_log_format
    self._custom_exception_handler = None

default_exception_handler(loop, context)

Default exception handler for event loop.

This is called when an exception occurs and no exception handler is set. By default, this will shut down the server gracefully.

This can be called by a custom exception handler that wants to defer to the default handler behavior.

Source code in kserve/model_server.py
253
254
255
256
257
258
259
260
261
262
263
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.
    By default, this will shut down the server gracefully.

    This can be called by a custom exception handler that wants to defer to the default handler behavior.
    """
    # gracefully shutdown the server
    loop.run_until_complete(self.stop())
    loop.default_exception_handler(context)

register_exception_handler(handler)

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).

Source code in kserve/model_server.py
242
243
244
245
246
247
248
249
250
251
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

register_model(model)

Register a model to the model server.

Parameters:

Name Type Description Default
model Model

The model object.

required
Source code in kserve/model_server.py
275
276
277
278
279
280
281
282
283
284
285
def register_model(self, model: Model):
    """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)

register_model_handle(name, model_handle)

Register a model handle to the model server.

Parameters:

Name Type Description Default
name str

The name of the model handle.

required
model_handle DeploymentHandle

The model handle object.

required
Source code in kserve/model_server.py
265
266
267
268
269
270
271
272
273
def register_model_handle(self, name: str, model_handle: DeploymentHandle):
    """Register a model handle to the model server.

    Args:
        name: The name of the model handle.
        model_handle: The model handle object.
    """
    self.registered_models.update_handle(name, model_handle)
    logger.info("Registering model handle: %s", name)

start(models)

Start the model server with a set of registered models.

Parameters:

Name Type Description Default
models Union[List[Model], Dict[str, Deployment]]

a list of models to register to the model server.

required
Source code in kserve/model_server.py
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def start(self, models: Union[List[Model], Dict[str, Deployment]]) -> 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):
        for model in models:
            if isinstance(model, Model):
                self.register_model(model)
                # pass whether to log request latency into the model
                model.enable_latency_logging = self.enable_latency_logging
            else:
                raise RuntimeError("Model type should be 'Model'")
    elif isinstance(models, dict):
        if all([isinstance(v, Deployment) for v in models.values()]):
            # TODO: make this port number a variable
            rayserve.start(detached=True, http_options={"host": "0.0.0.0", "port": 9071})
            for key in models:
                models[key].deploy()
                handle = models[key].get_handle()
                self.register_model_handle(key, handle)
        else:
            raise RuntimeError("Model type should be RayServe Deployment")
    else:
        raise RuntimeError("Unknown model collection types")

    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 serve():
        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)
        if self.workers == 1:
            self._rest_server = UvicornServer(self.http_port, [],
                                              self.dataplane, self.model_repository_extension,
                                              self.enable_docs_url,
                                              log_config=self.log_config,
                                              access_log_format=self.access_log_format)
            await self._rest_server.run()
        else:
            # Since py38 MacOS/Windows defaults to use spawn for starting multiprocessing.
            # https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods
            # Spawn does not work with FastAPI/uvicorn in multiprocessing mode, use fork for multiprocessing
            # https://github.com/tiangolo/fastapi/issues/1586
            serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            serversocket.bind(('0.0.0.0', self.http_port))
            serversocket.listen(5)
            multiprocessing.set_start_method('fork')
            self._rest_server = UvicornServer(self.http_port, [serversocket],
                                              self.dataplane, self.model_repository_extension,
                                              self.enable_docs_url, log_config=self.log_config,
                                              access_log_format=self.access_log_format)
            for _ in range(self.workers):
                p = Process(target=self._rest_server.run_sync)
                p.start()

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

    asyncio.run(servers_task())

stop(sig=None) async

Stop the instances of REST and gRPC model servers.

Parameters:

Name Type Description Default
sig Optional[int]

The signal to stop the server. Default: None.

None
Source code in kserve/model_server.py
228
229
230
231
232
233
234
235
236
237
238
239
240
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)

Model

Source code in kserve/model.py
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
class Model:
    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.
        """
        self.name = name
        self.ready = False
        # 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],
                       verb: InferenceVerb = InferenceVerb.PREDICT,
                       headers: Dict[str, str] = None) -> Union[Dict, InferResponse, List[str]]:
        """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):
        if self._http_client_instance is None:
            self._http_client_instance = httpx.AsyncClient()
        return self._http_client_instance

    @property
    def _grpc_client(self):
        if self._grpc_client_stub is None:
            # requires appending the port to the predictor host for gRPC to work
            if ":" not in self.predictor_host:
                port = 443 if self.use_ssl else 80
                self.predictor_host = f"{self.predictor_host}:{port}"
            if self.use_ssl:
                _channel = grpc.aio.secure_channel(self.predictor_host, grpc.ssl_channel_credentials())
            else:
                _channel = grpc.aio.insecure_channel(self.predictor_host)
            self._grpc_client_stub = grpc_predict_v2_pb2_grpc.GRPCInferenceServiceStub(_channel)
        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

    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 []

    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) -> Dict:
        protocol = "https" if self.use_ssl else "http"
        predict_url = PREDICTOR_URL_FORMAT.format(protocol, self.predictor_host, self.name)
        if self.protocol == PredictorProtocol.REST_V2.value:
            predict_url = PREDICTOR_V2_URL_FORMAT.format(protocol, self.predictor_host, self.name)

        # 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']
        if isinstance(payload, InferRequest):
            payload = payload.to_rest()
        data = orjson.dumps(payload)
        response = await self._http_client.post(
            predict_url,
            timeout=self.timeout,
            headers=predict_headers,
            content=data
        )
        if not response.is_success:
            message = (
                "{error_message}, '{0.status_code} {0.reason_phrase}' for url '{0.url}'"
            )
            error_message = ""
            if "content-type" in response.headers and response.headers["content-type"] == "application/json":
                error_message = response.json()
                if "error" in error_message:
                    error_message = error_message["error"]
            message = message.format(response, error_message=error_message)
            raise HTTPStatusError(message, request=response.request, response=response)
        return orjson.loads(response.content)

    async def _grpc_predict(self, payload: Union[ModelInferRequest, InferRequest], headers: Dict[str, str] = None) \
            -> ModelInferResponse:
        if isinstance(payload, InferRequest):
            payload = payload.to_grpc()
        async_result = await self._grpc_client.ModelInfer(
            request=payload,
            timeout=self.timeout,
            metadata=(('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]:
        """ 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:
            res = await self._grpc_predict(payload, headers)
            return InferResponse.from_grpc(res)
        else:
            res = await self._http_predict(payload, headers)
            # return an InferResponse if this is REST V2, otherwise just return the dictionary
            return InferResponse.from_rest(self.name, res) if is_v2(PredictorProtocol(self.protocol)) else res

    async def generate(self, payload: GenerateRequest,
                       headers: Dict[str, str] = None) -> Union[GenerateResponse, AsyncIterator[Any]]:
        """`generate` handler can be overridden to implement text generation.

        """
        raise NotImplementedError("generate is not implemented")

    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.")

        protocol = "https" if self.use_ssl else "http"
        # Currently explainer only supports the kserve v1 endpoints
        explain_url = EXPLAINER_URL_FORMAT.format(protocol, self.explainer_host, self.name)
        response = await self._http_client.post(
            url=explain_url,
            timeout=self.timeout,
            content=orjson.dumps(payload)
        )

        response.raise_for_status()
        return orjson.loads(response.content)

__call__(body, verb=InferenceVerb.PREDICT, headers=None) async

Method to call predictor or explainer with the given input.

Parameters:

Name Type Description Default
body Union[Dict, CloudEvent, InferRequest]

Request body.

required
verb InferenceVerb

The inference verb for predict/generate/explain

PREDICT
headers Dict[str, str]

Request headers.

None

Returns:

Type Description
Union[Dict, InferResponse, List[str]]

Response output from preprocess -> predict/generate/explain -> postprocess

Source code in kserve/model.py
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def __call__(self, body: Union[Dict, CloudEvent, InferRequest],
                   verb: InferenceVerb = InferenceVerb.PREDICT,
                   headers: Dict[str, str] = None) -> Union[Dict, InferResponse, List[str]]:
    """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

__init__(name, predictor_config=None)

KServe Model Public Interface

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

Parameters:

Name Type Description Default
name str

The name of the model.

required
predictor_config Optional[PredictorConfig]

The configurations for http call to the predictor.

None
Source code in kserve/model.py
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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.
    """
    self.name = name
    self.ready = False
    # 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

explain(payload, headers=None) async

explain handler can be overridden to implement the model explanation. The default implementation makes call to the explainer if explainer_host is specified.

Parameters:

Name Type Description Default
payload Dict

Explainer model inputs passed from preprocess handler.

required
headers Dict[str, str]

Request headers.

None

Returns:

Type Description
Dict

An Explanation for the inference result.

Source code in kserve/model.py
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
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.")

    protocol = "https" if self.use_ssl else "http"
    # Currently explainer only supports the kserve v1 endpoints
    explain_url = EXPLAINER_URL_FORMAT.format(protocol, self.explainer_host, self.name)
    response = await self._http_client.post(
        url=explain_url,
        timeout=self.timeout,
        content=orjson.dumps(payload)
    )

    response.raise_for_status()
    return orjson.loads(response.content)

generate(payload, headers=None) async

generate handler can be overridden to implement text generation.

Source code in kserve/model.py
327
328
329
330
331
332
async def generate(self, payload: GenerateRequest,
                   headers: Dict[str, str] = None) -> Union[GenerateResponse, AsyncIterator[Any]]:
    """`generate` handler can be overridden to implement text generation.

    """
    raise NotImplementedError("generate is not implemented")

load()

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:

Name Type Description
bool bool

True if model is ready, False otherwise

Source code in kserve/model.py
197
198
199
200
201
202
203
204
205
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

postprocess(result, headers=None) async

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.

Parameters:

Name Type Description Default
result Union[Dict, InferResponse]

The inference result passed from predict handler or the HTTP response from predictor.

required
headers Dict[str, str]

Request headers.

None

Returns:

Type Description
Union[Dict, InferResponse]

A Dict or InferResponse after post-process to return back to the client.

Source code in kserve/model.py
239
240
241
242
243
244
245
246
247
248
249
250
251
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

predict(payload, headers=None) async

The predict handler can be overridden for performing the inference. By default, the predict handler makes call to predictor for the inference step.

Parameters:

Name Type Description Default
payload Union[Dict, InferRequest, ModelInferRequest]

Model inputs passed from preprocess handler.

required
headers Dict[str, str]

Request headers.

None

Returns:

Type Description
Union[Dict, InferResponse]

Inference result or a Response from the predictor.

Source code in kserve/model.py
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
async def predict(self, payload: Union[Dict, InferRequest, ModelInferRequest],
                  headers: Dict[str, str] = None) -> Union[Dict, InferResponse]:
    """ 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:
        res = await self._grpc_predict(payload, headers)
        return InferResponse.from_grpc(res)
    else:
        res = await self._http_predict(payload, headers)
        # return an InferResponse if this is REST V2, otherwise just return the dictionary
        return InferResponse.from_rest(self.name, res) if is_v2(PredictorProtocol(self.protocol)) else res

preprocess(payload, headers=None) async

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.

Parameters:

Name Type Description Default
payload Union[Dict, InferRequest]

Payload of the request.

required
headers Dict[str, str]

Request headers.

None

Returns:

Type Description
Union[Dict, InferRequest]

A Dict or InferRequest in KServe Model Transformer mode which is transmitted on the wire to predictor.

Union[Dict, InferRequest]

Tensors in KServe Predictor mode which is passed to predict handler for performing the inference.

Source code in kserve/model.py
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
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

PredictorConfig

Source code in kserve/model.py
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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

__init__(predictor_host, predictor_protocol=PredictorProtocol.REST_V1.value, predictor_use_ssl=False, predictor_request_timeout_seconds=600)

The configuration for the http call to the predictor

Parameters:

Name Type Description Default
predictor_host str

The host name of the predictor

required
predictor_protocol str

The inference protocol used for predictor http call

REST_V1.value
predictor_use_ssl bool

Enable using ssl for http connection to the predictor

False
predictor_request_timeout_seconds int

The request timeout seconds for the predictor http call

600
Source code in kserve/model.py
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
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

InferInput

Source code in kserve/protocol/infer_type.py
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
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

    @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

    def set_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]]:
        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
        """
        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:
            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
        return True

data: Union[List, np.ndarray, InferTensorContents] property

Get the data of the inference input associated with this object.

Returns:

Type Description
Union[List, ndarray, InferTensorContents]

The data of the inference input.

datatype: str property

Get the datatype of inference input associated with this object.

Returns:

Type Description
str

The datatype of the inference input.

name: str property

Get the name of inference input associated with this object.

Returns:

Type Description
str

The name of the inference input

parameters: Union[Dict, MessageMap[str, InferParameter], None] property

Get the parameters of the inference input associated with this object.

Returns:

Type Description
Union[Dict, MessageMap[str, InferParameter], None]

The additional inference parameters

shape: List[int] property

Get the shape of inference input associated with this object.

Returns:

Type Description
List[int]

The shape of the inference input.

__init__(name, shape, datatype, data=None, parameters=None)

An object of InferInput class is used to describe the input tensor of an inference request.

Parameters:

Name Type Description Default
name str

The name of the inference input whose data will be described by this object.

required
shape

The shape of the associated inference input.

required
datatype

The data type of the associated inference input.

required
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.

None
parameters

The additional inference parameters.

None
Source code in kserve/protocol/infer_type.py
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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

as_numpy()

Decode the inference input data as numpy array.

Returns:

Type Description
ndarray

A numpy array of the inference input data

Source code in kserve/protocol/infer_type.py
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def as_numpy(self) -> np.ndarray:
    """ Decode the inference input data as numpy array.

    Returns:
        A numpy array of the inference input data
    """
    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:
        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)

set_data_from_numpy(input_tensor, binary_data=True)

Set the tensor data from the specified numpy array for input associated with this object.

Parameters:

Name Type Description Default
input_tensor

The tensor data in numpy array format.

required
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.

True
Source code in kserve/protocol/infer_type.py
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
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)

set_shape(shape)

Set the shape of inference input.

Parameters:

Name Type Description Default
shape

The shape of the associated inference input.

required
Source code in kserve/protocol/infer_type.py
143
144
145
146
147
148
149
def set_shape(self, shape: List[int]):
    """Set the shape of inference input.

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

InferOutput

Source code in kserve/protocol/infer_type.py
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
class InferOutput:
    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 InferOutput class is used to describe the output tensor for an inference response.

        Args:
            name : The name of inference output whose data will be described by this object.
            shape : The shape of the associated inference output.
            datatype : The data type of the associated inference output.
            data : The data of the inference output. When data is not set,
                   raw_data is used for gRPC with numpy array bytes by calling 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 output associated with this object.

        Returns:
            The name of inference output.
        """
        return self._name

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

        Returns:
            The data type of inference output.
        """
        return self._datatype

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

        Returns:
            The data of inference output.
        """
        return self._data

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

        Returns:
            The shape of inference output
        """
        return self._shape

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

        Returns:
            The additional inference parameters associated with the inference output.
        """
        return self._parameters

    @parameters.setter
    def parameters(self, params: Union[Dict, MessageMap[str, InferParameter]]):
        self._parameters = params

    def set_shape(self, shape: List[int]):
        """Set the shape of inference output.

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

    def as_numpy(self) -> numpy.ndarray:
        """ Decode the tensor output data as numpy array.

        Returns:
            The numpy array of the associated inference output data.
        """
        dtype = to_np_dtype(self.datatype)
        if dtype is None:
            raise InvalidInput("invalid datatype in the input")
        if self._raw_data is not None:
            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, output_tensor: np.ndarray, binary_data: bool = True):
        """Set the tensor data from the specified numpy array for the inference output associated with this object.

        Args:
            output_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 output tensor.
        """
        if not isinstance(output_tensor, (np.ndarray,)):
            raise InferenceError("input_tensor must be a numpy array")

        dtype = from_np_dtype(output_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(output_tensor.shape):
            valid_shape = False
        else:
            for i in range(len(self._shape)):
                if self._shape[i] != output_tensor.shape[i]:
                    valid_shape = False
        if not valid_shape:
            raise InferenceError(
                "got unexpected numpy array shape [{}], expected [{}]".format(
                    str(output_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 output_tensor.size > 0:
                        for obj in np.nditer(output_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 output_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 output_tensor.flatten()]
        else:
            self._data = None
            if self._datatype == "BYTES":
                serialized_output = serialize_byte_tensor(output_tensor)
                if serialized_output.size > 0:
                    self._raw_data = serialized_output.item()
                else:
                    self._raw_data = b''
            else:
                self._raw_data = output_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, InferOutput):
            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
        return True

data: Union[List, np.ndarray, InferTensorContents] property

Get the data of inference output associated with this object.

Returns:

Type Description
Union[List, ndarray, InferTensorContents]

The data of inference output.

datatype: str property

Get the data type of inference output associated with this object.

Returns:

Type Description
str

The data type of inference output.

name: str property

Get the name of inference output associated with this object.

Returns:

Type Description
str

The name of inference output.

parameters: Union[Dict, MessageMap[str, InferParameter], None] property writable

Get the parameters of inference output associated with this object.

Returns:

Type Description
Union[Dict, MessageMap[str, InferParameter], None]

The additional inference parameters associated with the inference output.

shape: List[int] property

Get the shape of inference output associated with this object.

Returns:

Type Description
List[int]

The shape of inference output

__init__(name, shape, datatype, data=None, parameters=None)

An object of InferOutput class is used to describe the output tensor for an inference response.

Parameters:

Name Type Description Default
name

The name of inference output whose data will be described by this object.

required
shape

The shape of the associated inference output.

required
datatype

The data type of the associated inference output.

required
data

The data of the inference output. When data is not set, raw_data is used for gRPC with numpy array bytes by calling set_data_from_numpy.

None
parameters

The additional inference parameters.

None
Source code in kserve/protocol/infer_type.py
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
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 InferOutput class is used to describe the output tensor for an inference response.

    Args:
        name : The name of inference output whose data will be described by this object.
        shape : The shape of the associated inference output.
        datatype : The data type of the associated inference output.
        data : The data of the inference output. When data is not set,
               raw_data is used for gRPC with numpy array bytes by calling 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

as_numpy()

Decode the tensor output data as numpy array.

Returns:

Type Description
ndarray

The numpy array of the associated inference output data.

Source code in kserve/protocol/infer_type.py
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
def as_numpy(self) -> numpy.ndarray:
    """ Decode the tensor output data as numpy array.

    Returns:
        The numpy array of the associated inference output data.
    """
    dtype = to_np_dtype(self.datatype)
    if dtype is None:
        raise InvalidInput("invalid datatype in the input")
    if self._raw_data is not None:
        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)

set_data_from_numpy(output_tensor, binary_data=True)

Set the tensor data from the specified numpy array for the inference output associated with this object.

Parameters:

Name Type Description Default
output_tensor

The tensor data in numpy array format.

required
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.

True
Source code in kserve/protocol/infer_type.py
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
def set_data_from_numpy(self, output_tensor: np.ndarray, binary_data: bool = True):
    """Set the tensor data from the specified numpy array for the inference output associated with this object.

    Args:
        output_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 output tensor.
    """
    if not isinstance(output_tensor, (np.ndarray,)):
        raise InferenceError("input_tensor must be a numpy array")

    dtype = from_np_dtype(output_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(output_tensor.shape):
        valid_shape = False
    else:
        for i in range(len(self._shape)):
            if self._shape[i] != output_tensor.shape[i]:
                valid_shape = False
    if not valid_shape:
        raise InferenceError(
            "got unexpected numpy array shape [{}], expected [{}]".format(
                str(output_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 output_tensor.size > 0:
                    for obj in np.nditer(output_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 output_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 output_tensor.flatten()]
    else:
        self._data = None
        if self._datatype == "BYTES":
            serialized_output = serialize_byte_tensor(output_tensor)
            if serialized_output.size > 0:
                self._raw_data = serialized_output.item()
            else:
                self._raw_data = b''
        else:
            self._raw_data = output_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)

set_shape(shape)

Set the shape of inference output.

Parameters:

Name Type Description Default
shape List[int]

The shape of the associated inference output.

required
Source code in kserve/protocol/infer_type.py
504
505
506
507
508
509
510
def set_shape(self, shape: List[int]):
    """Set the shape of inference output.

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

InferRequest

Source code in kserve/protocol/infer_type.py
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
class InferRequest:
    id: Optional[str]
    model_name: str
    parameters: Optional[Dict]
    inputs: List[InferInput]
    from_grpc: bool

    def __init__(self, model_name: str, infer_inputs: List[InferInput],
                 request_id: Optional[str] = None,
                 raw_inputs=None,
                 from_grpc: Optional[bool] = False,
                 parameters: Optional[Union[Dict, MessageMap[str, InferParameter]]] = None):
        """InferRequest Data Model.

        Args:
            model_name: The model name.
            infer_inputs: The inference inputs for the model.
            request_id: The id for the inference request.
            raw_inputs: The binary data for the inference inputs.
            from_grpc: Indicate if the data model is constructed from gRPC request.
            parameters: The additional inference parameters.
        """

        self.id = request_id
        self.model_name = model_name
        self.inputs = infer_inputs
        self.parameters = parameters
        self.from_grpc = from_grpc
        if raw_inputs:
            for i, raw_input in enumerate(raw_inputs):
                self.inputs[i]._raw_data = raw_input

    @classmethod
    def from_grpc(cls, request: ModelInferRequest):
        """ The class method to construct the InferRequest from a ModelInferRequest

        """
        infer_inputs = [InferInput(name=input_tensor.name, shape=list(input_tensor.shape),
                                   datatype=input_tensor.datatype,
                                   data=get_content(input_tensor.datatype, input_tensor.contents),
                                   parameters=input_tensor.parameters)
                        for input_tensor in request.inputs]
        return cls(request_id=request.id, model_name=request.model_name, infer_inputs=infer_inputs,
                   raw_inputs=request.raw_input_contents, from_grpc=True, parameters=request.parameters)

    def to_rest(self) -> Dict:
        """ Converts the InferRequest object to v2 REST InferRequest Dict.

        Returns:
            The InferRequest Dict converted from InferRequest object.
        """
        infer_inputs = []
        for infer_input in self.inputs:
            datatype = infer_input.datatype
            if isinstance(infer_input.datatype, numpy.dtype):
                datatype = from_np_dtype(infer_input.datatype)
            infer_input_dict = {
                "name": infer_input.name,
                "shape": infer_input.shape,
                "datatype": datatype
            }
            if infer_input.parameters:
                infer_input_dict["parameters"] = to_http_parameters(infer_input.parameters)
            if isinstance(infer_input.data, numpy.ndarray):
                infer_input.set_data_from_numpy(infer_input.data, binary_data=False)
                infer_input_dict["data"] = infer_input.data
            else:
                infer_input_dict["data"] = infer_input.data
            infer_inputs.append(infer_input_dict)
        infer_request = {
            'id': self.id if self.id else str(uuid.uuid4()),
            'inputs': infer_inputs
        }
        if self.parameters:
            infer_request['parameters'] = to_http_parameters(self.parameters)
        return infer_request

    def to_grpc(self) -> ModelInferRequest:
        """ Converts the InferRequest object to gRPC ModelInferRequest type.

        Returns:
            The ModelInferResponse gRPC type converted from InferRequest object.
        """
        infer_inputs = []
        raw_input_contents = []
        for infer_input in self.inputs:
            if isinstance(infer_input.data, numpy.ndarray):
                infer_input.set_data_from_numpy(infer_input.data, binary_data=True)
            infer_input_dict = {
                "name": infer_input.name,
                "shape": infer_input.shape,
                "datatype": infer_input.datatype,
            }
            if infer_input.parameters:
                infer_input_dict["parameters"] = to_grpc_parameters(infer_input.parameters)
            if infer_input._raw_data is not None:
                raw_input_contents.append(infer_input._raw_data)
            else:
                if not isinstance(infer_input.data, List):
                    raise InvalidInput("input data is not a List")
                infer_input_dict["contents"] = {}
                data_key = GRPC_CONTENT_DATATYPE_MAPPINGS.get(infer_input.datatype, None)
                if data_key is not None:
                    infer_input._data = [bytes(val, 'utf-8') if isinstance(val, str)
                                         else val for val in
                                         infer_input.data]  # str to byte conversion for grpc proto
                    infer_input_dict["contents"][data_key] = infer_input.data
                else:
                    raise InvalidInput("invalid input datatype")
            infer_inputs.append(infer_input_dict)

        return ModelInferRequest(id=self.id, model_name=self.model_name, inputs=infer_inputs,
                                 raw_input_contents=raw_input_contents,
                                 parameters=to_grpc_parameters(self.parameters) if self.parameters else None)

    def as_dataframe(self) -> pd.DataFrame:
        """ Decode the tensor inputs as pandas dataframe.

        Returns:
            The inference input data as pandas dataframe
        """
        dfs = []
        for input in self.inputs:
            input_data = input.data
            if input.datatype == "BYTES":
                input_data = [str(val, "utf-8") if isinstance(val, bytes)
                              else val for val in input.data]
            dfs.append(pd.DataFrame(input_data, columns=[input.name]))
        return pd.concat(dfs, axis=1)

    def __eq__(self, other):
        if not isinstance(other, InferRequest):
            return False
        if self.model_name != other.model_name:
            return False
        if self.id != other.id:
            return False
        if self.from_grpc != other.from_grpc:
            return False
        if self.parameters != other.parameters:
            return False
        if self.inputs != other.inputs:
            return False
        return True

__init__(model_name, infer_inputs, request_id=None, raw_inputs=None, from_grpc=False, parameters=None)

InferRequest Data Model.

Parameters:

Name Type Description Default
model_name str

The model name.

required
infer_inputs List[InferInput]

The inference inputs for the model.

required
request_id Optional[str]

The id for the inference request.

None
raw_inputs

The binary data for the inference inputs.

None
from_grpc Optional[bool]

Indicate if the data model is constructed from gRPC request.

False
parameters Optional[Union[Dict, MessageMap[str, InferParameter]]]

The additional inference parameters.

None
Source code in kserve/protocol/infer_type.py
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
def __init__(self, model_name: str, infer_inputs: List[InferInput],
             request_id: Optional[str] = None,
             raw_inputs=None,
             from_grpc: Optional[bool] = False,
             parameters: Optional[Union[Dict, MessageMap[str, InferParameter]]] = None):
    """InferRequest Data Model.

    Args:
        model_name: The model name.
        infer_inputs: The inference inputs for the model.
        request_id: The id for the inference request.
        raw_inputs: The binary data for the inference inputs.
        from_grpc: Indicate if the data model is constructed from gRPC request.
        parameters: The additional inference parameters.
    """

    self.id = request_id
    self.model_name = model_name
    self.inputs = infer_inputs
    self.parameters = parameters
    self.from_grpc = from_grpc
    if raw_inputs:
        for i, raw_input in enumerate(raw_inputs):
            self.inputs[i]._raw_data = raw_input

as_dataframe()

Decode the tensor inputs as pandas dataframe.

Returns:

Type Description
DataFrame

The inference input data as pandas dataframe

Source code in kserve/protocol/infer_type.py
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def as_dataframe(self) -> pd.DataFrame:
    """ Decode the tensor inputs as pandas dataframe.

    Returns:
        The inference input data as pandas dataframe
    """
    dfs = []
    for input in self.inputs:
        input_data = input.data
        if input.datatype == "BYTES":
            input_data = [str(val, "utf-8") if isinstance(val, bytes)
                          else val for val in input.data]
        dfs.append(pd.DataFrame(input_data, columns=[input.name]))
    return pd.concat(dfs, axis=1)

from_grpc(request) classmethod

The class method to construct the InferRequest from a ModelInferRequest

Source code in kserve/protocol/infer_type.py
319
320
321
322
323
324
325
326
327
328
329
330
@classmethod
def from_grpc(cls, request: ModelInferRequest):
    """ The class method to construct the InferRequest from a ModelInferRequest

    """
    infer_inputs = [InferInput(name=input_tensor.name, shape=list(input_tensor.shape),
                               datatype=input_tensor.datatype,
                               data=get_content(input_tensor.datatype, input_tensor.contents),
                               parameters=input_tensor.parameters)
                    for input_tensor in request.inputs]
    return cls(request_id=request.id, model_name=request.model_name, infer_inputs=infer_inputs,
               raw_inputs=request.raw_input_contents, from_grpc=True, parameters=request.parameters)

to_grpc()

Converts the InferRequest object to gRPC ModelInferRequest type.

Returns:

Type Description
ModelInferRequest

The ModelInferResponse gRPC type converted from InferRequest object.

Source code in kserve/protocol/infer_type.py
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def to_grpc(self) -> ModelInferRequest:
    """ Converts the InferRequest object to gRPC ModelInferRequest type.

    Returns:
        The ModelInferResponse gRPC type converted from InferRequest object.
    """
    infer_inputs = []
    raw_input_contents = []
    for infer_input in self.inputs:
        if isinstance(infer_input.data, numpy.ndarray):
            infer_input.set_data_from_numpy(infer_input.data, binary_data=True)
        infer_input_dict = {
            "name": infer_input.name,
            "shape": infer_input.shape,
            "datatype": infer_input.datatype,
        }
        if infer_input.parameters:
            infer_input_dict["parameters"] = to_grpc_parameters(infer_input.parameters)
        if infer_input._raw_data is not None:
            raw_input_contents.append(infer_input._raw_data)
        else:
            if not isinstance(infer_input.data, List):
                raise InvalidInput("input data is not a List")
            infer_input_dict["contents"] = {}
            data_key = GRPC_CONTENT_DATATYPE_MAPPINGS.get(infer_input.datatype, None)
            if data_key is not None:
                infer_input._data = [bytes(val, 'utf-8') if isinstance(val, str)
                                     else val for val in
                                     infer_input.data]  # str to byte conversion for grpc proto
                infer_input_dict["contents"][data_key] = infer_input.data
            else:
                raise InvalidInput("invalid input datatype")
        infer_inputs.append(infer_input_dict)

    return ModelInferRequest(id=self.id, model_name=self.model_name, inputs=infer_inputs,
                             raw_input_contents=raw_input_contents,
                             parameters=to_grpc_parameters(self.parameters) if self.parameters else None)

to_rest()

Converts the InferRequest object to v2 REST InferRequest Dict.

Returns:

Type Description
Dict

The InferRequest Dict converted from InferRequest object.

Source code in kserve/protocol/infer_type.py
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
def to_rest(self) -> Dict:
    """ Converts the InferRequest object to v2 REST InferRequest Dict.

    Returns:
        The InferRequest Dict converted from InferRequest object.
    """
    infer_inputs = []
    for infer_input in self.inputs:
        datatype = infer_input.datatype
        if isinstance(infer_input.datatype, numpy.dtype):
            datatype = from_np_dtype(infer_input.datatype)
        infer_input_dict = {
            "name": infer_input.name,
            "shape": infer_input.shape,
            "datatype": datatype
        }
        if infer_input.parameters:
            infer_input_dict["parameters"] = to_http_parameters(infer_input.parameters)
        if isinstance(infer_input.data, numpy.ndarray):
            infer_input.set_data_from_numpy(infer_input.data, binary_data=False)
            infer_input_dict["data"] = infer_input.data
        else:
            infer_input_dict["data"] = infer_input.data
        infer_inputs.append(infer_input_dict)
    infer_request = {
        'id': self.id if self.id else str(uuid.uuid4()),
        'inputs': infer_inputs
    }
    if self.parameters:
        infer_request['parameters'] = to_http_parameters(self.parameters)
    return infer_request

InferResponse

Source code in kserve/protocol/infer_type.py
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
class InferResponse:
    id: str
    model_name: str
    model_version: Optional[str]
    parameters: Optional[Dict]
    outputs: List[InferOutput]
    from_grpc: bool

    def __init__(self, response_id: str, model_name: str, infer_outputs: List[InferOutput],
                 model_version: Optional[str] = None, raw_outputs=None, from_grpc: Optional[bool] = False,
                 parameters: Optional[Union[Dict, MessageMap[str, InferParameter]]] = None):
        """The InferResponse Data Model

        Args:
            response_id: The id of the inference response.
            model_name: The name of the model.
            infer_outputs: The inference outputs of the inference response.
            model_version: The version of the model.
            raw_outputs: The raw binary data of the inference outputs.
            from_grpc: Indicate if the InferResponse is constructed from a gRPC response.
            parameters: The additional inference parameters.
        """

        self.id = response_id
        self.model_name = model_name
        self.model_version = model_version
        self.outputs = infer_outputs
        self.parameters = parameters
        self.from_grpc = from_grpc
        if raw_outputs:
            for i, raw_output in enumerate(raw_outputs):
                self.outputs[i]._raw_data = raw_output

    @classmethod
    def from_grpc(cls, response: ModelInferResponse) -> 'InferResponse':
        """ The class method to construct the InferResponse object from gRPC message type.
        """
        infer_outputs = [InferOutput(name=output.name, shape=list(output.shape),
                                     datatype=output.datatype,
                                     data=get_content(output.datatype, output.contents),
                                     parameters=output.parameters)
                         for output in response.outputs]
        return cls(model_name=response.model_name, model_version=response.model_version, response_id=response.id,
                   parameters=response.parameters, infer_outputs=infer_outputs,
                   raw_outputs=response.raw_output_contents, from_grpc=True)

    @classmethod
    def from_rest(cls, model_name: str, response: Dict) -> 'InferResponse':
        """ The class method to construct the InferResponse object from REST message type.

        """
        infer_outputs = [InferOutput(name=output['name'],
                                     shape=list(output['shape']),
                                     datatype=output['datatype'],
                                     data=output['data'],
                                     parameters=output.get('parameters', None))
                         for output in response['outputs']]
        return cls(model_name=model_name,
                   model_version=response.get('model_version', None),
                   response_id=response.get('id', None),
                   parameters=response.get('parameters', None),
                   infer_outputs=infer_outputs)

    def to_rest(self) -> Dict:
        """ Converts the InferResponse object to v2 REST InferResponse dict.

        Returns:
            The InferResponse Dict.
        """
        infer_outputs = []
        for i, infer_output in enumerate(self.outputs):
            infer_output_dict = {
                "name": infer_output.name,
                "shape": infer_output.shape,
                "datatype": infer_output.datatype
            }
            if infer_output.parameters:
                infer_output_dict["parameters"] = to_http_parameters(infer_output.parameters)
            if isinstance(infer_output.data, numpy.ndarray):
                infer_output.set_data_from_numpy(infer_output.data, binary_data=False)
                infer_output_dict["data"] = infer_output.data
            elif isinstance(infer_output._raw_data, bytes):
                infer_output_dict["data"] = infer_output.as_numpy().tolist()
            else:
                infer_output_dict["data"] = infer_output.data
            infer_outputs.append(infer_output_dict)
        res = {
            'id': self.id,
            'model_name': self.model_name,
            'model_version': self.model_version,
            'outputs': infer_outputs
        }
        if self.parameters:
            res['parameters'] = to_http_parameters(self.parameters)
        return res

    def to_grpc(self) -> ModelInferResponse:
        """ Converts the InferResponse object to gRPC ModelInferResponse type.

        Returns:
            The ModelInferResponse gRPC message.
        """
        infer_outputs = []
        raw_output_contents = []
        for infer_output in self.outputs:
            if isinstance(infer_output.data, numpy.ndarray):
                infer_output.set_data_from_numpy(infer_output.data, binary_data=True)
            infer_output_dict = {
                "name": infer_output.name,
                "shape": infer_output.shape,
                "datatype": infer_output.datatype,
            }
            if infer_output.parameters:
                infer_output_dict["parameters"] = to_grpc_parameters(infer_output.parameters)
            if infer_output._raw_data is not None:
                raw_output_contents.append(infer_output._raw_data)
            else:
                if not isinstance(infer_output.data, List):
                    raise InvalidInput("output data is not a List")
                infer_output_dict["contents"] = {}
                data_key = GRPC_CONTENT_DATATYPE_MAPPINGS.get(infer_output.datatype, None)
                if data_key is not None:
                    infer_output._data = [bytes(val, 'utf-8') if isinstance(val, str)
                                          else val for val in
                                          infer_output.data]  # str to byte conversion for grpc proto
                    infer_output_dict["contents"][data_key] = infer_output.data
                else:
                    raise InvalidInput("to_grpc: invalid output datatype")
            infer_outputs.append(infer_output_dict)

        return ModelInferResponse(id=self.id, model_name=self.model_name, model_version=self.model_version,
                                  outputs=infer_outputs, raw_output_contents=raw_output_contents,
                                  parameters=to_grpc_parameters(self.parameters) if self.parameters else None)

    def __eq__(self, other):
        if not isinstance(other, InferResponse):
            return False
        if self.model_name != other.model_name:
            return False
        if self.model_version != other.model_version:
            return False
        if self.id != other.id:
            return False
        if self.from_grpc != other.from_grpc:
            return False
        if self.parameters != other.parameters:
            return False
        if self.outputs != other.outputs:
            return False
        return True

__init__(response_id, model_name, infer_outputs, model_version=None, raw_outputs=None, from_grpc=False, parameters=None)

The InferResponse Data Model

Parameters:

Name Type Description Default
response_id str

The id of the inference response.

required
model_name str

The name of the model.

required
infer_outputs List[InferOutput]

The inference outputs of the inference response.

required
model_version Optional[str]

The version of the model.

None
raw_outputs

The raw binary data of the inference outputs.

None
from_grpc Optional[bool]

Indicate if the InferResponse is constructed from a gRPC response.

False
parameters Optional[Union[Dict, MessageMap[str, InferParameter]]]

The additional inference parameters.

None
Source code in kserve/protocol/infer_type.py
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
def __init__(self, response_id: str, model_name: str, infer_outputs: List[InferOutput],
             model_version: Optional[str] = None, raw_outputs=None, from_grpc: Optional[bool] = False,
             parameters: Optional[Union[Dict, MessageMap[str, InferParameter]]] = None):
    """The InferResponse Data Model

    Args:
        response_id: The id of the inference response.
        model_name: The name of the model.
        infer_outputs: The inference outputs of the inference response.
        model_version: The version of the model.
        raw_outputs: The raw binary data of the inference outputs.
        from_grpc: Indicate if the InferResponse is constructed from a gRPC response.
        parameters: The additional inference parameters.
    """

    self.id = response_id
    self.model_name = model_name
    self.model_version = model_version
    self.outputs = infer_outputs
    self.parameters = parameters
    self.from_grpc = from_grpc
    if raw_outputs:
        for i, raw_output in enumerate(raw_outputs):
            self.outputs[i]._raw_data = raw_output

from_grpc(response) classmethod

The class method to construct the InferResponse object from gRPC message type.

Source code in kserve/protocol/infer_type.py
654
655
656
657
658
659
660
661
662
663
664
665
@classmethod
def from_grpc(cls, response: ModelInferResponse) -> 'InferResponse':
    """ The class method to construct the InferResponse object from gRPC message type.
    """
    infer_outputs = [InferOutput(name=output.name, shape=list(output.shape),
                                 datatype=output.datatype,
                                 data=get_content(output.datatype, output.contents),
                                 parameters=output.parameters)
                     for output in response.outputs]
    return cls(model_name=response.model_name, model_version=response.model_version, response_id=response.id,
               parameters=response.parameters, infer_outputs=infer_outputs,
               raw_outputs=response.raw_output_contents, from_grpc=True)

from_rest(model_name, response) classmethod

The class method to construct the InferResponse object from REST message type.

Source code in kserve/protocol/infer_type.py
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
@classmethod
def from_rest(cls, model_name: str, response: Dict) -> 'InferResponse':
    """ The class method to construct the InferResponse object from REST message type.

    """
    infer_outputs = [InferOutput(name=output['name'],
                                 shape=list(output['shape']),
                                 datatype=output['datatype'],
                                 data=output['data'],
                                 parameters=output.get('parameters', None))
                     for output in response['outputs']]
    return cls(model_name=model_name,
               model_version=response.get('model_version', None),
               response_id=response.get('id', None),
               parameters=response.get('parameters', None),
               infer_outputs=infer_outputs)

to_grpc()

Converts the InferResponse object to gRPC ModelInferResponse type.

Returns:

Type Description
ModelInferResponse

The ModelInferResponse gRPC message.

Source code in kserve/protocol/infer_type.py
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
def to_grpc(self) -> ModelInferResponse:
    """ Converts the InferResponse object to gRPC ModelInferResponse type.

    Returns:
        The ModelInferResponse gRPC message.
    """
    infer_outputs = []
    raw_output_contents = []
    for infer_output in self.outputs:
        if isinstance(infer_output.data, numpy.ndarray):
            infer_output.set_data_from_numpy(infer_output.data, binary_data=True)
        infer_output_dict = {
            "name": infer_output.name,
            "shape": infer_output.shape,
            "datatype": infer_output.datatype,
        }
        if infer_output.parameters:
            infer_output_dict["parameters"] = to_grpc_parameters(infer_output.parameters)
        if infer_output._raw_data is not None:
            raw_output_contents.append(infer_output._raw_data)
        else:
            if not isinstance(infer_output.data, List):
                raise InvalidInput("output data is not a List")
            infer_output_dict["contents"] = {}
            data_key = GRPC_CONTENT_DATATYPE_MAPPINGS.get(infer_output.datatype, None)
            if data_key is not None:
                infer_output._data = [bytes(val, 'utf-8') if isinstance(val, str)
                                      else val for val in
                                      infer_output.data]  # str to byte conversion for grpc proto
                infer_output_dict["contents"][data_key] = infer_output.data
            else:
                raise InvalidInput("to_grpc: invalid output datatype")
        infer_outputs.append(infer_output_dict)

    return ModelInferResponse(id=self.id, model_name=self.model_name, model_version=self.model_version,
                              outputs=infer_outputs, raw_output_contents=raw_output_contents,
                              parameters=to_grpc_parameters(self.parameters) if self.parameters else None)

to_rest()

Converts the InferResponse object to v2 REST InferResponse dict.

Returns:

Type Description
Dict

The InferResponse Dict.

Source code in kserve/protocol/infer_type.py
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
def to_rest(self) -> Dict:
    """ Converts the InferResponse object to v2 REST InferResponse dict.

    Returns:
        The InferResponse Dict.
    """
    infer_outputs = []
    for i, infer_output in enumerate(self.outputs):
        infer_output_dict = {
            "name": infer_output.name,
            "shape": infer_output.shape,
            "datatype": infer_output.datatype
        }
        if infer_output.parameters:
            infer_output_dict["parameters"] = to_http_parameters(infer_output.parameters)
        if isinstance(infer_output.data, numpy.ndarray):
            infer_output.set_data_from_numpy(infer_output.data, binary_data=False)
            infer_output_dict["data"] = infer_output.data
        elif isinstance(infer_output._raw_data, bytes):
            infer_output_dict["data"] = infer_output.as_numpy().tolist()
        else:
            infer_output_dict["data"] = infer_output.data
        infer_outputs.append(infer_output_dict)
    res = {
        'id': self.id,
        'model_name': self.model_name,
        'model_version': self.model_version,
        'outputs': infer_outputs
    }
    if self.parameters:
        res['parameters'] = to_http_parameters(self.parameters)
    return res

serialize_byte_tensor(input_tensor)

Serializes a bytes tensor into a flat numpy array of length prepended bytes. The numpy array should use dtype of np.object_. For np.bytes_, numpy will remove trailing zeros at the end of byte sequence and because of this it should be avoided. Args: input_tensor : np.array of the bytes tensor to serialize. Returns: serialized_bytes_tensor : The 1-D numpy array of type uint8 containing the serialized bytes in 'C' order.

Source code in kserve/protocol/infer_type.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def serialize_byte_tensor(input_tensor: numpy.ndarray):
    """
    Serializes a bytes tensor into a flat numpy array of length prepended
    bytes. The numpy array should use dtype of np.object_. For np.bytes_,
    numpy will remove trailing zeros at the end of byte sequence and because
    of this it should be avoided.
    Args:
        input_tensor : np.array of the bytes tensor to serialize.
    Returns:
        serialized_bytes_tensor : The 1-D numpy array of type uint8 containing the serialized bytes in 'C' order.
    """

    if input_tensor.size == 0:
        return ()

    # If the input is a tensor of string/bytes objects, then must flatten those
    # into a 1-dimensional array containing the 4-byte byte size followed by the
    # actual element bytes. All elements are concatenated together in "C" order.
    if (input_tensor.dtype == np.object_) or (input_tensor.dtype.type == np.bytes_):
        flattened_ls = []
        for obj in np.nditer(input_tensor, flags=["refs_ok"], order="C"):
            # If directly passing bytes to BYTES type,
            # don't convert it to str as Python will encode the
            # bytes which may distort the meaning
            if input_tensor.dtype == np.object_:
                if type(obj.item()) == bytes:
                    s = obj.item()
                else:
                    s = str(obj.item()).encode("utf-8")
            else:
                s = obj.item()
            flattened_ls.append(struct.pack("<I", len(s)))
            flattened_ls.append(s)
        flattened = b"".join(flattened_ls)
        return flattened
    return None

to_grpc_parameters(parameters)

Converts REST parameters to GRPC InferParameter objects

:param parameters: parameters to be converted. :return: converted parameters as Dict[str, InferParameter] :raises InvalidInput: if the parameter type is not supported.

Source code in kserve/protocol/infer_type.py
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
def to_grpc_parameters(parameters: Union[Dict[str, Union[str, bool, int]], MessageMap[str, InferParameter]]) \
        -> Dict[str, InferParameter]:
    """
    Converts REST parameters to GRPC InferParameter objects

    :param parameters: parameters to be converted.
    :return: converted parameters as Dict[str, InferParameter]
    :raises InvalidInput: if the parameter type is not supported.
    """
    grpc_params: Dict[str, InferParameter] = {}
    for key, val in parameters.items():
        if isinstance(val, str):
            grpc_params[key] = InferParameter(string_param=val)
        elif isinstance(val, bool):
            grpc_params[key] = InferParameter(bool_param=val)
        elif isinstance(val, int):
            grpc_params[key] = InferParameter(int64_param=val)
        elif isinstance(val, InferParameter):
            grpc_params[key] = val
        else:
            raise InvalidInput(f"to_grpc: invalid parameter value: {val}")
    return grpc_params

to_http_parameters(parameters)

Converts GRPC InferParameter parameters to REST parameters

:param parameters: parameters to be converted. :return: converted parameters as Dict[str, Union[str, bool, int]]

Source code in kserve/protocol/infer_type.py
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
def to_http_parameters(parameters: Union[dict, MessageMap[str, InferParameter]]) -> Dict[str, Union[str, bool, int]]:
    """
        Converts GRPC InferParameter parameters to REST parameters

        :param parameters: parameters to be converted.
        :return: converted parameters as Dict[str, Union[str, bool, int]]
        """
    http_params: Dict[str, Union[str, bool, int]] = {}
    for key, val in parameters.items():
        if isinstance(val, InferParameter):
            if val.HasField("bool_param"):
                http_params[key] = val.bool_param
            elif val.HasField("int64_param"):
                http_params[key] = val.int64_param
            elif val.HasField("string_param"):
                http_params[key] = val.string_param
        else:
            http_params[key] = val
    return http_params
Back to top