Chapter 7: Metadata and Versioning

Model Lineage, Reproducibility, and Governance

弘益人間 · Benefit All Humanity

7.1 The Importance of Metadata

Metadata is data about your model - information that describes how it was created, what it does, and how to use it. Proper metadata enables reproducibility, facilitates collaboration, ensures compliance, and supports model governance in production environments.

7.2 Comprehensive Metadata Schema

A well-structured metadata schema for neural network models:

{
  "model_info": {
    "name": "ResNet50-ImageNet",
    "version": "2.1.0",
    "description": "ResNet-50 trained on ImageNet-1K dataset",
    "task": "image_classification",
    "domain": "computer_vision",
    "license": "Apache-2.0",
    "created_date": "2025-01-15T10:30:00Z",
    "modified_date": "2025-01-20T14:22:00Z"
  },

  "authors": [
    {
      "name": "AI Research Team",
      "email": "research@company.com",
      "organization": "Company AI Lab"
    }
  ],

  "training": {
    "framework": "pytorch",
    "framework_version": "2.1.0",
    "dataset": {
      "name": "ImageNet-1K",
      "version": "2012",
      "num_samples": 1281167,
      "num_classes": 1000,
      "split": "train",
      "preprocessing": "resize_224_normalize"
    },
    "hyperparameters": {
      "batch_size": 256,
      "learning_rate": 0.1,
      "optimizer": "SGD",
      "momentum": 0.9,
      "weight_decay": 0.0001,
      "epochs": 90,
      "lr_schedule": "step_decay",
      "augmentation": ["random_crop", "horizontal_flip"]
    },
    "hardware": {
      "gpus": "8x NVIDIA V100",
      "training_time_hours": 48
    }
  },

  "performance": {
    "metrics": {
      "top1_accuracy": 0.761,
      "top5_accuracy": 0.931,
      "loss": 0.932
    },
    "evaluation_dataset": "ImageNet-1K val",
    "inference_latency_ms": {
      "batch_1_cpu": 45.2,
      "batch_1_gpu": 3.8,
      "batch_32_gpu": 28.5
    }
  },

  "model_architecture": {
    "layers": 50,
    "parameters": 25557032,
    "trainable_parameters": 25557032,
    "flops": 4089184256,
    "memory_mb": 97.5
  },

  "input_output": {
    "inputs": [
      {
        "name": "image",
        "shape": [null, 3, 224, 224],
        "dtype": "float32",
        "preprocessing": {
          "normalize": true,
          "mean": [0.485, 0.456, 0.406],
          "std": [0.229, 0.224, 0.225]
        }
      }
    ],
    "outputs": [
      {
        "name": "logits",
        "shape": [null, 1000],
        "dtype": "float32",
        "postprocessing": "softmax"
      }
    ],
    "labels_file": "imagenet_labels.txt"
  },

  "deployment": {
    "target_platforms": ["cloud", "edge", "mobile"],
    "min_memory_mb": 256,
    "recommended_hardware": "GPU with 4GB+ VRAM",
    "serving_framework": ["torchserve", "tensorflow_serving", "onnx_runtime"]
  },

  "provenance": {
    "base_model": "ResNet50-v1.5",
    "fine_tuned": false,
    "quantized": false,
    "pruned": false,
    "modifications": [
      {
        "date": "2025-01-20",
        "type": "bug_fix",
        "description": "Fixed batch normalization inference mode"
      }
    ]
  },

  "compliance": {
    "approved": true,
    "approval_date": "2025-01-22",
    "approver": "model_governance_team",
    "risk_level": "low",
    "bias_assessment": "completed",
    "privacy_review": "passed"
  },

  "references": {
    "paper": "https://arxiv.org/abs/1512.03385",
    "code": "https://github.com/pytorch/vision",
    "documentation": "https://pytorch.org/vision/models.html"
  }
}

7.3 Model Versioning Strategies

Use semantic versioning (SemVer) for model versions:

Semantic Versioning Format

MAJOR.MINOR.PATCH

