본문으로 건너뛰기

078 로깅과 콜백 함수 활용

키워드: 로깅, 콜백, logging, callback

개요​

FLAML의 학습 과정을 모니터링하고 제어하려면 로깅과 콜백 함수가 필요합니다. 이 글에서는 학습 진행 상황 추적, 커스텀 로깅, 콜백 함수 구현 방법을 알아봅니다.

실습 환경​

  • Python 버전: 3.11 권장
  • 필요 패키지: flaml[automl]
pip install flaml[automl] pandas numpy matplotlib

기본 로깅 설정​

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

# 078 로깅 설정
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger('flaml_tutorial')

# 078 데이터 준비
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)

logger.info(f"데이터 준비 완료: {X_train.shape}")

FLAML verbose 레벨​

# 078 verbose 레벨별 출력
verbose_levels = {
    '레벨': [0, 1, 2, 3],
    '설명': [
        '출력 없음',
        '진행 상황만',
        '상세 정보 포함',
        '디버그 레벨'
    ],
    '권장 상황': [
        '배치 실행',
        '일반 사용',
        '문제 해결',
        '개발/디버깅'
    ]
}

print("FLAML verbose 레벨:")
print(pd.DataFrame(verbose_levels).to_string(index=False))

# 078 verbose=2로 실행
automl = AutoML()
automl.fit(
    X_train, y_train,
    task="classification",
    time_budget=30,
    verbose=2,  # 상세 로깅
)

print(f"\n최적 모델: {automl.best_estimator}")

파일 로깅​

import os
from datetime import datetime

# 078 로그 파일 설정
log_dir = "logs"
os.makedirs(log_dir, exist_ok=True)

timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = os.path.join(log_dir, f"flaml_{timestamp}.log")

# 078 파일 핸들러 추가
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))

# 078 FLAML 로거에 핸들러 추가
flaml_logger = logging.getLogger('flaml.automl')
flaml_logger.addHandler(file_handler)

# 078 학습 실행
automl_logged = AutoML()
automl_logged.fit(
    X_train, y_train,
    task="classification",
    time_budget=20,
    verbose=2
)

print(f"\n로그 파일 저장: {log_file}")

학습 이력 분석​

# 078 config_history 분석
history = automl.config_history

print(f"총 탐색 횟수: {len(history)}")
print("\n탐색 이력 (처음 5개):")
for i, (config_id, config_data) in enumerate(list(history.items())[:5]):
    print(f"  {config_id}: {config_data}")

# 078 손실 이력 추출
losses = []
times = []
estimators = []

for config_id, config_data in history.items():
    if 'loss' in config_data:
        losses.append(config_data.get('loss', float('inf')))
        times.append(config_data.get('time_total_s', 0))
        estimators.append(config_data.get('learner', 'unknown'))

# 078 시각화
import matplotlib.pyplot as plt

fig, axes = plt.subplots(1, 2, figsize=(14, 5))

# 078 손실 변화
axes[0].plot(losses, marker='o', alpha=0.7)
axes[0].set_xlabel('Trial')
axes[0].set_ylabel('Loss')
axes[0].set_title('Loss Over Trials')
axes[0].axhline(y=min(losses), color='red', linestyle='--', label=f'Best: {min(losses):.4f}')
axes[0].legend()

# 078 누적 시간
cumulative_time = np.cumsum(times)
axes[1].plot(cumulative_time, marker='s', alpha=0.7)
axes[1].set_xlabel('Trial')
axes[1].set_ylabel('Cumulative Time (s)')
axes[1].set_title('Cumulative Training Time')

plt.tight_layout()
plt.show()

커스텀 콜백 함수​

