Model Lineage, Reproducibility, and Governance
弘益人間 · Benefit All Humanity
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.
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"
}
}
Use semantic versioning (SemVer) for model versions:
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
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')}")
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)
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}")
Integrate with MLflow, Weights & Biases, or TensorBoard:
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")
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()
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']}")
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")
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"
}
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