Dynamic Pipelines

Write dynamic pipelines

Why Dynamic Pipelines?

Traditional ZenML pipelines require you to define the entire DAG structure at pipeline definition time. While this works well for many use cases, there are scenarios where you need more flexibility:

• Runtime-dependent workflows: When the number of steps or their configuration depends on data computed during pipeline execution
• Dynamic parallelization: When you need to spawn multiple parallel step executions based on runtime conditions
• Conditional execution: When the workflow structure needs to adapt based on intermediate results

Dynamic pipelines allow you to write pipelines that generate their DAG structure dynamically at runtime, giving you the power of Python's control flow (loops, conditionals) combined with ZenML's orchestration capabilities.

Dynamic pipelines are powerful but easy to get wrong (e.g., .load() vs .chunk(), mapping vs submit). If you use an AI coding agent, the zenml-pipeline-authoring skill can guide implementation step-by-step. See LLM tooling.

Basic Example

The simplest dynamic pipeline uses regular Python control flow to determine step execution:

from zenml import step, pipeline

@step
def generate_int() -> int:
    return 3

@step
def do_something(index: int) -> None:
    print(f"Processing index {index}")

@pipeline(dynamic=True)
def dynamic_pipeline() -> None:
    count = generate_int()
    # `count` is an artifact, we now load the data
    count_data = count.load()

    for idx in range(count_data):
        # This will run sequentially, like regular Python code would.
        do_something(idx)

if __name__ == "__main__":
    dynamic_pipeline()

In this example, the number of do_something steps executed depends on the value returned by generate_int(), which is only known at runtime.

Key Features

Dynamic Step Configuration

You can configure steps dynamically within your pipeline using with_options():

@pipeline(dynamic=True)
def dynamic_pipeline():
    some_step.with_options(enable_cache=False)()

This allows you to modify step behavior based on runtime conditions or data.

Artifact name substitutions in dynamic pipelines

Dynamic pipelines support the same artifact name substitutions as regular pipelines. This matters when a dynamically generated step has outputs whose names include runtime-friendly placeholders. The substituted artifact name is still a real output that you can pass to downstream steps.

from typing import Annotated

from zenml import ArtifactConfig, pipeline, step

@step(substitutions={"suffix": "validated"})
def produce() -> Annotated[int, ArtifactConfig(name="score_{suffix}")]:
    return 1

@step
def consume(score: int) -> None:
    print(score)

@pipeline(dynamic=True)
def dynamic_pipeline() -> None:
    score = produce()
    consume(score)

One caveat: when you use child_pipeline.embed(...), the child pipeline's own configuration is not applied. That includes child-level substitutions; the parent run's configuration controls the steps that execute inline.

Step inputs as parameters

Any value you pass to a step that is not the output of another step is uploaded to the artifact store as an external artifact, even a small int. Each upload costs a write to the artifact store and a request to the server, which adds up when a pipeline calls steps in a loop.

Set the ZENML_PARAMETER_SIZE_THRESHOLD environment variable to pass JSON-serializable inputs as step parameters instead, which skips the upload. The variable must be set in the environment in which the pipeline executes, not on the client that starts the run. See this page for how to configure environment variables for pipeline execution.

Value	Behavior
unset or 0	Every raw input is uploaded. This is the default.
a positive number	JSON-serializable inputs up to that many bytes become step parameters. Larger inputs are uploaded.
-1	Every JSON-serializable input becomes a step parameter.

Inputs that cannot be parameters are still uploaded, so raising the threshold is safe. Wrap an input with ExternalArtifact to keep uploading it.

Step Runtime Configuration

You can control where a step executes by specifying its runtime:

• runtime="inline": The step runs in the orchestration environment (same process/container as the orchestrator)
• runtime="isolated": The orchestrator spins up a separate step execution environment (new container/process)

@step(runtime="isolated")
def some_step() -> None:
    # This step will run in its own isolated environment
    ...

@step(runtime="inline")
def another_step() -> None:
    # This step will run in the orchestration environment
    ...

Use runtime="isolated" when you need:

