모델 계보, 재현성 및 거버넌스
弘益人間 · 널리 인간을 이롭게 하라
메타데이터는 모델에 대한 데이터입니다 - 모델이 어떻게 생성되었는지, 무엇을 하는지, 어떻게 사용하는지를 설명하는 정보입니다. 적절한 메타데이터는 재현성을 가능하게 하고, 협업을 촉진하며, 규정 준수를 보장하고, 프로덕션 환경에서 모델 거버넌스를 지원합니다.
신경망 모델을 위한 잘 구조화된 메타데이터 스키마:
{
"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"]
}
}모델 버전에 시맨틱 버저닝(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')}")모델 카드는 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으로 훈련된 모델은 클래스 불균형 존재
- 얼굴 인식이나 감시에 사용 금지
- 편향 평가 완료 (규정 준수 문서 참조)
## 주의사항 및 권장사항
- 최상의 결과를 위해 도메인별 데이터로 미세 조정
- 프로덕션에서 분포 이동 모니터링
- 정기적 재훈련 권장 (분기별)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}")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")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']}")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")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"
}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