run.pipeline.runs[1]  # index 0 is the current runAs shown in the example, we can get additional information about the current run using the StepContext, which is explained in more detail in the advanced docs.

Code example

This section combines all the code from this section into one simple script that you can use to see the concepts discussed above:

Putting it all together, this is how we can load the model trained by the svc_trainer step of our example pipeline from the previous sections:

from typing_extensions import Tuple, Annotated

import pandas as pd

from sklearn.datasets import load_iris

from sklearn.model_selection import train_test_split

from sklearn.base import ClassifierMixin

from sklearn.svm import SVC

from zenml import pipeline, step

from zenml.client import Client

@step

def training_data_loader() -> Tuple[

Annotated[pd.DataFrame, "X_train"],

Annotated[pd.DataFrame, "X_test"],

Annotated[pd.Series, "y_train"],

Annotated[pd.Series, "y_test"],

]:

"""Load the iris dataset as tuple of Pandas DataFrame / Series."""

iris = load_iris(as_frame=True)

X_train, X_test, y_train, y_test = train_test_split(

iris.data, iris.target, test_size=0.2, shuffle=True, random_state=42

return X_train, X_test, y_train, y_test

@step

def svc_trainer(

X_train: pd.DataFrame,

y_train: pd.Series,

gamma: float = 0.001,

) -> Tuple[

Annotated[ClassifierMixin, "trained_model"],

Annotated[float, "training_acc"],

]:

"""Train a sklearn SVC classifier and log to MLflow."""

model = SVC(gamma=gamma)

model.fit(X_train.to_numpy(), y_train.to_numpy())

train_acc = model.score(X_train.to_numpy(), y_train.to_numpy())

print(f"Train accuracy: {train_acc}")

return model, train_acc

@pipeline

def training_pipeline(gamma: float = 0.002):

X_train, X_test, y_train, y_test = training_data_loader()

svc_trainer(gamma=gamma, X_train=X_train, y_train=y_train)

if __name__ == "__main__":

# You can run the pipeline and get the run object directly

last_run = training_pipeline()

print(last_run.id)