8장: 모델 레지스트리와 버전 관리

모델 관리 없이 발생하는 문제:
 
  1. 재현 불가능: 어떤 설정으로 학습했는지 기록이 없음
  2. 추적 불가능: 프로덕션 모델의 학습 이력을 알 수 없음
  3. 롤백 불가능: 이전 버전으로 되돌릴 수 없음
  4. 비교 불가능: 여러 실험의 결과를 체계적으로 비교할 수 없음
  5. 협업 불가능: 팀원 간 모델 공유와 핸드오프가 어려움

모델 레지스트리 핵심 구성 요소:
 
  1. 모델 아티팩트 저장소
     - 모델 가중치, LoRA 어댑터
     - S3, GCS, 로컬 스토리지
 
  2. 메타데이터 저장소
     - 학습 설정, 데이터 정보, 평가 결과
     - 데이터베이스 또는 JSON 파일
 
  3. 버전 관리 시스템
     - 모델 버전 추적, 태그, 스테이지 관리
     - 자동 증가 버전 + 시맨틱 태그
 
  4. 라이프사이클 관리
     - staging, production, archived 상태 전환
     - 승인 프로세스, 롤백 기능

import mlflow
from mlflow.tracking import MlflowClient
 
# MLflow 서버 설정
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("fine-tuning-llama3-8b")
 
# 학습 실행 기록
with mlflow.start_run(run_name="qlora-v1-baseline") as run:
    # 1. 하이퍼파라미터 로깅
    mlflow.log_params({
        "base_model": "meta-llama/Llama-3.1-8B-Instruct",
        "method": "QLoRA",
        "lora_r": 32,
        "lora_alpha": 64,
        "learning_rate": 2e-4,
        "batch_size": 16,
        "epochs": 3,
        "max_seq_length": 2048,
        "dataset_size": 5000,
        "dataset_version": "v2.1",
    })
 
    # 2. 학습 실행
    # trainer.train() ...
 
    # 3. 메트릭 로깅
    mlflow.log_metrics({
        "train_loss": 0.45,
        "eval_loss": 0.52,
        "rouge_l": 0.78,
        "bertscore_f1": 0.85,
    })
 
    # 4. 모델 아티팩트 저장
    mlflow.log_artifacts("./best-adapter", artifact_path="lora-adapter")
 
    # 5. 데이터셋 정보 기록
    mlflow.log_artifact("data/train.jsonl", artifact_path="dataset")
 
    # 6. 평가 보고서 기록
    mlflow.log_artifact(
        "eval_output/eval_results.json",
        artifact_path="evaluation"
    )
 
    print("Run ID: " + run.info.run_id)

from mlflow.tracking import MlflowClient
 
client = MlflowClient()
 
# 모델 등록
model_name = "llama3-8b-code-review"
model_uri = "runs:/" + run_id + "/lora-adapter"
 
# 새 버전으로 등록
mv = mlflow.register_model(model_uri, model_name)
print("등록된 버전: " + str(mv.version))
 
# 버전 설명 추가
client.update_model_version(
    name=model_name,
    version=mv.version,
    description=(
        "QLoRA 학습, r=32, alpha=64, "
        "5000개 코드 리뷰 데이터, "
        "ROUGE-L: 0.78, BERTScore F1: 0.85"
    )
)

# 스테이지 전환: None -> Staging -> Production -> Archived
client.transition_model_version_stage(
    name=model_name,
    version=mv.version,
    stage="Staging"
)
 
# 평가 통과 후 프로덕션으로 승격
client.transition_model_version_stage(
    name=model_name,
    version=mv.version,
    stage="Production"
)
 
# 이전 프로덕션 버전은 자동으로 Archived 처리
# 또는 명시적으로 처리
client.transition_model_version_stage(
    name=model_name,
    version=old_version,
    stage="Archived"
)

모델 라이프사이클 스테이지:
 
  None:
    - 초기 등록 상태
    - 실험 단계의 모델
 
  Staging:
    - 평가 및 검증 중인 모델
    - 프로덕션 배포 전 테스트
 
  Production:
    - 현재 프로덕션에서 서빙 중인 모델
    - 동시에 하나의 버전만 권장
 
  Archived:
    - 더 이상 사용하지 않는 이전 버전
    - 롤백을 위해 보관

from huggingface_hub import HfApi
 
api = HfApi()
 
# LoRA 어댑터 업로드
api.upload_folder(
    folder_path="./best-adapter",
    repo_id="myorg/llama3-8b-code-review-lora",
    repo_type="model",
    commit_message="v1.0: QLoRA r=32, 5000개 데이터 학습",
)
 
