47.3 Building an Experiment Registry

Figure 47.16 illustrates how these three pipeline stages connect through hash-verified artifact links. Each stage consumes input artifacts (identified by their SHA-256 hashes), produces output artifacts (also hashed), and emits a PROV record that links inputs to outputs. The result is a fully traceable chain from raw data to final evaluation metrics.

1. Project Structure

A well-organized project structure is the first prerequisite for a functional registry. The following layout separates concerns cleanly: source code in src/, configurations in configs/, data in data/ (DVC-tracked), and outputs in outputs/ (also DVC-tracked). The dvc.yaml file at the root defines the pipeline DAG.

Six months after submission, a reviewer asks you to rerun the analysis in Table 3 of your paper, but the student who trained the model has graduated, the GPU server was rebuilt, and nobody can identify which preprocessing script produced the training set. An experiment registry turns that panic into a three-command operation by assigning a content-addressable hash to every input and output artifact at each pipeline stage and storing those hashes alongside run metadata (Git commit, config values, timestamps) in a queryable, machine-verifiable record. Build one whenever your project involves iterative experimentation with changing parameters or data; for one-off scripts that will never be rerun, a simple lab notebook entry suffices.

The project structure below shows how these principles translate into a concrete directory layout for a crystal property prediction pipeline. In short: if you cannot regenerate any result from its recorded inputs in three commands, your registry is incomplete.

crystal-property-prediction/
├── .git/                      # Git repository
├── .dvc/                      # DVC configuration
│   └── config                 # DVC remote settings
├── dvc.yaml                   # Pipeline DAG definition
├── dvc.lock                   # Frozen hashes for all pipeline stages
├── params.yaml                # Shared parameters (read by DVC)
├── configs/
│   ├── preprocess.yaml        # Preprocessing parameters
│   ├── train.yaml             # Training hyperparameters
│   └── evaluate.yaml          # Evaluation settings
├── src/
│   ├── __init__.py
│   ├── preprocess.py          # Stage 1: data preprocessing
│   ├── train.py               # Stage 2: model training
│   ├── evaluate.py            # Stage 3: evaluation
│   └── registry.py            # Provenance recording utilities
├── data/
│   ├── raw/                   # Raw data (DVC-tracked)
│   │   └── crystals.parquet.dvc
│   └── processed/             # Processed data (DVC-tracked)
│       └── .gitkeep
├── outputs/
│   ├── models/                # Trained models (DVC-tracked)
│   │   └── .gitkeep
│   └── metrics/               # Evaluation results
│       └── .gitkeep
├── provenance/                # PROV documents (Git-tracked)
│   └── .gitkeep
├── Dockerfile                 # Environment specification
├── requirements.txt           # Pinned dependencies
└── README.md

2. The DVC Pipeline Definition

