Diamonds and Iteration

What you’ll learn¶

Build diamond DAGs that diverge and reconverge
Implement iterative refinement loops with standard Python control flow

Prerequisites: Sources and Sequencing, Branching and Merging, Metrics and Filtering Estimated time: 15 minutes

The previous tutorials covered individual building blocks: sources, chains, fan-out, merging, filtering, and multi-input operations. This tutorial combines those primitives into composite patterns you’ll use in real pipelines.

Pattern	Combines	Description
Diamond DAG	Fan-out + Merge + Chain	Diverge into parallel branches, reconverge, then continue
Iterative refinement	Generation + Scoring + Filtering + Transformation in a loop	Repeated generate → score → filter → modify cycles

from __future__ import annotations

from artisan.operations.curator import Filter, Merge
from artisan.operations.examples import (
    DataGenerator,
    DataTransformer,
    MetricCalculator,
)
from artisan.orchestration import PipelineManager
from artisan.utils import tutorial_setup
from artisan.visualization import (
    build_macro_graph,
    build_micro_graph,
    inspect_metrics,
    inspect_pipeline,
)

Graph legend: See Sources and Sequencing for box/arrow key.

Diamond DAG¶

A diamond is the simplest composite pattern: one source fans out to two independent branches, both branches process the same inputs differently, then a Merge step reconverges them before downstream processing continues.

              ┌─── transform (seed=100) ───┐
generate ─────┤                             ├─── merge ─── metrics
              └─── transform (seed=200) ───┘

When to use: You need to apply two independent transformations to the same inputs and then process the combined results — for example, generating candidates via two strategies and then scoring them together.

env_diamond = tutorial_setup("diamond_dag")

pipeline = PipelineManager.create(
    name="diamond_dag",
    delta_root=env_diamond.delta_root,
    staging_root=env_diamond.staging_root,
    working_root=env_diamond.working_root,
)
output = pipeline.output

# Source: generate 2 datasets
pipeline.run(operation=DataGenerator, name="generate", params={"count": 2, "seed": 42})

# Branch A: transform with one strategy
pipeline.run(
    operation=DataTransformer,
    name="transform_a",
    inputs={"dataset": output("generate", "datasets")},
    params={"seed": 100, "output_prefix": "branch_a_"},
)

# Branch B: transform with a different strategy
pipeline.run(
    operation=DataTransformer,
    name="transform_b",
    inputs={"dataset": output("generate", "datasets")},
    params={"seed": 200, "output_prefix": "branch_b_"},
)

# Reconverge: merge both branches into a single stream
pipeline.run(
    operation=Merge,
    name="merge",
    inputs={
        "branch_a": output("transform_a", "dataset"),
        "branch_b": output("transform_b", "dataset"),
    },
)

# Continue: score the combined results
pipeline.run(
    operation=MetricCalculator,
    name="metrics",
    inputs={"dataset": output("merge", "merged")},
)

result_diamond = pipeline.finalize()

build_macro_graph(env_diamond.delta_root)

The macro graph shows the diamond shape: data_generator feeds both data_transformer steps (fan-out), which converge on merge, and then metric_calculator processes the unified stream. The Merge step produces no new artifacts — it unions the two branches using passthrough semantics, so metric_calculator receives all 4 datasets (2 from each branch).

inspect_pipeline(env_diamond.delta_root)

The pipeline overview confirms the flow: step 0 produced 2 datasets, steps 1 and 2 each transformed those 2 into 2 more, Merge passed all 4 through, and metric_calculator scored each one.

build_micro_graph(env_diamond.delta_root)

Many real workflows follow a loop: generate candidates, score them, keep the best, modify survivors, and repeat. Artisan handles this with a standard Python for loop around pipeline.run() calls — no special API needed.

generate ──→ score ──→ filter ──→ transform ──→ score ──→ filter ──→ ...

Each iteration adds new steps to the same pipeline. Provenance tracks lineage across all rounds automatically.

When to use: Design optimization, evolutionary search, iterative screening — any workflow where you repeatedly score, select, and regenerate.

Building one round¶

Before looping, let’s build a single round to understand each piece. The four steps are:

Generate — create initial candidate datasets
Score — compute metrics on each candidate
Filter — keep candidates that meet a threshold
Transform — modify survivors for the next round

env_single = tutorial_setup("cycling_single_round")

pipeline = PipelineManager.create(
    name="cycling_single_round",
    delta_root=env_single.delta_root,
    staging_root=env_single.staging_root,
    working_root=env_single.working_root,
)
output = pipeline.output

# 1. Generate 6 initial candidates
pipeline.run(operation=DataGenerator, name="generate", params={"count": 6, "seed": 42})
datasets = output("generate", "datasets")

# 2. Score each candidate
pipeline.run(operation=MetricCalculator, name="score", inputs={"dataset": datasets})