# 모델 카드 (README.md) 추가
model_card_content = """
---
license: apache-2.0
base_model: meta-llama/Llama-3.1-8B-Instruct
tags:
  - fine-tuned
  - code-review
  - lora
---
 
# Llama 3.1 8B Code Review (LoRA Adapter)
 
## 학습 정보
- 방법: QLoRA (r=32, alpha=64)
- 데이터: 5,000개 코드 리뷰 예제
- 에포크: 3
 
## 평가 결과
- ROUGE-L: 0.78
- BERTScore F1: 0.85
"""
 
api.upload_file(
    path_or_fileobj=model_card_content.encode(),
    path_in_repo="README.md",
    repo_id="myorg/llama3-8b-code-review-lora",
    repo_type="model",
)

# Git 태그로 버전 관리
api.create_tag(
    repo_id="myorg/llama3-8b-code-review-lora",
    tag="v1.0",
    tag_message="첫 번째 프로덕션 릴리스"
)
 
# 특정 태그의 모델 로드
from peft import PeftModel
from transformers import AutoModelForCausalLM
 
model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-3.1-8B-Instruct",
    device_map="auto"
)
model = PeftModel.from_pretrained(
    model,
    "myorg/llama3-8b-code-review-lora",
    revision="v1.0"  # 특정 버전 지정
)

# model_registry.py
import json
import shutil
from datetime import datetime
from pathlib import Path
from dataclasses import dataclass, asdict
 
 
@dataclass
class ModelVersion:
    version: int
    model_path: str
    created_at: str
    stage: str  # none, staging, production, archived
    base_model: str
    method: str
    training_config: dict
    evaluation_results: dict
    data_info: dict
    description: str
 
 
class LocalModelRegistry:
    """로컬 파일 기반 모델 레지스트리"""
 
    def __init__(self, registry_path: str):
        self.registry_path = Path(registry_path)
        self.registry_path.mkdir(parents=True, exist_ok=True)
        self.metadata_file = self.registry_path / "registry.json"
        self._load_metadata()
 
    def _load_metadata(self):
        if self.metadata_file.exists():
            with open(self.metadata_file) as f:
                self.metadata = json.load(f)
        else:
            self.metadata = {"models": {}}
 
    def _save_metadata(self):
        with open(self.metadata_file, "w") as f:
            json.dump(self.metadata, f, ensure_ascii=False, indent=2)
 
    def register_model(
        self,
        model_name: str,
        adapter_path: str,
        base_model: str,
        method: str,
        training_config: dict,
        evaluation_results: dict,
        data_info: dict,
        description: str = "",
    ) -> ModelVersion:
        """새 모델 버전 등록"""
        if model_name not in self.metadata["models"]:
            self.metadata["models"][model_name] = {
                "versions": [],
                "latest_version": 0
            }
 
        model_info = self.metadata["models"][model_name]
        version = model_info["latest_version"] + 1
 
        # 아티팩트 복사
        dest = self.registry_path / model_name / ("v" + str(version))
        dest.mkdir(parents=True, exist_ok=True)
        shutil.copytree(adapter_path, str(dest / "adapter"), dirs_exist_ok=True)
 
        # 메타데이터 기록
        model_version = ModelVersion(
            version=version,
            model_path=str(dest / "adapter"),
            created_at=datetime.now().isoformat(),
            stage="none",
            base_model=base_model,
            method=method,
            training_config=training_config,
            evaluation_results=evaluation_results,
            data_info=data_info,
            description=description,
        )
 
        model_info["versions"].append(asdict(model_version))
        model_info["latest_version"] = version
        self._save_metadata()
 
        print("등록 완료: " + model_name + " v" + str(version))
        return model_version
 
    def promote_to_production(
        self, model_name: str, version: int
    ):
        """모델을 프로덕션으로 승격"""
        model_info = self.metadata["models"][model_name]
 
        for v in model_info["versions"]:
            if v["stage"] == "production":
                v["stage"] = "archived"
            if v["version"] == version:
                v["stage"] = "production"
 
        self._save_metadata()
        print(model_name + " v" + str(version) + " -> Production")
 
    def get_production_model(self, model_name: str) -> dict:
        """현재 프로덕션 모델 정보 반환"""
        model_info = self.metadata["models"][model_name]
 
        for v in model_info["versions"]:
            if v["stage"] == "production":
                return v
 
        raise ValueError("프로덕션 모델이 없습니다: " + model_name)
 
    def list_versions(self, model_name: str) -> list[dict]:
        """모델의 모든 버전 나열"""
        return self.metadata["models"][model_name]["versions"]
 
    def compare_versions(
        self, model_name: str, v1: int, v2: int
    ) -> str:
        """두 버전의 평가 결과 비교"""
        versions = self.metadata["models"][model_name]["versions"]
        ver1 = next(v for v in versions if v["version"] == v1)
        ver2 = next(v for v in versions if v["version"] == v2)
 
        lines = [
            "버전 비교: v" + str(v1) + " vs v" + str(v2),
            "=" * 40,
        ]
 
        all_metrics = set(
            list(ver1["evaluation_results"].keys())
            + list(ver2["evaluation_results"].keys())
        )
 
        for metric in sorted(all_metrics):
            val1 = ver1["evaluation_results"].get(metric, "N/A")
            val2 = ver2["evaluation_results"].get(metric, "N/A")
            lines.append(
                "  " + metric + ": "
                + str(val1) + " vs " + str(val2)
            )
 
        return "\n".join(lines)

