はじめに
機械学習モデルの開発から運用までを効率的に管理する「MLOps（Machine Learning Operations）」の重要性が高まっています。
しかし、KubeflowやSageMakerのような本格的なMLOpsプラットフォームは、個人開発や小規模なプロジェクトにはオーバースペックなこともしばしばあります。
そこで今回は、MLflow × FastAPI × Dockerを使って構成した、「最小限のMLOps構成」 をご紹介します。
How：どうやって構築したか
全体構成
全体構成
[Jupyter Notebook]
↓ モデル学習・登録
[MLflow Tracking Server]
↓ ステージから選択
[FastAPI サーバー]
→ APIで推論提供 (Swagger UIで確認)
上記を一貫して構築できる、実践的なミニマル構成です。
Why：なぜこの構成を選んだのか
1. 最小限の構成で始められるから
| ツール | 用途 | 特徴 |
|---|---|---|
| MLflow | モデル管理・追跡・ステージ管理 | ローカルで完結、Web UI付き |
| FastAPI | API提供 | 自動ドキュメント生成（Swagger UI）、高速 |
| Jupyter | 環境構築 | 簡易かつ再現性の高い実行環境 |
2. 開発から運用まで一貫した管理ができる
- Jupyterで開発したモデルをそのままMLflowに登録・管理
- MLflowでのステージ管理により、APIのモデル切り替えが容易
- FastAPIでAPIを提供し、即座に検証・利用可能
実際の運用方法
1. モデルの開発とデプロイ（Jupyter Notebook）
from sklearn.ensemble import RandomForestRegressor
from sklearn.datasets import fetch_california_housing
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error
import mlflow
import mlflow.sklearn
from mlflow.models.signature import infer_signature
# データ準備
X, y = fetch_california_housing(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y)
# 学習
reg = RandomForestRegressor().fit(X_train, y_train)
y_pred = reg.predict(X_test)
# モデル登録
signature = infer_signature(X_train, y_pred)
input_example = X_train[:5]
with mlflow.start_run():
mlflow.log_param("n_estimators", 100)
mlflow.log_metric("rmse", mean_squared_error(y_test, y_pred, squared=False))
mlflow.sklearn.log_model(
sk_model=reg,
artifact_path="random_forest_regressor_model",
signature=signature,
input_example=input_example,
registered_model_name="CaliforniaHousingModel"
)
2. MLflowでのモデル管理とステージ設定
モデルレジストリ画面（例）
以下のようにモデルのバージョン管理がUIで行うことができます。
- Staging：テスト用
- Production：本番環境
- Archived：過去のバージョン
精度やハイパーパラメータも記録可能
3. FastAPIでモデル提供
from fastapi import FastAPI
from pydantic import BaseModel
import pandas as pd
import mlflow.sklearn
model = mlflow.sklearn.load_model("models:/CaliforniaHousingModel/Production")
app = FastAPI()
class InputData(BaseModel):
MedInc: float
HouseAge: float
AveRooms: float
AveBedrms: float
Population: float
AveOccup: float
Latitude: float
Longitude: float
@app.post("/predict")
def predict(data: InputData):
df = pd.DataFrame([data.dict()])
pred = model.predict(df)
return {"prediction": pred.tolist()}
なおFastAPIは、Swagger UI で即検証できます
工夫した点
1. MLflowトラッキングの活用
- signature, input_example を設定することで、モデルがどんな形式の入力を想定しているかを明示的に記録できます。
- log_param() によって、学習に使用したハイパーパラメータを追跡可能に。
- log_metric() を使えば、モデルごとの評価値（例：RMSEや精度）も簡単に比較可能です。
- 入力データのハッシュを記録することも推奨（例：hashlib.sha256(X_train).hexdigest()）
- これにより、学習時に使用したデータが後で完全に一致するかどうかを検証でき、モデルの再現性が担保されます。
特にチーム開発や将来の再学習時に、「同じデータを使ったか？」の確認が容易になります。
- これにより、学習時に使用したデータが後で完全に一致するかどうかを検証でき、モデルの再現性が担保されます。
2. FastAPIの設計工夫
- Pydantic によるスキーマ定義で、型安全かつエラー抑制
- /docsでAPIドキュメントが即座に自動生成され、開発効率が高い
- 非同期処理も可能なので、拡張性が高い
So what：この構成で何が得られるか？
- 最小限の構成でMLOpsの主要機能を体験可能
- PoCや個人プロジェクトに最適な軽量構成
- 将来的なCI/CDやモニタリングの拡張も視野に入れられる
今後の展望
今後は以下のような機能追加も検討可能です：
- モデルの自動再学習＋スケジューリング
- FastAPIにおけるログ収集と監視（Prometheus, Evidentlyなど）
- 入力チェック強化やエラーハンドリング
参考リンク
この構成が、「とりあえずMLOpsの流れを一周体験してみたい」という方にとって、実用的な最初のステップになれば幸いです！