• Better resource isolation
• Different environment requirements
• Parallel execution (see below)

Use runtime="inline" when you need:

• Faster execution (no container startup overhead)
• Shared resources with the orchestrator
• Sequential execution

Map/Reduce over collections

Dynamic pipelines support a high-level map/reduce pattern over sequence-like step outputs. This lets you fan out a step across items of a collection and then reduce the results without manually writing loops or loading data in the orchestration environment.

from zenml import pipeline, step

@step
def producer() -> list[int]:
    return [1, 2, 3]

@step
def worker(value: int) -> int:
    return value * 2

@step
def reducer(values: list[int]) -> int:
    return sum(values)

@pipeline(dynamic=True, enable_cache=False)
def map_reduce():
    values = producer()
    results = worker.map(values)   # fan out over collection
    reducer(results)               # pass list of artifacts directly

For a complete agentic workflow that combines dynamic mapping, reduction, and a human approval gate, see the agentic_hitl_pipeline example.

Key points:

• step.map(...) fans out a step over sequence-like inputs. These inputs can be either

  • a single list-like output artifact (see the code sample above)
  • a list of output artifacts.
  • the output of a .map(...) or .product(...) call if the respective step only returns a single output artifact
• Steps can accept lists of artifacts directly as inputs (useful for reducers).
• You can pass the mapped output directly to a downstream step without loading in the orchestration environment.

Mapping semantics: map vs product

• step.map(...): If multiple sequence-like inputs are provided, all must have the same length n. ZenML creates n mapped steps where the i-th step receives the i-th element from each input.
• step.product(...): Creates a mapped step for each combination of elements across all input sequences (cartesian product).

Example (cartesian product):

from zenml import pipeline, step

@step
def int_values() -> list[int]:
    return [1, 2]

@step
def str_values() -> list[str]:
    return ["a", "b", "c"]

@step
def do_something(a: int, b: str) -> int:
    ...

@pipeline(dynamic=True)
def cartesian_example():
    a = int_values()
    b = str_values()
    # Produces 2 * 3 = 6 mapped steps
    do_something.product(a=a, b=b)

Broadcasting inputs with unmapped(...)

If you want to pass a sequence-like artifact as a whole to each mapped invocation (i.e., avoid splitting), wrap it with unmapped(...):

from zenml import pipeline, step, unmapped

@step
def producer(length: int) -> list[int]:
    return [1] * length

@step
def consumer(a: int, b: list[int]) -> None:
    # `b` is the full list for every mapped call
    ...

@pipeline(dynamic=True)
def unmapped_example():
    a = producer(length=3)   # list of 3 ints
    b = producer(length=4)   # list of 4 ints
    consumer.map(a=a, b=unmapped(b))

Unpacking mapped outputs

If a mapped step returns multiple outputs, you can split them into separate lists (one per output) using unpack(). This returns a tuple of lists of artifact futures, aligned by mapped invocation.

from zenml import pipeline, step

@step
def create_int_list() -> list[int]:
    return [1, 2]

@step
def compute(a: int) -> tuple[int, int]:
    return a * 2, a * 3

@pipeline(dynamic=True)
def map_pipeline():
    ints = create_int_list()
    results = compute.map(a=ints)  # Map over [1, 2]

    # Unpack per-output across all mapped invocations
    double, triple = results.unpack()

    # Each element is an ArtifactFuture; load to get concrete values
    doubles = [f.load() for f in double]  # [2, 4]
    triples = [f.load() for f in triple]  # [3, 6]

Notes:

• results is a future that refers to all outputs of all steps, and unpack() works for both .map(...) and .product(...).
• Each list contains future objects that refer to a single artifact.

Manual Looping: .chunk() vs .load()

When looping over artifacts manually, you need two different operations:

Method	Purpose	When to Use
.load()	Gets the actual data	Making decisions, filtering, control flow
.chunk(idx)	Creates a DAG edge	Passing to downstream steps

Mental model: .chunk() is for wiring (tells the orchestrator "this step depends on item X from upstream"), .load() is for decisions (gets values for your Python logic). You typically need both: load to iterate and decide, chunk to wire up the DAG.

