第1章：MLOps基礎

解答：

1. データの扱い

   • DevOps: コードのバージョン管理が中心
   • MLOps: コード、データ、モデルの3つをバージョン管理する必要がある
   • 機械学習では、同じコードでもデータが異なれば結果が変わるため、データのバージョニングが必須

2. テストの複雑性

   • DevOps: 決定論的なテスト（同じ入力 → 同じ出力）
   • MLOps: 確率的なテスト（モデルの性能、データドリフト、バイアスなど）
   • モデルのテストは精度だけでなく、公平性、解釈可能性なども評価する必要がある

3. 継続的なモニタリング

   • DevOps: システムの稼働状態、エラーレートを監視
   • MLOps: モデルの性能劣化、データドリフト、予測分布の変化を監視
   • 時間経過によるモデルの劣化が避けられないため、自動再訓練の仕組みが必要

import pandas as pd
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
import hashlib
import json

# MLflow実験の設定
mlflow.set_experiment("customer_classification")

# データバージョンの計算
def compute_data_version(df):
    """データのハッシュ値を計算してバージョンとして使用"""
    data_str = pd.util.hash_pandas_object(df).values.tobytes()
    return hashlib.md5(data_str).hexdigest()[:8]

# データ読み込み
df = pd.read_csv('data.csv')
data_version = compute_data_version(df)

X = df.drop('target', axis=1)
y = df['target']

# 実験開始
with mlflow.start_run(run_name="rf_baseline"):

    # ハイパーパラメータの定義
    params = {
        'n_estimators': 100,
        'max_depth': 10,
        'min_samples_split': 5,
        'random_state': 42
    }

    # データ分割（固定シードで再現性確保）
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, stratify=y
    )

    # パラメータの記録
    mlflow.log_params(params)
    mlflow.log_param("data_version", data_version)
    mlflow.log_param("test_size", 0.2)
    mlflow.log_param("n_train_samples", len(X_train))
    mlflow.log_param("n_test_samples", len(X_test))

    # モデル訓練
    model = RandomForestClassifier(**params)
    model.fit(X_train, y_train)

    # Cross-validation
    cv_scores = cross_val_score(model, X_train, y_train, cv=5)
    mlflow.log_metric("cv_mean_accuracy", cv_scores.mean())
    mlflow.log_metric("cv_std_accuracy", cv_scores.std())

    # テストセットでの評価
    y_pred = model.predict(X_test)

    # 複数のメトリクスを記録
    metrics = {
        'test_accuracy': accuracy_score(y_test, y_pred),
        'test_precision': precision_score(y_test, y_pred, average='weighted'),
        'test_recall': recall_score(y_test, y_pred, average='weighted'),
        'test_f1': f1_score(y_test, y_pred, average='weighted')
    }

    mlflow.log_metrics(metrics)

    # 特徴量重要度の記録
    feature_importance = dict(zip(X.columns, model.feature_importances_))
    mlflow.log_dict(feature_importance, "feature_importance.json")

    # モデルの保存
    mlflow.sklearn.log_model(
        model,
        "model",
        registered_model_name="customer_classifier"
    )

    # タグの設定
    mlflow.set_tag("model_type", "RandomForest")
    mlflow.set_tag("framework", "scikit-learn")
    mlflow.set_tag("environment", "development")

    # 結果の表示
    print("=== 実験結果 ===")
    print(f"Run ID: {mlflow.active_run().info.run_id}")
    print(f"Data Version: {data_version}")
    print(f"\nMetrics:")
    for metric, value in metrics.items():
        print(f"  {metric}: {value:.4f}")
    print(f"\nCV Accuracy: {cv_scores.mean():.4f} ± {cv_scores.std():.4f}")

    # モデルの登録確認
    print(f"\n✓ モデルをMLflowに記録しました")
    print(f"✓ 実験名: customer_classification")

改善ポイント：

1. MLflowで実験を管理
2. データバージョンの記録
3. 全ハイパーパラメータの記録
4. 複数の評価メトリクスを記録
5. Cross-validationの実施
6. 特徴量重要度の保存
7. モデルレジストリへの登録
8. 再現性の確保（random_state、stratify）

import numpy as np
import pandas as pd
from scipy import stats
from datetime import datetime
import json

