Skip to content

pipekit-experiment

The boundary between pipekit pipelines and external orchestration tools — runtime-checkable protocols, a content-addressed model registry, JSON-roundtrippable artifacts, and thin per-tool adapters. Master plan reference: Report 12.

Protocols

pipekit_experiment.protocols

Two protocols — ExperimentTracker and ModelRegistry.

The seam between pipekit pipelines and external orchestration tools. Adapters (MLflow, W&B, DVC, Hydra, Metaflow) satisfy these protocols structurally — no inheritance from pipekit-experiment is required.

Both protocols are runtime-checkable.

See master plan Report 12, section 2.2.

ExperimentTracker

Bases: Protocol

Log metrics, parameters, and artifacts from a training run.

Implementations wrap external trackers — MLflow, W&B, ClearML, Neptune, ad-hoc CSV writers. Pipekit pipelines call these methods; the adapter translates to the tool's API.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@runtime_checkable
class ExperimentTracker(Protocol):
    """Log metrics, parameters, and artifacts from a training run.

    Implementations wrap external trackers — MLflow, W&B, ClearML,
    Neptune, ad-hoc CSV writers. Pipekit pipelines call these methods;
    the adapter translates to the tool's API.
    """

    def start_run(self, name: str, config: dict[str, Any]) -> Run:
        """Open a run named ``name`` with hyperparameters ``config``."""
        ...

    def log_metrics(self, run: Run, metrics: dict[str, float], step: int) -> None:
        """Record scalar ``metrics`` for ``run`` at global ``step``."""
        ...

    def log_artifact(self, run: Run, path: str, name: str) -> None:
        """Attach the file at ``path`` to ``run`` under ``name``."""
        ...

    def end_run(self, run: Run, status: str) -> None:
        """Close ``run`` with a final ``status`` (e.g. ``"ok"``/``"failed"``)."""
        ...

end_run(run, status)

Close run with a final status (e.g. "ok"/"failed").

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
45
46
47
def end_run(self, run: Run, status: str) -> None:
    """Close ``run`` with a final ``status`` (e.g. ``"ok"``/``"failed"``)."""
    ...

log_artifact(run, path, name)

Attach the file at path to run under name.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
41
42
43
def log_artifact(self, run: Run, path: str, name: str) -> None:
    """Attach the file at ``path`` to ``run`` under ``name``."""
    ...

log_metrics(run, metrics, step)

Record scalar metrics for run at global step.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
37
38
39
def log_metrics(self, run: Run, metrics: dict[str, float], step: int) -> None:
    """Record scalar ``metrics`` for ``run`` at global ``step``."""
    ...

start_run(name, config)

Open a run named name with hyperparameters config.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
33
34
35
def start_run(self, name: str, config: dict[str, Any]) -> Run:
    """Open a run named ``name`` with hyperparameters ``config``."""
    ...

ModelRegistry

Bases: Protocol

Store and retrieve trained models by content hash.

Implementations: filesystem-backed, S3-backed, MLflow-registry-backed. The canonical identifier is the content hash; name is a mutable tag that resolves to a hash.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@runtime_checkable
class ModelRegistry(Protocol):
    """Store and retrieve trained models by content hash.

    Implementations: filesystem-backed, S3-backed, MLflow-registry-backed.
    The canonical identifier is the content hash; ``name`` is a
    mutable tag that resolves to a hash.
    """

    def store(
        self,
        model_op: Operator,
        *,
        name: str | None = None,
        tags: dict[str, Any] | None = None,
        weights: bytes | None = None,
    ) -> str:
        """Persist ``model_op`` (and optional weights). Return the hash."""

    def load(self, ref: str) -> Operator:
        """Resolve ``ref`` (hash or name) and return the rebuilt operator."""

    def load_weights(self, ref: str) -> bytes | None:
        """Return the raw weight bytes stored for ``ref``, or ``None``."""

    def list(self, *, tags: dict[str, Any] | None = None) -> builtins.list[str]:
        """Return content hashes, optionally filtered by tag values."""

    def tag(self, hash: str, name: str, *, force: bool = False) -> None:
        """Bind ``name`` to ``hash``. ``force=True`` allows re-tagging."""

list(*, tags=None)

Return content hashes, optionally filtered by tag values.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
75
76
def list(self, *, tags: dict[str, Any] | None = None) -> builtins.list[str]:
    """Return content hashes, optionally filtered by tag values."""

load(ref)

Resolve ref (hash or name) and return the rebuilt operator.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
69
70
def load(self, ref: str) -> Operator:
    """Resolve ``ref`` (hash or name) and return the rebuilt operator."""

load_weights(ref)

Return the raw weight bytes stored for ref, or None.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
72
73
def load_weights(self, ref: str) -> bytes | None:
    """Return the raw weight bytes stored for ``ref``, or ``None``."""

store(model_op, *, name=None, tags=None, weights=None)

Persist model_op (and optional weights). Return the hash.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
59
60
61
62
63
64
65
66
67
def store(
    self,
    model_op: Operator,
    *,
    name: str | None = None,
    tags: dict[str, Any] | None = None,
    weights: bytes | None = None,
) -> str:
    """Persist ``model_op`` (and optional weights). Return the hash."""

tag(hash, name, *, force=False)

Bind name to hash. force=True allows re-tagging.

Source code in packages/pipekit-experiment/src/pipekit_experiment/protocols.py 
78
79
def tag(self, hash: str, name: str, *, force: bool = False) -> None:
    """Bind ``name`` to ``hash``. ``force=True`` allows re-tagging."""

Run records

pipekit_experiment.run

Run records, per-step metrics, and named artifacts.

  • Run — a single pipeline execution; identified by run_id and pipeline_hash.
  • RunMetrics — per-step metric series.
  • RunArtifacts — named artifact URIs attached to a run.

See master plan Report 12, section 2.3.

Run dataclass

A single execution of a pipeline (training or inference).

Attributes:

Name Type Description
run_id str

Tracker-assigned identifier (or a uuid for local runs).
pipeline_hash str

Pipekit-derived content hash of the pipeline.
config dict[str, Any]

Flat dict of pipeline parameters logged at start.
metrics dict[str, float]

Latest metric values (the running snapshot; full per-step history lives in RunMetrics).
artifacts dict[str, str]

name → URI map.
started_at datetime

UTC timestamp.
ended_at datetime | None

UTC timestamp; None until end_run.
status RunStatus

