Patterns

The shapes real pipelines take once they leave the notebook, which pipekit pieces each one uses, and what is still missing.

About this page

Covers	Fourteen deployment patterns and four habits that cut across them
Read first	A tour of the core
Worked examples	The tutorials, linked from each pattern
History	Distilled from Report 9 of the GeoStack design reports, which walks twenty cases through one domain (a methane plume from a satellite scene) and now points here

A pipeline is written once and then used in many places: a notebook, a batch job, a web service, a scheduled family of runs, an audit. pipekit's bet is that these are the same object run by different runners, so the patterns below are mostly about which runner to put around an Operator, and which of its properties (a config, a hash, pickleability) the runner relies on.

Each pattern names its pipekit pieces, says what works today and what does not, and links the tutorial that runs it end to end. The status column uses three marks: today (ships and is tested), partly (works with a workaround), missing (needs code that does not exist yet).

#	Pattern	Pieces	Status	Tutorial
1	Explore, then batch	Tap, Log, ShapeTrace, Profile, Snapshot	today	bigger graph
2	ETL over many inputs	Map, ProcessMap, Quarantine, Sink	today	ETL over many regions
3	Validation as operators	Assert* (AssertThat, AssertSignature, …), Quarantine, QCError	today	inference
4	Fusing several inputs	Graph, Input	today	bigger graph
5	One preprocessing object for training and serving	Operator config, registry	today	train, register, forecast
6	Cycles: time stepping and assimilation	StatefulOperator, pipekit-cycle	today	DA cycle, ensemble DA
7	An inference service	AsyncMap.run, BatchedMap, the config hash	today	inference
8	Promotion by tag	LocalModelRegistry.tag	today	none
9	Pipelines from users	register, loads_sandboxed, from_yaml_sandboxed	today	none
10	A pinned artifact	InferenceArtifact, TrainingArtifact, to_yaml	today	none
11	Orchestrator tasks with deterministic keys	pipeline_hash, Cache in a DirectoryStore	today	ETL over many regions
12	A/B comparison	Fanout	partly	none
13	Scheduled families, backfills and what-if reruns	the config hash	today	none
14	The feedback loop	pipekit-train, registry	today	train, register, forecast

The snippets share one small operator, a detector with a threshold:

from pipekit_ops import Operator


class Threshold(Operator):
    """x ↦ 1[x > k]."""

    def __init__(self, *, k: float) -> None:
        self.k = k

    def _apply(self, x):
        return x > self.k

Exploration and batch

1. Explore, then batch

The problem. A pipeline grows in a notebook, one step at a time, with prints and plots between the steps. Then it has to run on a thousand inputs, and the prints and plots have to go.

The pattern. Observers pass their input through unchanged, so you can leave them in while you explore and drop them, or point them at a logger, when you batch. ShapeTrace logs each step's shape and dtype, Profile times a step, Snapshot keeps a copy of an intermediate, and Tap calls any function on it. Log writes one record per call to a logging logger with no closure, so it can stay in the batch pipeline and in its config. The object you explored with is the object you batch.

from pipekit_ops import Profile, ShapeTrace, Snapshot, Tap

# today (pipekit-ops); clean, features, model and plot are user
snap, prof = Snapshot(), Profile()
explore = (
    clean
    | ShapeTrace()
    | snap.at("clean")  # log shape and dtype, keep a copy
    | features
    | Tap(plot)  # look at the features
    | prof.wrap(model)  # time the model
)
batch = clean | features | model  # the same steps, no observers
snap.captures["clean"], prof.report()  # what the exploration saw

Today. All of it. ShapeTrace(extra=...) adds a line for whatever else a carrier carries (units, a coordinate system).

2. ETL over many inputs

The problem. One pipeline, many inputs (regions, files, stations, days), some of them broken in ways you only find at scale.

The pattern. Extract with a list of keys, and transform with Map(pipeline, executor="process", on_error="log_and_continue") so one bad input leaves a None instead of killing the run. Set malformed records aside with Quarantine, and load with a Sink that writes one keyed output per input. Run check_pickleable(pipeline) before fanning out: closures do not cross process boundaries.

Today. All of it. log_and_continue reports each failure as a warning, with its traceback, on the pipekit_ops.parallel logger, so the run's logging setup decides where failures go. on_error="collect" instead leaves a Failed(index, exception) in each failed slot and raises one MapError at the end, holding every failure and the partial results. Map reads its input lazily and keeps at most max_in_flight items in flight, so a generator over a million keys does not land in memory.