class DataDriftMonitor:
    """データドリフト監視システム"""

    def __init__(self, baseline_data, threshold=0.05, alert_features=None):
        """
        Args:
            baseline_data: ベースラインデータ（DataFrame）
            threshold: p値の閾値（デフォルト: 0.05）
            alert_features: 監視する特徴量のリスト（Noneの場合は全特徴量）
        """
        self.baseline_data = baseline_data
        self.threshold = threshold
        self.alert_features = alert_features or baseline_data.columns.tolist()
        self.drift_history = []
        self.baseline_stats = self._compute_baseline_stats()

    def _compute_baseline_stats(self):
        """ベースラインの統計量を計算"""
        stats_dict = {}
        for col in self.baseline_data.columns:
            if pd.api.types.is_numeric_dtype(self.baseline_data[col]):
                stats_dict[col] = {
                    'mean': self.baseline_data[col].mean(),
                    'std': self.baseline_data[col].std(),
                    'min': self.baseline_data[col].min(),
                    'max': self.baseline_data[col].max(),
                    'median': self.baseline_data[col].median()
                }
        return stats_dict

    def detect_drift(self, new_data, feature):
        """単一特徴量のドリフト検出"""
        if feature not in self.baseline_data.columns:
            raise ValueError(f"特徴量 {feature} が見つかりません")

        # 数値型のみ処理
        if not pd.api.types.is_numeric_dtype(self.baseline_data[feature]):
            return None

        # KS検定
        baseline_values = self.baseline_data[feature].dropna()
        new_values = new_data[feature].dropna()

        statistic, p_value = stats.ks_2samp(baseline_values, new_values)

        is_drift = p_value < self.threshold

        # 統計量の変化を計算
        baseline_mean = baseline_values.mean()
        new_mean = new_values.mean()
        mean_shift = (new_mean - baseline_mean) / baseline_mean * 100

        drift_info = {
            'timestamp': datetime.now().isoformat(),
            'feature': feature,
            'ks_statistic': float(statistic),
            'p_value': float(p_value),
            'drift_detected': bool(is_drift),
            'baseline_mean': float(baseline_mean),
            'new_mean': float(new_mean),
            'mean_shift_pct': float(mean_shift),
            'n_baseline': len(baseline_values),
            'n_new': len(new_values)
        }

        return drift_info

    def monitor_all_features(self, new_data):
        """全特徴量のドリフト監視"""
        results = []
        alerts = []

        print(f"=== データドリフト監視実行 ===")
        print(f"時刻: {datetime.now()}")
        print(f"監視特徴量数: {len(self.alert_features)}")
        print(f"新データサンプル数: {len(new_data)}\n")

        for feature in self.alert_features:
            if feature not in new_data.columns:
                continue

            drift_info = self.detect_drift(new_data, feature)

            if drift_info is None:
                continue

            results.append(drift_info)
            self.drift_history.append(drift_info)

            # ドリフト検出時のアラート
            if drift_info['drift_detected']:
                alert_msg = (
                    f"⚠️  ドリフト検出: {feature}\n"
                    f"   KS統計量: {drift_info['ks_statistic']:.4f}\n"
                    f"   p値: {drift_info['p_value']:.4f}\n"
                    f"   平均値シフト: {drift_info['mean_shift_pct']:.2f}%"
                )
                alerts.append(alert_msg)
                print(alert_msg + "\n")

        # サマリー
        n_drift = sum(r['drift_detected'] for r in results)
        print(f"=== 監視結果 ===")
        print(f"ドリフト検出: {n_drift}/{len(results)}特徴量")

        if n_drift > len(results) * 0.3:  # 30%以上でアラート
            print("⚠️  警告: 多数の特徴量でドリフトが検出されました")
            print("   モデルの再訓練を推奨します")

        return results, alerts

    def generate_report(self):
        """ドリフトレポート生成"""
        if not self.drift_history:
            return "ドリフト履歴がありません"

        df_history = pd.DataFrame(self.drift_history)

        report = f"""
=== データドリフト監視レポート ===

監視期間: {df_history['timestamp'].min()} ~ {df_history['timestamp'].max()}
総監視回数: {len(df_history)}
ユニーク特徴量数: {df_history['feature'].nunique()}

ドリフト検出サマリー:
{df_history.groupby('feature')['drift_detected'].agg(['sum', 'count']).to_string()}

ドリフト検出率トップ5:
{df_history[df_history['drift_detected']].groupby('feature').size().sort_values(ascending=False).head().to_string()}

平均シフト率（絶対値）トップ5:
{df_history.groupby('feature')['mean_shift_pct'].apply(lambda x: abs(x).mean()).sort_values(ascending=False).head().to_string()}
        """

        return report

    def save_report(self, filepath):
        """レポートをJSONで保存"""
        report_data = {
            'baseline_stats': self.baseline_stats,
            'drift_history': self.drift_history,
            'summary': {
                'total_checks': len(self.drift_history),
                'total_drifts': sum(d['drift_detected'] for d in self.drift_history)
            }
        }

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

        print(f"✓ レポート保存: {filepath}")

