Writing a Composite

What you’ll learn¶

How to define a CompositeDefinition with inputs, outputs, and parameters
How compose() wires internal operations using CompositeContext
How to run your composite in collapsed and expanded mode
How to verify outputs in both modes

Prerequisites: Composable Operations
Estimated time: 20 minutes
GPU required: No.

The Composable Operations tutorial showed how to use composites. This tutorial shows how to write one from scratch.

A CompositeDefinition encapsulates a multi-operation workflow as a single reusable unit. You define what goes in, what comes out, and how the internal operations connect. The pipeline decides how to execute it — collapsed (in-memory) or expanded (separate steps).

Section	What you’ll build
Anatomy	The five parts of a `CompositeDefinition`
Build	A `GenerateAndFilter` composite from scratch
Test	Run collapsed and expanded, verify outputs match

from __future__ import annotations

from enum import StrEnum
from typing import ClassVar

from pydantic import BaseModel, Field

from artisan.composites import (
    CompositeContext,
    CompositeDefinition,
)
from artisan.operations.curator import Filter
from artisan.operations.examples import (
    DataGenerator,
    MetricCalculator,
)
from artisan.orchestration import PipelineManager
from artisan.schemas.specs.output_spec import OutputSpec
from artisan.utils import tutorial_setup
from artisan.visualization import (
    build_macro_graph,
    build_micro_graph,
    inspect_pipeline,
)

Anatomy of a CompositeDefinition¶

Every composite has five parts:

Part	Purpose
`name` / `description`	Metadata for registry and docs
`InputRole` / `OutputRole`	StrEnum classes naming external I/O
`inputs` / `outputs`	ClassVar dicts mapping roles to `InputSpec` / `OutputSpec`
`Params` (optional)	Pydantic `BaseModel` for composite-level configuration
`compose()`	Wiring method that calls `ctx.run()` to execute internal operations

The framework validates at class definition time that:

InputRole values match inputs keys
OutputRole values match outputs keys
compose() is implemented

Building `GenerateAndFilter`¶

We’ll build a composite that generates datasets, scores them, and filters to keep only high-quality ones. This is a common pattern: produce candidates, evaluate quality, keep the best.

DataGenerator ──→ MetricCalculator ──→ Filter
  (N datasets)      (N metrics)       (top datasets)

Start with the class skeleton — metadata and role enums:

class GenerateAndFilter(CompositeDefinition):
    """Generate datasets, score them, and filter by quality."""

    name = "generate_and_filter"
    description = "Generate, score, and quality-filter datasets."

    # External I/O roles
    class OutputRole(StrEnum):
        DATASETS = "datasets"

    # No InputRole — this is a generative composite (no external inputs)

    outputs: ClassVar[dict[str, OutputSpec]] = {
        OutputRole.DATASETS: OutputSpec(
            artifact_type="data",
            description="Quality-filtered datasets",
        ),
    }

    # Composite-level parameters
    class Params(BaseModel):
        count: int = Field(default=6, description="Datasets to generate")
        threshold: float = Field(
            default=0.5, description="Minimum median score to keep"
        )
        seed: int = Field(default=42, description="Random seed")

    params: Params = Params()

    def compose(self, ctx: CompositeContext) -> None:
        # Step 1: Generate candidate datasets
        generated = ctx.run(
            DataGenerator,
            params={"count": self.params.count, "seed": self.params.seed},
        )

        # Step 2: Score each dataset (creates provenance edges for Filter)
        ctx.run(
            MetricCalculator,
            inputs={"dataset": generated.output("datasets")},
        )

        # Step 3: Filter — keep datasets above threshold
        filtered = ctx.run(
            Filter,
            inputs={"passthrough": generated.output("datasets")},
            params={
                "criteria": [
                    {
                        "metric": "distribution.median",
                        "operator": "ge",
                        "value": self.params.threshold,
                    }
                ],
            },
        )

        # Map internal result to declared output
        ctx.output("datasets", filtered.output("passthrough"))

The composite has no inputs (it’s generative — DataGenerator creates data from nothing). The compose() method wires three operations:

ctx.run() executes an operation and returns a CompositeStepHandle
handle.output(role) returns a CompositeRef that wires to the next operation’s input
ctx.output(role, ref) maps an internal result to a declared composite output

Notice that MetricCalculator’s return value isn’t captured — its handle isn’t wired to any downstream input. It only needs to run so that provenance edges exist between datasets and their metrics. Filter discovers these metrics automatically by walking forward through provenance from its passthrough artifacts. It flattens nested metric keys with dot separators (e.g., distribution.median, summary.cv), which is why the criterion uses distribution.median rather than a top-level field name.

Run collapsed¶

In collapsed mode, all three internal operations execute as a single pipeline step. Artifacts pass in-memory between operations.

env_collapsed = tutorial_setup("composite_collapsed")

pipeline = PipelineManager.create(
    name="collapsed",
    delta_root=env_collapsed.delta_root,
    staging_root=env_collapsed.staging_root,
    working_root=env_collapsed.working_root,
)

# Run the composite as a single step
pipeline.run(
    operation=GenerateAndFilter,
    params={"count": 6, "threshold": 0.5, "seed": 42},
)

result_collapsed = pipeline.finalize()

inspect_pipeline(env_collapsed.delta_root)

build_macro_graph(env_collapsed.delta_root)

The pipeline shows a single step that produced the filtered datasets. The generation, scoring, and filtering all happened inside one worker with zero Delta Lake round-trips between internal operations.

build_micro_graph(env_collapsed.delta_root)

Run expanded¶

In expanded mode, each ctx.run() becomes its own pipeline step with independent caching and worker dispatch.

env_expanded = tutorial_setup("composite_expanded")

pipeline = PipelineManager.create(
    name="expanded",
    delta_root=env_expanded.delta_root,
    staging_root=env_expanded.staging_root,
    working_root=env_expanded.working_root,
)

# Expand the composite into separate steps
expanded = pipeline.expand(
    GenerateAndFilter,
    params={"count": 6, "threshold": 0.5, "seed": 42},
)

result_expanded = pipeline.finalize()

inspect_pipeline(env_expanded.delta_root)

build_macro_graph(env_expanded.delta_root)

Three steps instead of one — each prefixed with the composite name. The same GenerateAndFilter definition drives both modes. Choose collapsed when the operations always run together, expanded when you want per-operation caching or different resource allocations.

build_micro_graph(env_expanded.delta_root)

Summary¶

This tutorial built a CompositeDefinition from scratch and ran it in both execution modes.

The five parts of a composite:

Metadata — name and description for the registry
Roles — InputRole/OutputRole StrEnums naming external I/O
Specs — inputs/outputs ClassVar dicts with InputSpec/OutputSpec
Params — optional Pydantic BaseModel for configuration
compose() — wiring method using ctx.run() and ctx.output() (plus ctx.input() for composites that accept external inputs)

Next steps¶

Composites and Composition — design rationale for collapsed vs expanded execution
Writing Composite Operations — recipe-style patterns including multi-input, nesting, and curators
CompositeDefinition Reference — full API signatures and field tables