class TrainingCallback:
    """FLAML 학습 콜백 클래스"""

    def __init__(self):
        self.history = []
        self.best_loss = float('inf')
        self.best_config = None

    def on_trial_complete(self, trial_id, result, config):
        """각 trial 완료 시 호출"""
        loss = result.get('loss', float('inf'))

        self.history.append({
            'trial_id': trial_id,
            'loss': loss,
            'config': config,
            'time': result.get('time_total_s', 0)
        })

        if loss < self.best_loss:
            self.best_loss = loss
            self.best_config = config
            print(f"  [NEW BEST] Trial {trial_id}: loss={loss:.4f}")
        else:
            print(f"  Trial {trial_id}: loss={loss:.4f}")

    def get_summary(self):
        """학습 요약 반환"""
        return {
            'total_trials': len(self.history),
            'best_loss': self.best_loss,
            'best_config': self.best_config,
            'total_time': sum(h['time'] for h in self.history)
        }

# 078 콜백 사용 예시
callback = TrainingCallback()

# 078 FLAML은 직접 콜백을 지원하지 않으므로, 학습 후 이력 분석
automl_cb = AutoML()
automl_cb.fit(
    X_train, y_train,
    task="classification",
    time_budget=30,
    verbose=1
)

# 078 이력 분석으로 콜백 시뮬레이션
for i, (config_id, config_data) in enumerate(automl_cb.config_history.items()):
    callback.on_trial_complete(
        trial_id=i,
        result={'loss': config_data.get('loss', 1.0), 'time_total_s': config_data.get('time_total_s', 0)},
        config=config_data
    )

print("\n학습 요약:")
summary = callback.get_summary()
for key, value in summary.items():
    print(f"  {key}: {value}")

실시간 모니터링​

import time
from threading import Thread
import queue

class RealTimeMonitor:
    """실시간 학습 모니터링"""

    def __init__(self, update_interval=5):
        self.update_interval = update_interval
        self.is_running = False
        self.metrics_queue = queue.Queue()

    def start(self):
        """모니터링 시작"""
        self.is_running = True
        self.monitor_thread = Thread(target=self._monitor_loop)
        self.monitor_thread.start()

    def stop(self):
        """모니터링 종료"""
        self.is_running = False
        if hasattr(self, 'monitor_thread'):
            self.monitor_thread.join()

    def _monitor_loop(self):
        """모니터링 루프"""
        import psutil

        while self.is_running:
            metrics = {
                'timestamp': time.time(),
                'cpu_percent': psutil.cpu_percent(),
                'memory_percent': psutil.virtual_memory().percent
            }
            self.metrics_queue.put(metrics)
            time.sleep(self.update_interval)

    def get_metrics(self):
        """수집된 메트릭 반환"""
        metrics_list = []
        while not self.metrics_queue.empty():
            metrics_list.append(self.metrics_queue.get())
        return metrics_list

# 078 실시간 모니터링 사용
monitor = RealTimeMonitor(update_interval=2)
monitor.start()

# 078 학습 실행
automl_monitored = AutoML()
automl_monitored.fit(
    X_train, y_train,
    task="classification",
    time_budget=15,
    verbose=0
)

monitor.stop()

# 078 수집된 메트릭 확인
metrics = monitor.get_metrics()
print(f"\n수집된 메트릭 수: {len(metrics)}")
if metrics:
    print(f"평균 CPU: {np.mean([m['cpu_percent'] for m in metrics]):.1f}%")
    print(f"평균 메모리: {np.mean([m['memory_percent'] for m in metrics]):.1f}%")

tune.run 콜백​

from flaml import tune

# 078 tune.run에서 콜백 사용
class TuneCallback:
    def __init__(self):
        self.trials = []

    def __call__(self, result):
        """각 trial 결과 처리"""
        self.trials.append({
            'accuracy': result.get('accuracy', 0),
            'config': result.get('config', {})
        })
        print(f"Trial {len(self.trials)}: accuracy={result.get('accuracy', 0):.4f}")

def objective(config):
    """목적 함수"""
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.model_selection import cross_val_score

    model = RandomForestClassifier(
        n_estimators=config['n_estimators'],
        max_depth=config['max_depth'],
        random_state=42
    )
    scores = cross_val_score(model, X_train, y_train, cv=3)
    return {'accuracy': scores.mean()}