DVC pipelines (dvc.yaml) declare the stages of a pipeline, their dependencies, parameters, and outputs. DVC uses this declaration to build the artifact DAG, determine which stages need re-execution when inputs change, and lock the exact hashes of all artifacts in dvc.lock (an auto-generated lockfile that freezes each artifact's SHA-256 hash at the time of the last successful run, analogous to package-lock.json in Node.js).

Mental Model

Think of the artifact DAG as a restaurant kitchen's order ticket system. Each ticket (provenance record) lists the exact ingredients that went into a dish (input hashes), which cook prepared it (code version), the recipe card used (config hash), and the timestamp when it left the kitchen (output hash). If a customer complains about a dish served last Tuesday, the chef can pull the ticket, trace back to the specific bag of flour and carton of eggs, and determine whether the problem was a bad ingredient, a wrong recipe, or a substitution. The DAG links tickets across stations the same way: the prep station's output ticket becomes the grill station's input ticket, forming a chain from raw delivery to plated dish.

# dvc.yaml: Three-stage ML pipeline with full dependency tracking
stages:
  preprocess:
    cmd: python -m src.preprocess
    deps:
      - src/preprocess.py
      - data/raw/crystals.parquet
    params:
      - configs/preprocess.yaml:
          - feature_columns
          - normalize
          - train_fraction
          - random_seed
    outs:
      - data/processed/train.parquet
      - data/processed/test.parquet
    metrics:
      - outputs/metrics/preprocess_stats.json:
          cache: false

  train:
    cmd: python -m src.train
    deps:
      - src/train.py
      - data/processed/train.parquet
    params:
      - configs/train.yaml:
          - architecture
          - hidden_dim
          - num_layers
          - dropout
          - learning_rate
          - batch_size
          - max_epochs
          - early_stopping_patience
          - random_seed
    outs:
      - outputs/models/model.pt
    plots:
      - outputs/metrics/training_curves.json:
          cache: false
          x: epoch
          y: val_loss

  evaluate:
    cmd: python -m src.evaluate
    deps:
      - src/evaluate.py
      - data/processed/test.parquet
      - outputs/models/model.pt
    params:
      - configs/evaluate.yaml:
          - metrics_list
          - confidence_level
    metrics:
      - outputs/metrics/evaluation.json:
          cache: false
    plots:
      - outputs/metrics/predictions_vs_actual.json:
          cache: false
          x: actual
          y: predicted

The params entries are particularly important for provenance: DVC tracks not just which config file a stage depends on, but which specific parameters within that file. If you change the learning rate in train.yaml, DVC knows to re-run the training and evaluation stages but skip preprocessing (whose parameters did not change). This selective re-execution is the practical benefit of a well-specified artifact DAG.

3. Stage 1: Preprocessing with Provenance

Each pipeline stage is a Python module that reads its inputs, performs its computation, writes its outputs, and records provenance metadata. The preprocessing stage loads raw crystal structure data, extracts features, splits into train/test sets, and emits a PROV record documenting the transformation.

The three utility functions imported below (compute_file_hash, get_git_info, and ProvenanceRecorder) are defined in the shared src/registry.py module presented in Section 6. They appear here in order of use rather than definition so you can see each stage's logic before the underlying plumbing.

"""src/preprocess.py: Stage 1 of the provenance-tracked pipeline."""
import hashlib
import json
import yaml
import pandas as pd
import numpy as np
from pathlib import Path
from datetime import datetime, timezone
from sklearn.model_selection import train_test_split

from src.registry import (
    compute_file_hash,
    get_git_info,
    ProvenanceRecorder,
)


def load_config(path: str = "configs/preprocess.yaml") -> dict:
    """Load preprocessing configuration."""
    with open(path) as f:
        return yaml.safe_load(f)


def extract_features(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Extract features from raw crystal structure data.

    Selects configured feature columns, applies normalization,
    and handles missing values.
    """
    feature_cols = config["feature_columns"]
    target_col = config.get("target_column", "bandgap_eV")

    # Select features and target
    features = df[feature_cols + [target_col]].copy()

    # Drop rows with missing targets
    initial_count = len(features)
    features = features.dropna(subset=[target_col])
    dropped = initial_count - len(features)

    if config.get("normalize", True):
        for col in feature_cols:
            col_mean = features[col].mean()
            col_std = features[col].std()
            if col_std > 0:
                features[col] = (features[col] - col_mean) / col_std

    return features, dropped


def main():
    """Run the preprocessing stage with full provenance tracking."""
    config = load_config()
    git_info = get_git_info()
    recorder = ProvenanceRecorder(stage_name="preprocess")

    # Record start
    recorder.start(
        git_commit=git_info["commit_hash"],
        config_hash=compute_file_hash("configs/preprocess.yaml"),
    )

    # Load raw data
    raw_path = Path("data/raw/crystals.parquet")
    raw_hash = compute_file_hash(raw_path)
    recorder.add_input("raw_crystals", str(raw_path), raw_hash)

    df = pd.read_parquet(raw_path)
    print(f"Loaded {len(df)} raw crystal records")

    # Extract features
    features, dropped = extract_features(df, config)
    print(f"Extracted features: {len(features)} records "
          f"({dropped} dropped for missing targets)")

    # Train/test split with fixed random seed
    train_df, test_df = train_test_split(
        features,
        train_size=config["train_fraction"],
        random_state=config["random_seed"],
    )

    # Save processed data
    train_path = Path("data/processed/train.parquet")
    test_path = Path("data/processed/test.parquet")
    train_path.parent.mkdir(parents=True, exist_ok=True)

    train_df.to_parquet(train_path, index=False)
    test_df.to_parquet(test_path, index=False)

    train_hash = compute_file_hash(train_path)
    test_hash = compute_file_hash(test_path)

    recorder.add_output("train_data", str(train_path), train_hash)
    recorder.add_output("test_data", str(test_path), test_hash)

    # Save preprocessing statistics
    stats = {
        "raw_records": len(df),
        "dropped_records": dropped,
        "train_records": len(train_df),
        "test_records": len(test_df),
        "feature_columns": config["feature_columns"],
        "normalized": config.get("normalize", True),
        "random_seed": config["random_seed"],
        "raw_data_hash": raw_hash,
        "train_data_hash": train_hash,
        "test_data_hash": test_hash,
    }
    stats_path = Path("outputs/metrics/preprocess_stats.json")
    stats_path.parent.mkdir(parents=True, exist_ok=True)
    with open(stats_path, "w") as f:
        json.dump(stats, f, indent=2)

    # Complete provenance record
    recorder.complete(metrics=stats)
    recorder.save_prov("provenance/preprocess_prov.json")

    print(f"Preprocessing complete:")
    print(f"  Train: {len(train_df)} records [{train_hash[:12]}...]")
    print(f"  Test:  {len(test_df)} records [{test_hash[:12]}...]")


if __name__ == "__main__":
    main()

4. Stage 2: Training with Experiment Tracking

The training stage combines DVC pipeline integration (for artifact DAG tracking) with MLflow experiment tracking (for hyperparameter and metric logging). This dual integration is the key to a complete registry: DVC captures what artifacts exist and how they relate; MLflow captures the detailed experimental context (learning curves, parameter importance, model comparison dashboards).

"""src/train.py: Stage 2 with MLflow tracking and provenance."""
import json
import yaml
import torch
import torch.nn as nn
import torch.optim as optim
import mlflow
import numpy as np
import pandas as pd
from pathlib import Path
from torch.utils.data import DataLoader, TensorDataset

from src.registry import (
    compute_file_hash,
    get_git_info,
    ProvenanceRecorder,
    set_reproducibility_seed,
)


class CrystalPropertyNet(nn.Module):
    """A feedforward network for crystal property prediction."""

    def __init__(self, input_dim: int, hidden_dim: int,
                 num_layers: int, dropout: float):
        super().__init__()
        layers = []
        current_dim = input_dim

        for i in range(num_layers):
            layers.append(nn.Linear(current_dim, hidden_dim))
            layers.append(nn.ReLU())
            layers.append(nn.Dropout(dropout))
            current_dim = hidden_dim

        layers.append(nn.Linear(current_dim, 1))
        self.network = nn.Sequential(*layers)

    def forward(self, x):
        return self.network(x).squeeze(-1)


def load_config(path: str = "configs/train.yaml") -> dict:
    """Load training configuration."""
    with open(path) as f:
        return yaml.safe_load(f)


def create_dataloader(
    parquet_path: str,
    target_col: str,
    batch_size: int,
    shuffle: bool = True,
) -> tuple[DataLoader, int]:
    """Create a PyTorch DataLoader from a parquet file."""
    df = pd.read_parquet(parquet_path)
    feature_cols = [c for c in df.columns if c != target_col]

    X = torch.tensor(df[feature_cols].values, dtype=torch.float32)
    y = torch.tensor(df[target_col].values, dtype=torch.float32)

    dataset = TensorDataset(X, y)
    loader = DataLoader(dataset, batch_size=batch_size, shuffle=shuffle)
    return loader, X.shape[1]


def train_epoch(model, loader, optimizer, criterion, device):
    """Train for one epoch and return average loss."""
    model.train()
    total_loss = 0.0
    n_batches = 0

    for X_batch, y_batch in loader:
        X_batch, y_batch = X_batch.to(device), y_batch.to(device)
        optimizer.zero_grad()
        predictions = model(X_batch)
        loss = criterion(predictions, y_batch)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
        n_batches += 1

    return total_loss / n_batches


@torch.no_grad()
def validate(model, loader, criterion, device):
    """Validate and return average loss."""
    model.eval()
    total_loss = 0.0
    n_batches = 0

    for X_batch, y_batch in loader:
        X_batch, y_batch = X_batch.to(device), y_batch.to(device)
        predictions = model(X_batch)
        loss = criterion(predictions, y_batch)
        total_loss += loss.item()
        n_batches += 1

    return total_loss / n_batches


def main():
    """Run training with MLflow tracking and provenance recording."""
    config = load_config()
    git_info = get_git_info()
    recorder = ProvenanceRecorder(stage_name="train")

    # Set reproducibility
    set_reproducibility_seed(config["random_seed"])

    recorder.start(
        git_commit=git_info["commit_hash"],
        config_hash=compute_file_hash("configs/train.yaml"),
    )

    # Record input artifacts
    train_path = "data/processed/train.parquet"
    train_hash = compute_file_hash(train_path)
    recorder.add_input("train_data", train_path, train_hash)

    config_hash = compute_file_hash("configs/train.yaml")
    recorder.add_input("train_config", "configs/train.yaml", config_hash)

    # Setup device
    device = torch.device(
        "cuda" if torch.cuda.is_available() else "cpu"
    )
    print(f"Training on {device}")

    # Create data loaders
    target_col = config.get("target_column", "bandgap_eV")
    train_loader, input_dim = create_dataloader(
        train_path, target_col, config["batch_size"]
    )

    # For validation, use a portion of training data
    # (test data is reserved for Stage 3)
    val_loader, _ = create_dataloader(
        train_path, target_col, config["batch_size"], shuffle=False
    )

    # Create model
    model = CrystalPropertyNet(
        input_dim=input_dim,
        hidden_dim=config["hidden_dim"],
        num_layers=config["num_layers"],
        dropout=config["dropout"],
    ).to(device)

    optimizer = optim.Adam(model.parameters(), lr=config["learning_rate"])
    criterion = nn.MSELoss()

    # MLflow tracking
    mlflow.set_tracking_uri("sqlite:///mlflow.db")
    mlflow.set_experiment("crystal_property_prediction")

    with mlflow.start_run() as run:
        # Log provenance metadata as MLflow tags
        mlflow.set_tag("git_commit", git_info["commit_hash"])
        mlflow.set_tag("git_branch", git_info["branch"])
        mlflow.set_tag("git_dirty", str(git_info["is_dirty"]))
        mlflow.set_tag("train_data_hash", train_hash)
        mlflow.set_tag("config_hash", config_hash)

        # Log all hyperparameters
        mlflow.log_params({
            "architecture": config["architecture"],
            "hidden_dim": config["hidden_dim"],
            "num_layers": config["num_layers"],
            "dropout": config["dropout"],
            "learning_rate": config["learning_rate"],
            "batch_size": config["batch_size"],
            "max_epochs": config["max_epochs"],
            "random_seed": config["random_seed"],
            "input_dim": input_dim,
            "device": str(device),
        })

        # Training loop with early stopping
        best_val_loss = float("inf")
        patience_counter = 0
        training_curves = []

        for epoch in range(1, config["max_epochs"] + 1):
            train_loss = train_epoch(
                model, train_loader, optimizer, criterion, device
            )
            val_loss = validate(model, val_loader, criterion, device)

            # Log to MLflow
            mlflow.log_metrics({
                "train_loss": train_loss,
                "val_loss": val_loss,
            }, step=epoch)

            training_curves.append({
                "epoch": epoch,
                "train_loss": round(train_loss, 6),
                "val_loss": round(val_loss, 6),
            })

            # Early stopping
            if val_loss < best_val_loss:
                best_val_loss = val_loss
                patience_counter = 0
                # Save best model
                model_path = Path("outputs/models/model.pt")
                model_path.parent.mkdir(parents=True, exist_ok=True)
                torch.save({
                    "model_state_dict": model.state_dict(),
                    "optimizer_state_dict": optimizer.state_dict(),
                    "epoch": epoch,
                    "val_loss": val_loss,
                    "config": config,
                    "input_dim": input_dim,
                    "git_commit": git_info["commit_hash"],
                    "train_data_hash": train_hash,
                }, model_path)
            else:
                patience_counter += 1
                if patience_counter >= config["early_stopping_patience"]:
                    print(f"Early stopping at epoch {epoch}")
                    break

            if epoch % 10 == 0:
                print(f"Epoch {epoch}: train={train_loss:.6f}, "
                      f"val={val_loss:.6f}")

        # Log final metrics
        model_hash = compute_file_hash("outputs/models/model.pt")
        mlflow.log_metrics({
            "best_val_loss": best_val_loss,
            "final_epoch": epoch,
        })
        mlflow.set_tag("model_hash", model_hash)

        # Log the model to MLflow's artifact store
        mlflow.pytorch.log_model(
            model, "model",
            registered_model_name="CrystalPropertyPredictor",
        )

        # Save training curves for DVC plots
        curves_path = Path("outputs/metrics/training_curves.json")
        with open(curves_path, "w") as f:
            json.dump(training_curves, f, indent=2)

        print(f"\nTraining complete (MLflow run: {run.info.run_id})")
        print(f"  Best val loss: {best_val_loss:.6f} at epoch {epoch}")
        print(f"  Model hash: {model_hash[:12]}...")

    # Record output artifacts in provenance
    recorder.add_output(
        "trained_model", "outputs/models/model.pt", model_hash
    )
    recorder.complete(metrics={
        "best_val_loss": best_val_loss,
        "final_epoch": epoch,
        "mlflow_run_id": run.info.run_id,
    })
    recorder.save_prov("provenance/train_prov.json")


if __name__ == "__main__":
    main()

5. Stage 3: Evaluation with Verification

The evaluation stage loads the trained model and test data, computes metrics, and performs a critical step that most pipelines omit: hash verification. Before using any input artifact, the stage recomputes its hash and compares it against the hash recorded in the provenance chain (stored in both the PROV JSON files from previous stages and the frozen hashes in dvc.lock). This catches silent corruption, accidental overwrites, and any discrepancy between what the registry claims and what actually exists on disk.

"""src/evaluate.py: Stage 3 with hash verification and provenance."""
import json
import yaml
import torch
import numpy as np
import pandas as pd
from pathlib import Path
from sklearn.metrics import (
    mean_absolute_error,
    mean_squared_error,
    r2_score,
)

from src.registry import (
    compute_file_hash,
    get_git_info,
    ProvenanceRecorder,
    verify_artifact_chain,
)
from src.train import CrystalPropertyNet


def load_config(path: str = "configs/evaluate.yaml") -> dict:
    """Load evaluation configuration."""
    with open(path) as f:
        return yaml.safe_load(f)


def load_model_with_verification(
    model_path: str,
    expected_train_hash: str = None,
) -> tuple[CrystalPropertyNet, dict]:
    """Load a model checkpoint and verify its provenance metadata.

    Checks that the embedded data hash matches the expected value,
    catching cases where the model was trained on different data
    than the evaluation expects.
    """
    # weights_only=False is required to load the full metadata dict
    checkpoint = torch.load(model_path, weights_only=False)

    # Verify embedded provenance if expected hashes are provided
    if expected_train_hash:
        actual_hash = checkpoint.get("train_data_hash", "UNKNOWN")
        if actual_hash != expected_train_hash:
            raise ValueError(
                f"Model provenance mismatch!\n"
                f"  Expected train data hash: {expected_train_hash[:16]}...\n"
                f"  Model was trained on:     {actual_hash[:16]}...\n"
                f"  This model may have been trained on different data."
            )

    config = checkpoint["config"]
    input_dim = checkpoint["input_dim"]

    model = CrystalPropertyNet(
        input_dim=input_dim,
        hidden_dim=config["hidden_dim"],
        num_layers=config["num_layers"],
        dropout=config["dropout"],
    )
    model.load_state_dict(checkpoint["model_state_dict"])
    model.eval()

    return model, checkpoint


def compute_metrics(
    y_true: np.ndarray,
    y_pred: np.ndarray,
    confidence_level: float = 0.95,
) -> dict:
    """Compute evaluation metrics with bootstrap confidence intervals,
    where bootstrap resampling is a statistical technique that repeatedly
    draws random samples (with replacement) from the predictions to
    estimate the variability of a metric."""
    mae = mean_absolute_error(y_true, y_pred)
    rmse = np.sqrt(mean_squared_error(y_true, y_pred))
    r2 = r2_score(y_true, y_pred)

    # Bootstrap confidence intervals
    n_bootstrap = 1000
    rng = np.random.default_rng(42)
    n_samples = len(y_true)
    bootstrap_maes = []

    for _ in range(n_bootstrap):
        idx = rng.integers(0, n_samples, size=n_samples)
        bootstrap_maes.append(
            mean_absolute_error(y_true[idx], y_pred[idx])
        )

    alpha = (1 - confidence_level) / 2
    ci_lower = np.quantile(bootstrap_maes, alpha)
    ci_upper = np.quantile(bootstrap_maes, 1 - alpha)

    return {
        "mae": round(float(mae), 6),
        "rmse": round(float(rmse), 6),
        "r2": round(float(r2), 6),
        "mae_ci_lower": round(float(ci_lower), 6),
        "mae_ci_upper": round(float(ci_upper), 6),
        "confidence_level": confidence_level,
        "n_test_samples": n_samples,
    }


def main():
    """Run evaluation with hash verification and provenance."""
    config = load_config()
    git_info = get_git_info()
    recorder = ProvenanceRecorder(stage_name="evaluate")

    recorder.start(
        git_commit=git_info["commit_hash"],
        config_hash=compute_file_hash("configs/evaluate.yaml"),
    )

    # --- Hash Verification ---
    # Before using any artifact, verify its integrity against
    # the provenance chain from previous stages

    test_path = "data/processed/test.parquet"
    model_path = "outputs/models/model.pt"

    test_hash = compute_file_hash(test_path)
    model_hash = compute_file_hash(model_path)

    # Load preprocessing provenance to get expected test hash
    preprocess_prov_path = Path("provenance/preprocess_prov.json")
    if preprocess_prov_path.exists():
        with open(preprocess_prov_path) as f:
            preprocess_prov = json.load(f)
        expected_test_hash = preprocess_prov.get("outputs", {}).get(
            "test_data", {}
        ).get("hash")
        if expected_test_hash and expected_test_hash != test_hash:
            raise ValueError(
                f"Test data integrity check FAILED!\n"
                f"  Expected: {expected_test_hash[:16]}...\n"
                f"  Actual:   {test_hash[:16]}...\n"
                f"  The test data may have been modified since preprocessing."
            )
        print(f"Test data integrity: VERIFIED [{test_hash[:12]}...]")

    # Load training provenance to get expected train data hash
    train_prov_path = Path("provenance/train_prov.json")
    expected_train_hash = None
    if train_prov_path.exists():
        with open(train_prov_path) as f:
            train_prov = json.load(f)
        expected_train_hash = train_prov.get("inputs", {}).get(
            "train_data", {}
        ).get("hash")

    recorder.add_input("test_data", test_path, test_hash)
    recorder.add_input("trained_model", model_path, model_hash)

    # Load model with provenance verification
    model, checkpoint = load_model_with_verification(
        model_path, expected_train_hash
    )
    print(f"Model integrity: VERIFIED [{model_hash[:12]}...]")
    print(f"  Trained at commit: {checkpoint.get('git_commit', 'N/A')[:8]}...")
    print(f"  Best val loss: {checkpoint.get('val_loss', 'N/A')}")

    # Load test data
    target_col = config.get("target_column", "bandgap_eV")
    test_df = pd.read_parquet(test_path)
    feature_cols = [c for c in test_df.columns if c != target_col]

    X_test = torch.tensor(
        test_df[feature_cols].values, dtype=torch.float32
    )
    y_test = test_df[target_col].values

    # Run predictions
    with torch.no_grad():
        y_pred = model(X_test).numpy()

    # Compute metrics
    metrics = compute_metrics(
        y_test, y_pred,
        confidence_level=config.get("confidence_level", 0.95),
    )

    # Add provenance metadata to metrics
    metrics["model_hash"] = model_hash
    metrics["test_data_hash"] = test_hash
    metrics["git_commit"] = git_info["commit_hash"]
    metrics["model_trained_at_commit"] = checkpoint.get("git_commit", "N/A")

    # Save evaluation results
    eval_path = Path("outputs/metrics/evaluation.json")
    eval_path.parent.mkdir(parents=True, exist_ok=True)
    with open(eval_path, "w") as f:
        json.dump(metrics, f, indent=2)

    # Save predictions vs actuals for DVC plots
    predictions_path = Path("outputs/metrics/predictions_vs_actual.json")
    pred_records = [
        {"actual": round(float(a), 4), "predicted": round(float(p), 4)}
        for a, p in zip(y_test[:500], y_pred[:500])  # Sample for plotting
    ]
    with open(predictions_path, "w") as f:
        json.dump(pred_records, f, indent=2)

    # Record provenance
    eval_hash = compute_file_hash(eval_path)
    recorder.add_output("evaluation_metrics", str(eval_path), eval_hash)
    recorder.complete(metrics=metrics)
    recorder.save_prov("provenance/evaluate_prov.json")

    print(f"\nEvaluation Results:")
    print(f"  MAE:  {metrics['mae']:.4f} "
          f"({metrics['mae_ci_lower']:.4f}, {metrics['mae_ci_upper']:.4f})")
    print(f"  RMSE: {metrics['rmse']:.4f}")
    print(f"  R2:   {metrics['r2']:.4f}")
    print(f"  Test samples: {metrics['n_test_samples']}")


if __name__ == "__main__":
    main()

6. The Registry Utilities Module

The three pipeline stages share a set of provenance utilities: hash computation, Git state capture, and the PROV recorder. These live in src/registry.py.

"""src/registry.py: Shared provenance utilities for pipeline stages."""
import hashlib
import json
import subprocess
import random
import numpy as np
import torch
from pathlib import Path
from datetime import datetime, timezone
from typing import Optional


def compute_file_hash(file_path, algorithm: str = "sha256") -> str:
    """Compute cryptographic hash of a file, streaming for large files."""
    hasher = hashlib.new(algorithm)
    with open(file_path, "rb") as f:
        while chunk := f.read(8192):
            hasher.update(chunk)
    return hasher.hexdigest()


def get_git_info() -> dict:
    """Capture current Git state for provenance."""
    def git(args):
        r = subprocess.run(
            ["git"] + args, capture_output=True, text=True, check=True
        )
        return r.stdout.strip()

    status = git(["status", "--porcelain"])
    return {
        "commit_hash": git(["rev-parse", "HEAD"]),
        "branch": git(["rev-parse", "--abbrev-ref", "HEAD"]),
        "is_dirty": len(status) > 0,
        "dirty_files": status if status else None,
    }


def set_reproducibility_seed(seed: int):
    """Set random seeds for reproducibility across all frameworks."""
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)
        # Note: setting deterministic mode reduces performance
        torch.backends.cudnn.deterministic = True
        torch.backends.cudnn.benchmark = False


def verify_artifact_chain(provenance_dir: str = "provenance") -> dict:
    """Verify the integrity of the entire provenance chain.

    Checks that every output artifact recorded in provenance
    still matches its recorded hash on disk.
    """
    prov_dir = Path(provenance_dir)
    results = {"verified": [], "failed": [], "missing": []}

    for prov_file in sorted(prov_dir.glob("*_prov.json")):
        with open(prov_file) as f:
            prov = json.load(f)

        stage = prov.get("stage_name", prov_file.stem)

        for name, info in prov.get("outputs", {}).items():
            path = info.get("path")
            expected_hash = info.get("hash")

            if not path or not expected_hash:
                continue

            if not Path(path).exists():
                results["missing"].append({
                    "stage": stage,
                    "artifact": name,
                    "path": path,
                })
                continue

            actual_hash = compute_file_hash(path)
            if actual_hash == expected_hash:
                results["verified"].append({
                    "stage": stage,
                    "artifact": name,
                    "hash": actual_hash[:12],
                })
            else:
                results["failed"].append({
                    "stage": stage,
                    "artifact": name,
                    "expected": expected_hash[:12],
                    "actual": actual_hash[:12],
                })

    return results


class ProvenanceRecorder:
    """Records provenance for a single pipeline stage.

    Creates a JSON document compatible with conversion to W3C PROV
    (see Section 47.2's openlineage_to_prov converter).
    """

    def __init__(self, stage_name: str):
        self.stage_name = stage_name
        self.record = {
            "stage_name": stage_name,
            "inputs": {},
            "outputs": {},
            "started_at": None,
            "completed_at": None,
            "git_commit": None,
            "config_hash": None,
            "metrics": {},
            "status": "initialized",
        }

    def start(self, git_commit: str, config_hash: str):
        """Record the start of a pipeline stage."""
        self.record["started_at"] = datetime.now(timezone.utc).isoformat()
        self.record["git_commit"] = git_commit
        self.record["config_hash"] = config_hash
        self.record["status"] = "running"

    def add_input(self, name: str, path: str, content_hash: str):
        """Record an input artifact with its hash."""
        self.record["inputs"][name] = {
            "path": path,
            "hash": content_hash,
            "recorded_at": datetime.now(timezone.utc).isoformat(),
        }

    def add_output(self, name: str, path: str, content_hash: str):
        """Record an output artifact with its hash."""
        self.record["outputs"][name] = {
            "path": path,
            "hash": content_hash,
            "recorded_at": datetime.now(timezone.utc).isoformat(),
        }

    def complete(self, metrics: Optional[dict] = None):
        """Record successful completion of the stage."""
        self.record["completed_at"] = datetime.now(timezone.utc).isoformat()
        self.record["status"] = "completed"
        if metrics:
            self.record["metrics"] = metrics

    def save_prov(self, path: str):
        """Save the provenance record as JSON."""
        out_path = Path(path)
        out_path.parent.mkdir(parents=True, exist_ok=True)
        with open(out_path, "w") as f:
            json.dump(self.record, f, indent=2)
        print(f"Provenance saved: {path}")

7. Running the Pipeline

With all stages defined and the DVC pipeline configured, running the complete pipeline is a single command. DVC handles dependency resolution, selective re-execution, and artifact caching automatically.

# Run the entire pipeline
dvc repro

# DVC output shows the execution plan:
# Running stage 'preprocess':
# > python -m src.preprocess
# Loaded 47665 raw crystal records
# Extracted features: 47523 records (142 dropped)
# Preprocessing complete:
#   Train: 38018 records [d4e5f6a7b8c9...]
#   Test:  9505 records  [e5f6a7b8c9d0...]
#
# Running stage 'train':
# > python -m src.train
# Training on cuda
# Epoch 10: train=0.048231, val=0.052117
# ...
# Early stopping at epoch 87
# Training complete (MLflow run: a1b2c3d4...)
#
# Running stage 'evaluate':
# > python -m src.evaluate
# Test data integrity: VERIFIED [e5f6a7b8c9d0...]
# Model integrity: VERIFIED [c3b2a1f6e5d4...]
# Evaluation Results:
#   MAE:  0.0423 (0.0398, 0.0449)
#   RMSE: 0.0612
#   R2:   0.9387

# After pipeline runs, DVC creates dvc.lock with frozen hashes
cat dvc.lock | head -30

# Verify the entire provenance chain after a run
python -c "
from src.registry import verify_artifact_chain
results = verify_artifact_chain()
print(f'Verified: {len(results[\"verified\"])} artifacts')
print(f'Failed:   {len(results[\"failed\"])} artifacts')
print(f'Missing:  {len(results[\"missing\"])} artifacts')
for v in results['verified']:
    print(f'  OK: {v[\"stage\"]}/{v[\"artifact\"]} [{v[\"hash\"]}...]')
for f in results['failed']:
    print(f'  FAIL: {f[\"stage\"]}/{f[\"artifact\"]} '
          f'expected={f[\"expected\"]}... actual={f[\"actual\"]}...')
"

8. Reproducing Historical Experiments

The entire point of this infrastructure is reproducibility. Given a specific experiment (identified by its Git commit), we can reconstruct the exact state of code, data, config, and environment, then re-execute the pipeline:

"""Reproduce a historical experiment from its provenance record."""
import subprocess
import json
from pathlib import Path

from src.registry import compute_file_hash


def reproduce_experiment(git_ref: str, verify: bool = True):
    """Reproduce an experiment from a specific Git commit.

    Steps:
    1. Checkout the code at the specified commit
    2. Pull the corresponding data versions from DVC
    3. Re-run the pipeline
    4. Compare output hashes against the original provenance
    """
    print(f"Reproducing experiment at {git_ref[:8]}...")

    # Step 1: Checkout code (and DVC pointer files)
    subprocess.run(["git", "checkout", git_ref], check=True)
    print(f"  Code checked out at {git_ref[:8]}")

    # Step 2: Pull data versions that match the DVC pointers
    subprocess.run(["dvc", "checkout"], check=True)
    subprocess.run(["dvc", "pull"], check=True)
    print("  Data versions restored")

    # Step 3: Re-run the pipeline
    result = subprocess.run(
        ["dvc", "repro"],
        capture_output=True, text=True,
    )
    print(f"  Pipeline {'completed' if result.returncode == 0 else 'FAILED'}")

    if result.returncode != 0:
        print(f"  Error: {result.stderr}")
        return False

    # Step 4: Verify outputs match original provenance
    if verify:
        prov_dir = Path("provenance")
        if not prov_dir.exists():
            print("  No provenance records found; skipping verification")
            return True

        all_match = True
        for prov_file in sorted(prov_dir.glob("*_prov.json")):
            with open(prov_file) as f:
                original_prov = json.load(f)

            stage = original_prov["stage_name"]
            for name, info in original_prov.get("outputs", {}).items():
                path = info["path"]
                original_hash = info["hash"]
                current_hash = compute_file_hash(path)

                if current_hash == original_hash:
                    print(f"  MATCH: {stage}/{name} [{current_hash[:12]}...]")
                else:
                    print(f"  MISMATCH: {stage}/{name}")
                    print(f"    Original:  {original_hash[:16]}...")
                    print(f"    Reproduced: {current_hash[:16]}...")
                    all_match = False

        if all_match:
            print("\n  REPRODUCTION SUCCESSFUL: all artifacts match")
        else:
            print("\n  REPRODUCTION PARTIAL: some artifacts differ")
            print("  (Differences may be due to GPU non-determinism)")

        return all_match

    return True


# Reproduce the experiment from commit abc123
# reproduce_experiment("abc123def456")

9. Discovery Workbench Integration

With a clear reproducibility standard in place, the remaining challenge is making the registry accessible to every member of the team through a shared interface. The experiment registry integrates with the Discovery Workbench (introduced in Chapter 6) as a core infrastructure service. The Workbench uses the registry to track all experiments initiated through its interface, enabling researchers to browse experiment histories, compare runs, and reproduce results directly from the Workbench UI.

"""Discovery Workbench experiment registry integration."""
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
import json

from src.registry import compute_file_hash, get_git_info


@dataclass
class WorkbenchExperiment:
    """A complete experiment record for the Discovery Workbench."""
    experiment_id: str
    name: str
    description: str
    created_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )
    git_commit: str = ""
    config_hashes: dict = field(default_factory=dict)
    data_hashes: dict = field(default_factory=dict)
    model_hashes: dict = field(default_factory=dict)
    metrics: dict = field(default_factory=dict)
    provenance_files: list = field(default_factory=list)
    tags: dict = field(default_factory=dict)
    status: str = "created"


class WorkbenchRegistry:
    """Experiment registry for the Discovery Workbench.

    Provides a high-level API for registering, querying, and
    reproducing experiments through the Workbench interface.
    """

    def __init__(self, registry_path: str = "workbench_registry.json"):
        self.registry_path = Path(registry_path)
        self.experiments: dict[str, WorkbenchExperiment] = {}
        if self.registry_path.exists():
            self._load()

    def _load(self):
        """Load the registry from disk."""
        with open(self.registry_path) as f:
            data = json.load(f)
        for exp_id, exp_data in data.items():
            self.experiments[exp_id] = WorkbenchExperiment(**exp_data)

    def _save(self):
        """Persist the registry to disk."""
        data = {
            exp_id: asdict(exp)
            for exp_id, exp in self.experiments.items()
        }
        with open(self.registry_path, "w") as f:
            json.dump(data, f, indent=2)

    def register_experiment(
        self,
        name: str,
        description: str,
        tags: dict = None,
    ) -> WorkbenchExperiment:
        """Register a new experiment in the Workbench."""
        git_info = get_git_info()
        exp_id = f"exp_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}"

        experiment = WorkbenchExperiment(
            experiment_id=exp_id,
            name=name,
            description=description,
            git_commit=git_info["commit_hash"],
            tags=tags or {},
        )
        self.experiments[exp_id] = experiment
        self._save()
        return experiment

    def record_pipeline_run(
        self,
        experiment_id: str,
        provenance_dir: str = "provenance",
    ):
        """Record a completed pipeline run in the experiment."""
        exp = self.experiments[experiment_id]
        prov_dir = Path(provenance_dir)

        for prov_file in sorted(prov_dir.glob("*_prov.json")):
            with open(prov_file) as f:
                prov = json.load(f)

            exp.provenance_files.append(str(prov_file))

            # Aggregate hashes from all stages
            for name, info in prov.get("inputs", {}).items():
                if "data" in name:
                    exp.data_hashes[name] = info["hash"]
            for name, info in prov.get("outputs", {}).items():
                if "model" in name:
                    exp.model_hashes[name] = info["hash"]
                if "data" in name:
                    exp.data_hashes[name] = info["hash"]

            # Collect metrics
            exp.metrics.update(prov.get("metrics", {}))

        exp.status = "completed"
        self._save()

    def compare_experiments(
        self, exp_id_a: str, exp_id_b: str
    ) -> dict:
        """Compare two experiments: what changed, what improved?"""
        a = self.experiments[exp_id_a]
        b = self.experiments[exp_id_b]

        # Find data differences
        data_changed = {
            k: (a.data_hashes.get(k, "N/A"), b.data_hashes.get(k, "N/A"))
            for k in set(a.data_hashes) | set(b.data_hashes)
            if a.data_hashes.get(k) != b.data_hashes.get(k)
        }

        # Find metric differences
        metric_deltas = {}
        for key in set(a.metrics) | set(b.metrics):
            val_a = a.metrics.get(key)
            val_b = b.metrics.get(key)
            if isinstance(val_a, (int, float)) and isinstance(val_b, (int, float)):
                metric_deltas[key] = {
                    "a": val_a,
                    "b": val_b,
                    "delta": round(val_b - val_a, 6),
                }

        return {
            "experiments": (exp_id_a, exp_id_b),
            "data_changed": data_changed,
            "model_changed": a.model_hashes != b.model_hashes,
            "git_commits": (a.git_commit[:8], b.git_commit[:8]),
            "metric_deltas": metric_deltas,
        }

    def list_experiments(
        self, tag_filter: dict = None
    ) -> list[WorkbenchExperiment]:
        """List experiments, optionally filtered by tags."""
        results = list(self.experiments.values())
        if tag_filter:
            results = [
                exp for exp in results
                if all(
                    exp.tags.get(k) == v
                    for k, v in tag_filter.items()
                )
            ]
        return sorted(results, key=lambda e: e.created_at, reverse=True)


# Example usage within the Discovery Workbench
registry = WorkbenchRegistry()

# Register and run an experiment
exp = registry.register_experiment(
    name="Crystal bandgap prediction v12",
    description="Transformer with 6 layers, lr=0.0005",
    tags={"domain": "materials", "target": "bandgap"},
)
print(f"Registered: {exp.experiment_id}")

# After running the pipeline (dvc repro)...
registry.record_pipeline_run(exp.experiment_id)

# Compare with a previous experiment
# comparison = registry.compare_experiments("exp_20260414_103000", exp.experiment_id)

10. Putting It All Together: The Complete Workflow

Here is the end-to-end workflow that a researcher follows when using the provenance-complete pipeline:

1. Configure: edit configs/train.yaml with new hyperparameters.
2. Commit: git add configs/ && git commit -m "Increase layers to 8"
3. Run: dvc repro (DVC skips unchanged stages, runs only what is needed).
4. Review: dvc metrics show and mlflow ui to inspect results.
5. Compare: dvc metrics diff against the previous commit.
6. Version: git add dvc.lock outputs/metrics/ provenance/ && git commit
7. Push: git push && dvc push to share code and data.
8. Reproduce: any collaborator runs git pull && dvc pull && dvc repro.

Every step maintains the artifact DAG, verifies hashes, and generates PROV records. The cost is modest: roughly 20 lines of instrumentation per stage (the ProvenanceRecorder calls). In return, every result traces back to its raw inputs, verifies through hash comparison, and reproduces through pipeline replay.

11. Advanced Topics

Distributed Provenance

When experiments span multiple machines (local development, cloud training, cluster evaluation), you must assemble the provenance chain from distributed records. The OpenLineage event model from Section 47.2 handles this without extra plumbing because its event-based architecture already expects multi-source ingestion: each machine emits events to a centralized lineage backend (Marquez, an open source metadata service that collects and serves OpenLineage events, Atlan, or a custom collector), and the backend assembles the complete run graph. The one requirement: every event must include the same run_id so the backend can link distributed stages into a coherent pipeline execution.

Provenance for LLM Experiments

Large language model experiments introduce additional provenance challenges. The "data" artifact is often a prompt template, a set of few-shot examples, or a system message, none of which fit the traditional "dataset on disk" model. The "model" is often an API endpoint (GPT-4, Claude) whose weights you cannot access for hashing. Record four elements for each LLM experiment: the exact prompt text (hashed for content addressing), the API model identifier and version, the sampling parameters (temperature, top-p), and the raw API response (including token counts and finish reason). The LLMOps infrastructure from Chapter 22 provides the tracking layer; the provenance standards from this chapter provide the interchange format.

Regulatory Compliance

Regulated domains (clinical trials, financial modeling, safety-critical AI) require provenance that meets specific standards: Food and Drug Administration (FDA) 21 CFR Part 11 (the FDA regulation governing electronic records and electronic signatures, requiring audit trails, access controls, and tamper-evident storage) for electronic records, General Data Protection Regulation (GDPR) Article 22 for automated decision-making, and the European Union (EU) AI Act for high-risk AI systems. The tamper-evident provenance chain from Section 47.2 provides the foundation. For full compliance, add: (1) digital signatures on provenance documents (not just hash chains), (2) external timestamping through a trusted authority, (3) role-based access controls on the registry, and (4) audit log retention policies that match your regulatory timeline (typically 10+ years for clinical data).