uns[0]

# OR: last_run = my_pipeline.get_runs()[0]model_trainer_step: StepView = last_run.get_step("model_trainer")

model: ArtifactView = model_trainer_step.output

loaded_model = model.read()

pipeline: PipelineResponseModel = zenml.client.Client().get_pipeline("first_pipeline")

# OR: pipeline = pipeline_instance.model

last_run: PipelineRunResponseModel = pipeline.last_run

# OR: last_run = pipeline.runs[0]

# OR: last_run = pipeline.get_runs(custom_filters)[0]

# OR: last_run = pipeline.last_successful_run

model_trainer_step: StepRunResponseModel = last_run.steps["model_trainer"]

model: ArtifactResponseModel = model_trainer_step.output

loaded_model = model.load()

Check out this page for more information on how to programmatically fetch information about previous pipeline runs.

Controlling the step execution order

from zenml.pipelines import pipeline

@pipeline

def my_pipeline(step_1, step_2, step_3):

step_1()

step_2()

step_3()

step_3.after(step_1)  # Old: Use the `step.after(...)` method

step_3.after(step_2)

from zenml import pipeline

@pipeline

def my_pipeline():

step_1()

step_2()

step_3(after=["step_1", "step_2"])  # New: Pass the `after` argument when calling a step

Check out this page for more information on how to control the step execution order.

Defining steps with multiple outputs

# Old: Use the `Output` class

from zenml.steps import step, Output

@step

def my_step() -> Output(int_output=int, str_output=str):

...

# New: Use a `Tuple` annotation and optionally assign custom output names

from typing_extensions import Annotated

from typing import Tuple

from zenml import step

# Default output names `output_0`, `output_1`

@step

def my_step() -> Tuple[int, str]:

...

# Custom output names

@step

def my_step() -> Tuple[

Annotated[int, "int_output"],

Annotated[str, "str_output"],

]:

...

Check out this page for more information on how to annotate your step outputs.

Accessing run information inside steps

from zenml.steps import StepContext, step