Mímisbrunnr知恵の泉

← MLOps 一覧

🎓 レベル：標準　|　重要度：A（必須）

📎 前提：再学習と継続的トレーニング（CT）　|　関連：MLパイプラインの全体設計

要点（BLUF）

1. オーケストレーションとは

ステージ（取り込み→検証→特徴量→学習→評価→登録、MLパイプラインの全体設計）を、依存関係つきの **DAG（有向非巡回グラフ）**として宣言し、スケジューラが順序・並列・リトライ・トリガーを管理します。手作業のノートブック実行から、機械が再現実行できる自動パイプラインへの移行です。

flowchart LR
  I["データ取り込み"] --> DV["データ検証ゲート"]
  DV --> F["特徴量生成"]
  F --> TR["モデル学習"]
  TR --> MV["モデル検証ゲート"]
  MV -->|"合格"| REG["モデル登録"]
  MV -->|"不合格"| STOP["パイプライン停止と通知"]
  REG --> DEP["デプロイ（カナリア）"]
ツール特徴
Airflow汎用ワークフロー。DAG をコードで定義。重い学習は外部計算に投げる
Kubeflow PipelinesKubernetes ネイティブ。GPU 対応・ML向けの成果物管理

2. CI/CD/CT の三位一体

flowchart TB
  CI["CI：コードのテスト＋データ・モデルのテスト"] --> CD["CD：パイプラインを本番へ自動デプロイ"]
  CD --> CT["CT：新データで自動再学習・再デプロイ"]
  CT -->|"監視がトリガー（06-03）"| CT

3. ML特化のテスト：通常のソフトに「足される」もの

テスト種別何を確認するか例
データ検証スキーマ・型・範囲・欠損・分布列が揃っているか・PSIが基準内か
モデル検証精度・前モデル比較・最低保証新版が旧版を下回らないか
スキュー検査学習と推論で特徴量が一致するか学習推論スキューの防止
振る舞いテスト既知の入力に妥当な出力不変条件・方向性テスト

パイプラインの各ゲートでこれらが落ちたら、先に進めず停止・通知します。

4. 動く最小例：データ検証ゲートとモデル検証ゲート

パイプラインの2つのゲート（データ検証・モデル検証）を関数で実装し、合否で進行を止める様子を示します。

import numpy as np

def validate_data(X, schema):
    """データ検証ゲート：形状・欠損・範囲・分布を確認"""
    errors = []
    if X.shape[1] != schema["n_features"]:
        errors.append(f"列数不一致 {X.shape[1]} != {schema['n_features']}")
    if np.isnan(X).any():
        errors.append(f"欠損あり {int(np.isnan(X).sum())}件")
    if (X < schema["min"]).any() or (X > schema["max"]).any():
        errors.append("範囲外の値あり")
    return (len(errors) == 0), errors

def validate_model(new_acc, prev_acc, min_acc, tolerance=0.01):
    """モデル検証ゲート：最低保証と前モデル比較"""
    errors = []
    if new_acc < min_acc:
        errors.append(f"最低精度未達 {new_acc:.3f} < {min_acc}")
    if new_acc < prev_acc - tolerance:
        errors.append(f"前モデルから悪化 {new_acc:.3f} < {prev_acc:.3f}-{tolerance}")
    return (len(errors) == 0), errors

schema = {"n_features": 4, "min": -10.0, "max": 10.0}
rng = np.random.default_rng(0)

# データ検証：正常データと異常データ
ok_data = rng.normal(0, 1, (100, 4))
bad_data = rng.normal(0, 1, (100, 4)); bad_data[0, 0] = 999.0   # 範囲外
print("データ検証(正常):", validate_data(ok_data, schema))
print("データ検証(異常):", validate_data(bad_data, schema))

# モデル検証：改善版と劣化版
print("モデル検証(改善):", validate_model(new_acc=0.93, prev_acc=0.91, min_acc=0.85))
print("モデル検証(劣化):", validate_model(new_acc=0.88, prev_acc=0.93, min_acc=0.85))

出力：

データ検証(正常): (True, [])
データ検証(異常): (False, ['範囲外の値あり'])
モデル検証(改善): (True, [])
モデル検証(劣化): (False, ['前モデルから悪化 0.880 < 0.930-0.01'])

出力の意味：データ検証ゲートは範囲外の値（999.0）を捉えてパイプラインを止め、汚れたデータが学習に流れ込むのを防ぎます。モデル検証ゲートは、最低精度0.85は満たすものの前モデル（0.93）から悪化した0.88を不合格にしました——再学習しても改善しないモデルは本番に出しません（再学習と継続的トレーニング（CT）の「昇格は慎重に」を自動で強制）。これらのゲートが DAG の各ステージに埋め込まれ、CT で自動再学習が走っても品質が守られます。

5. 運用の勘所

なぜそうするのか

ML に CT と ML特化テストを足すのは、ML の品質はコードだけでは保証できないからです。通常のソフトはコードが正しければ動きますが、ML はデータが変われば壊れます（MLOpsとMLライフサイクル）。だから CI に「データ検証・モデル検証」を足し、コードが変わらなくても新データで再学習する CT を回す必要があります。ゲートで止める設計にするのは、自動化が「壊れたものを高速に本番へ流す装置」になるのを防ぐためです——自動化と品質ゲートはセットで初めて意味を持ちます。

⚠️ よくある落とし穴

対応 lab

関連ノート