089 모델 저장과 불러오기

키워드: 저장, 불러오기, pickle, joblib

개요

학습된 FLAML 모델을 저장하고 불러오는 것은 배포와 재사용에 필수적입니다. 이 글에서는 다양한 직렬화 방법과 모범 사례를 알아봅니다.

실습 환경

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

pip install flaml[automl] pandas numpy

데이터 준비

import numpy as np
import pandas as pd
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from flaml import AutoML

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
)

# 089 FLAML 모델 학습
automl = AutoML()
automl.fit(
    X_train, y_train,
    task="classification",
    time_budget=30,
    verbose=1
)

print(f"최적 모델: {automl.best_estimator}")
print(f"학습 완료")

1. pickle로 저장

import pickle

# 089 pickle 저장
with open('model_pickle.pkl', 'wb') as f:
    pickle.dump(automl, f)

print("pickle 저장 완료: model_pickle.pkl")

# 089 pickle 불러오기
with open('model_pickle.pkl', 'rb') as f:
    loaded_pickle = pickle.load(f)

# 089 검증
y_pred_pickle = loaded_pickle.predict(X_test)
from sklearn.metrics import accuracy_score
print(f"pickle 로드 후 정확도: {accuracy_score(y_test, y_pred_pickle):.4f}")

2. joblib으로 저장 (권장)

import joblib

# 089 joblib 저장
joblib.dump(automl, 'model_joblib.pkl')
print("joblib 저장 완료: model_joblib.pkl")

# 089 joblib 불러오기
loaded_joblib = joblib.load('model_joblib.pkl')

# 089 검증
y_pred_joblib = loaded_joblib.predict(X_test)
print(f"joblib 로드 후 정확도: {accuracy_score(y_test, y_pred_joblib):.4f}")

# 089 압축 저장
joblib.dump(automl, 'model_compressed.pkl', compress=3)
print("압축 저장 완료: model_compressed.pkl")

3. 저장 방법 비교

import os

# 089 파일 크기 비교
sizes = {
    '방법': ['pickle', 'joblib', 'joblib (compress=3)'],
    '파일': ['model_pickle.pkl', 'model_joblib.pkl', 'model_compressed.pkl'],
    '크기 (KB)': []
}

for file in sizes['파일']:
    if os.path.exists(file):
        size_kb = os.path.getsize(file) / 1024
        sizes['크기 (KB)'].append(f"{size_kb:.1f}")
    else:
        sizes['크기 (KB)'].append("N/A")

print("\n=== 저장 방법 비교 ===")
print(pd.DataFrame(sizes).to_string(index=False))

# 089 권장 사항
print("\n권장: joblib (대용량 numpy 배열에 효율적)")

4. 내부 모델만 저장

# 089 FLAML의 내부 모델만 저장 (더 작은 크기)
internal_model = automl.model.estimator

joblib.dump(internal_model, 'internal_model.pkl')
print("내부 모델 저장: internal_model.pkl")

# 089 내부 모델 불러오기
loaded_internal = joblib.load('internal_model.pkl')

# 089 예측 (FLAML 래퍼 없이)
y_pred_internal = loaded_internal.predict(X_test)
print(f"내부 모델 정확도: {accuracy_score(y_test, y_pred_internal):.4f}")

# 089 크기 비교
print(f"\nFLAML 전체: {os.path.getsize('model_joblib.pkl')/1024:.1f} KB")
print(f"내부 모델만: {os.path.getsize('internal_model.pkl')/1024:.1f} KB")

5. 설정과 함께 저장

import json

def save_model_with_config(automl, base_path='model_package'):
    """모델과 설정을 함께 저장"""

    # 모델 저장
    model_path = f'{base_path}_model.pkl'
    joblib.dump(automl, model_path)

    # 설정 저장
    config = {
        'best_estimator': automl.best_estimator,
        'best_config': {k: str(v) for k, v in automl.best_config.items()},
        'best_loss': float(automl.best_loss),
        'n_trials': len(automl.config_history)
    }

    config_path = f'{base_path}_config.json'
    with open(config_path, 'w') as f:
        json.dump(config, f, indent=2)

    print(f"모델 저장: {model_path}")
    print(f"설정 저장: {config_path}")

    return model_path, config_path

# 089 저장
save_model_with_config(automl, 'my_model')

# 089 설정 확인
with open('my_model_config.json', 'r') as f:
    config = json.load(f)
print(f"\n저장된 설정: {config}")

6. 모델 로더 클래스

class ModelLoader:
    """모델 로딩 유틸리티"""

    def __init__(self, model_path, config_path=None):
        self.model_path = model_path
        self.config_path = config_path
        self.model = None
        self.config = None

    def load(self):
        """모델과 설정 로드"""
        # 모델 로드
        self.model = joblib.load(self.model_path)

        # 설정 로드 (있으면)
        if self.config_path and os.path.exists(self.config_path):
            with open(self.config_path, 'r') as f:
                self.config = json.load(f)

        return self

    def predict(self, X):
        """예측"""
        if self.model is None:
            raise ValueError("모델이 로드되지 않았습니다. load()를 먼저 호출하세요.")
        return self.model.predict(X)

    def predict_proba(self, X):
        """확률 예측"""
        if self.model is None:
            raise ValueError("모델이 로드되지 않았습니다.")
        return self.model.predict_proba(X)

    def get_info(self):
        """모델 정보"""
        if self.config:
            return self.config
        return {"model_type": type(self.model).__name__}

