peline

from zenml.client import Client

@pipelinedef feature_engineering_pipeline():

dataset = load_data()

# This returns artifacts called "iris_training_dataset" and "iris_testing_dataset"

train_data, test_data = prepare_data()

@pipeline

def training_pipeline():

client = Client()

# Fetch by name alone - uses the latest version of this artifact

train_data = client.get_artifact_version(name="iris_training_dataset")

# For test, we want a particular version

test_data = client.get_artifact_version(name="iris_testing_dataset", version="raw_2023")

# We can now send these directly into ZenML steps

sklearn_classifier = model_trainer(train_data)

model_evaluator(model, sklearn_classifier)

materialized in memory in the

Pattern 2: Artifact exchange between pipelines through a Model

While passing around artifacts with IDs or names is very useful, it is often desirable to have the ZenML Model be the point of reference instead.

ZenML Model. Each time the

On the other side, the do_predictions pipeline simply picks up the latest promoted model and runs batch inference on it. It need not know of the IDs or names of any of the artifacts produced by the training pipeline's many runs. This way these two pipelines can independently be run, but can rely on each other's output.

In code, this is very simple. Once the pipelines are configured to use a particular model, we can use get_step_context to fetch the configured model within a step directly. Assuming there is a predict step in the do_predictions pipeline, we can fetch the production model like so:

from zenml import step, get_step_context

# IMPORTANT: Cache needs to be disabled to avoid unexpected behavior

@step(enable_cache=False)

def predict(

data: pd.DataFrame,

) -> Annotated[pd.Series, "predictions"]:

# model name and version are derived from pipeline context

model = get_step_context().model

# Fetch the model directly from the model control plane

model = model.get_model_artifact("trained_model")

# Make predictions