# 078 탐색 공간
search_space = {
    'n_estimators': tune.randint(50, 200),
    'max_depth': tune.randint(3, 15),
}

# 078 콜백과 함께 실행
tune_callback = TuneCallback()

analysis = tune.run(
    objective,
    config=search_space,
    metric='accuracy',
    mode='max',
    num_samples=10,
    verbose=0
)

print(f"\n최적 설정: {analysis.best_config}")
print(f"최적 정확도: {analysis.best_result['accuracy']:.4f}")

프로그레스 바​

from tqdm import tqdm
import time

class ProgressTracker:
    """학습 진행률 추적"""

    def __init__(self, time_budget):
        self.time_budget = time_budget
        self.start_time = None
        self.pbar = None

    def start(self):
        """진행률 추적 시작"""
        self.start_time = time.time()
        self.pbar = tqdm(total=self.time_budget, desc="Training", unit="s")

    def update(self):
        """진행률 업데이트"""
        if self.pbar:
            elapsed = time.time() - self.start_time
            self.pbar.n = min(elapsed, self.time_budget)
            self.pbar.refresh()

    def close(self):
        """진행률 추적 종료"""
        if self.pbar:
            self.pbar.n = self.time_budget
            self.pbar.close()

# 078 프로그레스 바 사용
print("\n프로그레스 바 예시:")
tracker = ProgressTracker(time_budget=10)
tracker.start()

automl_progress = AutoML()
# 078 별도 스레드에서 진행률 업데이트 필요
automl_progress.fit(
    X_train, y_train,
    task="classification",
    time_budget=10,
    verbose=0
)

tracker.close()
print(f"완료: {automl_progress.best_estimator}")

로깅 데코레이터​

import functools
import time

def log_execution(func):
    """함수 실행 로깅 데코레이터"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logger = logging.getLogger(func.__module__)

        logger.info(f"Starting {func.__name__}")
        start = time.time()

        try:
            result = func(*args, **kwargs)
            elapsed = time.time() - start
            logger.info(f"Completed {func.__name__} in {elapsed:.2f}s")
            return result
        except Exception as e:
            logger.error(f"Error in {func.__name__}: {e}")
            raise

    return wrapper

@log_execution
def train_automl(X, y, time_budget=30):
    """AutoML 학습 함수"""
    automl = AutoML()
    automl.fit(
        X, y,
        task="classification",
        time_budget=time_budget,
        verbose=0
    )
    return automl

# 078 데코레이터 사용
model = train_automl(X_train, y_train, time_budget=15)
print(f"결과: {model.best_estimator}")

로깅 베스트 프랙티스​

best_practices = {
    '항목': ['로그 레벨', '파일 로깅', '구조화', '성능'],
    '권장사항': [
        'DEBUG: 개발, INFO: 운영, WARNING: 문제',
        '별도 파일에 상세 로그 저장',
        'JSON 형식으로 구조화된 로깅',
        '대용량 로그는 rotate 설정'
    ],
    '이유': [
        '상황에 맞는 정보량 제공',
        '문제 발생 시 분석 용이',
        '로그 파싱 및 분석 용이',
        '디스크 공간 절약'
    ]
}

print("\n로깅 베스트 프랙티스:")
print(pd.DataFrame(best_practices).to_string(index=False))

정리​

  • verbose: 0(없음) ~ 3(디버그) 레벨 조절
  • 파일 로깅: FileHandler로 로그 파일 저장
  • config_history: 모든 탐색 이력 확인
  • 콜백: 학습 진행 중 이벤트 처리
  • 모니터링: 실시간 시스템 리소스 추적
  • 로깅은 문제 해결과 성능 분석에 필수

다음 글 예고​

다음 글에서는 AutoML 결과 재현성 확보에 대해 알아보겠습니다. 랜덤 시드와 환경 설정으로 동일한 결과를 얻는 방법을 다룹니다.

FLAML AutoML 마스터 시리즈 #078