7장: 메타데이터 및 버전 관리

모델 계보, 재현성 및 거버넌스

弘益人間 · 널리 인간을 이롭게 하라

7.1 메타데이터의 중요성

메타데이터는 모델에 대한 데이터입니다 - 모델이 어떻게 생성되었는지, 무엇을 하는지, 어떻게 사용하는지를 설명하는 정보입니다. 적절한 메타데이터는 재현성을 가능하게 하고, 협업을 촉진하며, 규정 준수를 보장하고, 프로덕션 환경에서 모델 거버넌스를 지원합니다.

7.2 포괄적 메타데이터 스키마

신경망 모델을 위한 잘 구조화된 메타데이터 스키마:

{
  "model_info": {
    "name": "ResNet50-ImageNet",
    "version": "2.1.0",
    "description": "ImageNet-1K 데이터셋으로 훈련된 ResNet-50",
    "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 연구팀",
      "email": "research@company.com",
      "organization": "회사 AI 연구소"
    }
  ],

  "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
    }
  },

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

7.3 모델 버전 관리 전략

모델 버전에 시맨틱 버저닝(SemVer)을 사용하세요:

시맨틱 버저닝 형식

MAJOR.MINOR.PATCH

예시:
  1.0.0 → 초기 릴리스
  1.1.0 → 새로운 기능 (데이터 증강 추가)
  1.1.1 → 버그 수정 (전처리 수정)
  2.0.0 → 호환성 없는 변경 (다른 입력 형식)

규칙:
  MAJOR: 호환되지 않는 API/아키텍처 변경
  MINOR: 하위 호환되는 기능 추가
  PATCH: 하위 호환되는 버그 수정

버전 관리 구현

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, ensure_ascii=False)

    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

# 사용 예
registry = ModelRegistry()
version = ModelVersion(1, 0, 0)

metadata = {
    "description": "초기 ResNet50 모델",
    "accuracy": 0.761,
    "model_path": "models/resnet50_v1.0.0.pth"
}

registry.register_model("ResNet50", version, metadata)
print(f"최신 버전: {registry.get_latest_version('ResNet50')}")

7.4 모델 카드

모델 카드는 Google의 모델 카드 프레임워크를 따르는 표준화된 문서를 제공합니다:

# 모델 카드: ResNet50-ImageNet

## 모델 세부사항
- **개발자:** 회사 AI 연구소
- **모델 날짜:** 2025년 1월
- **모델 버전:** 2.1.0
- **모델 유형:** 합성곱 신경망 (CNN)
- **라이선스:** Apache 2.0

## 의도된 사용
- **주요 사용 목적:** 1000개 ImageNet 클래스의 이미지 분류
- **주요 사용자:** 연구자, 개발자, 기업
- **범위 외 사용:** 의료 진단, 감시, 해악

## 훈련 데이터
- 데이터셋: ImageNet-1K (ILSVRC 2012)
- 128만 훈련 이미지, 1000개 클래스
- 데이터 증강: 랜덤 크롭, 수평 뒤집기, 색상 지터

## 평가 데이터
- ImageNet-1K 검증 세트 (5만 이미지)
- 훈련과 동일한 전처리

## 성능
| 메트릭 | 값 |
|--------|-------|
| Top-1 정확도 | 76.1% |
| Top-5 정확도 | 93.1% |
| 추론 (GPU) | 3.8ms |

## 한계
- ImageNet 클래스로만 훈련됨
- 분포 외 이미지에서 성능 저하
- ImageNet 데이터셋의 편향 포함 가능

## 윤리적 고려사항
- ImageNet으로 훈련된 모델은 클래스 불균형 존재
- 얼굴 인식이나 감시에 사용 금지
- 편향 평가 완료 (규정 준수 문서 참조)

## 주의사항 및 권장사항
- 최상의 결과를 위해 도메인별 데이터로 미세 조정
- 프로덕션에서 분포 이동 모니터링
- 정기적 재훈련 권장 (분기별)

7.5 모델 계보 추적

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):
        """이 모델 버전의 고유 지문 계산"""
        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, ensure_ascii=False)

    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):
        """전체 조상 체인 가져오기"""
        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

# 사용 예
tracker = LineageTracker()

# 초기 모델 기록
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=["ImageNet에서 초기 훈련"]
)
tracker.record(initial)

# 미세 조정된 버전 기록
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=["도메인별 데이터로 미세 조정", "데이터 증강 추가"]
)
tracker.record(finetuned)

# 조상 가져오기
ancestry = tracker.get_ancestry("resnet50", "1.1.0")
for i, lineage in enumerate(ancestry):
    print(f"세대 {i}: {lineage.version} by {lineage.created_by}")

7.6 실험 추적 통합

MLflow 통합

import mlflow
import mlflow.pytorch

# 실험 시작
mlflow.set_experiment("resnet50-imagenet")

with mlflow.start_run(run_name="v2.1.0"):
    # 파라미터 로깅
    mlflow.log_param("batch_size", 256)
    mlflow.log_param("learning_rate", 0.1)
    mlflow.log_param("optimizer", "SGD")

    # 훈련 루프
    for epoch in range(num_epochs):
        train_loss = train_one_epoch()
        val_acc = validate()

        # 메트릭 로깅
        mlflow.log_metric("train_loss", train_loss, step=epoch)
        mlflow.log_metric("val_accuracy", val_acc, step=epoch)

    # 모델 로깅
    mlflow.pytorch.log_model(
        model,
        "model",
        registered_model_name="ResNet50"
    )

    # 아티팩트 로깅
    mlflow.log_artifact("config.yaml")
    mlflow.log_artifact("training_log.txt")

    # 태그 설정
    mlflow.set_tag("version", "2.1.0")
    mlflow.set_tag("task", "image_classification")
    mlflow.set_tag("status", "production")

7.7 모델 파일에 메타데이터 포함

PyTorch

import torch

# 메타데이터와 함께 모델 저장
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')

# 로드 및 메타데이터 액세스
checkpoint = torch.load('model_with_metadata.pth')
print(f"모델 버전: {checkpoint['metadata']['version']}")
print(f"정확도: {checkpoint['metadata']['accuracy']}")

ONNX

import onnx
from onnx import helper

# 모델 로드
model = onnx.load("model.onnx")

# 메타데이터 추가 (doc_string 사용)
model.doc_string = """
모델: ResNet50
버전: 2.1.0
정확도: 76.1%
데이터셋: ImageNet-1K
생성일: 2025-01-15
"""

# 저장
onnx.save(model, "model_with_metadata.onnx")

7.8 데이터 버전 관리

import hashlib
from pathlib import Path

class DataVersioning:
    """데이터셋 버전 관리"""

    @staticmethod
    def hash_dataset(dataset_path):
        """전체 데이터셋 해시 계산"""
        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):
        """파일 목록 해시 계산 (대용량 데이터셋용)"""
        hasher = hashlib.sha256()
        for path in sorted(file_list):
            hasher.update(str(path).encode())
        return hasher.hexdigest()

# 모델과 함께 데이터 버전 추적
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 재현성 체크리스트

7.10 거버넌스 및 규정 준수

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()
        }

        # 전체 상태 업데이트
        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