from zenml import pipeline, step

@step
def create_int_list() -> list[int]:
    return [1, 2, 3, 4]

@step
def compute(a: int) -> int:
    return a * 2

@pipeline(dynamic=True)
def custom_loop():
    ints = create_int_list()

    # .load() to get values for Python control flow (iteration + filtering)
    for index, value in enumerate(ints.load()):
        if value % 2 == 0:
            # .chunk() to create DAG edge (wiring to downstream step)
            chunk = ints.chunk(index=index)
            compute(chunk)

Parallel Step Execution

Dynamic pipelines support true parallel execution using step.submit(). This method returns a StepFuture that you can use to wait for results or pass to downstream steps:

from zenml import step, pipeline

@step
def some_step(arg: int) -> int:
    return arg * 2

@pipeline(dynamic=True)
def dynamic_pipeline():
    # Submit a step for parallel execution
    future = some_step.submit(arg=1)
    
    # Wait and get artifact response(s)
    artifact = future.result()
    
    # Wait and load artifact data
    data = future.load()
    
    # Pass the output to another step
    downstream_step(future)

    # Run multiple steps in parallel
    for idx in range(3):
        some_step.submit(arg=idx)

The StepFuture object provides several methods:

• result(): Wait for the step to complete and return the artifact response(s)
• load(): Wait for the step to complete and load the actual artifact data
• Pass directly: You can pass a StepFuture directly to downstream steps, and ZenML will automatically wait for it

When using step.submit(), steps with runtime="isolated" will execute in separate containers/processes, while steps with runtime="inline" will execute in separate threads within the orchestration environment.

Ordering submitted steps

A submitted step starts as soon as its inputs are available. To impose an order between steps that have no data dependency, use after or start_after:

• after= waits for the upstream step to finish before starting.
• start_after= waits for the upstream step to start before starting.

start_after is useful when an upstream step is long-running and you want a dependent to run alongside it rather than after it. A common case is a step that brings up a service and a second step that uses it:

@pipeline(dynamic=True)
def serve_and_query():
    server = serve_model.submit()  # long-running
    # Starts once the server step is running, not when it finishes.
    query = run_inference.submit(start_after=server)
    query.wait()

Using after=server here would deadlock, since the dependent would wait for the long-running server step to finish. Both parameters accept a single future or a list, and you can combine them: run_inference.submit(after=preprocess, start_after=server). start_after is available on step.submit(...), step.map(...), step.product(...), and on a direct synchronous step(...) call, where the entrypoint blocks until the upstream has started. The upstream can be another step or a submitted child pipeline, for example run_inference(start_after=serve_pipeline.submit()).

start_after orders execution, it does not probe readiness. The dependent starts once the upstream step has launched (for isolated steps, once it is submitted to the infrastructure), which does not guarantee that whatever the upstream sets up is ready to serve. Add your own connection retries if the dependent needs to reach a service the upstream starts. A failed upstream counts as started, so a start_after dependent is released rather than blocked when the upstream fails. Circular start_after dependencies are not detected and will stall the involved steps.

Child pipelines inside dynamic pipelines

Dynamic pipelines can call other dynamic pipelines from their @pipeline body. This is useful for composing larger workflows out of reusable dynamic building blocks.

Key behavior:

• Only dynamic pipelines can be called as child pipelines.
• Child pipelines run on the same stack as the parent run.
• Child pipelines can run synchronously (child(...)) or concurrently (child.submit(...)).
• Child pipeline calls are only allowed in pipeline bodies, not inside step functions.
• Child pipelines reuse the parent run's Docker image — they don't trigger a new build. The child snapshot inherits the parent's build, code reference, and code path so the child runs against the exact same image and source bundle as the parent.

Child pipeline outputs are returned as artifact references:

• None
• A single output artifact
• A tuple of output artifacts

These outputs can be passed directly to downstream steps.

from zenml import pipeline, step

@step
def produce_number() -> int:
    return 42

@pipeline(dynamic=True)
def child_pipeline():
    return produce_number()