Examples:
  1.0.0 → Initial release
  1.1.0 → New feature (added data augmentation)
  1.1.1 → Bug fix (fixed preprocessing)
  2.0.0 → Breaking change (different input format)

Rules:
  MAJOR: Incompatible API/architecture changes
  MINOR: Backward-compatible functionality additions
  PATCH: Backward-compatible bug fixes

Version Control Implementation

import json
from pathlib import Path
from datetime import datetime

class ModelVersion:
    def __init__(self, major, minor, patch):
        self.major = major
        self.minor = minor
        self.patch = patch

    def __str__(self):
        return f"{self.major}.{self.minor}.{self.patch}"

    def bump_major(self):
        return ModelVersion(self.major + 1, 0, 0)

    def bump_minor(self):
        return ModelVersion(self.major, self.minor + 1, 0)

    def bump_patch(self):
        return ModelVersion(self.major, self.minor, self.patch + 1)

class ModelRegistry:
    def __init__(self, registry_path="models/registry.json"):
        self.registry_path = Path(registry_path)
        self.load()

    def load(self):
        if self.registry_path.exists():
            with open(self.registry_path) as f:
                self.data = json.load(f)
        else:
            self.data = {"models": {}}

    def save(self):
        self.registry_path.parent.mkdir(parents=True, exist_ok=True)
        with open(self.registry_path, 'w') as f:
            json.dump(self.data, f, indent=2)

    def register_model(self, name, version, metadata):
        if name not in self.data["models"]:
            self.data["models"][name] = {"versions": {}}

        version_str = str(version)
        self.data["models"][name]["versions"][version_str] = {
            **metadata,
            "registered_at": datetime.utcnow().isoformat()
        }
        self.save()

    def get_latest_version(self, name):
        if name not in self.data["models"]:
            return None

        versions = self.data["models"][name]["versions"].keys()
        latest = max(versions, key=lambda v: tuple(map(int, v.split('.'))))
        return latest

# Usage
registry = ModelRegistry()
version = ModelVersion(1, 0, 0)

metadata = {
    "description": "Initial ResNet50 model",
    "accuracy": 0.761,
    "model_path": "models/resnet50_v1.0.0.pth"
}

registry.register_model("ResNet50", version, metadata)
print(f"Latest version: {registry.get_latest_version('ResNet50')}")

7.4 Model Cards

Model Cards provide standardized documentation following Google's Model Card framework:

# MODEL CARD: ResNet50-ImageNet

## Model Details
- **Developed by:** Company AI Lab
- **Model date:** January 2025
- **Model version:** 2.1.0
- **Model type:** Convolutional Neural Network (CNN)
- **License:** Apache 2.0

## Intended Use
- **Primary intended uses:** Image classification for 1000 ImageNet classes
- **Primary intended users:** Researchers, developers, enterprises
- **Out-of-scope uses:** Medical diagnosis, surveillance, harm

## Training Data
- Dataset: ImageNet-1K (ILSVRC 2012)
- 1.28M training images, 1000 classes
- Data augmentation: random crop, horizontal flip, color jitter

## Evaluation Data
- ImageNet-1K validation set (50K images)
- Same preprocessing as training

## Performance
| Metric | Value |
|--------|-------|
| Top-1 Accuracy | 76.1% |
| Top-5 Accuracy | 93.1% |
| Inference (GPU) | 3.8ms |

## Limitations
- Trained only on ImageNet classes
- Performance degrades on out-of-distribution images
- May have biases present in ImageNet dataset

## Ethical Considerations
- Model trained on ImageNet which has known class imbalances
- Should not be used for facial recognition or surveillance
- Bias assessment completed (see compliance documentation)

## Caveats and Recommendations
- Fine-tune on domain-specific data for best results
- Monitor for distribution shift in production
- Regular retraining recommended (quarterly)

7.5 Model Lineage Tracking

Track the complete history of model development:

import hashlib
import json
from dataclasses import dataclass, asdict
from typing import List, Optional