# 使用例
from sklearn.datasets import make_classification

# ベースラインデータ（訓練時のデータ）
X_baseline, _ = make_classification(
    n_samples=1000, n_features=10, random_state=42
)
df_baseline = pd.DataFrame(
    X_baseline,
    columns=[f'feature_{i}' for i in range(10)]
)

# 新しいデータ（ドリフトあり）
X_new, _ = make_classification(
    n_samples=500, n_features=10, random_state=43
)
# いくつかの特徴量にシフトを追加
X_new[:, 0] += 2.0  # feature_0に大きなシフト
X_new[:, 3] += 0.5  # feature_3に小さなシフト

df_new = pd.DataFrame(
    X_new,
    columns=[f'feature_{i}' for i in range(10)]
)

# ドリフト監視の実行
monitor = DataDriftMonitor(df_baseline, threshold=0.05)
results, alerts = monitor.monitor_all_features(df_new)

# レポート生成
print("\n" + monitor.generate_report())

# レポート保存
monitor.save_report('drift_report.json')

出力例：

=== データドリフト監視実行 ===
時刻: 2025-10-21 10:30:45.123456
監視特徴量数: 10
新データサンプル数: 500

⚠️  ドリフト検出: feature_0
   KS統計量: 0.8920
   p値: 0.0000
   平均値シフト: 412.34%

⚠️  ドリフト検出: feature_3
   KS統計量: 0.2145
   p値: 0.0023
   平均値シフト: 87.56%

=== 監視結果 ===
ドリフト検出: 2/10特徴量

import pandas as pd
import numpy as np
import mlflow
import mlflow.sklearn
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.metrics import accuracy_score, classification_report
from datetime import datetime
import logging

