hsml.default_predictor #

DefaultPredict #

Predictor for deployments whose requests are described by a deployment schema.

Validates each request against the schema, looks up and transforms the feature vector through the feature view, runs the model when the deployment has one, and logs the request when the feature view has logging enabled. Without a model it returns the transformed vectors. Subclass it and override load_model or model_predict for models the default loader cannot handle.

Example

# predictor.py, deployed with model.deploy(script_file="predictor.py",
# default_predictor=True) so the schema is still inferred
from hsml.default_predictor import DefaultPredict


class Predict(DefaultPredict):
    def load_model(self, model_files_path):
        return my_framework.load(model_files_path)

    def model_predict(self, feature_vectors):
        return self.model.predict_proba(
            feature_vectors[self.model_input_columns]
        )

check_model_input_contract #

check_model_input_contract() -> None

Fail unless the model's input schema is served by the transformed feature view schema.

Columnar inputs must all exist among the transformed columns with a compatible type family; a tensor input must be a single tensor whose last dimension is the number of transformed columns. Nothing is guessed: a missing or incompatible input schema raises.

RAISES	DESCRIPTION
`hopsworks.client.exceptions.ModelServingException`	On any mismatch, naming the columns or shapes involved.

close #

close(timeout: float = 5.0) -> bool

Drain the queued log requests.

Registered to run at interpreter exit; call it earlier for an orderly shutdown.

PARAMETER	DESCRIPTION
`timeout`	Seconds to wait for the queue to empty and the worker to stop. TYPE: `float` DEFAULT: `5.0`

RETURNS	DESCRIPTION
`bool`	`True` when the queue emptied in time.

fetch_feature_vectors #

fetch_feature_vectors(rows: list[dict[str, Any]]) -> Any

Look up and transform the feature vectors for rows.

Blocking. A deployment served by KServe uses fetch_feature_vectors_async instead, so the event loop is not held for the round trip.

PARAMETER	DESCRIPTION
`rows`	Validated rows as returned by `prepare_rows`. TYPE: `list[dict[str, Any]]`

RETURNS	DESCRIPTION
`Any`	A pandas DataFrame in the transformed schema order, one row per input row.

RAISES	DESCRIPTION
`fastapi.HTTPException`	404 when an entity is missing, 422 when a transformation fails, 503 when the feature store is unreachable.

fetch_feature_vectors_async `async` #

fetch_feature_vectors_async(
    rows: list[dict[str, Any]],
) -> Any

Look up and transform the feature vectors for rows, without holding the loop.

The lookup is a round trip to the online store, and it is most of what a request waits for: a cluster measurement put it at 2.5 ms of a 12.6 ms mean request, of which only 0.2 ms was CPU, and at p99 it was 71 of 90 ms. Holding the event loop for that is what capped a deployment's throughput.

PARAMETER	DESCRIPTION
`rows`	Validated rows as returned by `prepare_rows`. TYPE: `list[dict[str, Any]]`

RETURNS	DESCRIPTION
`Any`	A pandas DataFrame in the transformed schema order, one row per input row.

RAISES	DESCRIPTION
`fastapi.HTTPException`	the same errors as the blocking method.

load_model #

load_model(model_files_path: str) -> Any

Load the single pickle or joblib file under model_files_path.

PARAMETER	DESCRIPTION
`model_files_path`	Directory holding the model files (`MODEL_FILES_PATH` in the pod). TYPE: `str`

RETURNS	DESCRIPTION
`Any`	The loaded model object.

RAISES	DESCRIPTION
`hopsworks.client.exceptions.ModelServingException`	If there is not exactly one candidate file; override this method or pass your own script then.

log #

log(
    rows: list[dict[str, Any]],
    feature_vectors: Any,
    predictions: Any,
    request_id: str | None = None,
) -> None

Queue the request for logging through the feature view when logging is enabled.

The logging frame is built and handed to the wrapper's async logger on a background thread, so this returns as soon as the request is queued. A full queue drops the request's rows and counts the drop; a failure on the worker is counted and logged and never fails a prediction.

PARAMETER	DESCRIPTION
`rows`	Validated rows as returned by `prepare_rows`. TYPE: `list[dict[str, Any]]`
`feature_vectors`	The DataFrame returned by `fetch_feature_vectors`, carrying the logging metadata. TYPE: `Any`
`predictions`	The model output, ignored without a model. TYPE: `Any`
`request_id`	Correlation id stored with every logged row. TYPE: `str \| None` DEFAULT: `None`

model_predict #

model_predict(feature_vectors: Any) -> Any

Run the model on the transformed feature vectors, or return them unchanged when there is no model.

PARAMETER	DESCRIPTION
`feature_vectors`	The DataFrame returned by `fetch_feature_vectors`. TYPE: `Any`

RETURNS	DESCRIPTION
`Any`	The model's predictions as a list, or `feature_vectors` itself without a model.

RAISES	DESCRIPTION
`fastapi.HTTPException`	500 when the model raises.

predict `async` #

predict(inputs: Any, request_id: str | None = None) -> Any

Serve one request: validate, look up, predict or return the vectors, log.

A coroutine, because the model server awaits it and the feature lookup inside is a round trip to the online store. A blocking predict runs on the server's event loop, so that round trip stops every other request in the deployment.

A subclass that overrides predict must be async def

The wrapper awaits the component's own predict only when that method is a coroutine function. A synchronous override calling super().predict(...) therefore returns the coroutine rather than its result, and the deployment answers with an unawaited object on the first request. Declare the override async def and await super().predict(...), or call predict_blocking, which runs the same body to completion.

PARAMETER	DESCRIPTION
`inputs`	The request rows, as objects keyed by field name, arrays in schema order, or the v2 tensors of a gRPC request. TYPE: `Any`
`request_id`	Correlation id from the `x-request-id` header; generated when absent. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Any`	The predictions list with a model, else `{"predictions": [...vectors...], "columns": [...]}`,
`Any`	or the same response as v2 output tensors when the request carried tensors.

predict_blocking #

predict_blocking(
    inputs: Any, request_id: str | None = None
) -> Any

predict for a caller with no event loop.

Same request, same response, same errors; the lookup blocks the calling thread instead of yielding. For a script or a notebook driving the predictor directly, where there is no loop to hold up and nothing else waiting on it.

PARAMETER	DESCRIPTION
`inputs`	As for `predict`. TYPE: `Any`
`request_id`	As for `predict`. TYPE: `str \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Any`	As for `predict`.

prepare_rows #

prepare_rows(inputs: Any) -> list[dict[str, Any]]

Validate the request rows against the schema.

PARAMETER	DESCRIPTION
`inputs`	The request rows, as objects keyed by field name or arrays in schema order. TYPE: `Any`

RETURNS	DESCRIPTION
`list[dict[str, Any]]`	The rows as dicts keyed by field name.

RAISES	DESCRIPTION
`fastapi.HTTPException`	400 with the structured errors on a mismatch, 413 on an oversize batch.

PredictionError #

Bases: ModelServingException

A request could not be served; carries the HTTP status and the structured error detail.

run_kserve_wrapper #

run_kserve_wrapper() -> None

Replace this process with the KServe wrapper that serves the Predict class of SCRIPT_PATH.

Deployments without a model artifact are started as python <script>; calling this from the script's __main__ block gives them the same server, payload handling, and feature logger as model deployments.