@dataclass
class ModelLineage:
    model_id: str
    version: str
    parent_model_id: Optional[str]
    parent_version: Optional[str]
    training_data_hash: str
    code_commit_hash: str
    created_at: str
    created_by: str
    modifications: List[str]

    def compute_fingerprint(self):
        """Compute unique fingerprint for this model version"""
        data = f"{self.model_id}{self.version}{self.training_data_hash}"
        return hashlib.sha256(data.encode()).hexdigest()

class LineageTracker:
    def __init__(self, storage_path="lineage.json"):
        self.storage_path = storage_path
        self.lineage_db = {}
        self.load()

    def load(self):
        try:
            with open(self.storage_path) as f:
                data = json.load(f)
                self.lineage_db = {
                    k: ModelLineage(**v) for k, v in data.items()
                }
        except FileNotFoundError:
            pass

    def save(self):
        with open(self.storage_path, 'w') as f:
            data = {k: asdict(v) for k, v in self.lineage_db.items()}
            json.dump(data, f, indent=2)

    def record(self, lineage: ModelLineage):
        key = f"{lineage.model_id}:{lineage.version}"
        self.lineage_db[key] = lineage
        self.save()

    def get_ancestry(self, model_id: str, version: str):
        """Get full ancestry chain"""
        ancestry = []
        current_id, current_version = model_id, version

        while current_id:
            key = f"{current_id}:{current_version}"
            if key not in self.lineage_db:
                break

            lineage = self.lineage_db[key]
            ancestry.append(lineage)

            current_id = lineage.parent_model_id
            current_version = lineage.parent_version

        return ancestry

# Usage
tracker = LineageTracker()

# Record initial model
initial = ModelLineage(
    model_id="resnet50",
    version="1.0.0",
    parent_model_id=None,
    parent_version=None,
    training_data_hash="a1b2c3...",
    code_commit_hash="xyz789",
    created_at="2025-01-15T10:00:00Z",
    created_by="researcher@company.com",
    modifications=["Initial training on ImageNet"]
)
tracker.record(initial)

# Record fine-tuned version
finetuned = ModelLineage(
    model_id="resnet50",
    version="1.1.0",
    parent_model_id="resnet50",
    parent_version="1.0.0",
    training_data_hash="d4e5f6...",
    code_commit_hash="abc123",
    created_at="2025-01-20T14:00:00Z",
    created_by="engineer@company.com",
    modifications=["Fine-tuned on domain-specific data", "Added data augmentation"]
)
tracker.record(finetuned)

# Get ancestry
ancestry = tracker.get_ancestry("resnet50", "1.1.0")
for i, lineage in enumerate(ancestry):
    print(f"Generation {i}: {lineage.version} by {lineage.created_by}")

7.6 Experiment Tracking Integration

Integrate with MLflow, Weights & Biases, or TensorBoard:

MLflow Integration

import mlflow
import mlflow.pytorch

# Start experiment
mlflow.set_experiment("resnet50-imagenet")

with mlflow.start_run(run_name="v2.1.0"):
    # Log parameters
    mlflow.log_param("batch_size", 256)
    mlflow.log_param("learning_rate", 0.1)
    mlflow.log_param("optimizer", "SGD")

    # Training loop
    for epoch in range(num_epochs):
        train_loss = train_one_epoch()
        val_acc = validate()

        # Log metrics
        mlflow.log_metric("train_loss", train_loss, step=epoch)
        mlflow.log_metric("val_accuracy", val_acc, step=epoch)

    # Log model
    mlflow.pytorch.log_model(
        model,
        "model",
        registered_model_name="ResNet50"
    )

    # Log artifacts
    mlflow.log_artifact("config.yaml")
    mlflow.log_artifact("training_log.txt")

    # Set tags
    mlflow.set_tag("version", "2.1.0")
    mlflow.set_tag("task", "image_classification")
    mlflow.set_tag("status", "production")

Weights & Biases Integration

import wandb

# Initialize run
wandb.init(
    project="resnet50-imagenet",
    name="v2.1.0",
    config={
        "batch_size": 256,
        "learning_rate": 0.1,
        "architecture": "ResNet50"
    },
    tags=["production", "image-classification"]
)

# Training loop
for epoch in range(num_epochs):
    train_loss = train_one_epoch()
    val_acc = validate()

    wandb.log({
        "epoch": epoch,
        "train_loss": train_loss,
        "val_accuracy": val_acc
    })