# ロギング設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AutoMLPipeline:
    """自動ML訓練パイプライン（MLOps Level 1）"""

    def __init__(
        self,
        experiment_name,
        model_name,
        accuracy_threshold=0.75,
        cv_folds=5
    ):
        """
        Args:
            experiment_name: MLflow実験名
            model_name: モデル登録名
            accuracy_threshold: モデル登録の精度閾値
            cv_folds: Cross-validationのフォールド数
        """
        self.experiment_name = experiment_name
        self.model_name = model_name
        self.accuracy_threshold = accuracy_threshold
        self.cv_folds = cv_folds

        # MLflow実験の設定
        mlflow.set_experiment(experiment_name)

        logger.info(f"自動訓練パイプライン初期化完了")
        logger.info(f"実験名: {experiment_name}")
        logger.info(f"精度閾値: {accuracy_threshold}")

    def load_data(self, data_source):
        """データの自動取得"""
        logger.info(f"データ読み込み開始: {data_source}")

        # 実際の環境では、DBやAPIからデータを取得
        # ここではデモ用のデータ生成
        from sklearn.datasets import make_classification

        X, y = make_classification(
            n_samples=1000,
            n_features=20,
            n_informative=15,
            n_redundant=5,
            random_state=42
        )

        df = pd.DataFrame(
            X,
            columns=[f'feature_{i}' for i in range(20)]
        )
        df['target'] = y

        # 意図的に欠損値を追加
        missing_mask = np.random.random(df.shape) < 0.05
        df = df.mask(missing_mask)

        logger.info(f"データ読み込み完了: {len(df)}行, {len(df.columns)}列")
        logger.info(f"欠損値: {df.isnull().sum().sum()}個")

        return df

    def create_preprocessing_pipeline(self):
        """前処理パイプライン作成"""
        return Pipeline([
            ('imputer', SimpleImputer(strategy='mean')),
            ('scaler', StandardScaler())
        ])

    def create_model_pipeline(self, preprocessing_pipeline):
        """モデルパイプライン作成"""
        return Pipeline([
            ('preprocessing', preprocessing_pipeline),
            ('classifier', RandomForestClassifier(
                n_estimators=100,
                max_depth=10,
                random_state=42
            ))
        ])

    def evaluate_model(self, pipeline, X_train, X_test, y_train, y_test):
        """モデル評価"""
        # Cross-validation
        cv_scores = cross_val_score(
            pipeline,
            X_train,
            y_train,
            cv=self.cv_folds,
            scoring='accuracy'
        )

        # テストセット評価
        y_pred = pipeline.predict(X_test)
        test_accuracy = accuracy_score(y_test, y_pred)

        metrics = {
            'cv_mean_accuracy': cv_scores.mean(),
            'cv_std_accuracy': cv_scores.std(),
            'test_accuracy': test_accuracy
        }

        logger.info("モデル評価完了")
        logger.info(f"CV精度: {cv_scores.mean():.4f} ± {cv_scores.std():.4f}")
        logger.info(f"テスト精度: {test_accuracy:.4f}")

        return metrics, y_pred

    def should_register_model(self, metrics):
        """モデル登録判定"""
        test_acc = metrics['test_accuracy']
        cv_acc = metrics['cv_mean_accuracy']

        # 条件1: テスト精度が閾値以上
        # 条件2: CV精度との差が大きすぎない（過学習チェック）
        meets_threshold = test_acc >= self.accuracy_threshold
        not_overfitting = abs(test_acc - cv_acc) < 0.1

        should_register = meets_threshold and not_overfitting

        if should_register:
            logger.info(f"✓ モデル登録条件を満たしました")
        else:
            logger.warning(f"⚠️  モデル登録条件を満たしていません")
            if not meets_threshold:
                logger.warning(f"   精度不足: {test_acc:.4f} < {self.accuracy_threshold}")
            if not not_overfitting:
                logger.warning(f"   過学習の可能性: テスト精度とCV精度の差が大きい")

        return should_register

    def run_training(self, data_source):
        """訓練パイプライン実行"""
        logger.info("=" * 50)
        logger.info(f"自動訓練開始: {datetime.now()}")
        logger.info("=" * 50)

        with mlflow.start_run(run_name=f"auto_training_{datetime.now().strftime('%Y%m%d_%H%M%S')}"):

            # 1. データ読み込み
            df = self.load_data(data_source)

            # データ情報の記録
            mlflow.log_param("n_samples", len(df))
            mlflow.log_param("n_features", len(df.columns) - 1)
            mlflow.log_param("n_missing", df.isnull().sum().sum())

            # 2. データ分割
            X = df.drop('target', axis=1)
            y = df['target']

            X_train, X_test, y_train, y_test = train_test_split(
                X, y,
                test_size=0.2,
                random_state=42,
                stratify=y
            )

            mlflow.log_param("test_size", 0.2)
            mlflow.log_param("n_train", len(X_train))
            mlflow.log_param("n_test", len(X_test))

            # 3. パイプライン作成
            preprocessing_pipeline = self.create_preprocessing_pipeline()
            model_pipeline = self.create_model_pipeline(preprocessing_pipeline)

            # ハイパーパラメータ記録
            model_params = model_pipeline.named_steps['classifier'].get_params()
            mlflow.log_params({
                f"model_{k}": v for k, v in model_params.items()
                if not k.startswith('_')
            })

            # 4. モデル訓練
            logger.info("モデル訓練開始")
            model_pipeline.fit(X_train, y_train)
            logger.info("モデル訓練完了")

            # 5. モデル評価
            metrics, y_pred = self.evaluate_model(
                model_pipeline,
                X_train, X_test,
                y_train, y_test
            )

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

            # 分類レポート
            report = classification_report(y_test, y_pred, output_dict=True)
            mlflow.log_dict(report, "classification_report.json")

            # 6. モデル保存
            mlflow.sklearn.log_model(
                model_pipeline,
                "model",
                signature=mlflow.models.signature.infer_signature(X_train, y_train)
            )

            # 7. モデル登録判定
            if self.should_register_model(metrics):
                model_uri = f"runs:/{mlflow.active_run().info.run_id}/model"

                mlflow.register_model(
                    model_uri,
                    self.model_name
                )

                logger.info(f"✓ モデルをレジストリに登録: {self.model_name}")
                logger.info(f"  精度: {metrics['test_accuracy']:.4f}")

                # タグ設定
                mlflow.set_tag("registered", "true")
                mlflow.set_tag("stage", "staging")
            else:
                logger.warning("モデルは登録されませんでした")
                mlflow.set_tag("registered", "false")

            # 8. 共通タグ
            mlflow.set_tag("training_type", "automatic")
            mlflow.set_tag("timestamp", datetime.now().isoformat())

            logger.info("=" * 50)
            logger.info(f"自動訓練完了: {datetime.now()}")
            logger.info(f"Run ID: {mlflow.active_run().info.run_id}")
            logger.info("=" * 50)

            return metrics