@step
def consume_number(value: int) -> None:
    print(value)

@pipeline(dynamic=True)
def parent_pipeline():
    child_output = child_pipeline()
    consume_number(child_output)

For concurrent execution, use submit() and wait on the future:

@pipeline(dynamic=True)
def parent_pipeline_concurrent():
    future = child_pipeline.submit()
    child_output = future.result()
    consume_number(child_output)

Inline child pipelines with embed(...)

Use child_pipeline.embed(...) if you want to reuse another dynamic pipeline's body without creating a child pipeline run.

@pipeline(dynamic=True)
def parent_pipeline_inline():
    # Executes child steps in the parent run context
    child_output = child_pipeline.embed()
    consume_number(child_output)

embed(...) behavior:

• It executes the child pipeline entrypoint inline as part of the parent run.
• It does not create a separate child run in the dashboard.
• It is only valid inside a dynamic pipeline body.
• It is not allowed inside @step functions.

Limitations of embed(...). Unlike child_pipeline(...) and child_pipeline.submit(...), the inline form does not apply the child pipeline's own configuration. The parent run's configuration governs every step that runs inline:

• Child-level settings, retry, enable_cache, enable_step_logs, environment, secrets, tags, substitutions, model, and on_init / on_success / on_failure / on_cleanup hooks are ignored.
• Per-step Docker overrides on the child pipeline are also ignored — the parent's image is used for any inline isolated step.
• depends_on config templates declared on the child pipeline are not picked up.
• There is no failure isolation: an exception inside the inline body aborts the parent run.

If any of these matter to your use case, call the child as child_pipeline(...) (sync) or child_pipeline.submit(...) (concurrent) instead. Both create a real child run with its own configuration applied.

In short, use:

• child_pipeline(...) for a synchronous child run
• child_pipeline.submit(...) for a concurrent child run
• child_pipeline.embed(...) for embedded execution in the parent run

Resume idempotency depends on submit order. Child pipeline child runs are identified by the order of child_pipeline(...) / child_pipeline.submit(...) calls in the parent body: the first call to my_pipeline becomes pipeline:my_pipeline, the second becomes pipeline:my_pipeline_2, and so on. On resume, ZenML reuses an existing child run only if the same call appears in the same position. If you reorder, insert, or remove child pipeline calls before existing ones, every subsequent ID shifts and previously completed children are re-executed. Same caveat applies to step invocation IDs.

Build, code, and Docker settings inheritance

Child runs share the parent's orchestration environment, image, and code bundle. This has two consequences worth knowing:

• No new Docker build. The child snapshot inherits the parent's build, code_reference, and code_path. The child runs against the exact same image and source bundle as the parent — there is no separate build step, and the child's code/dependencies must already be installed in the parent's image.
• Pipeline-level Docker settings on the child are ignored. When a child pipeline (or a child step) declares non-default docker_settings, those settings are silently overridden by the parent's. If you need a different image for a step inside a child pipeline, configure that step with a step_operator or use runtime="isolated" together with stack-level resource configuration on the parent.

This applies to all three call modes (child(...), child.submit(...), and child.embed(...)).

Permissions and authentication for nested runs

Nested runs orchestrate from the parent's environment, so they share the parent's API token. The token must be scoped to the root run of the nesting tree — the root orchestrator can mint per-child-run tokens for any descendant. Child runs cannot mint tokens for their siblings; only descendants of the same root tree are reachable from a given parent token.

This is transparent for the default flow (the root orchestrator launches everything in the same environment). It matters if you build automation on top of ZENML_PIPELINE_RUN_ID tokens — those tokens give you read/update access to the run they were minted for and any of its descendants, but not to siblings or unrelated runs.

Config Templates with depends_on

You can use YAML configuration files to provide default parameters for steps using the depends_on parameter:

# config.yaml
steps:
  some_step:
    parameters:
      arg: 3

# run.py
from zenml import step, pipeline

@step
def some_step(arg: int) -> None:
    print(f"arg is {arg}")

@pipeline(dynamic=True, depends_on=[some_step])
def dynamic_pipeline():
    some_step()

