MLOps 모델 배포 자동화: GitHub Actions + Kubeflow로 CI/CD/CT 파이프라인 구축하기

[1. 데이터 수집 & 전처리]
        ↓
[2. 모델 학습 & 실험 추적]  ← MLflow / W&B
        ↓
[3. 모델 검증 & 등록]  ← 품질 게이트(AUC ≥ 0.85 등)
        ↓
[4. 배포(Serving)]  ← REST/gRPC API
        ↓
[5. 모니터링 & 피드백]  ← 드리프트 감지 → 자동 재학습 트리거

# .github/workflows/mlops-pipeline.yml
name: MLOps CI/CD/CT Pipeline
 
on:
  push:
    branches: [main]
    paths:
      - 'src/models/**'
      - 'src/data/**'
  schedule:
    - cron: '0 2 * * 1'  # 매주 월요일 새벽 2시 정기 재학습
 
jobs:
  data-validation:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Validate data schema and drift
        # drift_monitor.py가 예시 3에서 구현한 Evidently AI 기반 스크립트입니다
        run: |
          python src/monitoring/drift_monitor.py \
            --reference data/reference_dataset.parquet \
            --current data/current_dataset.parquet \
            --threshold 0.05
 
  train-and-evaluate:
    needs: data-validation
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Train model
        run: python src/train.py --experiment-name ${{ github.sha }}
      - name: Evaluate quality gate
        id: eval
        run: |
          AUC=$(python src/evaluate.py --model-path outputs/model.pkl)
          echo "auc=$AUC" >> $GITHUB_OUTPUT
          # bc -l은 Linux 전용이므로, 이식성을 위해 Python으로 비교합니다
          python -c "import sys; sys.exit(0 if float('$AUC') >= 0.85 else 1)" || \
            (echo "Quality gate failed: AUC=$AUC" && exit 1)
 
  canary-deploy:
    needs: train-and-evaluate
    runs-on: ubuntu-latest
    steps:
      - name: Deploy canary (10% traffic)
        run: |
          kubectl apply -f k8s/seldon-canary-deployment.yaml
          # 30분짜리 모니터링을 Actions 잡 내에서 블로킹으로 돌리면
          # 비용·타임아웃 문제가 생깁니다. 실무에서는 Argo Rollouts나
          # 별도 스케줄러에 위임하거나, 아래처럼 비동기 모니터링을 트리거합니다.
          python scripts/trigger_canary_monitor.py --run-id ${{ github.run_id }}

# src/train.py — MLflow 실험 추적 및 모델 등록
# 데이터 로딩 구현은 src/data/loader.py 참조
import mlflow
import mlflow.sklearn
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import roc_auc_score
 
def train_and_register(X_train, y_train, X_val, y_val, params: dict):
    with mlflow.start_run() as run:
        mlflow.log_params(params)
 
        model = GradientBoostingClassifier(**params)
        model.fit(X_train, y_train)
 
        auc = roc_auc_score(y_val, model.predict_proba(X_val)[:, 1])
        mlflow.log_metric("auc", auc)
 
        if auc >= 0.85:
            mlflow.sklearn.log_model(
                model,
                artifact_path="model",
                registered_model_name="fraud-detector"
            )
            client = mlflow.tracking.MlflowClient()
 
            # run_id 기반으로 조회해야 동시 실험 시 레이스 컨디션을 방지할 수 있습니다
            # get_latest_versions(stages=["None"]) 패턴은 동시 실행 환경에서 위험합니다
            versions = client.search_model_versions(
                f"run_id='{run.info.run_id}'"
            )
            client.transition_model_version_stage(
                name="fraud-detector",
                version=versions[0].version,
                stage="Production"
            )
            print(f"[OK] 모델 등록 완료 -- AUC: {auc:.4f}")
        else:
            print(f"[FAIL] 품질 게이트 실패 -- AUC: {auc:.4f} (기준: 0.85)")
 
        return auc

# k8s/seldon-deployment.yaml — MLflow 모델 레지스트리와 연동
# kubectl apply -f k8s/seldon-deployment.yaml 로 클러스터에 적용합니다
# 참고: Seldon Core는 2024년 이후 유지보수가 축소되는 추세입니다.
#       현재는 KServe가 커뮤니티에서 더 활발히 유지되고 있으니 신규 도입 시 참고하세요.
apiVersion: machinelearning.seldon.io/v1
kind: SeldonDeployment
metadata:
  name: fraud-detector
spec:
  predictors:
    - name: default
      traffic: 90
      graph:
        name: classifier
        implementation: MLFLOW_SERVER
        modelUri: "models:/fraud-detector/Production"
    - name: canary
      traffic: 10
      graph:
        name: classifier-canary
        implementation: MLFLOW_SERVER
        modelUri: "models:/fraud-detector/Staging"

# src/monitoring/drift_monitor.py
import requests
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, ModelPerformancePreset
import pandas as pd
 
def trigger_retraining(reason: str):
    """Kubeflow 파이프라인 API 또는 Slack 웹훅으로 재학습을 트리거합니다."""
    webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    if webhook_url:
        requests.post(webhook_url, json={"text": f"[MLOps] 재학습 트리거: {reason}"})
    # Kubeflow 파이프라인 연동 예시:
    # kfp.Client().create_run_from_pipeline_func(training_pipeline, arguments={})
 
def run_drift_report(reference_path: str, current_path: str) -> dict:
    reference = pd.read_parquet(reference_path)
    current = pd.read_parquet(current_path)
 
    report = Report(metrics=[
        DataDriftPreset(drift_share_threshold=0.3),  # 30% 이상 피처 드리프트 시 경고
        ModelPerformancePreset(),
    ])
    report.run(reference_data=reference, current_data=current)
 
    result = report.as_dict()
    drift_detected = result["metrics"][0]["result"]["dataset_drift"]
 
    if drift_detected:
        trigger_retraining(reason="data_drift_detected")
 
    report.save_html("reports/drift_report.html")
    return {"drift_detected": drift_detected, "report": "reports/drift_report.html"}