# 使用例
if __name__ == "__main__":
    # 自動訓練パイプラインの作成
    auto_pipeline = AutoMLPipeline(
        experiment_name="auto_training_demo",
        model_name="auto_classifier",
        accuracy_threshold=0.75,
        cv_folds=5
    )

    # 訓練実行
    metrics = auto_pipeline.run_training(data_source="production_db")

    print("\n=== 最終結果 ===")
    print(f"テスト精度: {metrics['test_accuracy']:.4f}")
    print(f"CV精度: {metrics['cv_mean_accuracy']:.4f} ± {metrics['cv_std_accuracy']:.4f}")

スケジュール実行の例：

import schedule
import time

def scheduled_training():
    """スケジュール訓練ジョブ"""
    auto_pipeline = AutoMLPipeline(
        experiment_name="scheduled_training",
        model_name="production_model",
        accuracy_threshold=0.80
    )

    try:
        metrics = auto_pipeline.run_training("production_db")
        logger.info(f"スケジュール訓練成功: 精度={metrics['test_accuracy']:.4f}")
    except Exception as e:
        logger.error(f"スケジュール訓練失敗: {str(e)}")

# 毎日午前2時に実行
schedule.every().day.at("02:00").do(scheduled_training)

# 毎週月曜日に実行
# schedule.every().monday.at("02:00").do(scheduled_training)

print("スケジュール訓練を開始します...")
# while True:
#     schedule.run_pending()
#     time.sleep(60)

解答：