# Save model
wandb.save("model.pth")

# Log model as artifact
artifact = wandb.Artifact('resnet50', type='model')
artifact.add_file('model.pth')
artifact.metadata = {
    "accuracy": 0.761,
    "framework": "pytorch",
    "version": "2.1.0"
}
wandb.log_artifact(artifact)

wandb.finish()

7.7 Metadata Embedding in Model Files

PyTorch

import torch

# Save model with metadata
torch.save({
    'model_state_dict': model.state_dict(),
    'optimizer_state_dict': optimizer.state_dict(),
    'epoch': epoch,
    'metadata': {
        'version': '2.1.0',
        'accuracy': 0.761,
        'training_data': 'ImageNet-1K',
        'created_at': '2025-01-15',
        'framework_version': torch.__version__
    }
}, 'model_with_metadata.pth')

# Load and access metadata
checkpoint = torch.load('model_with_metadata.pth')
print(f"Model version: {checkpoint['metadata']['version']}")
print(f"Accuracy: {checkpoint['metadata']['accuracy']}")

ONNX

import onnx
from onnx import helper

# Load model
model = onnx.load("model.onnx")

# Add metadata
model.metadata_props.append(helper.make_tensor_value_info(
    "version", onnx.TensorProto.STRING, []
))
model.metadata_props.append(helper.make_tensor_value_info(
    "accuracy", onnx.TensorProto.FLOAT, []
))

# Or use doc_string
model.doc_string = """
Model: ResNet50
Version: 2.1.0
Accuracy: 76.1%
Dataset: ImageNet-1K
"""

# Save
onnx.save(model, "model_with_metadata.onnx")

7.8 Data Versioning

Track dataset versions alongside model versions:

import hashlib
from pathlib import Path

class DataVersioning:
    @staticmethod
    def hash_dataset(dataset_path):
        """Compute hash of entire dataset"""
        hasher = hashlib.sha256()

        for file_path in sorted(Path(dataset_path).rglob('*')):
            if file_path.is_file():
                with open(file_path, 'rb') as f:
                    while chunk := f.read(8192):
                        hasher.update(chunk)

        return hasher.hexdigest()

    @staticmethod
    def hash_file_list(file_list):
        """Compute hash of file list (for large datasets)"""
        hasher = hashlib.sha256()
        for path in sorted(file_list):
            hasher.update(str(path).encode())
        return hasher.hexdigest()

# Track data version with model
data_hash = DataVersioning.hash_dataset("./data/imagenet")

model_metadata = {
    "model_version": "2.1.0",
    "data_version": data_hash,
    "data_source": "ImageNet-1K",
    "data_download_date": "2025-01-10"
}

7.9 Reproducibility Checklist

7.10 Governance and Compliance

Enterprise model governance framework:

class ModelGovernance:
    def __init__(self):
        self.approval_workflow = [
            "technical_review",
            "security_review",
            "bias_assessment",
            "privacy_review",
            "legal_approval"
        ]

    def submit_for_approval(self, model_id, version):
        approval_record = {
            "model_id": model_id,
            "version": version,
            "submitted_at": datetime.utcnow().isoformat(),
            "status": "pending",
            "reviews": {}
        }

        for review_type in self.approval_workflow:
            approval_record["reviews"][review_type] = {
                "status": "pending",
                "reviewer": None,
                "comments": None,
                "approved_at": None
            }

        return approval_record

    def complete_review(self, approval_record, review_type, approved, reviewer, comments):
        approval_record["reviews"][review_type] = {
            "status": "approved" if approved else "rejected",
            "reviewer": reviewer,
            "comments": comments,
            "approved_at": datetime.utcnow().isoformat()
        }

        # Update overall status
        all_approved = all(
            review["status"] == "approved"
            for review in approval_record["reviews"].values()
        )

        if all_approved:
            approval_record["status"] = "approved"
        elif any(review["status"] == "rejected" for review in approval_record["reviews"].values()):
            approval_record["status"] = "rejected"

        return approval_record