090 모델 버전 관리

키워드: 버전 관리, versioning, 모델 관리

개요

프로덕션 환경에서는 여러 버전의 모델을 관리하고 추적해야 합니다. 이 글에서는 FLAML 모델의 버전 관리 전략과 구현 방법을 알아봅니다.

실습 환경

• Python 버전: 3.11 권장
• 필요 패키지: flaml[automl]

pip install flaml[automl] pandas numpy

버전 관리의 필요성

import numpy as np
import pandas as pd

reasons = {
    '상황': ['모델 롤백', '성능 비교', '규정 준수', '협업'],
    '필요성': [
        '새 모델 문제 시 이전 버전 복구',
        '버전별 성능 추이 분석',
        '감사 추적(Audit Trail)',
        '팀원 간 모델 공유'
    ],
    '해결책': [
        '버전별 모델 저장',
        '메트릭 기록',
        '변경 이력 관리',
        '중앙 저장소'
    ]
}

print("모델 버전 관리의 필요성:")
print(pd.DataFrame(reasons).to_string(index=False))

간단한 버전 관리

import os
import json
import joblib
from datetime import datetime
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from flaml import AutoML

# 090 데이터 준비
np.random.seed(42)
X, y = make_classification(n_samples=2000, n_features=20, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 090 버전 관리 디렉토리
VERSION_DIR = "model_versions"
os.makedirs(VERSION_DIR, exist_ok=True)

def generate_version():
    """버전 생성 (시간 기반)"""
    return datetime.now().strftime("%Y%m%d_%H%M%S")

def save_versioned_model(automl, metrics, description=""):
    """버전이 있는 모델 저장"""
    version = generate_version()
    version_dir = os.path.join(VERSION_DIR, version)
    os.makedirs(version_dir, exist_ok=True)

    # 모델 저장
    model_path = os.path.join(version_dir, "model.pkl")
    joblib.dump(automl, model_path)

    # 메타데이터 저장
    metadata = {
        "version": version,
        "timestamp": datetime.now().isoformat(),
        "best_estimator": automl.best_estimator,
        "best_loss": float(automl.best_loss),
        "metrics": metrics,
        "description": description
    }

    meta_path = os.path.join(version_dir, "metadata.json")
    with open(meta_path, 'w') as f:
        json.dump(metadata, f, indent=2)

    print(f"모델 저장 완료: v{version}")
    return version

# 090 첫 번째 버전
automl_v1 = AutoML()
automl_v1.fit(X_train, y_train, task="classification", time_budget=30, verbose=0)

from sklearn.metrics import accuracy_score
y_pred = automl_v1.predict(X_test)
metrics_v1 = {"accuracy": accuracy_score(y_test, y_pred)}

version1 = save_versioned_model(automl_v1, metrics_v1, "Initial model")

버전 목록 조회

def list_versions():
    """저장된 모든 버전 조회"""
    versions = []

    if not os.path.exists(VERSION_DIR):
        return pd.DataFrame()

    for version_name in sorted(os.listdir(VERSION_DIR)):
        meta_path = os.path.join(VERSION_DIR, version_name, "metadata.json")
        if os.path.exists(meta_path):
            with open(meta_path, 'r') as f:
                metadata = json.load(f)
                versions.append({
                    "version": metadata["version"],
                    "timestamp": metadata["timestamp"],
                    "estimator": metadata["best_estimator"],
                    "accuracy": metadata["metrics"].get("accuracy", "N/A"),
                    "description": metadata["description"]
                })

    return pd.DataFrame(versions)

# 090 버전 목록
print("\n=== 모델 버전 목록 ===")
print(list_versions().to_string(index=False))

버전 로드

def load_version(version):
    """특정 버전 로드"""
    version_dir = os.path.join(VERSION_DIR, version)

    if not os.path.exists(version_dir):
        raise FileNotFoundError(f"버전 {version} 없음")

    # 모델 로드
    model_path = os.path.join(version_dir, "model.pkl")
    model = joblib.load(model_path)

    # 메타데이터 로드
    meta_path = os.path.join(version_dir, "metadata.json")
    with open(meta_path, 'r') as f:
        metadata = json.load(f)

    return model, metadata

def load_latest():
    """최신 버전 로드"""
    versions = list_versions()
    if len(versions) == 0:
        raise FileNotFoundError("저장된 버전 없음")

    latest = versions.iloc[-1]["version"]
    return load_version(latest)

# 090 버전 로드 테스트
loaded_model, loaded_meta = load_version(version1)
print(f"\n로드된 버전: {loaded_meta['version']}")
print(f"모델: {loaded_meta['best_estimator']}")

모델 비교

# 090 두 번째 버전 생성 (다른 설정)
import time
time.sleep(1)  # 버전 구분을 위해

automl_v2 = AutoML()
automl_v2.fit(X_train, y_train, task="classification", time_budget=30,
              estimator_list=['xgboost'], verbose=0)

y_pred_v2 = automl_v2.predict(X_test)
metrics_v2 = {"accuracy": accuracy_score(y_test, y_pred_v2)}

version2 = save_versioned_model(automl_v2, metrics_v2, "XGBoost only")

def compare_versions(version_a, version_b):
    """두 버전 비교"""
    _, meta_a = load_version(version_a)
    _, meta_b = load_version(version_b)

    comparison = {
        "항목": ["버전", "모델", "정확도", "설명"],
        "Version A": [
            meta_a["version"],
            meta_a["best_estimator"],
            f"{meta_a['metrics'].get('accuracy', 'N/A'):.4f}",
            meta_a["description"]
        ],
        "Version B": [
            meta_b["version"],
            meta_b["best_estimator"],
            f"{meta_b['metrics'].get('accuracy', 'N/A'):.4f}",
            meta_b["description"]
        ]
    }

    return pd.DataFrame(comparison)

# 090 버전 비교
print("\n=== 버전 비교 ===")
print(compare_versions(version1, version2).to_string(index=False))

모델 버전 매니저 클래스

class ModelVersionManager:
    """모델 버전 관리자"""

    def __init__(self, base_dir="models"):
        self.base_dir = base_dir
        os.makedirs(base_dir, exist_ok=True)
        self.current_version = None

    def save(self, model, metrics, description="", tags=None):
        """새 버전 저장"""
        version = datetime.now().strftime("%Y%m%d_%H%M%S")
        version_dir = os.path.join(self.base_dir, version)
        os.makedirs(version_dir, exist_ok=True)

        # 모델 저장
        joblib.dump(model, os.path.join(version_dir, "model.pkl"))

        # 메타데이터
        metadata = {
            "version": version,
            "timestamp": datetime.now().isoformat(),
            "metrics": metrics,
            "description": description,
            "tags": tags or [],
            "model_type": type(model).__name__
        }

        # FLAML 모델인 경우 추가 정보
        if hasattr(model, 'best_estimator'):
            metadata["best_estimator"] = model.best_estimator
            metadata["best_loss"] = float(model.best_loss)

        with open(os.path.join(version_dir, "metadata.json"), 'w') as f:
            json.dump(metadata, f, indent=2)

        self.current_version = version
        return version

    def load(self, version=None):
        """버전 로드 (없으면 최신)"""
        if version is None:
            versions = self.list_versions()
            if len(versions) == 0:
                raise FileNotFoundError("저장된 버전 없음")
            version = versions.iloc[-1]["version"]

        version_dir = os.path.join(self.base_dir, version)
        model = joblib.load(os.path.join(version_dir, "model.pkl"))

        with open(os.path.join(version_dir, "metadata.json"), 'r') as f:
            metadata = json.load(f)

        return model, metadata

    def list_versions(self):
        """버전 목록"""
        versions = []
        for v in sorted(os.listdir(self.base_dir)):
            meta_path = os.path.join(self.base_dir, v, "metadata.json")
            if os.path.exists(meta_path):
                with open(meta_path, 'r') as f:
                    meta = json.load(f)
                    versions.append({
                        "version": meta["version"],
                        "timestamp": meta["timestamp"][:10],
                        "model_type": meta.get("best_estimator", meta.get("model_type")),
                        "metrics": str(meta.get("metrics", {})),
                        "tags": ",".join(meta.get("tags", []))
                    })
        return pd.DataFrame(versions)

    def delete(self, version):
        """버전 삭제"""
        import shutil
        version_dir = os.path.join(self.base_dir, version)
        if os.path.exists(version_dir):
            shutil.rmtree(version_dir)
            print(f"버전 {version} 삭제됨")

    def tag(self, version, tags):
        """버전에 태그 추가"""
        meta_path = os.path.join(self.base_dir, version, "metadata.json")
        with open(meta_path, 'r') as f:
            meta = json.load(f)

        meta["tags"] = list(set(meta.get("tags", []) + tags))

        with open(meta_path, 'w') as f:
            json.dump(meta, f, indent=2)

    def get_by_tag(self, tag):
        """태그로 버전 검색"""
        versions = self.list_versions()
        return versions[versions["tags"].str.contains(tag, na=False)]

# 090 매니저 사용
manager = ModelVersionManager("managed_models")

# 090 모델 저장
v1 = manager.save(automl_v1, {"accuracy": 0.85}, "Base model", tags=["baseline"])
time.sleep(1)
v2 = manager.save(automl_v2, {"accuracy": 0.87}, "Improved model", tags=["candidate"])

# 090 목록 확인
print("\n=== 관리되는 버전 목록 ===")
print(manager.list_versions().to_string(index=False))

# 090 태그로 검색
manager.tag(v2, ["production"])
print("\n태그 'production' 검색:")
print(manager.get_by_tag("production"))

프로덕션 배포 워크플로우

class ProductionModelManager:
    """프로덕션 모델 관리"""

    def __init__(self, base_dir="production"):
        self.base_dir = base_dir
        os.makedirs(base_dir, exist_ok=True)
        self.staging_dir = os.path.join(base_dir, "staging")
        self.production_dir = os.path.join(base_dir, "production")
        self.archive_dir = os.path.join(base_dir, "archive")

        for d in [self.staging_dir, self.production_dir, self.archive_dir]:
            os.makedirs(d, exist_ok=True)

    def promote_to_staging(self, model, metrics, version=None):
        """스테이징 환경으로 승격"""
        version = version or datetime.now().strftime("%Y%m%d_%H%M%S")

        joblib.dump(model, os.path.join(self.staging_dir, "model.pkl"))

        meta = {
            "version": version,
            "stage": "staging",
            "promoted_at": datetime.now().isoformat(),
            "metrics": metrics
        }
        with open(os.path.join(self.staging_dir, "metadata.json"), 'w') as f:
            json.dump(meta, f, indent=2)

        print(f"스테이징 승격: v{version}")
        return version

    def promote_to_production(self):
        """스테이징 → 프로덕션 승격"""
        # 현재 프로덕션 백업
        prod_model = os.path.join(self.production_dir, "model.pkl")
        if os.path.exists(prod_model):
            # 아카이브로 이동
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            archive_path = os.path.join(self.archive_dir, f"model_{timestamp}.pkl")
            import shutil
            shutil.copy(prod_model, archive_path)
            print("기존 프로덕션 모델 아카이브됨")

        # 스테이징 → 프로덕션
        staging_model = os.path.join(self.staging_dir, "model.pkl")
        if os.path.exists(staging_model):
            import shutil
            shutil.copy(staging_model, prod_model)

            # 메타데이터 업데이트
            with open(os.path.join(self.staging_dir, "metadata.json"), 'r') as f:
                meta = json.load(f)

            meta["stage"] = "production"
            meta["deployed_at"] = datetime.now().isoformat()

            with open(os.path.join(self.production_dir, "metadata.json"), 'w') as f:
                json.dump(meta, f, indent=2)

            print(f"프로덕션 배포 완료: v{meta['version']}")
        else:
            raise FileNotFoundError("스테이징 모델 없음")

    def rollback(self, archive_version=None):
        """프로덕션 롤백"""
        archives = sorted(os.listdir(self.archive_dir))
        if not archives:
            raise FileNotFoundError("아카이브 없음")

        # 가장 최근 아카이브로 롤백
        if archive_version is None:
            archive_file = archives[-1]
        else:
            archive_file = f"model_{archive_version}.pkl"

        archive_path = os.path.join(self.archive_dir, archive_file)
        if os.path.exists(archive_path):
            import shutil
            shutil.copy(archive_path, os.path.join(self.production_dir, "model.pkl"))
            print(f"롤백 완료: {archive_file}")

    def load_production(self):
        """프로덕션 모델 로드"""
        model_path = os.path.join(self.production_dir, "model.pkl")
        return joblib.load(model_path)

# 090 프로덕션 워크플로우 테스트
prod_manager = ProductionModelManager()

# 1. 스테이징 배포
prod_manager.promote_to_staging(automl_v1, {"accuracy": 0.85})

# 2. 프로덕션 배포
prod_manager.promote_to_production()

# 3. 프로덕션 모델 사용
production_model = prod_manager.load_production()
print(f"프로덕션 모델 타입: {type(production_model).__name__}")

정리

• 버전 관리: 타임스탬프 기반 버전 생성
• 메타데이터: 메트릭, 설명, 태그 저장
• ModelVersionManager: 버전 저장, 로드, 검색, 삭제
• 프로덕션 워크플로우: 스테이징 → 프로덕션 → 아카이브
• 롤백: 이전 버전으로 빠른 복구

다음 글 예고

다음 글에서는 FastAPI로 모델 배포하기를 알아봅니다. FLAML 모델을 REST API로 서빙하는 방법을 다룹니다.

---

FLAML AutoML 마스터 시리즈 #090