MLOps: 実験から実稼働へ
すべてのデータ サイエンティストがこの瞬間を経験したことがあります。モデルは Jupyter ノートブックで完璧に機能し、 メトリクスは素晴らしく、チームはデモ中に拍手を送りました。次に、致命的な質問が来ます。 「いつこれを製品化しますか?」。そして沈黙が始まる。業界の推計によると、 最大 85% の機械学習プロジェクトが本番環境に到達することはありません。 モデルが機能しないからではなく、インフラストラクチャ、プロセス、 彼らを働かせるための規律 確実かつ継続的に.
MLOps (機械学習オペレーション) は、まさにこのギャップを埋めるために作成されました。 それは単一のテクノロジーではなく、一連の実践、ツール、文化です。 彼らは、孤立した実験を堅牢な実稼働 ML システムに変えます。この記事では、 MLOps が何を意味するのか、なぜそれが不可欠になったのか、具体的にどのように適用し始めるのか、 限られた予算の中でも。
何を学ぶか
- ほとんどの ML プロジェクトが本番環境に移行しない理由と、MLOps がその問題をどのように解決するか
- DevOps と MLOps の主な違い
- Google モデルに基づく MLOps 成熟度の 3 つのレベル
- 本番環境における ML モデルの完全なライフサイクル
- MLflow を使用して実験的な追跡を行う方法
- FastAPI と Docker を使用してモデルを提供する方法
- 年間 5,000 ユーロ未満で開始できるオープンソース スタック
MLOps とは何か、なぜ必要なのか
MLOps とその原則の適用 DevOps 機械学習のライフサイクルに。 DevOps が従来のソフトウェアの開発と運用を統合するのと同じように、MLOps も統合します ML システムのデータ サイエンス、エンジニアリング、運用。目標は、電子メールを自動化することです。 データの準備からトレーニング、検証まで、すべてのフェーズを再現可能にします。 導入から監視、再トレーニングまで。
DevOps と MLOps: 主な違い
ソフトウェア業界出身者は、同じ DevOps プラクティスを適用すれば十分だと考えるかもしれません。 ML モデルへ。実際には、MLOps 自体が分野となる根本的な違いがあります。
| 待ってます | DevOps | MLOps |
|---|---|---|
| アーチファクト | ソースコード | コード + データ + モデル |
| バージョン管理 | コードの Git | データとモデルのための Git + DVC |
| テスト | 単体テスト、結合テスト | データ検証、モデル検証、A/B テスト |
| CI/CD | コードのビルド、テスト、デプロイ | モデルのトレーニング、検証、デプロイ |
| 監視 | レイテンシー、エラー、稼働時間 | データドリフト、コンセプトドリフト、モデルパフォーマンス |
| 劣化 | 明示的なバグ | 時間の経過とともに静かに劣化 |
| 再現性 | 同じコード = 同じ出力 | 同じコード + 同じデータ + 同じシード = 同じ出力 |
最も重要な違いは、 静かな劣化。ソフトウェアサービス 従来の方法は機能するか機能しません。バグがある場合はエラーがスローされます。 ML モデルでできることは、 技術的なエラーがなく、漸進的な精度で予測を返し続けます。 トレーニングと比較して入力データが変更されているため、さらに悪くなります。監視なし 具体的には、ユーザーが苦情を言い始めるまで誰も気づきません。
「デスバレー」問題 ML
Gartner は、生成 AI プロジェクトの 30% がフェーズ後に放棄されると推定しています データ品質、制御が不十分なため、2025 年末までに概念実証が完了する予定です。 不十分なリスク、増大するコスト、または不透明なビジネス価値。 MLOps アドレス これらの問題のそれぞれを体系的に説明します。
MLOps 市場: 数字と傾向
MLOps 市場は目覚ましい速度で成長しています。業界分析によると、 MLOps市場の世界的な価値は、2025年には20億ドルから30億ドルになると推定されています。 2035年までに250億ドルから560億ドルに達すると予測されています。 情報源に応じて、CAGR (年平均成長率) は 29% ～ 42% です。
これらの数字は、企業が巨額の投資を行っているという具体的な現実を反映しています。 ML モデルを実稼働環境に導入します。市場の推定によると、70％以上 北米の大企業は本番環境で AI ワークロードを実行しており、55% 以上が 統合された自動モデル監視システム。ただし、3分の2近くは、 組織は依然としてパイロット段階で行き詰まり、AI を拡張できません エンタープライズレベルで。
MLOps 成熟度の 3 つのレベル
Google は標準となっている 3 層の MLOps 成熟度モデルを定義しました 事実上のこの分野の。各レベルは自動化の度合いの増加を表しており、 ML ライフサイクルにおける信頼性。
レベル 0: 手動プロセス
レベル 0 では、各ステージは手動です。データサイエンティストはノートで作業し、トレーニングします モデルをローカルに保存し、ファイルとしてエクスポートしてエンジニアリング チームに渡します。 API でラップします。自動化、監視、自動再トレーニングはありません。
| 特性 | レベル0 |
|---|---|
| トレーニング | マニュアル、ノートに |
| 展開する | 手動、.pkl または .h5 ファイルの配信 |
| 監視 | なしまたは手動 |
| 再訓練 | 明示的なリクエストがあった場合のみ |
| 再現性 | 貧乏か不在か |
このレベルは、ML をユースケースに適用し始めている組織では一般的です。 モデルがめったに更新されず、データの変更がほとんどない場合は、これで十分な場合があります。 しかし、スケールがありません。
レベル 1: ML パイプラインの自動化
レベル 1 では、トレーニングは自動化されます。 ML パイプライン。そうではありません 単一のモデルではなく、それを生成するパイプライン全体がデプロイされます。これにより、 継続的なトレーニング: 新しいデータが到着すると、パイプラインが再トレーニングされます モデルが自動的に作成されます。
| 特性 | レベル1 |
|---|---|
| トレーニング | パイプライン経由で自動 |
| 展開する | 自動化されたパイプライン |
| 監視 | モデルのパフォーマンス + トリガーの再トレーニング |
| 再訓練 | 新しいデータまたは劣化時に自動 |
| 再現性 | 良好 (パイプラインのバージョン管理) |
データが頻繁に変更されるが、ML アプローチが残る場合は、レベル 1 で十分です。 安定しています。パイプラインは同じですが、新しいデータを使用して定期的に再実行されます。
レベル 2: 機械学習のための CI/CD
レベル 2 では、完全なシステムが追加されます ML 固有の CI/CD。 データが変更されるだけでなく、パイプライン コード、機能、ハイパーパラメーターも変更されます。 モデルのアーキテクチャ。各変更は自動テスト、検証、 制御された展開。
| 特性 | レベル2 |
|---|---|
| トレーニング | パイプライン自体の自動 + CI/CD |
| 展開する | ブルー/グリーン、カナリア、A/B テスト |
| 監視 | 包括的: データ ドリフト、コンセプト ドリフト、パフォーマンス、レイテンシ |
| 再訓練 | 検証とロールバックによる自動化 |
| 再現性 | 完全 (コード + データ + 環境のバージョン管理) |
成熟した組織のレベル 2 と目標を達成します。投資が必要です インフラストラクチャと文化において重要ですが、それが数十、数百を管理する唯一の方法です 持続可能な生産モデルの構築。
MLOps のライフサイクル
本番環境における ML モデルのライフサイクルは、反復的なプロセスです。 6 つの主要なフェーズ。従来のソフトウェア開発とは異なり、このサイクルでは 終わりはありません: 運用中のモデルには継続的なメンテナンスが必要です。
+----------+ +---------+ +----------+
| DATA |---->| TRAIN |---->| EVALUATE |
| Collect | | Feature | | Validate |
| Clean | | Train | | Compare |
| Version | | Tune | | Approve |
+----------+ +---------+ +----------+
^ |
| v
+----------+ +---------+ +----------+
| RETRAIN |<----| MONITOR |<----| DEPLOY |
| Trigger | | Drift | | Stage |
| Schedule | | Metrics | | Canary |
| Auto | | Alert | | Release |
+----------+ +---------+ +----------+
1. 日付: 収集、クリーニング、バージョン管理
すべてはデータから始まります。このフェーズでは、生データが収集され、クリーニングされます。 (欠損値、外れ値、重複の処理)、それらは有用な機能に変換され、 彼らは自分自身をバージョン管理します。データのバージョン管理はモデルを再現するための基本です 正確に知る必要があります どのデータ 訓練に使われていました。 のようなツール DVC (データバージョン管理) バージョン管理を許可する Git に似た大規模なデータセット。
2. トレーニング: 特徴量エンジニアリングとトレーニング
データの準備ができたら、特徴を構築し、アルゴリズムを選択してトレーニングします。 モデル。各実験（ハイパーパラメータ、機能、アーキテクチャの組み合わせ） 関連するパラメータとメトリクスを使用して追跡されます。のようなツール MLフロー このプロセスを体系的かつ再現可能にします。
3. 評価: 検証と比較
トレーニングされたモデルは、事前定義された指標 (精度、F1 スコア、 RMSE、AUC) を使用し、現在運用中のバージョンと比較します。新しいモデルなら 最低閾値を超えていない場合、または以前の閾値より改善していない場合、彼は昇進しません。
4. デプロイ: ステージング、カナリア、リリース
承認されたモデルは、テストのためのステージングなどの先進的な環境を通過します。 統合、限られた実際のトラフィックでの検証用のカナリア、そして最後に本番環境 完了しました。のような戦略 ブルー/グリーン展開 e カナリアリリース リスクを最小限に抑えます。
5. 監視: ドリフト、メトリクス、アラート
運用環境では、モデルは継続的に監視されます。メトリクスが追跡される 技術 (レイテンシ、スループット、エラー) と ML メトリクス (実際のデータの精度、 予測の分布、データドリフト）。メトリクスが以下の場合にアラートがトリガーされます。 それらはしきい値を下回ります。
6. 再トレーニング: トリガーと自動化
監視によって劣化が検出されると、再トレーニングがアクティブ化されます。これは可能です スケジュール（例: 毎週）、トリガーベース（例: 90% 未満の精度）、または手動。 新しいモデルは、再度評価フェーズと展開フェーズを経ます。
オープンソースの MLOps スタック
MLOps の利点の 1 つは、MLOps をカバーする成熟したオープンソース エコシステムがあることです。 ライフサイクルの各段階。高価なエンタープライズ プラットフォームを購入する必要はありません 始めるには: 適切なツールを使用すると、完全な MLOps パイプラインを構築できます。
| 段階 | 楽器 | 関数 |
|---|---|---|
| データのバージョン管理 | DVC | Git と統合されたデータセットとモデルのバージョン管理 |
| 実験の追跡 | MLフロー | 各実験のパラメータ、メトリクス、アーティファクトのログ記録 |
| モデルレジストリ | MLflow モデル レジストリ | モデルのバージョン管理とプロモーション (ステージング/実稼働) |
| パイプライン オーケストレーション | プリフェクト / エアフロー | ワークフローのオーケストレーション、スケジュール設定、再試行 |
| モデルの提供 | FastAPI + Docker | コンテナ化された予測を提供するための REST API |
| コンテナ化 | ドッカー + K8 | 再現可能な環境、水平方向のスケーラビリティ |
| 監視 | プロメテウス + グラファナ | メトリクス、ダッシュボード、アラート |
| データの検証 | 大きな期待 | 自動データ品質テスト |
ノートブックからパイプラインへ: 実践例
すべての ML チームが直面する最も一般的なステップ、つまりコードの変換を見てみましょう。 モジュール式で再現可能なパイプラインの Jupyter ノートブックに書かれています。を取ってみましょう 分類の実際の例とそれを再構成する方法を見てみましょう。
前: モノリシック ノートブック
これは典型的なノートブックで、すべてが 1 つのファイル内に存在し、分割されていません。 ログやバージョン管理を行わずに説明責任を負います。
# Cella 1: Tutto nello stesso notebook
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score
import pickle
# Caricamento dati
df = pd.read_csv("data/customers.csv")
# Feature engineering inline
df["age_group"] = pd.cut(df["age"], bins=[0, 25, 45, 65, 100],
labels=["young", "adult", "senior", "elderly"])
df["total_spend"] = df["orders"] * df["avg_order_value"]
# Split
X = df[["age", "total_spend", "visits", "days_since_last"]]
y = df["churned"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
# Training - parametri hardcoded
model = RandomForestClassifier(n_estimators=100, max_depth=10)
model.fit(X_train, y_train)
# Valutazione - print a schermo
y_pred = model.predict(X_test)
print(f"Accuracy: {accuracy_score(y_test, y_pred)}")
print(f"F1: {f1_score(y_test, y_pred)}")
# Salvataggio - pickle senza versioning
with open("model.pkl", "wb") as f:
pickle.dump(model, f)
print("Modello salvato!")
モノリシックノートブックの問題
- 再生不可: シードもデータのバージョン管理もありません
- 追跡されていない: パラメーターとメトリックはノートブックの出力にのみ存在します。
- テスト不可能: テストする分離された関数がない
- 導入不可: ピクルスは API ではありません
- 保守不可能: 機能を変更するにはすべてを再実行する必要があります
変更後: モジュラーパイプライン
コードを個別のモジュールに再構成し、それぞれが特定の責任を負います。 すべての関数はテスト可能で、すべてのパラメータは構成可能で、すべてのメトリクスが追跡されます。
"""Modulo per la preparazione e trasformazione dei dati."""
import pandas as pd
from pathlib import Path
from typing import Tuple
def load_data(path: str) -> pd.DataFrame:
"""Carica il dataset dal path specificato."""
filepath = Path(path)
if not filepath.exists():
raise FileNotFoundError(f"Dataset non trovato: {path}")
return pd.read_csv(filepath)
def create_features(df: pd.DataFrame) -> pd.DataFrame:
"""Crea le feature derivate per il modello."""
result = df.copy()
result["age_group"] = pd.cut(
result["age"],
bins=[0, 25, 45, 65, 100],
labels=["young", "adult", "senior", "elderly"]
)
result["total_spend"] = result["orders"] * result["avg_order_value"]
return result
def split_data(
df: pd.DataFrame,
target_col: str = "churned",
feature_cols: list = None,
test_size: float = 0.2,
random_state: int = 42
) -> Tuple[pd.DataFrame, pd.DataFrame, pd.Series, pd.Series]:
"""Split dei dati in train/test con seed fisso per riproducibilità."""
from sklearn.model_selection import train_test_split
if feature_cols is None:
feature_cols = ["age", "total_spend", "visits", "days_since_last"]
X = df[feature_cols]
y = df[target_col]
return train_test_split(X, y, test_size=test_size, random_state=random_state)
"""Modulo per il training e la valutazione del modello."""
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
from typing import Dict, Any
import pandas as pd
def train_model(
X_train: pd.DataFrame,
y_train: pd.Series,
n_estimators: int = 100,
max_depth: int = 10,
random_state: int = 42
) -> RandomForestClassifier:
"""Addestra un RandomForestClassifier con parametri configurabili."""
model = RandomForestClassifier(
n_estimators=n_estimators,
max_depth=max_depth,
random_state=random_state
)
model.fit(X_train, y_train)
return model
def evaluate_model(
model: RandomForestClassifier,
X_test: pd.DataFrame,
y_test: pd.Series
) -> Dict[str, float]:
"""Valuta il modello e restituisce un dizionario di metriche."""
y_pred = model.predict(X_test)
return {
"accuracy": accuracy_score(y_test, y_pred),
"f1_score": f1_score(y_test, y_pred),
"precision": precision_score(y_test, y_pred),
"recall": recall_score(y_test, y_pred),
}
"""Pipeline principale che orchestra tutte le fasi."""
from src.data.preprocessing import load_data, create_features, split_data
from src.models.trainer import train_model, evaluate_model
import yaml
from pathlib import Path
def run_pipeline(config_path: str = "config.yaml") -> None:
"""Esegue l'intera pipeline ML con configurazione esterna."""
# 1. Carica configurazione
with open(config_path) as f:
config = yaml.safe_load(f)
# 2. Data preparation
print("[1/4] Caricamento dati...")
df = load_data(config["data"]["path"])
df = create_features(df)
# 3. Split
print("[2/4] Split train/test...")
X_train, X_test, y_train, y_test = split_data(
df,
test_size=config["data"]["test_size"],
random_state=config["data"]["random_state"]
)
# 4. Training
print("[3/4] Training modello...")
model = train_model(
X_train, y_train,
n_estimators=config["model"]["n_estimators"],
max_depth=config["model"]["max_depth"],
random_state=config["model"]["random_state"]
)
# 5. Evaluation
print("[4/4] Valutazione...")
metrics = evaluate_model(model, X_test, y_test)
for name, value in metrics.items():
print(f" {name}: {value:.4f}")
if __name__ == "__main__":
run_pipeline()
# config.yaml - Tutti i parametri in un unico file
data:
path: "data/customers.csv"
test_size: 0.2
random_state: 42
feature_cols:
- age
- total_spend
- visits
- days_since_last
model:
algorithm: "random_forest"
n_estimators: 100
max_depth: 10
random_state: 42
evaluation:
metrics:
- accuracy
- f1_score
- precision
- recall
min_accuracy: 0.85
モジュラーパイプラインの利点
- 再現可能: 固定シード、アウトソーシングされた構成、バージョン管理されたデータ
- テスト可能: 各機能は分離されており、専用の単体テストを行うことができます
- 保守可能: 特徴を変更してもトレーニングには影響しません、またその逆も同様です
- 設定可能: コードに触れずにハイパーパラメータを変更する
- 自動化可能: パイプラインは CI/CD で実行できます
MLflow を使用した実験の追跡
ハイパーパラメータを変更したのに、その組み合わせを思い出せなくなったことは何回ありますか 最良の結果が得られたとしたら？ザ」実験的な追跡 この問題を解決します 各実験のパラメータ、メトリクス、アーティファクトを自動的に記録します。
MLフロー 実験的な追跡用の最も人気のあるオープンソース ツールです。 実験を表示および比較するための Web UI を備えたサーバー、Python API を提供します。 ログ記録用と、モデルのライフサイクルを管理するためのモデル レジストリ。
セットアップと最初の実験
# Installazione
pip install mlflow
# Avvio del tracking server locale
mlflow server --host 127.0.0.1 --port 5000
"""Pipeline ML con experiment tracking via MLflow."""
import mlflow
import mlflow.sklearn
from src.data.preprocessing import load_data, create_features, split_data
from src.models.trainer import train_model, evaluate_model
def run_tracked_pipeline(config: dict) -> None:
"""Esegue la pipeline tracciando tutto con MLflow."""
# Imposta il tracking URI (server locale o remoto)
mlflow.set_tracking_uri("http://127.0.0.1:5000")
mlflow.set_experiment("churn-prediction")
with mlflow.start_run(run_name="rf-baseline") as run:
# Log dei parametri
mlflow.log_param("algorithm", "RandomForest")
mlflow.log_param("n_estimators", config["model"]["n_estimators"])
mlflow.log_param("max_depth", config["model"]["max_depth"])
mlflow.log_param("test_size", config["data"]["test_size"])
mlflow.log_param("random_state", config["data"]["random_state"])
# Data preparation
df = load_data(config["data"]["path"])
df = create_features(df)
X_train, X_test, y_train, y_test = split_data(
df,
test_size=config["data"]["test_size"],
random_state=config["data"]["random_state"]
)
# Log dimensioni dataset
mlflow.log_param("train_samples", len(X_train))
mlflow.log_param("test_samples", len(X_test))
mlflow.log_param("n_features", X_train.shape[1])
# Training
model = train_model(
X_train, y_train,
n_estimators=config["model"]["n_estimators"],
max_depth=config["model"]["max_depth"]
)
# Evaluation
metrics = evaluate_model(model, X_test, y_test)
# Log delle metriche
for name, value in metrics.items():
mlflow.log_metric(name, value)
# Log del modello come artefatto
mlflow.sklearn.log_model(
model,
artifact_path="model",
registered_model_name="churn-classifier"
)
# Log della configurazione come artefatto
mlflow.log_artifact("config.yaml")
print(f"Run ID: {run.info.run_id}")
print(f"Metriche: {metrics}")
いくつかの実験を実行した後、 http://127.0.0.1:5000 ブラウザで。
MLflow UI にはすべての実験を含む表が表示され、比較できるようになります。
メトリクスをパフォーマンス順に並べ替え、メトリクスとメトリクスのグラフを表示します。
モデル レジストリ: モデルのバージョン管理
コードが Git でバージョン管理されるのと同じように、ML モデルもバージョン管理される必要があります と モデルレジストリ。 MLflow モデル レジストリは集中システムを提供します モデルのライフサイクルを 3 つの段階で管理します。
| スタジアム | 説明 | 誰が使うのか |
|---|---|---|
| なし / ステージング | テストおよび検証中のモデル | データサイエンティスト、QA |
| 生産 | 承認済みモデル、実際のトラフィックに対応 | API の提供、エンドユーザー |
| アーカイブ済み | モデルは廃止され、監査のために保管されます | コンプライアンス、ロールバック |
"""Gestione del ciclo di vita del modello con MLflow Model Registry."""
from mlflow.tracking import MlflowClient
client = MlflowClient("http://127.0.0.1:5000")
# Recupera l'ultima versione del modello in staging
latest_versions = client.get_latest_versions(
name="churn-classifier",
stages=["Staging"]
)
if latest_versions:
version = latest_versions[0].version
print(f"Modello in staging: v{version}")
# Promuovi a Production dopo validazione
client.transition_model_version_stage(
name="churn-classifier",
version=version,
stage="Production",
archive_existing_versions=True # Archivia versione precedente
)
print(f"Modello v{version} promosso a Production")
# Carica il modello in produzione per inference
import mlflow.pyfunc
model = mlflow.pyfunc.load_model("models:/churn-classifier/Production")
prediction = model.predict(new_data)
デプロイ: FastAPI + Docker
実稼働環境の ML モデルは通常、次のように公開されます。 REST API. ファストAPI Python の理想的な選択: そして高速 (ASGI に基づく)、 自動ドキュメント (OpenAPI/Swagger) を生成し、優れた検証機能を備えています Pydantic 経由のデータ。によるコンテナ化 ドッカー、私たちは得ます どこにでも展開できるアーティファクト。
"""API REST per servire predizioni del modello ML."""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import mlflow.pyfunc
import pandas as pd
import logging
from typing import List
# Configurazione logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(
title="Churn Prediction API",
description="API per predizioni di churn basata su ML",
version="1.0.0"
)
class PredictionRequest(BaseModel):
"""Schema della richiesta di predizione."""
age: int = Field(..., ge=0, le=120, description="Eta del cliente")
total_spend: float = Field(..., ge=0, description="Spesa totale")
visits: int = Field(..., ge=0, description="Numero visite")
days_since_last: int = Field(..., ge=0, description="Giorni dall'ultima visita")
class PredictionResponse(BaseModel):
"""Schema della risposta di predizione."""
prediction: int
probability: float
model_version: str
# Caricamento modello all'avvio
MODEL_NAME = "churn-classifier"
MODEL_STAGE = "Production"
model = None
model_version = "unknown"
@app.on_event("startup")
async def load_model():
"""Carica il modello MLflow all'avvio del server."""
global model, model_version
try:
model_uri = f"models:/{MODEL_NAME}/{MODEL_STAGE}"
model = mlflow.pyfunc.load_model(model_uri)
model_version = model.metadata.run_id[:8]
logger.info(f"Modello caricato: {MODEL_NAME} ({model_version})")
except Exception as e:
logger.error(f"Errore caricamento modello: {e}")
raise
@app.get("/health")
async def health_check():
"""Endpoint di health check."""
return {"status": "healthy", "model_loaded": model is not None}
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
"""Genera una predizione di churn per un cliente."""
if model is None:
raise HTTPException(status_code=503, detail="Modello non caricato")
try:
input_data = pd.DataFrame([request.model_dump()])
prediction = model.predict(input_data)
probability = float(prediction[0]) if hasattr(prediction[0], '__float__') else 0.0
return PredictionResponse(
prediction=int(prediction[0]),
probability=probability,
model_version=model_version
)
except Exception as e:
logger.error(f"Errore predizione: {e}")
raise HTTPException(status_code=500, detail="Errore nella predizione")
@app.post("/predict/batch", response_model=List[PredictionResponse])
async def predict_batch(requests: List[PredictionRequest]):
"""Genera predizioni batch per più clienti."""
if model is None:
raise HTTPException(status_code=503, detail="Modello non caricato")
input_data = pd.DataFrame([r.model_dump() for r in requests])
predictions = model.predict(input_data)
return [
PredictionResponse(
prediction=int(p),
probability=float(p),
model_version=model_version
)
for p in predictions
]
# Dockerfile per il serving del modello ML
FROM python:3.11-slim
WORKDIR /app
# Dipendenze di sistema
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc \
&& rm -rf /var/lib/apt/lists/*
# Dipendenze Python
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Codice applicazione
COPY src/serving/ ./serving/
COPY config.yaml .
# Porta del servizio
EXPOSE 8000
# Healthcheck
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# Avvio con uvicorn
CMD ["uvicorn", "serving.app:app", "--host", "0.0.0.0", "--port", "8000"]
# Build dell'immagine
docker build -t churn-api:v1.0.0 .
# Avvio del container
docker run -d \
--name churn-api \
-p 8000:8000 \
-e MLFLOW_TRACKING_URI=http://mlflow-server:5000 \
churn-api:v1.0.0
# Test dell'API
curl -X POST http://localhost:8000/predict \
-H "Content-Type: application/json" \
-d '{"age": 35, "total_spend": 1250.50, "visits": 12, "days_since_last": 45}'
本番環境でのモニタリング
導入は作業の終わりではなく、新たな重要な段階の始まりです。 監視。生産中のモデルは時間の経過とともに劣化します。 変更とそれに伴うデータ。モニタリングは 3 つの主要な領域をカバーする必要があります。
追跡する指標
| カテゴリ | メトリクス | 楽器 |
|---|---|---|
| インフラストラクチャー | レイテンシー (p50、p95、p99)、スループット、HTTP エラー、CPU/RAM | プロメテウス + グラファナ |
| モデル | 精度、F1 スコア、予測分布、信頼度 | MLflow + カスタムメトリクス |
| データ | データドリフト、特徴ドリフト、欠損値、入力分布 | 明らかにAI / 大きな期待 |
データドリフトとコンセプトドリフト
次の 2 つのタイプのモデル劣化を区別することが重要です。
- 日付のドリフト: 入力データの分布は次のように変化します。 トレーニングセット。例: 25 ～ 45 歳の顧客を対象にトレーニングされたモデルは、 60歳以上のお客様へのお願い。
- コンセプトドリフト: 入力と出力の関係が変わります。例: 後 パンデミックでは顧客離れのパターンはまったく異なりますが、その特徴は 入力時には同じ分布になります。
"""Rilevamento data drift con test statistici."""
import numpy as np
from scipy import stats
from typing import Dict, Tuple
def detect_drift(
reference_data: np.ndarray,
production_data: np.ndarray,
feature_names: list,
threshold: float = 0.05
) -> Dict[str, Dict]:
"""
Rileva data drift confrontando distribuzioni con il test KS.
Args:
reference_data: dati di training (riferimento)
production_data: dati di produzione (attuali)
feature_names: nomi delle feature
threshold: soglia p-value per il drift (default 0.05)
Returns:
Report di drift per ogni feature
"""
drift_report = {}
for i, feature in enumerate(feature_names):
ref_values = reference_data[:, i]
prod_values = production_data[:, i]
# Test Kolmogorov-Smirnov
ks_stat, p_value = stats.ks_2samp(ref_values, prod_values)
drift_detected = p_value < threshold
drift_report[feature] = {
"ks_statistic": round(ks_stat, 4),
"p_value": round(p_value, 4),
"drift_detected": drift_detected,
"ref_mean": round(float(np.mean(ref_values)), 4),
"prod_mean": round(float(np.mean(prod_values)), 4),
}
if drift_detected:
print(f"DRIFT RILEVATO su '{feature}': "
f"KS={ks_stat:.4f}, p={p_value:.4f}")
return drift_report
再トレーニングを再開するタイミング
すべてのドリフトが直ちに再トレーニングを必要とするわけではありません。明確なしきい値の定義: データ ドリフト 重要な機能、精度の 5% を超える低下、または予測の分布 著しくアンバランス。過剰な再トレーニングを避ける 不安定。
年間 5,000 ユーロ未満の予算で始める方法
MLOps は、必ずしも数十万ユーロの費用がかかるエンタープライズ プラットフォームを意味するわけではありません。 イタリアの中小企業または小規模チームの場合、MLOps インフラストラクチャを構築することが可能です オープンソース ツールと最小限のコストで効果的です。
中小企業に提案するスタック
| 成分 | 解決 | 年間コスト |
|---|---|---|
| コード | GitHub 無料 / GitLab CE | 0ユーロ |
| データのバージョン管理 | DVC + Google Cloud Storage (5 GB 無料) | 0～50ユーロ |
| 実験の追跡 | 安価な VM 上の MLflow | 200～500ユーロ |
| トレーニング | Google Colab Pro / スポット VM | 120～600ユーロ |
| 給仕 | VM 上の FastAPI (2 vCPU、4 GB RAM) | 300 - 800 ユーロ |
| 監視 | Prometheus + Grafana (自己ホスト型) | 0 ユーロ (同じ VM 上) |
| CI/CD | GitHub アクション (2000 分/月無料) | 0ユーロ |
| コンテナレジストリ | GitHub コンテナ レジストリ | 0ユーロ |
推定合計額: 620 ～ 1,950 ユーロ/年、5,000ユーロの基準を大幅に下回っています。 このスタックは、中程度のトラフィック量で最大 5 ～ 10 個のモデルを実稼働環境でサポートします。 (1 日あたり数千件の予測)。
コスト削減のヒント
- スポット/プリエンプティブル VM: 緊急でないトレーニングを最大 70% 節約
- 自動スケーリング: リクエストがない場合はゼロにスケールします
- モデルの圧縮: より小さいモデル = 提供するリソースが少なくなる
- バッチ推論: リアルタイムの予測が必要ない場合は、夜間バッチを使用してください
- マルチテナント: すべてのプロジェクトに単一の MLflow/Grafana インフラストラクチャを適用
MLOps プロジェクトの構造
最後にすぐに使えるものを紹介します。フォルダー構造は次のとおりです。 MLOps プロジェクトに推奨されます。この組織は、 説明責任と自動化、テスト、コラボレーションの容易さ。
churn-prediction/
data/
raw/ # Dati grezzi (versionati con DVC)
processed/ # Dati trasformati
data.dvc # File di tracking DVC
src/
data/
preprocessing.py # Pulizia e feature engineering
validation.py # Validazione qualità dati
models/
trainer.py # Logica di training
evaluator.py # Valutazione e metriche
serving/
app.py # FastAPI application
schemas.py # Pydantic schemas
monitoring/
drift_detector.py # Rilevamento drift
metrics.py # Metriche custom
pipeline.py # Orchestrazione pipeline
tests/
test_preprocessing.py
test_trainer.py
test_api.py
config.yaml # Configurazione pipeline
Dockerfile # Container per serving
docker-compose.yaml # Stack locale completo
requirements.txt # Dipendenze Python
.dvc/ # Configurazione DVC
.github/
workflows/
train.yaml # CI/CD per training
deploy.yaml # CI/CD per deployment
mlflow/ # Artefatti MLflow (locale)
README.md
結論と次のステップ
MLOps は、大手テクノロジー企業だけが使える贅沢品ではありません。そして必需品 信頼性が高く持続可能な方法で ML モデルを本番環境に導入したいと考えている人なら誰でも。 この記事では、問題の理解から基本的な事項について説明しました。 （ML プロジェクトが失敗する理由）から具体的な解決策（モジュール式パイプライン、実験）まで 追跡、モデル レジストリ、コンテナ化された提供と監視)。
重要なのは、段階的に始めることです。モデルのレベル 2 に到達する必要はありません 初日から Google の成熟度を実感できます。レベル 0 から適切な実践を始めてください。
- すぐに： ノートブックのコードをモジュールに分割します。 config.yaml を使用します。
- 1週目： MLflow を追加して実験を追跡します。
- 2週目: FastAPI + Docker を使用してモデルをコンテナ化します。
- 1 か月目: GitHub Actions を使用して CI/CD パイプラインを実装します。
- 2 か月目: Prometheus によるモニタリングと基本的なアラートを追加します。
- 3 か月目: データのバージョン管理のために DVC を実装します。
シリーズの次の記事では、データ管理の各コンポーネントをさらに詳しく掘り下げていきます。 DVC を使用した、ML 固有の CI/CD パイプラインの作成、高度なモニタリング 明らかに、Kubernetes 上の AI とスケーラブルなデプロイメントです。どのアイテムも実用的で、 実用的なコードと詳しい手順。
シリーズロードマップ
- 第2条： DVC - ML のデータのバージョニング
- 第3条： MLflow の詳細 - 高度な実験追跡
- 第4条： GitHub Actions を使用した機械学習のための CI/CD
- 第5条： 本番環境での機能ストアと機能エンジニアリング
- 第6条： Kubernetes で提供されるスケーラブルなモデル
- 第7条： 高度な監視: データドリフトと明らかに AI
- 第8条： ガバナンス、コンプライアンス、責任ある ML