One of "running", "completed", "failed".
Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@dataclass
class Run:
    """A single execution of a pipeline (training or inference).

    Attributes:
        run_id: Tracker-assigned identifier (or a uuid for local runs).
        pipeline_hash: Pipekit-derived content hash of the pipeline.
        config: Flat dict of pipeline parameters logged at start.
        metrics: Latest metric values (the running snapshot;
            full per-step history lives in `RunMetrics`).
        artifacts: ``name → URI`` map.
        started_at: UTC timestamp.
        ended_at: UTC timestamp; ``None`` until `end_run`.
        status: One of ``"running"``, ``"completed"``, ``"failed"``.
    """

    run_id: str
    pipeline_hash: str
    config: dict[str, Any] = field(default_factory=dict)
    metrics: dict[str, float] = field(default_factory=dict)
    artifacts: dict[str, str] = field(default_factory=dict)
    started_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    ended_at: datetime | None = None
    status: RunStatus = "running"

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-friendly dict snapshot (datetimes → isoformat)."""
        return {
            "run_id": self.run_id,
            "pipeline_hash": self.pipeline_hash,
            "config": dict(self.config),
            "metrics": dict(self.metrics),
            "artifacts": dict(self.artifacts),
            "started_at": self.started_at.isoformat(),
            "ended_at": self.ended_at.isoformat() if self.ended_at else None,
            "status": self.status,
        }

to_dict()

Return a JSON-friendly dict snapshot (datetimes → isoformat).

Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
46
47
48
49
50
51
52
53
54
55
56
57
def to_dict(self) -> dict[str, Any]:
    """Return a JSON-friendly dict snapshot (datetimes → isoformat)."""
    return {
        "run_id": self.run_id,
        "pipeline_hash": self.pipeline_hash,
        "config": dict(self.config),
        "metrics": dict(self.metrics),
        "artifacts": dict(self.artifacts),
        "started_at": self.started_at.isoformat(),
        "ended_at": self.ended_at.isoformat() if self.ended_at else None,
        "status": self.status,
    }

RunArtifacts

Named artifacts (files, models, plots) attached to a Run.

Attributes:

Name Type Description
items dict[str, str]

name → URI. URI is whatever the storage backend understands — a local path, an s3:// URI, etc.
Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class RunArtifacts:
    """Named artifacts (files, models, plots) attached to a `Run`.

    Attributes:
        items: ``name → URI``. URI is whatever the storage backend
            understands — a local path, an ``s3://`` URI, etc.
    """

    def __init__(self) -> None:
        self.items: dict[str, str] = {}

    def attach(self, name: str, uri: str) -> None:
        """Record ``uri`` under ``name``. Re-attaching overwrites."""
        self.items[name] = uri

    def get(self, name: str) -> str | None:
        """Return the artifact stored under ``name``, or ``None``."""
        return self.items.get(name)

    def clear(self) -> None:
        """Drop all recorded entries."""
        self.items.clear()

attach(name, uri)

Record uri under name. Re-attaching overwrites.

Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
94
95
96
def attach(self, name: str, uri: str) -> None:
    """Record ``uri`` under ``name``. Re-attaching overwrites."""
    self.items[name] = uri

clear()

Drop all recorded entries.

Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
102
103
104
def clear(self) -> None:
    """Drop all recorded entries."""
    self.items.clear()

get(name)

Return the artifact stored under name, or None.

Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
 98
 99
100
def get(self, name: str) -> str | None:
    """Return the artifact stored under ``name``, or ``None``."""
    return self.items.get(name)

RunMetrics

Per-step metric series.

Attributes:

Name Type Description
values dict[str, list[tuple[int, float]]]

metric_name → [(step, value), …].
Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class RunMetrics:
    """Per-step metric series.

    Attributes:
        values: ``metric_name → [(step, value), …]``.
    """

    def __init__(self) -> None:
        self.values: dict[str, list[tuple[int, float]]] = {}

    def log(self, name: str, value: float, step: int) -> None:
        """Append ``(step, value)`` to the ``name`` series."""
        self.values.setdefault(name, []).append((int(step), float(value)))

    def latest(self) -> dict[str, float]:
        """Return ``{name: latest_value}`` across all series."""
        return {n: pairs[-1][1] for n, pairs in self.values.items() if pairs}

    def clear(self) -> None:
        """Drop all recorded entries."""
        self.values.clear()

clear()

Drop all recorded entries.

Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
78
79
80
def clear(self) -> None:
    """Drop all recorded entries."""
    self.values.clear()

latest()

Return {name: latest_value} across all series.

Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
74
75
76
def latest(self) -> dict[str, float]:
    """Return ``{name: latest_value}`` across all series."""
    return {n: pairs[-1][1] for n, pairs in self.values.items() if pairs}

log(name, value, step)

Append (step, value) to the name series.

Source code in packages/pipekit-experiment/src/pipekit_experiment/run.py 
70
71
72
def log(self, name: str, value: float, step: int) -> None:
    """Append ``(step, value)`` to the ``name`` series."""
    self.values.setdefault(name, []).append((int(step), float(value)))

Model registry

pipekit_experiment.registry

Content-addressed model registries.

A trained model is identified by hash(operator_config, weights); loading by hash is the canonical path. Names ("methane_emulator_v3") are tags that resolve to hashes.

  • LocalModelRegistry — local-filesystem backend. Useful for dev, CI, and as the reference implementation.
  • S3ModelRegistry — fsspec-backed (works for s3, gcs, az, file …). Behind the [s3] extra; raises a clean ImportError if fsspec isn't installed.

Layout under root/ (or s3://bucket/prefix/):

<hash>/operator.json   # pipekit operator state (round-trip via Operator.from_state)
<hash>/weights.bin     # optional raw weight bytes
<hash>/metadata.json   # tags, training run id, …
_tags/<name>           # one-line file holding the target hash

The weights.bin blob is opaque to the registry — domain operators round-trip their own weights via a separate channel (pipekit_jax.JaxModelOp.with_weights etc.). The registry just persists / retrieves the bytes alongside the operator config.

Both backends share their control flow via _ModelRegistryBase; the subclasses only supply the storage primitives (local Path I/O vs fsspec). References (hashes) and tag names are restricted to single path segments so a caller-supplied string can never escape the registry root.

Loading verifies content against its address: load and load_weights recompute the hash of what is on disk and raise HashMismatchError if it differs from the requested hash, and load also checks that the rebuilt operator hashes back to it (pipeline_hash(load(h), load_weights(h)) == h).

See master plan Report 12, section 2.4 and section 5.

HashMismatchError

Bases: ValueError

Stored content does not match the hash it is stored under.

Raised on load when the bytes on disk (operator.json, weights.bin) were modified after store, or when the operator rebuilt from them hashes to something else. Subclass of ValueError.

Source code in packages/pipekit-experiment/src/pipekit_experiment/registry.py 
52
53
54
55
56
57
58
class HashMismatchError(ValueError):
    """Stored content does not match the hash it is stored under.

    Raised on load when the bytes on disk (``operator.json``,
    ``weights.bin``) were modified after `store`, or when the operator
    rebuilt from them hashes to something else. Subclass of `ValueError`.
    """

LocalModelRegistry

Bases: _ModelRegistryBase

Local-filesystem model registry.

Writes are atomic (temp file + os.replace), so a crash mid-write can never leave a truncated operator.json or tag file behind.

Parameters:

Name Type Description Default
root str | PathLike[str]

Directory under which models are stored. Created on construction if it doesn't exist.

 required

Storage layout (under root):

``<hash>/operator.json`` — operator state
``<hash>/weights.bin``   — optional weight bytes
``<hash>/metadata.json`` — tags + arbitrary metadata
``_tags/<name>``         — file containing the target hash
Source code in packages/pipekit-experiment/src/pipekit_experiment/registry.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
class LocalModelRegistry(_ModelRegistryBase):
    """Local-filesystem model registry.

    Writes are atomic (temp file + ``os.replace``), so a crash mid-write
    can never leave a truncated ``operator.json`` or tag file behind.

    Args:
        root: Directory under which models are stored. Created on
            construction if it doesn't exist.

    Storage layout (under ``root``):

        ``<hash>/operator.json`` — operator state
        ``<hash>/weights.bin``   — optional weight bytes
        ``<hash>/metadata.json`` — tags + arbitrary metadata
        ``_tags/<name>``         — file containing the target hash
    """

    def __init__(self, root: str | os.PathLike[str]) -> None:
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)
        (self.root / "_tags").mkdir(exist_ok=True)

    def _exists(self, *parts: str) -> bool:
        return self.root.joinpath(*parts).exists()

    def _read_text(self, *parts: str) -> str:
        return self.root.joinpath(*parts).read_text()

    def _write_text(self, text: str, *parts: str) -> None:
        self._atomic_write(text.encode("utf-8"), *parts)

    def _read_bytes(self, *parts: str) -> bytes:
        return self.root.joinpath(*parts).read_bytes()

    def _write_bytes(self, data: bytes, *parts: str) -> None:
        self._atomic_write(data, *parts)

    def _atomic_write(self, data: bytes, *parts: str) -> None:
        path = self.root.joinpath(*parts)
        path.parent.mkdir(parents=True, exist_ok=True)
        tmp = path.with_name(path.name + ".tmp")
        tmp.write_bytes(data)
        os.replace(tmp, path)

    def _children(self) -> list[str]:
        return sorted(p.name for p in self.root.iterdir() if p.is_dir())

    def _where(self) -> str:
        return str(self.root)

S3ModelRegistry

Bases: _ModelRegistryBase

fsspec-backed model registry.

Works with any fsspec-supported scheme (s3://, gs://, az://, memory://, plain file:// paths). Behind the [s3] extra.

Unlike LocalModelRegistry, writes are only as atomic as the backing store makes them (object stores typically are; plain file:// through fsspec is not).

Parameters:

Name Type Description Default
uri str

Root URI (e.g. "s3://bucket/models/").

 required
storage_options dict[str, Any] | None

Passed straight to fsspec.open. Use this for credentials / endpoints.

 None

Raises:

Type Description
ImportError

If fsspec is not installed.
Source code in packages/pipekit-experiment/src/pipekit_experiment/registry.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
class S3ModelRegistry(_ModelRegistryBase):
    """fsspec-backed model registry.

    Works with any fsspec-supported scheme (``s3://``, ``gs://``,
    ``az://``, ``memory://``, plain ``file://`` paths). Behind the
    ``[s3]`` extra.

    Unlike `LocalModelRegistry`, writes are only as atomic as the
    backing store makes them (object stores typically are; plain
    ``file://`` through fsspec is not).

    Args:
        uri: Root URI (e.g. ``"s3://bucket/models/"``).
        storage_options: Passed straight to ``fsspec.open``. Use this
            for credentials / endpoints.

    Raises:
        ImportError: If ``fsspec`` is not installed.
    """

    def __init__(
        self,
        uri: str,
        storage_options: dict[str, Any] | None = None,
    ) -> None:
        try:
            import fsspec  # noqa: F401
        except ImportError as e:
            raise ImportError(
                "S3ModelRegistry requires fsspec. Install with "
                "`pip install pipekit-experiment[s3]`."
            ) from e
        self.uri = uri.rstrip("/") + "/"
        self.storage_options = dict(storage_options) if storage_options else {}

    def _fs(self):
        import fsspec

        fs, _ = fsspec.core.url_to_fs(self.uri, **self.storage_options)
        return fs

    def _path(self, *parts: str) -> str:
        return self.uri + "/".join(parts)

    def _exists(self, *parts: str) -> bool:
        return self._fs().exists(self._path(*parts))

    def _read_text(self, *parts: str) -> str:
        with self._fs().open(self._path(*parts), "r") as f:
            return f.read()

    def _write_text(self, text: str, *parts: str) -> None:
        with self._fs().open(self._path(*parts), "w") as f:
            f.write(text)

    def _read_bytes(self, *parts: str) -> bytes:
        with self._fs().open(self._path(*parts), "rb") as f:
            return f.read()

    def _write_bytes(self, data: bytes, *parts: str) -> None:
        with self._fs().open(self._path(*parts), "wb") as f:
            f.write(data)

    def _children(self) -> list[str]:
        # detail=False explicitly: several backends (memory, local) default
        # to detail=True and return dicts instead of path strings.
        entries = self._fs().ls(self.uri.rstrip("/"), detail=False)
        return sorted(entry.rstrip("/").split("/")[-1] for entry in entries)

    def _where(self) -> str:
        return self.uri

Artifacts

pipekit_experiment.artifacts

Reproducibility artifacts — training and inference snapshots.

  • TrainingArtifact — captures everything needed to rebuild a trained model: the training pipeline YAML, dataset hash, trained model hash, tracker run id, registry URI, backend info, dep lock.
  • InferenceArtifact — pins the trained-model hashes referenced by the inference pipeline. Subclass of TrainingArtifact is overkill; it's a separate dataclass.

Both serialise as JSON via save / load. They're plain data carriers — rerun and reload_model are convenience helpers that delegate to a supplied registry / loader.

See master plan Report 12, section 2.5.

InferenceArtifact dataclass

Bases: _JsonArtifact

Reproducibility snapshot for an inference pipeline.

Pins the trained-model hashes referenced by the pipeline so a pipekit YAML round-trip resolves to the exact same weights.

Attributes:

Name Type Description
pipeline_yaml str

Serialised inference pipeline.
pinned_model_hashes dict[str, str]

operator-path → hash for every registry-resolved operator in the pipeline.
backend_info dict[str, Any]

Backend-stack info.
deps_lock str

Lockfile contents.
metadata dict[str, Any]

Arbitrary metadata.
Source code in packages/pipekit-experiment/src/pipekit_experiment/artifacts.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@dataclass
class InferenceArtifact(_JsonArtifact):
    """Reproducibility snapshot for an inference pipeline.

    Pins the trained-model hashes referenced by the pipeline so a
    pipekit YAML round-trip resolves to the exact same weights.

    Attributes:
        pipeline_yaml: Serialised inference pipeline.
        pinned_model_hashes: ``operator-path → hash`` for every
            registry-resolved operator in the pipeline.
        backend_info: Backend-stack info.
        deps_lock: Lockfile contents.
        metadata: Arbitrary metadata.
    """

    pipeline_yaml: str
    pinned_model_hashes: dict[str, str] = field(default_factory=dict)
    backend_info: dict[str, Any] = field(default_factory=dict)
    deps_lock: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)

    def reload_models(self, registry: Any) -> dict[str, Operator]:
        """Resolve every pinned model via ``registry.load``.

        Returns a ``{operator-path: Operator}`` dict for callers to
        substitute back into the pipeline.
        """
        return {path: registry.load(h) for path, h in self.pinned_model_hashes.items()}

reload_models(registry)

Resolve every pinned model via registry.load.

Returns a {operator-path: Operator} dict for callers to substitute back into the pipeline.

Source code in packages/pipekit-experiment/src/pipekit_experiment/artifacts.py 
110
111
112
113
114
115
116
def reload_models(self, registry: Any) -> dict[str, Operator]:
    """Resolve every pinned model via ``registry.load``.

    Returns a ``{operator-path: Operator}`` dict for callers to
    substitute back into the pipeline.
    """
    return {path: registry.load(h) for path, h in self.pinned_model_hashes.items()}

TrainingArtifact dataclass

Bases: _JsonArtifact

Reproducibility snapshot for a training run.

Attributes:

Name Type Description
training_pipeline_yaml str

Serialised training pipeline (the pipekit-train TrainingLoop config, typically as YAML or JSON text).
dataset_hash str

Content hash of the training dataset.
trained_model_hash str

Content hash of the trained model (registry identifier).
tracker_run_id str | None

External-tracker run reference (e.g. "mlflow://runs/abc123"). None if no tracker was attached.
model_registry_uri str

URI under which the model was stored.
backend_info dict[str, Any]

Free-form dict describing the training stack (lightning version, JAX version, GPUs, …).
deps_lock str

Contents of the uv / poetry lockfile.
metadata dict[str, Any]

Optional arbitrary metadata (author, date, notes).
Source code in packages/pipekit-experiment/src/pipekit_experiment/artifacts.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@dataclass
class TrainingArtifact(_JsonArtifact):
    """Reproducibility snapshot for a training run.

    Attributes:
        training_pipeline_yaml: Serialised training pipeline (the
            pipekit-train `TrainingLoop` config, typically as YAML
            or JSON text).
        dataset_hash: Content hash of the training dataset.
        trained_model_hash: Content hash of the trained model
            (registry identifier).
        tracker_run_id: External-tracker run reference (e.g.
            ``"mlflow://runs/abc123"``). ``None`` if no tracker was
            attached.
        model_registry_uri: URI under which the model was stored.
        backend_info: Free-form dict describing the training stack
            (lightning version, JAX version, GPUs, …).
        deps_lock: Contents of the uv / poetry lockfile.
        metadata: Optional arbitrary metadata (author, date, notes).
    """

    training_pipeline_yaml: str
    dataset_hash: str
    trained_model_hash: str
    tracker_run_id: str | None
    model_registry_uri: str
    backend_info: dict[str, Any] = field(default_factory=dict)
    deps_lock: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)

    def reload_model(self, registry: Any) -> Operator:
        """Resolve the stored model via ``registry.load``.

        Args:
            registry: Anything satisfying the `ModelRegistry` protocol
                (typically a `LocalModelRegistry` or `S3ModelRegistry`).
        """
        return registry.load(self.trained_model_hash)

reload_model(registry)

Resolve the stored model via registry.load.

Parameters:

Name Type Description Default
registry Any

Anything satisfying the ModelRegistry protocol (typically a LocalModelRegistry or S3ModelRegistry).

 required
Source code in packages/pipekit-experiment/src/pipekit_experiment/artifacts.py 
78
79
80
81
82
83
84
85
def reload_model(self, registry: Any) -> Operator:
    """Resolve the stored model via ``registry.load``.

    Args:
        registry: Anything satisfying the `ModelRegistry` protocol
            (typically a `LocalModelRegistry` or `S3ModelRegistry`).
    """
    return registry.load(self.trained_model_hash)

Adapters

Each adapter is gated behind its tool's optional extra and raises a clean ImportError on first use if the underlying tool isn't installed. The pure-pipekit surface remains importable either way.

DVC

Requires pipekit-experiment[dvc].

pipekit_experiment.adapters.dvc

DVC adapter — dataset content-hash tracking.

DVC is fundamentally about data versioning, not metrics. This adapter focuses on dataset content-hash tracking via the DVC CLI:

  • add(path) — stage path into DVC's content-addressed store, producing a .dvc pointer file.
  • pull(path) — fetch the data referenced by path.dvc from the configured remote.
  • hash(path) — return the content hash recorded in path.dvc.

The adapter doesn't import the dvc Python package — it shells out to the dvc CLI. That keeps the dependency surface minimal: any DVC version installed on the system works. The [dvc] extra declares the Python package for environments that prefer in-process APIs; future versions can switch to the Python API without changing the public interface here.

Behind pipekit-experiment[dvc] extra.

See master plan Report 12, section 3.2.

DVCDatasetVersioning

Track datasets via the DVC CLI.

Parameters:

Name Type Description Default
repo str | Path

Path to the DVC-initialised git repository. Defaults to the current working directory.

 '.'
dvc_executable str

Override the dvc binary location (useful in CI / virtualenv setups).

 'dvc'

Raises:

Type Description
RuntimeError

at first use if the dvc CLI isn't on PATH.
Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/dvc.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class DVCDatasetVersioning:
    """Track datasets via the DVC CLI.

    Args:
        repo: Path to the DVC-initialised git repository. Defaults to
            the current working directory.
        dvc_executable: Override the ``dvc`` binary location (useful
            in CI / virtualenv setups).

    Raises:
        RuntimeError: at first use if the ``dvc`` CLI isn't on PATH.
    """

    def __init__(
        self,
        repo: str | Path = ".",
        dvc_executable: str = "dvc",
    ) -> None:
        self.repo = Path(repo)
        self.dvc_executable = dvc_executable

    def _check(self) -> None:
        if shutil.which(self.dvc_executable) is None:
            raise RuntimeError(
                f"DVC executable {self.dvc_executable!r} not found on PATH. "
                "Install with `pip install pipekit-experiment[dvc]` and "
                "ensure the `dvc` CLI is available."
            )

    def _run(self, *args: str, capture: bool = True) -> subprocess.CompletedProcess:
        self._check()
        return subprocess.run(
            [self.dvc_executable, *args],
            cwd=self.repo,
            check=True,
            capture_output=capture,
            text=True,
        )

    def add(self, path: str | Path) -> Path:
        """Stage ``path`` into DVC. Returns the ``.dvc`` pointer file path.

        The returned `Path` is anchored under ``self.repo`` when ``path``
        is given as a relative path, so callers can rely on
        ``add(...).exists()`` regardless of the process cwd.
        """
        self._run("add", str(path))
        return self._pointer_path(path)

    def pull(self, path: str | Path) -> None:
        """Fetch the data referenced by ``path`` (or ``path.dvc``)."""
        self._run("pull", str(path))

    def push(self, path: str | Path) -> None:
        """Send the data referenced by ``path`` to the remote."""
        self._run("push", str(path))

    def hash(self, path: str | Path) -> str:
        """Return the content hash recorded for ``path``.

        Reads the ``.dvc`` pointer file and extracts the ``md5`` /
        ``hash`` field. Cheap (no DVC subprocess) once the pointer
        exists.

        Raises:
            FileNotFoundError: if no ``.dvc`` pointer is found.
            ValueError: if the pointer file is malformed.
        """
        pointer_path = self._pointer_path(path)
        try:
            text = pointer_path.read_text()
        except FileNotFoundError as e:
            raise FileNotFoundError(
                f"No .dvc pointer for {path!r} (looked at {pointer_path})."
            ) from e
        # DVC pointer files are small YAMLs. Avoid pulling in PyYAML —
        # the hash lives in a top-level ``outs:`` list entry, which
        # YAML renders as ``- md5: <hash>`` or ``- hash: <hash>``.
        for line in text.splitlines():
            stripped = line.strip().lstrip("-").strip()
            if stripped.startswith("md5:"):
                return stripped.split(":", 1)[1].strip()
            if stripped.startswith("hash:"):
                return stripped.split(":", 1)[1].strip()
        raise ValueError(
            f"Could not find a content hash in {pointer_path}; "
            "expected an `md5:` or `hash:` line."
        )

    def status(self, path: str | Path | None = None) -> str:
        """Return the textual ``dvc status`` output."""
        args = ["status"]
        if path is not None:
            args.append(str(path))
        return self._run(*args).stdout

    def _pointer_path(self, path: str | Path) -> Path:
        p = Path(path)
        target = p if p.suffix == ".dvc" else Path(str(p) + ".dvc")
        if not target.is_absolute():
            target = self.repo / target
        return target

    def record_hash_in(self, path: str | Path, sink: dict[str, Any]) -> str:
        """Convenience: stash the dataset hash into ``sink`` and return it.

        Pairs nicely with `TrainingArtifact` construction::

            meta: dict[str, str] = {}
            dvc.record_hash_in("data/train/", meta)
            artifact = TrainingArtifact(..., dataset_hash=meta["data/train/"], ...)
        """
        h = self.hash(path)
        sink[str(path)] = h
        return h

add(path)

Stage path into DVC. Returns the .dvc pointer file path.

The returned Path is anchored under self.repo when path is given as a relative path, so callers can rely on add(...).exists() regardless of the process cwd.

Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/dvc.py 
71
72
73
74
75
76
77
78
79
def add(self, path: str | Path) -> Path:
    """Stage ``path`` into DVC. Returns the ``.dvc`` pointer file path.

    The returned `Path` is anchored under ``self.repo`` when ``path``
    is given as a relative path, so callers can rely on
    ``add(...).exists()`` regardless of the process cwd.
    """
    self._run("add", str(path))
    return self._pointer_path(path)

hash(path)

Return the content hash recorded for path.

Reads the .dvc pointer file and extracts the md5 / hash field. Cheap (no DVC subprocess) once the pointer exists.

Raises:

Type Description
FileNotFoundError

if no .dvc pointer is found.
ValueError

if the pointer file is malformed.
Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/dvc.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def hash(self, path: str | Path) -> str:
    """Return the content hash recorded for ``path``.

    Reads the ``.dvc`` pointer file and extracts the ``md5`` /
    ``hash`` field. Cheap (no DVC subprocess) once the pointer
    exists.

    Raises:
        FileNotFoundError: if no ``.dvc`` pointer is found.
        ValueError: if the pointer file is malformed.
    """
    pointer_path = self._pointer_path(path)
    try:
        text = pointer_path.read_text()
    except FileNotFoundError as e:
        raise FileNotFoundError(
            f"No .dvc pointer for {path!r} (looked at {pointer_path})."
        ) from e
    # DVC pointer files are small YAMLs. Avoid pulling in PyYAML —
    # the hash lives in a top-level ``outs:`` list entry, which
    # YAML renders as ``- md5: <hash>`` or ``- hash: <hash>``.
    for line in text.splitlines():
        stripped = line.strip().lstrip("-").strip()
        if stripped.startswith("md5:"):
            return stripped.split(":", 1)[1].strip()
        if stripped.startswith("hash:"):
            return stripped.split(":", 1)[1].strip()
    raise ValueError(
        f"Could not find a content hash in {pointer_path}; "
        "expected an `md5:` or `hash:` line."
    )

pull(path)

Fetch the data referenced by path (or path.dvc).

Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/dvc.py 
81
82
83
def pull(self, path: str | Path) -> None:
    """Fetch the data referenced by ``path`` (or ``path.dvc``)."""
    self._run("pull", str(path))

push(path)

Send the data referenced by path to the remote.

Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/dvc.py 
85
86
87
def push(self, path: str | Path) -> None:
    """Send the data referenced by ``path`` to the remote."""
    self._run("push", str(path))

record_hash_in(path, sink)

Convenience: stash the dataset hash into sink and return it.

Pairs nicely with TrainingArtifact construction::

meta: dict[str, str] = {}
dvc.record_hash_in("data/train/", meta)
artifact = TrainingArtifact(..., dataset_hash=meta["data/train/"], ...)
Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/dvc.py 
135
136
137
138
139
140
141
142
143
144
145
146
def record_hash_in(self, path: str | Path, sink: dict[str, Any]) -> str:
    """Convenience: stash the dataset hash into ``sink`` and return it.

    Pairs nicely with `TrainingArtifact` construction::

        meta: dict[str, str] = {}
        dvc.record_hash_in("data/train/", meta)
        artifact = TrainingArtifact(..., dataset_hash=meta["data/train/"], ...)
    """
    h = self.hash(path)
    sink[str(path)] = h
    return h

status(path=None)

Return the textual dvc status output.

Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/dvc.py 
121
122
123
124
125
126
def status(self, path: str | Path | None = None) -> str:
    """Return the textual ``dvc status`` output."""
    args = ["status"]
    if path is not None:
        args.append(str(path))
    return self._run(*args).stdout

Hydra + hydra-zen

Requires pipekit-experiment[hydra].

pipekit_experiment.adapters.hydra

Hydra + hydra-zen adapter.

Bridges pipekit operator graphs and Hydra configs. Two directions:

  • from_hydra_cfg(cfg) — instantiate a pipekit Operator from a Hydra/OmegaConf config (or any object with a _target_ key). Uses hydra.utils.instantiate.
  • to_hydra_cfg(op) — build a hydra-zen builds(...) spec from an operator's state record. The resulting object round-trips through hydra-zen's to_yaml / instantiate.
  • from_yaml_sandboxed(text) — build a pipeline from untrusted _target_ YAML, allowing only classes registered with pipekit.register. It never calls hydra.utils.instantiate (which will call any dotted path): the YAML is translated into a pipekit state record and built by pipekit.from_state_sandboxed.

The adapter doesn't import hydra or hydra_zen at module load — each method imports lazily so a user can from … import hydra as hydra_adapter in code that branches on what's installed.

Behind pipekit-experiment[hydra] extra.

See master plan Report 12, section 3.3.

HydraConfigLoader

Bidirectional bridge between pipekit operators and Hydra configs.

Stateless — every method is a @staticmethod. Construct once and reuse, or call the static methods directly.

Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/hydra.py 
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class HydraConfigLoader:
    """Bidirectional bridge between pipekit operators and Hydra configs.

    Stateless — every method is a ``@staticmethod``. Construct once
    and reuse, or call the static methods directly.
    """

    @staticmethod
    def from_hydra_cfg(cfg: Any) -> Operator:
        """Instantiate an `Operator` from a Hydra-flavoured config.

        ``cfg`` may be:
          - An ``omegaconf.DictConfig`` / ``ListConfig``.
          - A plain ``dict`` with a ``_target_`` key.
          - Any object that ``hydra.utils.instantiate`` accepts.

        Raises:
            ImportError: if ``hydra-core`` isn't installed.
            TypeError: if the instantiation result isn't an `Operator`.
        """
        hu = _require_hydra()
        obj = hu.instantiate(cfg)
        if not isinstance(obj, Operator):
            raise TypeError(
                "from_hydra_cfg expected the config to instantiate to an "
                f"Operator, got {type(obj).__name__}."
            )
        return obj

    @staticmethod
    def to_hydra_cfg(op: Operator) -> Any:
        """Produce a hydra-zen ``builds(...)`` spec for ``op``.

        The returned object is the hydra-zen-generated dataclass that
        hydra-zen's ``to_yaml`` and ``instantiate`` understand. For a
        plain operator with primitive-typed config the round-trip is
        lossless; operators with closures / callables (Lambda, Tap, …)
        emit a debug-only spec.

        Raises:
            ImportError: if ``hydra-zen`` isn't installed.
            TypeError: if ``op`` isn't an `Operator`.
        """
        if not isinstance(op, Operator):
            raise TypeError(
                f"to_hydra_cfg expects an Operator, got {type(op).__name__}."
            )
        hz = _require_hydra_zen()
        return hz.builds(type(op), populate_full_signature=False, **op.get_config())

    @staticmethod
    def to_yaml(op: Operator) -> str:
        """Render ``op``, nested operators included, as ``_target_`` YAML.

        Written from ``op.state``: each operator becomes a mapping with
        its ``_target_`` and its config, nested operators become nested
        mappings, an exception class becomes
        ``{_target_: hydra.utils.get_class, path: module.Name}``, and an
        operator whose config is not a constructor call (a `Graph`) is
        written as ``_target_: <Class>.from_config`` with a ``config:``
        mapping. The text loads back with `from_yaml` (Hydra) or
        `from_yaml_sandboxed` to an operator with the same `state`.

        Raises:
            ImportError: if PyYAML isn't installed.
            TypeError: if ``op`` isn't an `Operator`.
            ValueError: if any operator in the tree holds a closure
                (``forbid_in_yaml``): its config could not be loaded back.
        """
        if not isinstance(op, Operator):
            raise TypeError(f"to_yaml expects an Operator, got {type(op).__name__}.")
        try:
            import yaml
        except ImportError as e:
            raise ImportError(
                "to_yaml requires PyYAML. Install with "
                "`pip install pipekit-experiment[hydra]`."
            ) from e
        document = _to_targets(op.state)
        return yaml.safe_dump(document, sort_keys=False)

    @staticmethod
    def from_yaml(text: str) -> Operator:
        """Parse a Hydra YAML string back into an `Operator`.

        Uses OmegaConf to load the YAML, then `from_hydra_cfg` to
        instantiate. Requires both ``hydra-core`` and ``omegaconf``
        (the latter is a transitive dep of ``hydra-core``).
        """
        try:
            from omegaconf import OmegaConf  # ty: ignore[unresolved-import]
        except ImportError as e:
            raise ImportError(
                "from_yaml requires omegaconf (a transitive dep of "
                "hydra-core). Install with "
                "`pip install pipekit-experiment[hydra]`."
            ) from e
        cfg = OmegaConf.create(text)
        return HydraConfigLoader.from_hydra_cfg(cfg)

    @staticmethod
    def from_yaml_sandboxed(text: str) -> Operator:
        """Build a pipeline from untrusted ``_target_`` YAML, safely.

        Only classes registered with `pipekit.register` can be built, at
        any depth (pipekit's own containers included); every class is
        checked before anything is constructed. The YAML is parsed with
        ``yaml.safe_load``, interpolations (``${...}``) are not resolved,
        and ``hydra.utils.instantiate`` is never called. Each ``_target_``
        must name an `Operator` class (or ``Class.from_config``) in an
        already-imported module; ``{_target_: hydra.utils.get_class,
        path: module.Name}`` names an exception class.

        Args:
            text: A YAML document whose root has a ``_target_``.

        Raises:
            ImportError: if PyYAML isn't installed.
            PermissionError: if a ``_target_`` isn't an `Operator` in a
                loaded module, or names a class that isn't registered.
            ValueError: if a node uses ``_args_`` or ``_partial_``, or the
                document is not a mapping with a ``_target_``.
        """
        try:
            import yaml
        except ImportError as e:
            raise ImportError(
                "from_yaml_sandboxed requires PyYAML. Install with "
                "`pip install pipekit-experiment[hydra]`."
            ) from e
        document = yaml.load(text, Loader=_yaml_12_bool_loader(yaml))
        if not isinstance(document, dict) or "_target_" not in document:
            raise ValueError(
                "from_yaml_sandboxed expects a YAML mapping with a `_target_`."
            )
        return from_state_sandboxed(_to_state(document, "<root>"))

from_hydra_cfg(cfg) staticmethod

Instantiate an Operator from a Hydra-flavoured config.

cfg may be: - An omegaconf.DictConfig / ListConfig. - A plain dict with a _target_ key. - Any object that hydra.utils.instantiate accepts.

Raises:

Type Description
ImportError

if hydra-core isn't installed.
TypeError

if the instantiation result isn't an Operator.
Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/hydra.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@staticmethod
def from_hydra_cfg(cfg: Any) -> Operator:
    """Instantiate an `Operator` from a Hydra-flavoured config.

    ``cfg`` may be:
      - An ``omegaconf.DictConfig`` / ``ListConfig``.
      - A plain ``dict`` with a ``_target_`` key.
      - Any object that ``hydra.utils.instantiate`` accepts.

    Raises:
        ImportError: if ``hydra-core`` isn't installed.
        TypeError: if the instantiation result isn't an `Operator`.
    """
    hu = _require_hydra()
    obj = hu.instantiate(cfg)
    if not isinstance(obj, Operator):
        raise TypeError(
            "from_hydra_cfg expected the config to instantiate to an "
            f"Operator, got {type(obj).__name__}."
        )
    return obj

from_yaml(text) staticmethod

Parse a Hydra YAML string back into an Operator.

Uses OmegaConf to load the YAML, then from_hydra_cfg to instantiate. Requires both hydra-core and omegaconf (the latter is a transitive dep of hydra-core).

Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/hydra.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
@staticmethod
def from_yaml(text: str) -> Operator:
    """Parse a Hydra YAML string back into an `Operator`.

    Uses OmegaConf to load the YAML, then `from_hydra_cfg` to
    instantiate. Requires both ``hydra-core`` and ``omegaconf``
    (the latter is a transitive dep of ``hydra-core``).
    """
    try:
        from omegaconf import OmegaConf  # ty: ignore[unresolved-import]
    except ImportError as e:
        raise ImportError(
            "from_yaml requires omegaconf (a transitive dep of "
            "hydra-core). Install with "
            "`pip install pipekit-experiment[hydra]`."
        ) from e
    cfg = OmegaConf.create(text)
    return HydraConfigLoader.from_hydra_cfg(cfg)

from_yaml_sandboxed(text) staticmethod

Build a pipeline from untrusted _target_ YAML, safely.

Only classes registered with pipekit.register can be built, at any depth (pipekit's own containers included); every class is checked before anything is constructed. The YAML is parsed with yaml.safe_load, interpolations (${...}) are not resolved, and hydra.utils.instantiate is never called. Each _target_ must name an Operator class (or Class.from_config) in an already-imported module; {_target_: hydra.utils.get_class, path: module.Name} names an exception class.

Parameters:

Name Type Description Default
text str

A YAML document whose root has a _target_.

 required

Raises:

Type Description
ImportError

if PyYAML isn't installed.
PermissionError

if a _target_ isn't an Operator in a loaded module, or names a class that isn't registered.
ValueError

if a node uses _args_ or _partial_, or the document is not a mapping with a _target_.
Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/hydra.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@staticmethod
def from_yaml_sandboxed(text: str) -> Operator:
    """Build a pipeline from untrusted ``_target_`` YAML, safely.

    Only classes registered with `pipekit.register` can be built, at
    any depth (pipekit's own containers included); every class is
    checked before anything is constructed. The YAML is parsed with
    ``yaml.safe_load``, interpolations (``${...}``) are not resolved,
    and ``hydra.utils.instantiate`` is never called. Each ``_target_``
    must name an `Operator` class (or ``Class.from_config``) in an
    already-imported module; ``{_target_: hydra.utils.get_class,
    path: module.Name}`` names an exception class.

    Args:
        text: A YAML document whose root has a ``_target_``.

    Raises:
        ImportError: if PyYAML isn't installed.
        PermissionError: if a ``_target_`` isn't an `Operator` in a
            loaded module, or names a class that isn't registered.
        ValueError: if a node uses ``_args_`` or ``_partial_``, or the
            document is not a mapping with a ``_target_``.
    """
    try:
        import yaml
    except ImportError as e:
        raise ImportError(
            "from_yaml_sandboxed requires PyYAML. Install with "
            "`pip install pipekit-experiment[hydra]`."
        ) from e
    document = yaml.load(text, Loader=_yaml_12_bool_loader(yaml))
    if not isinstance(document, dict) or "_target_" not in document:
        raise ValueError(
            "from_yaml_sandboxed expects a YAML mapping with a `_target_`."
        )
    return from_state_sandboxed(_to_state(document, "<root>"))

to_hydra_cfg(op) staticmethod

Produce a hydra-zen builds(...) spec for op.

The returned object is the hydra-zen-generated dataclass that hydra-zen's to_yaml and instantiate understand. For a plain operator with primitive-typed config the round-trip is lossless; operators with closures / callables (Lambda, Tap, …) emit a debug-only spec.

Raises:

Type Description
ImportError

if hydra-zen isn't installed.
TypeError

if op isn't an Operator.
Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/hydra.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@staticmethod
def to_hydra_cfg(op: Operator) -> Any:
    """Produce a hydra-zen ``builds(...)`` spec for ``op``.

    The returned object is the hydra-zen-generated dataclass that
    hydra-zen's ``to_yaml`` and ``instantiate`` understand. For a
    plain operator with primitive-typed config the round-trip is
    lossless; operators with closures / callables (Lambda, Tap, …)
    emit a debug-only spec.

    Raises:
        ImportError: if ``hydra-zen`` isn't installed.
        TypeError: if ``op`` isn't an `Operator`.
    """
    if not isinstance(op, Operator):
        raise TypeError(
            f"to_hydra_cfg expects an Operator, got {type(op).__name__}."
        )
    hz = _require_hydra_zen()
    return hz.builds(type(op), populate_full_signature=False, **op.get_config())

to_yaml(op) staticmethod

Render op, nested operators included, as _target_ YAML.

Written from op.state: each operator becomes a mapping with its _target_ and its config, nested operators become nested mappings, an exception class becomes {_target_: hydra.utils.get_class, path: module.Name}, and an operator whose config is not a constructor call (a Graph) is written as _target_: <Class>.from_config with a config: mapping. The text loads back with from_yaml (Hydra) or from_yaml_sandboxed to an operator with the same state.

Raises:

Type Description
ImportError

if PyYAML isn't installed.
TypeError

if op isn't an Operator.
ValueError

if any operator in the tree holds a closure (forbid_in_yaml): its config could not be loaded back.
Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/hydra.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@staticmethod
def to_yaml(op: Operator) -> str:
    """Render ``op``, nested operators included, as ``_target_`` YAML.

    Written from ``op.state``: each operator becomes a mapping with
    its ``_target_`` and its config, nested operators become nested
    mappings, an exception class becomes
    ``{_target_: hydra.utils.get_class, path: module.Name}``, and an
    operator whose config is not a constructor call (a `Graph`) is
    written as ``_target_: <Class>.from_config`` with a ``config:``
    mapping. The text loads back with `from_yaml` (Hydra) or
    `from_yaml_sandboxed` to an operator with the same `state`.

    Raises:
        ImportError: if PyYAML isn't installed.
        TypeError: if ``op`` isn't an `Operator`.
        ValueError: if any operator in the tree holds a closure
            (``forbid_in_yaml``): its config could not be loaded back.
    """
    if not isinstance(op, Operator):
        raise TypeError(f"to_yaml expects an Operator, got {type(op).__name__}.")
    try:
        import yaml
    except ImportError as e:
        raise ImportError(
            "to_yaml requires PyYAML. Install with "
            "`pip install pipekit-experiment[hydra]`."
        ) from e
    document = _to_targets(op.state)
    return yaml.safe_dump(document, sort_keys=False)

Metaflow

Requires pipekit-experiment[metaflow].

pipekit_experiment.adapters.metaflow

Metaflow adapter — wrap a pipekit operator as a Metaflow @step.

Pipekit handles the compute; Metaflow handles scheduling and cross-step artifact tracking. The adapter is pure glue:

  • as_step(op, name, inputs) — returns a function suitable for decoration with @step inside a FlowSpec.
  • MetaflowStepAdapter.run(op, inputs) — invoke an operator with inputs read from self.<input_name> and write outputs to self.<name>.

The adapter doesn't import metaflow at module load — the import happens only when the user actually constructs a Metaflow flow with the produced step.

Behind pipekit-experiment[metaflow] extra.

See master plan Report 12, section 3.5.

MetaflowStepAdapter

Adapt a pipekit Operator for use as a Metaflow @step.

Use either:

  • as_step(op, name, inputs) to obtain a callable that reads its inputs from self.<input_name> and writes the operator's output to self.<name>. Apply @step to it inside a FlowSpec.
  • run(op, namespace, name, inputs) to invoke the same logic synchronously, passing any attribute-style object (e.g. a SimpleNamespace; not a plain dict) as namespace. Useful for testing flows outside Metaflow.

The adapter never inspects Metaflow internals; it only relies on the self namespace conventions and the @step decorator.

Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/metaflow.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class MetaflowStepAdapter:
    """Adapt a pipekit `Operator` for use as a Metaflow ``@step``.

    Use either:

    - `as_step(op, name, inputs)` to obtain a callable that reads its
      inputs from ``self.<input_name>`` and writes the operator's
      output to ``self.<name>``. Apply ``@step`` to it inside a
      ``FlowSpec``.
    - `run(op, namespace, name, inputs)` to invoke the same logic
      synchronously, passing any attribute-style object (e.g. a
      ``SimpleNamespace``; *not* a plain dict) as ``namespace``.
      Useful for testing flows outside Metaflow.

    The adapter never inspects Metaflow internals; it only relies on
    the ``self`` namespace conventions and the ``@step`` decorator.
    """

    @staticmethod
    def run(
        op: Operator,
        namespace: Any,
        *,
        name: str,
        inputs: list[str],
    ) -> Any:
        """Invoke ``op`` against attributes of ``namespace`` and persist its output.

        Reads each ``input_name`` from ``namespace`` and calls
        ``op(*args)``. Writes the result back as ``namespace.<name>``.
        Returns the result.

        Args:
            op: The pipekit operator to invoke.
            namespace: An object exposing ``inputs`` as attributes
                (typically ``self`` inside a Metaflow ``@step``).
            name: Attribute name under which to store the output.
            inputs: List of input attribute names to read from
                ``namespace`` in order.
        """
        if not isinstance(op, Operator):
            raise TypeError(
                f"MetaflowStepAdapter.run expects an Operator, got {type(op).__name__}."
            )
        args = [getattr(namespace, n) for n in inputs]
        out = op(*args)
        setattr(namespace, name, out)
        return out

    @staticmethod
    def as_step(
        op: Operator,
        *,
        name: str,
        inputs: list[str],
        decorate: bool = True,
    ) -> Callable[..., Any]:
        """Build a Metaflow step function around ``op``.

        Returns a function with signature ``step(self)`` that reads
        ``self.<input>`` for each ``inputs`` entry, applies ``op``,
        and writes the result to ``self.<name>``.

        Args:
            op: The pipekit operator to wrap.
            name: Attribute name for the output (also used as
                ``__name__`` of the returned function).
            inputs: Input attribute names, in argument order.
            decorate: If ``True`` (default), wrap the returned
                function with ``metaflow.step`` so it's drop-in for a
                ``FlowSpec``. Set ``False`` to obtain the bare callable
                for testing.

        Raises:
            ImportError: if ``decorate=True`` and ``metaflow`` isn't
                installed.
        """
        if not isinstance(op, Operator):
            raise TypeError(
                f"MetaflowStepAdapter.as_step expects an Operator, "
                f"got {type(op).__name__}."
            )

        def _step(self: Any) -> None:
            MetaflowStepAdapter.run(op, self, name=name, inputs=inputs)

        _step.__name__ = name
        _step.__doc__ = (
            f"Metaflow step wrapping {type(op).__name__}; "
            f"reads {inputs!r}, writes {name!r}."
        )
        if not decorate:
            return _step
        metaflow = _require_metaflow()
        return metaflow.step(_step)

as_step(op, *, name, inputs, decorate=True) staticmethod

Build a Metaflow step function around op.

Returns a function with signature step(self) that reads self.<input> for each inputs entry, applies op, and writes the result to self.<name>.

Parameters:

Name Type Description Default
op Operator

The pipekit operator to wrap.

 required
name str

Attribute name for the output (also used as __name__ of the returned function).

 required
inputs list[str]

Input attribute names, in argument order.

 required
decorate bool

If True (default), wrap the returned function with metaflow.step so it's drop-in for a FlowSpec. Set False to obtain the bare callable for testing.

 True

Raises:

Type Description
ImportError

if decorate=True and metaflow isn't installed.
Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/metaflow.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@staticmethod
def as_step(
    op: Operator,
    *,
    name: str,
    inputs: list[str],
    decorate: bool = True,
) -> Callable[..., Any]:
    """Build a Metaflow step function around ``op``.

    Returns a function with signature ``step(self)`` that reads
    ``self.<input>`` for each ``inputs`` entry, applies ``op``,
    and writes the result to ``self.<name>``.

    Args:
        op: The pipekit operator to wrap.
        name: Attribute name for the output (also used as
            ``__name__`` of the returned function).
        inputs: Input attribute names, in argument order.
        decorate: If ``True`` (default), wrap the returned
            function with ``metaflow.step`` so it's drop-in for a
            ``FlowSpec``. Set ``False`` to obtain the bare callable
            for testing.

    Raises:
        ImportError: if ``decorate=True`` and ``metaflow`` isn't
            installed.
    """
    if not isinstance(op, Operator):
        raise TypeError(
            f"MetaflowStepAdapter.as_step expects an Operator, "
            f"got {type(op).__name__}."
        )

    def _step(self: Any) -> None:
        MetaflowStepAdapter.run(op, self, name=name, inputs=inputs)

    _step.__name__ = name
    _step.__doc__ = (
        f"Metaflow step wrapping {type(op).__name__}; "
        f"reads {inputs!r}, writes {name!r}."
    )
    if not decorate:
        return _step
    metaflow = _require_metaflow()
    return metaflow.step(_step)

run(op, namespace, *, name, inputs) staticmethod

Invoke op against attributes of namespace and persist its output.

Reads each input_name from namespace and calls op(*args). Writes the result back as namespace.<name>. Returns the result.

Parameters:

Name Type Description Default
op Operator

The pipekit operator to invoke.

 required
namespace Any

An object exposing inputs as attributes (typically self inside a Metaflow @step).

 required
name str

Attribute name under which to store the output.

 required
inputs list[str]

List of input attribute names to read from namespace in order.

 required
Source code in packages/pipekit-experiment/src/pipekit_experiment/adapters/metaflow.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@staticmethod
def run(
    op: Operator,
    namespace: Any,
    *,
    name: str,
    inputs: list[str],
) -> Any:
    """Invoke ``op`` against attributes of ``namespace`` and persist its output.

    Reads each ``input_name`` from ``namespace`` and calls
    ``op(*args)``. Writes the result back as ``namespace.<name>``.
    Returns the result.

    Args:
        op: The pipekit operator to invoke.
        namespace: An object exposing ``inputs`` as attributes
            (typically ``self`` inside a Metaflow ``@step``).
        name: Attribute name under which to store the output.
        inputs: List of input attribute names to read from
            ``namespace`` in order.
    """
    if not isinstance(op, Operator):
        raise TypeError(
            f"MetaflowStepAdapter.run expects an Operator, got {type(op).__name__}."
        )
    args = [getattr(namespace, n) for n in inputs]
    out = op(*args)
    setattr(namespace, name, out)
    return out