はじめに

機械学習モデルの開発から運用までを効率的に管理する「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の流れを一周体験してみたい」という方にとって、実用的な最初のステップになれば幸いです！