# 레지스트리 초기화
registry = LocalModelRegistry("./model-registry")
 
# 모델 등록
registry.register_model(
    model_name="code-review-assistant",
    adapter_path="./best-adapter",
    base_model="meta-llama/Llama-3.1-8B-Instruct",
    method="QLoRA",
    training_config={
        "lora_r": 32,
        "learning_rate": 2e-4,
        "epochs": 3,
        "dataset_size": 5000,
    },
    evaluation_results={
        "eval_loss": 0.52,
        "rouge_l": 0.78,
        "bertscore_f1": 0.85,
    },
    data_info={
        "train_file": "data/v2/train.jsonl",
        "train_size": 4500,
        "val_size": 500,
    },
    description="QLoRA 베이스라인, 코드 리뷰 5000개 데이터",
)
 
# 프로덕션 승격
registry.promote_to_production("code-review-assistant", version=1)
 
# 프로덕션 모델 조회
prod = registry.get_production_model("code-review-assistant")
print("프로덕션 모델: " + prod["model_path"])

저장 전략 비교:
 
  Full Model 저장:
    크기: 14 GB (7B FP16)
    장점: 독립적으로 사용 가능
    단점: 저장 공간 과다, 업로드/다운로드 시간
 
  LoRA 어댑터만 저장:
    크기: 100~500 MB
    장점: 극적으로 작은 크기, 빠른 전송
    단점: 베이스 모델이 별도로 필요
 
  권장: LoRA 어댑터만 저장 + 베이스 모델 버전 기록

import boto3
from pathlib import Path
 
 
class S3ModelStore:
    """S3 기반 모델 저장소"""
 
    def __init__(self, bucket: str, prefix: str = "models"):
        self.s3 = boto3.client("s3")
        self.bucket = bucket
        self.prefix = prefix
 
    def upload_adapter(
        self,
        local_path: str,
        model_name: str,
        version: int
    ):
        """LoRA 어댑터를 S3에 업로드"""
        s3_prefix = (
            self.prefix + "/" + model_name + "/v" + str(version) + "/"
        )
        local = Path(local_path)
 
        for file_path in local.rglob("*"):
            if file_path.is_file():
                s3_key = s3_prefix + str(file_path.relative_to(local))
                self.s3.upload_file(str(file_path), self.bucket, s3_key)
 
        print("업로드 완료: s3://" + self.bucket + "/" + s3_prefix)
 
    def download_adapter(
        self,
        model_name: str,
        version: int,
        local_path: str
    ):
        """S3에서 LoRA 어댑터 다운로드"""
        s3_prefix = (
            self.prefix + "/" + model_name + "/v" + str(version) + "/"
        )
        local = Path(local_path)
        local.mkdir(parents=True, exist_ok=True)
 
        paginator = self.s3.get_paginator("list_objects_v2")
        for page in paginator.paginate(
            Bucket=self.bucket, Prefix=s3_prefix
        ):
            for obj in page.get("Contents", []):
                key = obj["Key"]
                relative = key[len(s3_prefix):]
                dest = local / relative
                dest.parent.mkdir(parents=True, exist_ok=True)
                self.s3.download_file(self.bucket, key, str(dest))
 
        print("다운로드 완료: " + local_path)

@dataclass
class DataLineage:
    dataset_id: str
    version: str
    source: str
    preprocessing_steps: list[str]
    total_examples: int
    train_examples: int
    val_examples: int
    test_examples: int
    created_at: str
    checksum: str  # 데이터 무결성 확인용
 
 
def compute_dataset_checksum(file_path: str) -> str:
    """데이터셋 체크섬 계산"""
    import hashlib
    sha256 = hashlib.sha256()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            sha256.update(chunk)
    return sha256.hexdigest()
 
 
def record_data_lineage(
    dataset_path: str,
    preprocessing_steps: list[str]
) -> DataLineage:
    """데이터 계보 기록"""
    import json
 
    with open(dataset_path) as f:
        data = [json.loads(line) for line in f]
 
    return DataLineage(
        dataset_id="ds-" + datetime.now().strftime("%Y%m%d-%H%M%S"),
        version="1.0",
        source=dataset_path,
        preprocessing_steps=preprocessing_steps,
        total_examples=len(data),
        train_examples=int(len(data) * 0.8),
        val_examples=int(len(data) * 0.1),
        test_examples=int(len(data) * 0.1),
        created_at=datetime.now().isoformat(),
        checksum=compute_dataset_checksum(dataset_path),
    )