 with same pipeline name, step name and model nameexisting_services = model_deployer.find_model_server(

pipeline_name=pipeline_name,

pipeline_step_name=pipeline_step_name,

model_name=model_name,

if not existing_services:

raise RuntimeError(

f"No MLflow prediction service deployed by step "

f"'{pipeline_step_name}' in pipeline '{pipeline_name}' with name "

f"'{model_name}' is currently running."

service = existing_services[0]

# Let's try run a inference request against the prediction service

payload = json.dumps(

"inputs": {"messages": [{"role": "user", "content": "Tell a joke!"}]},

"params": {

"temperature": 0.5,

"max_tokens": 20,

},

response = requests.post(

url=service.get_prediction_url(),

data=payload,

headers={"Content-Type": "application/json"},

response.json()

Within the same pipeline, use the service from previous step to run inference this time using pre-built predict method

from typing_extensions import Annotated

import numpy as np

from zenml import step

from zenml.integrations.mlflow.services import MLFlowDeploymentService

# Use the service for inference

@step

def predictor(

service: MLFlowDeploymentService,

data: np.ndarray,

) -> Annotated[np.ndarray, "predictions"]:

"""Run a inference request against a prediction service"""

prediction = service.predict(data)

prediction = prediction.argmax(axis=-1)

return prediction

For more information and a full list of configurable attributes of the MLflow Model Deployer, check out the SDK Docs .

PreviousModel Deployers

NextSeldon

Last updated 19 days ago