Map(executor="process") starts its workers with spawn. ProcessMap(pipeline, on_error=...) is the same map as a preset, but it keeps the platform's default start method and raises a FutureWarning when that is not spawn, as on Linux (fork); the preset adopts spawn in the next minor release. The executor can also be chosen at deployment time: Map(pipeline, executor="scope") runs on whatever executor_scope(...) binds around the call.

3. Validation as operators

The problem. Checks scattered through glue code are skipped, duplicated, or run in the wrong place.

The pattern. A check is a step. AssertShape, AssertDType, AssertHasAttribute, AssertCallable, AssertThat and AssertSignature raise QCError when an input is wrong. Quarantine(check, sentinel, on_quarantine) replaces a bad input with a sentinel and hands it to a callback, for streams where one bad record should not stop the rest. Put the hard checks at the boundaries (after loading, before the model, before writing) and the soft ones on the records.

Today. All of it. AssertThat(pred, message) takes the predicate as an operator, so the check stays in the pipeline's config and hash; AssertCallable takes a plain function and is flagged forbid_in_yaml. AssertSignature("... y x", dtype=...) checks named dims and dtype, at build time through compute_output_signature as well as at run time. Array-aware checks (finite values, value ranges, across array libraries) are planned for pipekit-array; until then AssertThat or AssertCallable takes any predicate.

4. Fusing several inputs

The problem. A product needs two or more sources at once: two sensors, a measurement and a reference, a feed and its timestamp.

The pattern. A Graph. Each source is an Input, each step is called on the nodes it needs, and the outputs are named. Steps that take two carriers (_apply(self, a, b)) are ordinary operators.

from pipekit_ops import Graph, Input

a, b = Input("a"), Input("b")
fused = Align()(a, b)
g = Graph(inputs={"a": a, "b": b}, outputs={"product": Model()(fused)})
g(a=scene_a, b=scene_b)["product"]

Today. All of it, stateful steps included: call the graph with state= and it returns (outputs, state). Shape inference stops at the graph's edge. Get, Pick, Merge and Rename reshape the dicts a graph returns without a Lambda.

Machine learning

5. One preprocessing object for training and serving

The problem. Training normalises with one piece of code and serving with another, and they drift: the classic train/serve skew.

The pattern. The fitted preprocessing is an operator whose fitted values (a mean, a standard deviation, a vocabulary) are its configuration. Store it in the registry next to the model, with its hash in the model's tags, and load both by name at serving time. There is then exactly one Standardize(mean=…, std=…) in the world, and the hash proves which one ran.

scaler_hash = registry.store(scaler, name="scaler")
registry.store(model, name="model", weights=blob, tags={"scaler": scaler_hash})

serve = registry.load("scaler") | model_template.with_weights(
    registry.load_weights("model")
)

Today. All of it (LocalModelRegistry, S3ModelRegistry, JaxModelOp.with_weights). The tutorial fits the scaler on the training split only, so validation never sees its statistics.

6. Cycles: time stepping and assimilation

The problem. Forecast, observe, correct, repeat: the carrier is a state, and every step needs the previous one.

The pattern. pipekit-cycle. A Cycle steps a forward model; a DACycle or EnsembleDACycle alternates forecast and analysis, with observation operators and noise models as Protocols, so the filter can come from any library. Pre-processing (quality control, thinning, unit conversion of the observations) and post-processing (verification, spread-skill) are ordinary operators around the cycle.

Today. All of it.

Serving

7. An inference service

The problem. A web handler receives a request, runs a pipeline and must say exactly what produced its answer.

The pattern. Build the pipeline once at start-up. Inside the handler, which already runs in an event loop, use await AsyncMap(op).run(items) (not AsyncMap(op)(items), which starts its own loop). Batch for a GPU with BatchedMap, or let the model batch. Return the pipeline's content hash with every answer, so a response can be traced to a configuration.

Today. All of it. Timeout(op, seconds) bounds one slow call and RateLimit(op, calls, period) bounds how often an upstream service is called. pipekit provides no server: LitServe, FastAPI or anything else hosts the operator.

8. Promotion by tag

The problem. Several candidate versions exist; one of them is "the" model that serving uses, and promoting a new one (or rolling back) must not need a redeploy.

The pattern. Store every candidate by content hash; point a human-readable tag at the one in service. Serving loads by tag; promotion and rollback move the tag. The same configuration always gets the same hash, so storing it twice is harmless.