# 3. Filter: keep candidates with max value >= 0
pipeline.run(
    operation=Filter,
    name="filter",
    inputs={"passthrough": datasets},
    params={
        "criteria": [
            {"metric": "distribution.max", "operator": "ge", "value": 0},
        ]
    },
)

# 4. Transform survivors
pipeline.run(
    operation=DataTransformer,
    name="transform",
    inputs={"dataset": output("filter", "passthrough")},
    params={"seed": 100},
)

result_single = pipeline.finalize()

inspect_metrics(env_single.delta_root)

The metrics table shows each candidate’s scores. The Filter step used distribution.max >= 0 to decide which candidates survive. Filter auto-discovers the metrics from the score step via forward provenance walk from the passthrough datasets — no explicit metric input wiring needed. Candidates that pass continue to the Transform step; those that fail are dropped.

build_macro_graph(env_single.delta_root)

build_micro_graph(env_single.delta_root)

The graphs show the single-round flow: generate → score → filter → transform. Notice that Filter has no direct edge from the score step — it discovers the metrics automatically via forward provenance walk from the passthrough datasets.

Looping over multiple rounds¶

Now wrap the score → filter → transform sequence in a for loop. Each round’s output feeds into the next round’s input. The key insight: pipeline.output() returns OutputReference objects that can be passed directly to the next pipeline.run() call. A Python variable reassignment (datasets = ...) is all you need to chain rounds together.

env_cycle = tutorial_setup("cycling_multi_round")

N_ROUNDS = 3

pipeline = PipelineManager.create(
    name="cycling_multi_round",
    delta_root=env_cycle.delta_root,
    staging_root=env_cycle.staging_root,
    working_root=env_cycle.working_root,
)
output = pipeline.output

# Initial generation
pipeline.run(operation=DataGenerator, name="generate", params={"count": 6, "seed": 42})
datasets = output("generate", "datasets")

# Iterative refinement: score → filter → transform, repeated N rounds
for round_num in range(N_ROUNDS):
    # Score current candidates
    scored_name = f"score_r{round_num}"
    pipeline.run(
        operation=MetricCalculator, name=scored_name, inputs={"dataset": datasets}
    )

    # Filter: keep candidates above threshold
    filter_name = f"filter_r{round_num}"
    pipeline.run(
        operation=Filter,
        name=filter_name,
        inputs={"passthrough": datasets},
        params={
            "criteria": [
                {"metric": "distribution.max", "operator": "ge", "value": 0},
            ]
        },
    )

    # Transform survivors for the next round
    transform_name = f"transform_r{round_num}"
    pipeline.run(
        operation=DataTransformer,
        name=transform_name,
        inputs={"dataset": output(filter_name, "passthrough")},
        params={"seed": (round_num + 1) * 100},
    )

    # Feed modified datasets into the next round
    datasets = output(transform_name, "dataset")

result_cycle = pipeline.finalize()

Each round adds 3 steps (score, filter, transform), so the total grows linearly with N_ROUNDS. Each round’s Transform output became the next round’s Score input through the datasets variable reassignment.

No special framework API was needed — the loop is plain Python. The framework handles provenance tracking across rounds automatically.

inspect_pipeline(env_cycle.delta_root)

The pipeline overview shows all the steps. Each group of three (metric_calculator → filter → data_transformer) is one round. Watch the artifact counts: if the filter drops candidates in early rounds, later rounds process fewer artifacts.

inspect_metrics(env_cycle.delta_root)

The metrics table shows scores from every round. Metrics from round 0 are computed on the initial candidates; round 1 metrics come from the transformed survivors, and so on. Comparing across rounds reveals whether the iterative process is improving candidates.

build_macro_graph(env_cycle.delta_root)

The macro graph shows the repeating pattern across rounds. Each round’s score → filter → transform subgraph is connected to the previous round’s output, forming a chain of refinement cycles. Provenance arrows trace lineage from the initial candidates through every transformation.

build_micro_graph(env_cycle.delta_root)

Use the stepper to walk through rounds one step at a time. This is the best way to verify that lineage is tracked correctly across round boundaries — each round’s Transform outputs should trace back through the Filter to the previous round’s Transform outputs.

Summary¶

This tutorial covered two composite patterns that build on the fundamentals from previous tutorials:

Diamond DAG — fan-out into parallel branches, merge, and continue. Combines fan-out and merge with downstream processing.
Iterative refinement — a Python for loop around pipeline.run() calls implements score → filter → transform cycles. No special API needed; provenance tracks lineage across all rounds automatically.

Key takeaway: Artisan’s composability comes from Python, not from framework abstractions. OutputReference objects are the wiring mechanism, and standard control flow (loops, conditionals) is how you build complex topologies.

Next: Composable Operations — defining reusable operation compositions with CompositeDefinition

Deeper understanding:

Execution Flow — how steps dispatch and execute
Provenance System — how lineage tracking works across rounds