13. Staged pipelines

Runnable companion: examples/10_nested_pipeline_compare.py

The last chapter. Nothing new is introduced — this is how the previous eleven compose into a real multi-stage pipeline: an upstream model labels the data, a downstream model trains on those labels and emits an intermediate signal, and two final decoding options are scored against ground truth.

base dataset ──▶ model 1 ──attach teacher.cls──▶ base store
                                                     │
                            workset pins teacher.cls = model1
                                                     ▼
                       model 2 (trains on a workset) ──attach score.cls──▶ base store
                                                     │
                       add(score.cls, label.cls) into its own store
                                                     ▼
                                              final store
                                    ┌────────────────┴────────────────┐
                              optA ─attach decision.cls          optB ─attach decision.cls
                                    └──── accuracy ── compare ────┘

Chaining producers

Each hop is chapter 6 with chapter 9's pin in front of it:

# model 2's training workset consumes model 1's output explicitly
workset = client.create_workset(
    sources=[SourceSpec(sample_store="base", keys=["audio.wav", "teacher.cls"],
                        producer_pins={"teacher.cls": "model1"})],
    key_groups=[KeyGroupSpec(name="inputs", keys=["audio.wav"]),
                KeyGroupSpec(name="teacher", keys=["teacher.cls"])],
    shard_plan=ShardPlan(shard_size=64),
)
# … compute, then attach score.cls under producer "model2"

The pin is what makes the chain explicit: stage 2 consumes exactly the upstream producer it was built against, rather than whichever revision resolves.

Choosing a workset or a dataset per stage

This is the judgement call the chapter exists to demonstrate, and it is per stage rather than per pipeline:

Stage	Source	Why
Base model training	dataset	the run you will reproduce, publish and compare against
Follow-up stages	workset	cheap, live, iterated often; the exact membership is not the artifact
Final scoring	dataset	numbers you will quote need a frozen member set

Training on a workset is a legitimate choice rather than a shortcut: for an exploratory or intermediate stage, a frozen dataset buys nothing. Freeze exactly where reproducibility is a requirement.

Moving an intermediate into its own store

When a stage's output becomes the substrate for the next fan-out, add it across (chapter 5). A workset is a first-class add source:

client.create_store(name="final")
client.add_to_store(
    "final",
    operation_id="seed-from-stage2",
    from_workset=stage2_workset.workset_id,     # or from_store / from_dataset
    key_subset=["score.cls", "label.cls"],
)
client.wait_for_add("final", "seed-from-stage2", timeout_seconds=300)

In the new store, score.cls is an ordinary key on ordinary samples — no pin needed to read it, because the producer identity was resolved when the add was planned. That is what makes it a clean substrate for the next stage.

Fanning out options

Two decoding options are two producers writing the same key in one store — chapter 9's shape:

for option in ("optA", "optB"):
    ...  # attach decision.cls under producer `option`

# compare: one workset per option, each pinning its producer, scored vs the label

Then score each with a pinned workset and compare aggregates (chapter 12). An unpinned read cannot separate them: two producers, one key name.

Assembling this for real

One thing lives outside this guide because it is about where your code runs rather than about Datavo's model:

• Producing and training on one machine — skipping the round trip via local shard IO: local shards.

And the recurring rule for a pipeline that will run more than once: every stage is idempotent by construction if you let it be. An attach workset drains itself, an import and an add replay on operation_id. That is the two mechanisms — convergence and replay — working together: pass stable ids and a re-run costs only the work that is genuinely missing. See idempotency.

Pitfalls

• A shared producer id across stages. Two stages under one id cannot be compared, pinned apart, or removed separately. One producer per stage per configuration.
• Changing what a producer computes. A producer's output keys are recorded from its attaches, so adding a key under the same id is fine — both are recorded, and coverage spans both. But new values for a key you already wrote are not: convergence skips samples that producer has covered. Changed weights or settings mean a new producer id.
• Chaining without pins. Stage 2 then reads whichever revision resolves, which is a silent correctness bug the day a third producer appears.
• Freezing every stage. Datasets are permanent and named; converting at every hop accumulates artifacts that nobody owns.

---

That is the complete model. From here: the reference for CLI, caching and integration, and the internals for how it works underneath.