import tempfile
from pipekit_experiment import LocalModelRegistry

registry = LocalModelRegistry(tempfile.mkdtemp())
v1 = registry.store(Threshold(k=3.0), name="detector-candidate")
v2 = registry.store(Threshold(k=2.5), name="detector-candidate")

registry.tag(v1, "detector", force=True)  # v1 in service
registry.tag(v2, "detector", force=True)  # promote v2
registry.load("detector")  # → Threshold(k=2.5)
registry.tag(v1, "detector", force=True)  # roll back
registry.store(Threshold(k=3.0)) == v1  # → True: same config, same hash

Today. All of it.

9. Pipelines from users

The problem. Users (or another team, or a queue) send pipelines as data. Building arbitrary classes from a string is code execution.

The pattern. Register the classes you allow, containers included, and build untrusted input with loads_sandboxed (JSON) or HydraConfigLoader.from_yaml_sandboxed (_target_ YAML). Every class in the tree is checked before anything is built, and anything unregistered raises PermissionError. Operators holding closures are forbid_in_yaml and never deserialise, registered or not.

import pipekit_ops as pk
from pipekit_experiment.adapters.hydra import HydraConfigLoader

pk.register(pk.Sequential)
pk.register(Threshold)

text = """
_target_: pipekit_ops.Sequential
operators:
  - _target_: __main__.Threshold
    k: 2.0
"""
# a Sequential holding Threshold(k=2.0)
detector = HydraConfigLoader.from_yaml_sandboxed(text)

The YAML loader never calls Hydra's instantiate, which would call any dotted path (_target_: os.system): it translates the YAML into a state record and builds that through the allowlist. Interpolations (${...}) are not resolved, so a sandboxed config cannot read the environment.

Today. All of it.

Reproducibility and operations

10. A pinned artifact

The problem. A result has to be reproducible months later, for a review, an audit or a paper: the exact pipeline, the exact weights, the exact dependencies.

The pattern. pipekit-experiment's InferenceArtifact(pipeline_yaml, pinned_model_hashes, backend_info, deps_lock, metadata) records what ran; TrainingArtifact records how the model was made (the training pipeline, the dataset's hash, the model's hash, the tracker run). Keep the artifact with the output.

For the pipeline_yaml field, HydraConfigLoader.to_yaml(pipeline) writes the pipeline that actually ran, however deeply nested, and from_yaml loads it back with the same state.

Today. All of it. TrainingLoop.run() returns a TrainingArtifact when pipekit-experiment is installed.

11. Orchestrator tasks with deterministic keys

The problem. An orchestrator (Prefect, Dagster, Airflow, a cron job) retries, resumes and backfills. Every task must know whether it has already run.

The pattern. One task is one operator call on one input. Its key is a hash of the operator's configuration and the input's identity, so a rerun with the same pipeline and input finds its output and skips; a change to either gives a new key.

from pipekit_ops import Operator, pipeline_hash


def task_key(op: Operator, input_id: str) -> str:
    """One key per (pipeline configuration, input)."""
    return f"{pipeline_hash(op)[:16]}-{input_id}"


# the same pipeline on the same day: the same key, so the rerun skips
task_key(Threshold(k=3.0), "2026-10-16") == task_key(Threshold(k=3.0), "2026-10-16")
# one argument changed: a new key
task_key(Threshold(k=3.0), "2026-10-16") != task_key(Threshold(k=2.5), "2026-10-16")

Today. All of it. pipeline_hash hashes the whole configuration: every nested operator and, in a Graph, every node and its wiring. It is the hash the registry stores under, so a task key and a registry entry agree.

Inside one task, Cache does the same bookkeeping per input. Its key is a hash of the input's key and pipeline_hash of the cached operator, and key=Get("id") makes the input's key one field of it. With a DirectoryStore bound by name, the entries survive a restart and are shared by the processes on one file system:

import tempfile

from pipekit_ops import Cache, DirectoryStore, Get, Map, store_scope

# today (pipekit-ops)
# item ↦ 1[item["value"] > 3], cached under item["id"]
detect = Cache(Get("value") | Threshold(k=3.0), store="results", key=Get("id"))
# today (pipekit-ops)
results = DirectoryStore(tempfile.mkdtemp())
items = [{"id": "2026-10-16", "value": 4.0}, {"id": "2026-10-17", "value": 1.0}]
with store_scope(results=results):
    # a rerun after a crash finds both entries on disk and skips the work
    flags = Map(detect)(items)  # → [True, False]

