molforge.reproducibility¶

reproducibility ¶

Emit a citable pipeline.yaml from a molforge output's provenance.

Most papers in this space don't ship reproducible code. molforge already records what produced an output — every engine wrapper attaches a :class:~molforge.core.provenance.Provenance (engine, version, parameters, inputs, and a pointer to the step it consumed) to result.metadata["provenance"]. This module turns that chain into a single, human-readable manifest — the artifact a methods section can point at:

from molforge.reproducibility import emit_pipeline

folded = esmfold.predict(sequence)
docked = vina.dock(folded, ligand)
emit_pipeline(docked, "pipeline.yaml")

The resulting file linearizes the provenance chain into ordered steps and adds a consolidated environment block (molforge / Python / platform versions and the engine versions that ran)::

molforge_pipeline: 1
generated: "2026-07-15T12:00:00+00:00"
environment:
  molforge_version: "0.6.0"
  python_version: "3.12.13"
  platform: "macOS-14.3-arm64"
  engines: {ESMFold: "1.0.3", Vina: "1.2.5"}
steps:
  - step: 1
    engine: ESMFold
    engine_version: "1.0.3"
    inputs: {sequence: "MKT..."}
    parameters: {num_recycles: 4}
  - step: 2
    engine: Vina
    ...
output: {type: DockingResult}

The in-memory :class:PipelineManifest and its to_dict / to_json forms need no third-party dependency. Reading and writing the .yaml form needs PyYAML — an opt-in extra (pip install "molforge[repro]") so molforge's core stays numpy-only.

Replay¶

:func:replay re-executes a manifest's chain, threading each step's output into the next::

from molforge.reproducibility import load_pipeline, replay

manifest = load_pipeline("pipeline.yaml")
output = replay(manifest, context={"ligand": "aspirin.sdf"})

It resolves each step's engine from the registry (molforge's own wrappers plus anything under :mod:molforge.plugins), reconstructs the call with a per-operation replay handler (molforge ships predict / dock), and runs it. Handlers own the reconstruction, so the fragile "which recorded input is upstream vs. a literal" wiring is contained per operation rather than guessed globally.