class MLOpsMaturityAssessment:
    """MLOps成熟度評価システム"""

    def __init__(self):
        self.criteria = {
            'Level 0': {
                'データ管理': [
                    'データはローカルファイルで管理',
                    'データのバージョン管理なし',
                    'データ品質チェックは手動'
                ],
                'モデル開発': [
                    'Jupyter Notebookで開発',
                    '実験の記録は手動（Excelなど）',
                    'ハイパーパラメータは試行錯誤'
                ],
                'デプロイメント': [
                    'モデルのデプロイは手作業',
                    'デプロイ頻度は月〜年単位',
                    '本番環境との整合性確認なし'
                ],
                'モニタリング': [
                    'モデル性能の監視なし',
                    'アラート機能なし',
                    '問題発生時の対応は事後的'
                ]
            },
            'Level 1': {
                'データ管理': [
                    'データバージョニングツール使用（DVC等）',
                    '自動データ品質チェック',
                    'データ系譜の記録'
                ],
                'モデル開発': [
                    '実験管理ツール使用（MLflow等）',
                    'パラメータとメトリクスの自動記録',
                    'モデルレジストリでバージョン管理'
                ],
                'デプロイメント': [
                    '訓練パイプラインの自動化',
                    'モデルのパッケージング',
                    'ステージング環境でのテスト'
                ],
                'モニタリング': [
                    '基本的な性能監視',
                    'データドリフト検出',
                    '手動での再訓練トリガー'
                ]
            },
            'Level 2': {
                'データ管理': [
                    'データパイプラインの完全自動化',
                    '特徴量ストアの活用',
                    'データ品質の継続的監視'
                ],
                'モデル開発': [
                    'CI/CDパイプライン統合',
                    '自動テスト（データ、モデル、インフラ）',
                    '自動ハイパーパラメータ最適化'
                ],
                'デプロイメント': [
                    'コード変更からデプロイまで完全自動',
                    'カナリアデプロイメント/A/Bテスト',
                    '自動ロールバック機能'
                ],
                'モニタリング': [
                    '包括的な可観測性',
                    '自動再訓練トリガー',
                    '異常検知と自動対応'
                ]
            }
        }

        self.scores = {
            'Level 0': 0,
            'Level 1': 0,
            'Level 2': 0
        }

    def assess(self):
        """対話的な成熟度評価"""
        print("=" * 60)
        print("MLOps成熟度評価")
        print("=" * 60)
        print("各質問に 'yes' または 'no' で答えてください\n")

        for level, categories in self.criteria.items():
            print(f"\n### {level}の評価 ###\n")
            level_score = 0
            total_questions = 0

            for category, questions in categories.items():
                print(f"[{category}]")
                for i, question in enumerate(questions, 1):
                    total_questions += 1
                    while True:
                        answer = input(f"  {i}. {question}\n     → ").strip().lower()
                        if answer in ['yes', 'no', 'y', 'n']:
                            if answer in ['yes', 'y']:
                                level_score += 1
                            break
                        else:
                            print("     'yes'または'no'で答えてください")
                print()

            self.scores[level] = (level_score / total_questions) * 100
            print(f"{level}達成度: {self.scores[level]:.1f}%")

        return self.scores

    def determine_level(self):
        """成熟度レベルの判定"""
        # Level 2の基準を70%以上満たす場合
        if self.scores['Level 2'] >= 70:
            return 'Level 2', 'CI/CD完全自動化'

        # Level 1の基準を70%以上満たす場合
        elif self.scores['Level 1'] >= 70:
            return 'Level 1', 'MLパイプライン自動化'

        # それ以外
        else:
            return 'Level 0', '手動プロセス'

    def generate_recommendations(self, current_level):
        """改善推奨事項の生成"""
        recommendations = {
            'Level 0': [
                '1. 実験管理ツール（MLflow）の導入',
                '2. データバージョニング（DVC）の開始',
                '3. 訓練パイプラインのスクリプト化',
                '4. モデルレジストリの構築'
            ],
            'Level 1': [
                '1. CI/CDパイプラインの構築',
                '2. 自動テストの実装',
                '3. モデルモニタリングシステムの強化',
                '4. 自動再訓練の仕組み構築'
            ],
            'Level 2': [
                '1. 高度なA/Bテストフレームワーク',
                '2. マルチモデル管理の最適化',
                '3. MLOpsプラットフォームの統合',
                '4. 組織全体へのベストプラクティス展開'
            ]
        }

        return recommendations.get(current_level, [])

    def print_report(self):
        """評価レポートの出力"""
        print("\n" + "=" * 60)
        print("評価レポート")
        print("=" * 60)

        # スコア表示
        print("\n### 達成度スコア ###")
        for level, score in self.scores.items():
            bar = "█" * int(score / 5) + "░" * (20 - int(score / 5))
            print(f"{level}: {bar} {score:.1f}%")

        # レベル判定
        current_level, description = self.determine_level()
        print(f"\n### 現在の成熟度レベル ###")
        print(f"レベル: {current_level} - {description}")

        # 推奨事項
        print(f"\n### 次のステップへの推奨事項 ###")
        recommendations = self.generate_recommendations(current_level)
        for rec in recommendations:
            print(f"  {rec}")

        print("\n" + "=" * 60)

# 使用例（デモ用の自動評価）
def demo_assessment():
    """デモ用の評価（自動回答）"""
    assessment = MLOpsMaturityAssessment()

    # 模擬スコア（実際は対話的に評価）
    assessment.scores = {
        'Level 0': 90.0,  # Level 0の基準はほぼ満たしている
        'Level 1': 60.0,  # Level 1は部分的に達成
        'Level 2': 20.0   # Level 2はまだ初期段階
    }

    assessment.print_report()

# 実際の評価を実行する場合
# assessment = MLOpsMaturityAssessment()
# assessment.assess()
# assessment.print_report()

# デモ実行
demo_assessment()

出力例：

============================================================
評価レポート
============================================================

### 達成度スコア ###
Level 0: ██████████████████░░ 90.0%
Level 1: ████████████░░░░░░░░ 60.0%
Level 2: ████░░░░░░░░░░░░░░░░ 20.0%

### 現在の成熟度レベル ###
レベル: Level 0 - 手動プロセス

### 次のステップへの推奨事項 ###
  1. 実験管理ツール（MLflow）の導入
  2. データバージョニング（DVC）の開始
  3. 訓練パイプラインのスクリプト化
  4. モデルレジストリの構築

============================================================

評価基準の詳細：