演習：MLパイプラインを構築しよう

# pipeline/train_pipeline.py
import yaml
import mlflow
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import roc_auc_score, f1_score, classification_report

def load_config(config_path: str) -> dict:
    with open(config_path) as f:
        return yaml.safe_load(f)

def load_data(data_path: str) -> pd.DataFrame:
    df = pd.read_parquet(data_path)
    assert len(df) > 0, "データが空です"
    assert "churn" in df.columns, "ラベル列 'churn' が存在しません"
    print(f"Loaded {len(df)} records")
    return df

def preprocess(df: pd.DataFrame, config: dict) -> tuple:
    feature_cols = [c for c in df.columns if c not in ["churn", "customer_id"]]
    X = df[feature_cols]
    y = df["churn"]
    return train_test_split(
        X, y,
        test_size=config["data"]["test_size"],
        random_state=config["data"]["random_state"]
    )

def train(X_train, y_train, params: dict):
    model = GradientBoostingClassifier(**params)
    model.fit(X_train, y_train)
    return model

def evaluate(model, X_test, y_test) -> dict:
    y_pred = model.predict(X_test)
    y_prob = model.predict_proba(X_test)[:, 1]
    metrics = {
        "auc_roc": roc_auc_score(y_test, y_prob),
        "f1_score": f1_score(y_test, y_pred),
    }
    print(classification_report(y_test, y_pred))
    return metrics

def run_pipeline(config_path: str):
    config = load_config(config_path)

    with mlflow.start_run():
        # パラメータ記録
        mlflow.log_params(config["model"]["params"])
        mlflow.log_param("model_type", config["model"]["type"])

        # パイプライン実行
        df = load_data(config["data"]["path"])
        X_train, X_test, y_train, y_test = preprocess(df, config)
        model = train(X_train, y_train, config["model"]["params"])
        metrics = evaluate(model, X_test, y_test)

        # メトリクス記録
        mlflow.log_metrics(metrics)

        # 品質ゲート
        gates = config["quality_gates"]
        passed = all(
            metrics[k] >= v for k, v in gates.items() if k in metrics
        )

        if passed:
            mlflow.sklearn.log_model(model, "model")
            print("Quality gates PASSED - model logged")
        else:
            print("Quality gates FAILED - model NOT logged")
            for k, v in gates.items():
                if k in metrics and metrics[k] < v:
                    print(f"  FAIL: {k} = {metrics[k]:.4f} < {v}")

        return metrics, passed

if __name__ == "__main__":
    run_pipeline("config/training_config.yml")

# .github/workflows/ml-pipeline.yml
name: ML Training Pipeline

on:
  push:
    branches: [main]
    paths:
      - "src/**"
      - "pipeline/**"
      - "config/**"
  schedule:
    - cron: "0 2 * * 1"
  workflow_dispatch:

jobs:
  validate-data:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.10"
      - run: pip install -r requirements.txt
      - name: Validate training data
        run: python pipeline/validate_data.py

  train-and-evaluate:
    needs: validate-data
    runs-on: ubuntu-latest
    outputs:
      quality_gate: ${{ steps.train.outputs.quality_gate }}
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.10"
      - run: pip install -r requirements.txt
      - name: Train and evaluate
        id: train
        run: |
          result=$(python pipeline/train_pipeline.py)
          echo "quality_gate=$result" >> $GITHUB_OUTPUT
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_URI }}

  deploy-staging:
    needs: train-and-evaluate
    if: needs.train-and-evaluate.outputs.quality_gate == 'passed'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Deploy to staging
        run: python pipeline/deploy.py --stage staging
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_URI }}
      - name: Run integration tests
        run: pytest tests/integration/ -v

  notify:
    needs: [train-and-evaluate, deploy-staging]
    if: always()
    runs-on: ubuntu-latest
    steps:
      - name: Notify Slack
        uses: slackapi/slack-github-action@v1
        with:
          payload: |
            {
              "text": "ML Pipeline: ${{ needs.deploy-staging.result }}"
            }
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK }}

# pipeline/rollback.py
import mlflow
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

ROLLBACK_CONDITIONS = {
    "auc_roc_min": 0.75,
    "error_rate_max": 0.05,
    "latency_p99_max_ms": 200,
}

def check_rollback_needed(current_metrics: dict) -> tuple[bool, list[str]]:
    """ロールバックが必要か判定する"""
    reasons = []

    if current_metrics.get("auc_roc", 1.0) < ROLLBACK_CONDITIONS["auc_roc_min"]:
        reasons.append(
            f"AUC-ROC {current_metrics['auc_roc']:.4f} < "
            f"{ROLLBACK_CONDITIONS['auc_roc_min']}"
        )

    if current_metrics.get("error_rate", 0) > ROLLBACK_CONDITIONS["error_rate_max"]:
        reasons.append(
            f"Error rate {current_metrics['error_rate']:.4f} > "
            f"{ROLLBACK_CONDITIONS['error_rate_max']}"
        )

    if current_metrics.get("latency_p99_ms", 0) > ROLLBACK_CONDITIONS["latency_p99_max_ms"]:
        reasons.append(
            f"Latency {current_metrics['latency_p99_ms']}ms > "
            f"{ROLLBACK_CONDITIONS['latency_p99_max_ms']}ms"
        )

    return len(reasons) > 0, reasons

def execute_rollback(model_name: str, reasons: list[str]):
    """ロールバックを実行する"""
    client = mlflow.tracking.MlflowClient()

    # 現在のProductionバージョン取得
    prod = client.get_latest_versions(model_name, stages=["Production"])
    archived = client.get_latest_versions(model_name, stages=["Archived"])

    if not prod or not archived:
        logger.error("Rollback failed: missing versions")
        return False

    current_v = prod[0].version
    rollback_v = archived[0].version

    # ステージ変更
    client.transition_model_version_stage(
        name=model_name, version=current_v, stage="Archived"
    )
    client.transition_model_version_stage(
        name=model_name, version=rollback_v, stage="Production"
    )

    # ログ記録
    logger.info(f"ROLLBACK: {model_name} v{current_v} -> v{rollback_v}")
    logger.info(f"Reasons: {'; '.join(reasons)}")
    logger.info(f"Timestamp: {datetime.now().isoformat()}")

    return True

if __name__ == "__main__":
    # モニタリングから取得したメトリクス（例）
    metrics = {"auc_roc": 0.72, "error_rate": 0.02, "latency_p99_ms": 150}
    needed, reasons = check_rollback_needed(metrics)
    if needed:
        execute_rollback("churn_model", reasons)