if __name__ == "__main__":
    dynamic_pipeline.with_options(config_path="config.yaml")()

The depends_on parameter tells ZenML which steps can be configured via the YAML file. This is particularly useful when you want to allow users to configure pipeline behavior without modifying code.

Pass pipeline parameters when running snapshots from the server

When running a snapshot from the server (either via the UI or the SDK/Rest API), you can now pass pipeline parameters for your dynamic pipelines.

For example:

from zenml.client import Client

Client().trigger_pipeline(snapshot_id=<ID>, run_configuration={"parameters": {"my_param": 3}})

Limitations and Known Issues

Execution modes

When using the FAIL_FAST execution mode, failure of a step does not immediately cancel other inline steps. Instead, they continue executing until finished. Isolated steps on the other hand will be shut down immediately.

Orchestrator Support

Dynamic pipelines are currently only supported by:

Orchestrator	Isolated steps	Handles orchestration environment failures
LocalOrchestrator	❌	❌
LocalDockerOrchestrator	❌	❌
KubernetesOrchestrator	✅	✅
VertexOrchestrator	✅	❌
SagemakerOrchestrator	✅	❌
AzureMLOrchestrator	✅	❌

Artifact Loading

When you call .load() on an artifact in a dynamic pipeline, it synchronously loads the data. For large artifacts or when you want to maintain parallelism, consider passing the step outputs (future or artifact) directly to downstream steps instead of loading them.

Mapping Limitations

• Mapping is currently supported only over artifacts produced within the same pipeline run (mapping over raw data or external artifacts is not supported).
• Chunk size for mapped collection loading defaults to 1 and is not yet configurable.

Best Practices

1. Use runtime="isolated" for parallel steps: This ensures better resource isolation and prevents interference between concurrent step executions.
2. Handle step outputs appropriately: If you need the data immediately, use .load(). If you're just passing to another step, pass the output directly.
3. Be mindful of resource usage: Running many steps in parallel can consume significant resources. Monitor your orchestrator's resource limits.
4. Test incrementally: Start with simple dynamic pipelines and gradually add complexity. Dynamic pipelines can be harder to debug than static ones.
5. Use config templates for flexibility: The depends_on feature allows you to make pipelines configurable without code changes.

When to Use Dynamic Pipelines

Dynamic pipelines are ideal for:

• AI agent orchestration: Coordinating multiple autonomous agents (e.g., retrieval or reasoning agents) whose interactions or number of invocations are determined at runtime
• Hyperparameter tuning: Spawning multiple training runs with different configurations
• Data processing: Processing variable numbers of data chunks in parallel
• Conditional workflows: Adapting pipeline structure based on runtime data
• Dynamic batching: Creating batches based on available data
• Multi-agent and collaborative AI workflows: Building flexible, adaptive workflows where agents or LLM-driven components can be dynamically spawned, routed, or looped based on outputs, results, or user input

For most standard ML workflows, traditional static pipelines are simpler and more maintainable. Use dynamic pipelines when you specifically need runtime flexibility that static pipelines cannot provide.

Real-World Example: Hierarchical Document Search

The examples/hierarchical_doc_search_agent example combines dynamic pipelines with Pydantic AI agents for intelligent document traversal. It demonstrates:

• Using .with_options() to pass parameters vs artifacts
• The .chunk() vs .load() pattern: chunks for wiring the DAG, loads for making traversal decisions
• Spawning steps dynamically based on AI agent decisions

Each traverse_node call appears as a separate step in the DAG, created at runtime based on what the agent decides to explore.

Two other examples are useful when you want to see dynamic pipelines in more specialized settings:

• examples/rlm_document_analysis shows a Recursive Language Model style document-analysis workflow. ZenML decides how many chunk-processing steps to create at runtime, while the LLM loop inside each chunk decides which typed search tools to use.
• examples/optuna_hyperparameter_tuning combines Optuna's ask API with ZenML dynamic pipelines. Optuna decides which hyperparameters to try next; ZenML runs the trials, tracks their artifacts and metadata, and can fan the work out in parallel.