12. A/B comparison

The problem. A new pipeline version must be compared with the current one on the same inputs before it replaces it.

The pattern. Fanout({"a": current, "b": candidate}) runs both on each input and returns both outputs; a post-processing step compares them. Compare the distribution of the differences, not only their mean: two versions can agree on average and disagree badly on the inputs that matter.

from pipekit_ops import Fanout

ab = Fanout({"a": Threshold(k=3.0), "b": Threshold(k=2.5)})
ab(2.8)  # → {'a': False, 'b': True}

Partly. The side-by-side run ships; a diff of two pipelines' configurations, and a report of where their outputs disagree, do not. pipekit-evaluate's scorers and EvaluationReport are the natural home.

13. Scheduled families, backfills and what-if reruns

The problem. The same pipeline runs every day (monitoring), over a past period (backfill), and again with one setting changed (what-if).

The pattern. Key every output by the pipeline's hash and the input's date or identity (pattern 11). A daily run and a backfill are the same task over different dates. A what-if is a new operator with one constructor argument changed, so it has a new hash and writes beside the original, never over it.

Today. All of it; the schedule belongs to the orchestrator.

14. The feedback loop

The problem. Reviewers accept or reject the pipeline's outputs, and those verdicts should improve the next model.

The pattern. Verdicts become labels; a frozen label set gets a content hash; pipekit-train's TrainingLoop trains on it and returns a TrainingArtifact; the new model is stored, compared with the current one (pattern 12) and promoted by tag (pattern 8). Every model can be traced back to the labels it saw.

Today. All of it, given a label store that hashes its frozen sets.

Habits that cut across the patterns

Composition in the core, infrastructure next door

The core composes and nothing else; deployment infrastructure lives in the sister packages, each in one place.

Concern	Lives in
Operators, composition, parallel maps, flow limits, nested JSON, the sandbox allowlist	pipekit-ops
pipeline_hash and the hashing primitives	pipekit_ops.hashing
Where a Cache keeps its entries (MemoryStore, DirectoryStore)	pipekit_ops.store
Content-addressed storage, tags, artifacts, YAML loading	pipekit-experiment
Training loops and backends	pipekit-train
Time stepping and assimilation	pipekit-cycle
Benchmarks, scorers and reports	pipekit-evaluate
JAX models with weight round-trips	pipekit-jax
Domain operators, readers and writers	your library, subclassing Operator

Parallelism is chosen per deployment

Nothing is parallel by default; each deployment picks its primitive. All of them are one engine, Map(op, executor=...), and the presets are Map with fixed arguments and their own config.

Deployment	Primitive
CPU-bound batch over many inputs	Map(op, executor="process"), or the ProcessMap preset
I/O-bound reads	Map(op, executor="thread"), or ThreadMap
I/O inside a running event loop (a server)	await AsyncMap(op).run(items)
GPU inference	BatchedMap, or the model batches
Chosen at deployment, not in the pipeline	Map(op, executor="scope") under executor_scope(...)
Across machines	the orchestrator, calling pickleable operators

YAML is the shared language, both ways

Every operator, in any library, subclasses Operator, so a Hydra YAML with _target_ paths can name steps from several libraries side by side. HydraConfigLoader.from_yaml builds the nested pipeline, to_yaml writes one built in Python, and from_yaml_sandboxed loads YAML you did not write. A pipeline can start in Python or in YAML; either way the other form is one call away.

JAX speeds the model, not the pipeline

pipekit-jax wraps an Equinox model as an operator (JaxModelOp), with byte-identical weight round-trips through the registry, and adds diffrax forward models for pipekit-cycle. Inside the model, jit and vmap do their work; the pipeline around it stays eager Python. That is a deliberate split: pre- and post-processing are I/O and bookkeeping, where tracing buys little and costs flexibility. When the numerical core is itself a pipeline, importing pipekit_jax registers Sequential and Graph as pytrees, so a pipeline of JaxOperators traces under jit as a whole (pipekit under jit); Map refuses JAX tracers on its non-serial executors.

One domain, all the patterns

The GeoStack design reports walk every pattern above through one example, a methane plume from a satellite scene: read the scene, retrieve the enhancement, detect the plume, estimate its emission rate, review it, and retrain. Report 9 has the twenty cases, the libraries each one needs, and the proposals that close the gaps marked partly and missing here.