Replay is inherently partial: engines must be installed, GPU steps need the hardware (replay orchestrates, it doesn't provide compute), and inputs that aren't literals (a docking receptor is really the previous step's output; a ligand may be a path that no longer exists) come from the previous step or a supplied context. An unresolvable input, an unknown engine, or an operation with no registered handler raises a clear :class:ReplayError. Register a handler for a custom operation with :func:register_replay_handler.

A manifest is single-output and linear (provenance has one parent pointer); merging several outputs' chains is a future extension.

PipelineStep `dataclass` ¶

PipelineStep(
    step: int,
    engine: str,
    operation: str = "",
    engine_version: str = "",
    timestamp: str = "",
    inputs: dict[str, Any] = dict(),
    parameters: dict[str, Any] = dict(),
)

One step of a pipeline — a single provenance entry, linearized.

Attributes:

Name	Type	Description
`step`	`int`	1-indexed position in the pipeline (1 = oldest / originating).
`engine`	`str`	Producer name (engine name or molforge function path).
`operation`	`str`	The engine method that produced the output (`"predict"` / `"dock"` / ...); `""` when unrecorded.
`engine_version`	`str`	Producer version, `""` when not exposed.
`timestamp`	`str`	ISO-8601 UTC time the step ran.
`inputs`	`dict[str, Any]`	Input identifiers (sequence, paths, hashes).
`parameters`	`dict[str, Any]`	Engine arguments that drove the step.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to a plain, ordered dict for serialization.

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> PipelineStep

Rebuild from :meth:to_dict output; tolerant of missing keys.

PipelineManifest `dataclass` ¶

PipelineManifest(
    environment: dict[str, Any],
    steps: list[PipelineStep],
    generated: str = "",
    output: dict[str, Any] = dict(),
    schema_version: int = SCHEMA_VERSION,
)

A citable description of the workflow that produced an output.

Attributes:

Name	Type	Description
`environment`	`dict[str, Any]`	molforge / Python / platform versions plus an `engines` map of the engine versions that ran.
`steps`	`list[PipelineStep]`	The pipeline steps, oldest-first.
`generated`	`str`	ISO-8601 UTC time the manifest was emitted.
`output`	`dict[str, Any]`	A short descriptor of the terminal output (`{"type": ..., "name": ...}`).
`schema_version`	`int`	The on-disk schema version.

describe ¶

describe() -> str

A compact human-readable summary, one line per step.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to the on-disk dict shape (ordered, JSON/YAML-native).

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> PipelineManifest

Rebuild from :meth:to_dict output; tolerant of missing keys.

to_json ¶

to_json(*, indent: int | None = 2) -> str

Serialize to JSON text. No third-party dependency.

from_json `classmethod` ¶

from_json(text: str) -> PipelineManifest

Deserialize from JSON text.

to_yaml ¶

to_yaml() -> str

Serialize to YAML text. Requires the repro extra (PyYAML).

from_yaml `classmethod` ¶

from_yaml(text: str) -> PipelineManifest

Deserialize from YAML text. Requires the repro extra (PyYAML).

ReplayError ¶

Bases: RuntimeError

Raised when a manifest can't be replayed — a missing engine, an operation with no handler, or an input that can't be resolved.

pipeline_manifest ¶

pipeline_manifest(
    obj: Provenance | object,
) -> PipelineManifest

Build a :class:PipelineManifest from an output or a provenance.

Parameters:

Name	Type	Description	Default
`obj`	`Provenance \| object`	A :class:`~molforge.core.provenance.Provenance`, or any molforge output carrying one at `metadata["provenance"]` (a :class:`~molforge.core.Protein`, `DockingResult`, `Pose`, `DesignedSequence`, ...).	required

Returns:

Type	Description
`PipelineManifest`	A manifest with the provenance chain linearized oldest-first and the
`PipelineManifest`	environment consolidated.

Raises:

Type	Description
`ValueError`	If no provenance can be found on `obj`.

emit_pipeline ¶

emit_pipeline(
    obj: Provenance | object,
    path: str | PathLike[str],
    *,
    fmt: str = "yaml",
) -> PipelineManifest

Write a pipeline.yaml (or .json) describing how obj was made.

Parameters:

Name	Type	Description	Default
`obj`	`Provenance \| object`	An output carrying provenance, or a :class:`~molforge.core.provenance.Provenance`.	required
`path`	`str \| PathLike[str]`	Destination file path.	required
`fmt`	`str`	`"yaml"` (default; needs the `repro` extra) or `"json"` (no extra).	`'yaml'`

Returns:

Name	Type	Description
`The`	`PipelineManifest`	class:`PipelineManifest` that was written (handy for inspection).

Raises:

Type	Description
`ValueError`	If no provenance is found, or `fmt` is unrecognized.
`ImportError`	If `fmt="yaml"` and PyYAML isn't installed.

load_pipeline ¶

load_pipeline(
    path: str | PathLike[str],
) -> PipelineManifest

Load a manifest from a .yaml / .json file.

The format is chosen by suffix: .json is parsed as JSON (no extra); anything else is parsed as YAML (needs the repro extra). Since YAML is a superset of JSON, a .yaml loader also reads JSON content.

Raises:

Type	Description
`ImportError`	If a YAML file is loaded without PyYAML installed.

register_replay_handler ¶

register_replay_handler(
    operation: str,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator registering a replay handler for operation.

A handler has the signature (engine_factory, step, upstream_output, context) -> output — it reconstructs the engine from step.parameters and calls the right method, using upstream_output (the previous step's result) and context (user-supplied inputs) as needed::

@register_replay_handler("dock")
def _dock(factory, step, upstream, context): ...

replay ¶

replay(
    source: PipelineManifest | Provenance | object,
    *,
    context: dict[str, Any] | None = None,
) -> Any

Re-execute a pipeline's chain, returning the terminal output.

Parameters:

Name	Type	Description	Default
`source`	`PipelineManifest \| Provenance \| object`	A :class:`PipelineManifest` (e.g. from :func:`load_pipeline`), a :class:`~molforge.core.provenance.Provenance`, or any output carrying one.	required
`context`	`dict[str, Any] \| None`	Objects to resolve recorded inputs that aren't literals — keyed by the input name (`{"ligand": "aspirin.sdf"}`). A step's primary input is threaded from the previous step automatically; `context` covers the rest.	`None`

Returns:

Type	Description
`Any`	The output of the final step.

Raises:

Type	Description
`ReplayError`	If a step has no recorded operation, its engine can't be resolved, no handler is registered for its operation, or a required input can't be resolved.

molforge.reproducibility¶

reproducibility ¶

Replay¶

PipelineStep dataclass ¶

to_dict ¶

from_dict classmethod ¶

PipelineManifest dataclass ¶

describe ¶

to_dict ¶

from_dict classmethod ¶

to_json ¶

from_json classmethod ¶

to_yaml ¶

from_yaml classmethod ¶

ReplayError ¶

pipeline_manifest ¶

emit_pipeline ¶

load_pipeline ¶

register_replay_handler ¶

replay ¶

PipelineStep `dataclass` ¶

from_dict `classmethod` ¶

PipelineManifest `dataclass` ¶

from_dict `classmethod` ¶

from_json `classmethod` ¶

from_yaml `classmethod` ¶