# 089 사용 예시
loader = ModelLoader('my_model_model.pkl', 'my_model_config.json')
loader.load()

print("\n=== 모델 로더 테스트 ===")
print(f"모델 정보: {loader.get_info()}")
y_pred = loader.predict(X_test)
print(f"예측 정확도: {accuracy_score(y_test, y_pred):.4f}")

7. 버전 정보 포함

import sys
import platform

def save_with_environment(automl, path_prefix='model'):
    """환경 정보와 함께 저장"""

    # 모델 저장
    joblib.dump(automl, f'{path_prefix}.pkl')

    # 환경 정보 저장
    env_info = {
        'python_version': sys.version,
        'platform': platform.platform(),
        'packages': {}
    }

    # 주요 패키지 버전
    for pkg in ['flaml', 'numpy', 'pandas', 'scikit-learn', 'lightgbm', 'xgboost']:
        try:
            module = __import__(pkg.replace('-', '_'))
            env_info['packages'][pkg] = getattr(module, '__version__', 'unknown')
        except ImportError:
            pass

    # 모델 설정
    env_info['model'] = {
        'best_estimator': automl.best_estimator,
        'best_loss': float(automl.best_loss)
    }

    with open(f'{path_prefix}_env.json', 'w') as f:
        json.dump(env_info, f, indent=2)

    print(f"모델 저장: {path_prefix}.pkl")
    print(f"환경 정보: {path_prefix}_env.json")

# 089 환경 정보와 함께 저장
save_with_environment(automl, 'versioned_model')

8. 모델 호환성 검사

def check_model_compatibility(env_file):
    """모델 호환성 검사"""
    import warnings

    with open(env_file, 'r') as f:
        saved_env = json.load(f)

    issues = []

    # Python 버전 검사
    current_python = sys.version.split()[0]
    saved_python = saved_env['python_version'].split()[0]

    if current_python.split('.')[:2] != saved_python.split('.')[:2]:
        issues.append(f"Python 버전 불일치: 저장={saved_python}, 현재={current_python}")

    # 패키지 버전 검사
    for pkg, saved_ver in saved_env['packages'].items():
        try:
            module = __import__(pkg.replace('-', '_'))
            current_ver = getattr(module, '__version__', 'unknown')
            if current_ver != saved_ver:
                issues.append(f"{pkg}: 저장={saved_ver}, 현재={current_ver}")
        except ImportError:
            issues.append(f"{pkg}: 설치되지 않음")

    if issues:
        print("=== 호환성 경고 ===")
        for issue in issues:
            print(f"  - {issue}")
    else:
        print("호환성 검사 통과")

    return len(issues) == 0

# 089 호환성 검사
check_model_compatibility('versioned_model_env.json')

9. 모델 패키지 클래스

class ModelPackage:
    """완전한 모델 패키지"""

    def __init__(self):
        self.model = None
        self.config = None
        self.environment = None
        self.feature_names = None

    def save(self, automl, path, feature_names=None):
        """패키지 저장"""
        self.model = automl
        self.feature_names = feature_names

        package = {
            'model': automl,
            'config': {
                'best_estimator': automl.best_estimator,
                'best_config': automl.best_config,
                'best_loss': automl.best_loss
            },
            'environment': {
                'python': sys.version,
                'platform': platform.platform()
            },
            'feature_names': feature_names
        }

        joblib.dump(package, path)
        print(f"패키지 저장: {path}")

    @classmethod
    def load(cls, path):
        """패키지 로드"""
        package_data = joblib.load(path)

        instance = cls()
        instance.model = package_data['model']
        instance.config = package_data['config']
        instance.environment = package_data['environment']
        instance.feature_names = package_data.get('feature_names')

        return instance

    def predict(self, X):
        return self.model.predict(X)

    def predict_proba(self, X):
        return self.model.predict_proba(X)

# 089 패키지 사용
package = ModelPackage()
feature_names = [f'feature_{i}' for i in range(20)]
package.save(automl, 'full_package.pkl', feature_names=feature_names)

# 089 로드
loaded_package = ModelPackage.load('full_package.pkl')
print(f"\n로드된 설정: {loaded_package.config}")
print(f"특성 이름: {loaded_package.feature_names[:5]}...")

10. 저장 모범 사례

best_practices = {
    '항목': ['방법', '압축', '메타데이터', '버전 관리', '테스트'],
    '권장': [
        'joblib (sklearn 모델에 최적화)',
        'compress=3 (크기/속도 균형)',
        '설정, 환경 정보 함께 저장',
        '패키지 버전 기록',
        '로드 후 예측 검증'
    ],
    '이유': [
        '대용량 numpy 배열 효율적 처리',
        '저장 공간 절약',
        '재현성 및 디버깅',
        '호환성 문제 추적',
        '모델 무결성 확인'
    ]
}

print("\n=== 모델 저장 모범 사례 ===")
print(pd.DataFrame(best_practices).to_string(index=False))

정리

• pickle: 기본 Python 직렬화
• joblib: sklearn 모델에 권장 (압축 지원)
• 내부 모델: 크기 최소화 시 estimator만 저장
• 메타데이터: 설정, 환경 정보 함께 저장
• 호환성: 패키지 버전 기록 및 검사

다음 글 예고

다음 글에서는 모델 버전 관리를 알아봅니다. 여러 버전의 모델을 체계적으로 관리하는 방법을 다룹니다.

---

FLAML AutoML 마스터 시리즈 #089