ノートブックからサービスへ: パッケージング、FastAPIエンドポイント、スモークテスト

ノートブックの中のモデルは1人の役に立ちます。HTTPエンドポイントの向こうにあるモデルは、要求を送れるあらゆるプログラムの役に立ちます。本レッスンは、その2つの間の最短の正直な道筋をたどります。scikit-learnパイプラインを学習し、ファイルに保存し、小さなFastAPIアプリケーションで読み込み、すべての要求を検証し、サーバーを起動する前に全体をプロセス内でテストします。ノートブックで動くモデルがあり、「アプリから呼べる場所に置いて」と頼まれた人のためのレッスンです。すべてノートPCのCPUで数秒で動き、scikit-learn以外に必要なパッケージはFastAPIとuvicornだけです。

モデルは意図的に小さく、irisデータセットに対するロジスティック回帰です。レッスンのどこもモデルの話にならないようにするためです。同じ構成は、predictの本体を変えるだけでレッスン05の勾配ブースティングモデルやレッスン07のネットワークにも使えます。

推定器ではなくパイプラインを保存する

サービングの第一の規則は、前処理はモデルの一部だということです。レッスン02では、スケーラーを学習データだけで当てはめることを学びました。スケーラーが分類器と別のオブジェクトなら、サービスはそれを正しい順序で、正しい学習済み統計量で適用することを覚えておかなければならず、それはバグの温床です。scikit-learnのPipelineは複数のステップを1つのpredictを持つ1つのオブジェクトにまとめるので、保存するものと提供するものが、評価したものと同じになります。

下のブロックはパイプラインを学習し、確認し、特徴量名、クラス名、バージョン文字列とともにjoblibで保存します。メタデータは重要です。半年後にこのファイルを読み込むのは、どの列が何だったか覚えていない人だからです。

import joblib
import numpy as np
from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

iris = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
    iris.data, iris.target, test_size=0.25, stratify=iris.target, random_state=0
)
pipeline = Pipeline([
    ("scale", StandardScaler()),
    ("clf", LogisticRegression(max_iter=1000)),
])
pipeline.fit(X_train, y_train)
print(f"test accuracy: {pipeline.score(X_test, y_test):.3f}")

bundle = {
    "pipeline": pipeline,
    "feature_names": list(iris.feature_names),
    "class_names": [str(c) for c in iris.target_names],
    "version": "2026-11-13",
}
joblib.dump(bundle, "iris_model.joblib")
loaded = joblib.load("iris_model.joblib")
print("reloaded classes:", loaded["class_names"])
print("prediction for one flower:", loaded["pipeline"].predict([[5.1, 3.5, 1.4, 0.2]]))

test accuracy: 0.974
reloaded classes: ['setosa', 'versicolor', 'virginica']
prediction for one flower: [0]

joblibはscikit-learnが推奨するシリアライザです。中身はpickleで、大きなnumpy配列の扱いが改善されています。これには2つの警告が付きます。pickleファイルは読み込み時にコードを実行するので、信頼できない出所のものを決して読み込んではいけません。そしてpickleはそれを書いたライブラリのバージョンに縛られます。scikit-learnは別のバージョンで書かれたファイルに対して警告を出し、読み込めることをまったく約束しません。requirements.txtでバージョンを固定し、バンドルにも記録してください。ライブラリのバージョンを超えて生き残らなければならないモデルにはONNXが可搬な形式で、変換の価値がある場面はMLOpsとデプロイのハブで扱います。

サービス

図は3つの部分と、それぞれの境界を越えるものを示しています。クライアントはHTTPでJSONを送ります。APIはそれを検証し、モデルが期待する配列に変換し、モデルの出力をJSONに戻します。モデルはHTTPを見ることがなく、クライアントはnumpyを見ることがありません。

これがアプリケーションファイルです。uvicornで実行されるアプリを定義するものであり、このサイトのすべてのPythonブロックを実行するガードは接続を待ち受けるサーバーを起動できないので、# snippetの印が付いています。実行可能なバージョンは次の節にあります。

# snippet
# app.py
import joblib
from fastapi import FastAPI
from pydantic import BaseModel, Field

bundle = joblib.load("iris_model.joblib")       # load once, at import, not per request
app = FastAPI(title="iris-service")


class Measurements(BaseModel):
    sepal_length: float = Field(gt=0, lt=20)    # centimetres
    sepal_width: float = Field(gt=0, lt=20)
    petal_length: float = Field(gt=0, lt=20)
    petal_width: float = Field(gt=0, lt=20)


@app.get("/health")
def health() -> dict:
    return {"status": "ok", "model_version": bundle["version"]}


@app.post("/predict")
def predict(m: Measurements) -> dict:
    row = [[m.sepal_length, m.sepal_width, m.petal_length, m.petal_width]]
    probs = bundle["pipeline"].predict_proba(row)[0]
    best = int(probs.argmax())
    return {"species": bundle["class_names"][best], "probability": round(float(probs[best]), 3)}

上から下へ読んでください。モデルはモジュールのインポート時に一度だけ読み込まれます。これは小さなサービスで最も重要な性能上の判断です。要求ごとにjoblibファイルを読み込めば、1ミリ秒の予測が100ミリ秒の予測になります。Measurementsはpydanticモデルです。FastAPIはこれを使ってJSONボディを解析し、合致しないボディをすべての問題点を列挙した422応答で拒否し、/docsに対話的なドキュメントを生成します。各フィールドの範囲制限はゴミ入力に対する最初の防衛線です。負の花弁長は花ではありませんが、モデルはそれに対しても平然と自信に満ちた答えを返します。

predict関数は、検証済みのフィールドを学習時と同じ列順でパイプラインが期待する1 x 4のリストに変換し、predict_probaを呼び、最良のクラスをその確率とともに返します。明示的なint(...)とfloat(...)が重要です。numpyのスカラーはJSONにシリアライズできず、それなしで出るエラーは役に立たないからです。

サーバーなしのスモークテスト

FastAPIにはTestClientが付属しており、アプリをプロセス内で呼び出します。ポートもネットワークもサブプロセスも不要です。下のブロックは完結していて実行可能です。モデルを学習して保存し（ブロック単体で成立するよう最初のブロックを再掲しています）、同じアプリを定義し、意図的に不正な要求1つを含めて両方のエンドポイントを試します。

import joblib
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field
from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

# Train and save, as in the first block, so this block stands alone.
iris = load_iris()
pipeline = Pipeline([("scale", StandardScaler()), ("clf", LogisticRegression(max_iter=1000))])
pipeline.fit(iris.data, iris.target)
joblib.dump({"pipeline": pipeline, "feature_names": list(iris.feature_names),
             "class_names": [str(c) for c in iris.target_names], "version": "2026-11-13"}, "iris_model.joblib")

# The same app as app.py, loaded from the saved file.
bundle = joblib.load("iris_model.joblib")
app = FastAPI(title="iris-service")


class Measurements(BaseModel):
    sepal_length: float = Field(gt=0, lt=20)
    sepal_width: float = Field(gt=0, lt=20)
    petal_length: float = Field(gt=0, lt=20)
    petal_width: float = Field(gt=0, lt=20)


@app.get("/health")
def health() -> dict:
    return {"status": "ok", "model_version": bundle["version"]}


@app.post("/predict")
def predict(m: Measurements) -> dict:
    row = [[m.sepal_length, m.sepal_width, m.petal_length, m.petal_width]]
    probs = bundle["pipeline"].predict_proba(row)[0]
    best = int(probs.argmax())
    return {"species": bundle["class_names"][best], "probability": round(float(probs[best]), 3)}


# The smoke test: no server, no network, the client calls the app in-process.
client = TestClient(app)

r = client.get("/health")
print(r.status_code, r.json())

r = client.post("/predict", json={"sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2})
print(r.status_code, r.json())

r = client.post("/predict", json={"sepal_length": 6.7, "sepal_width": 3.0, "petal_length": 5.2, "petal_width": 2.3})
print(r.status_code, r.json())

r = client.post("/predict", json={"sepal_length": -1, "sepal_width": 3.0, "petal_length": 5.2})
print(r.status_code, r.json()["detail"][0]["msg"], "|", r.json()["detail"][1]["msg"])

200 {'status': 'ok', 'model_version': '2026-11-13'}
200 {'species': 'setosa', 'probability': 0.985}
200 {'species': 'virginica', 'probability': 0.961}
422 Input should be greater than 0 | Field required

最後の要求は負の長さと欠けたフィールドを持ち、応答は最初の問題で止まるのではなく、1つの422で両方の問題を列挙しています。モデルには何も届いていません。テストで確認すべきなのはこの振る舞いです。学習時に見たことのない入力からモデルを守るのがこの振る舞いだからです。

実際のプロジェクトでは、スモークテストはtests/test_app.pyにステータスコードと品種をassertする関数として置かれ、デプロイのたびにpytestで実行されます。1秒で終わり、モデルファイルの欠落、フィールド名の変更、壊れた依存関係を捕まえるテストは、事後のどんな監視よりも価値があります。表形式データ分類器プロジェクトは、より大きなモデルとDockerfileでこれを発展させます。

プロジェクトの構成

他の人が実行できるサービスには、2つのファイル以上のものが必要です。最小の構成はこうです。

iris-service/
  app.py              the FastAPI application above
  train.py            the first block: fits the pipeline and writes iris_model.joblib
  iris_model.joblib   the artefact (commit it for a tiny model, or build it in CI for a large one)
  requirements.txt    pinned dependencies
  tests/
    test_app.py       the smoke test, as pytest functions
  README.md           how to train, run and call it

requirements.txtは正確なバージョンを固定します。モデルファイルとそれを読むライブラリは一致していなければならないからです。バージョンは、モデルを学習した環境でpip freezeした結果に置き換えてください。

scikit-learn==1.9.1
numpy==2.5.4
joblib==1.6.0
fastapi==0.162.7
uvicorn==0.54.0
pydantic==2.14.3
httpx==0.28.1
pytest

httpxがあるのは、TestClientがその上に構築されているからです。テストランナーはモデルファイルに触れないので固定していません。上の正確なバージョンは本稿執筆時点のField Guideの環境のもので、これを読むころには古くなっています。pip freezeを実行して自分のものを使ってください。

実行する

一度学習し、サーバーを起動し、別のターミナルから呼び出します。--reloadはapp.pyが変わるとサーバーを再起動します。開発中は便利ですが、本番で使ってはいけません。

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python train.py
uvicorn app:app --host 127.0.0.1 --port 8000 --reload

curl -s http://127.0.0.1:8000/health

curl -s -X POST http://127.0.0.1:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"sepal_length": 6.7, "sepal_width": 3.0, "petal_length": 5.2, "petal_width": 2.3}'

最初の呼び出しは{"status":"ok","model_version":"2026-11-13"}を返し、2つ目は{"species":"virginica","probability":0.961}を返します。ブラウザでhttp://127.0.0.1:8000/docsを開くと生成されたドキュメントが見られ、pydanticモデルから作られた各エンドポイントのフォームが含まれています。

サーバーは127.0.0.1にバインドされているので、同じマシン上のプログラムだけが到達できます。0.0.0.0にバインドするとネットワークに公開され、その時点でサービスの前段に認証、レート制限、HTTPSが必要になりますが、本レッスンはそのどれも提供しません。それらはアプリケーションではなく、リバースプロキシやAPIゲートウェイの仕事です。

意図的に省いたもの

これは最小構成であり、次のステップが何かを知っておけば、必要になったときに、必要になる前ではなく、追加できます。クライアントが多数の行を一度に送るバッチ予測は、Measurementsのリストを受け取る2つ目のエンドポイントです。すべての要求とその予測をモデルバージョンとともにログに記録することは、後のあらゆる監視の基礎であり、標準のloggingモジュールで5行で済みます。Dockerfileは別のマシンでも環境を再現可能にします。インポート時ではなくFastAPIのlifespanハンドラでモデルを読み込めば、ファイルの読み込み中にヘルスチェックに失敗するのではなく、起動中であると報告できます。そしてPyTorchモデルでは、torch.set_num_threadsと単一ワーカープロセスで、サーバーとモデルがCPUコアを奪い合うのを防げます。MLOpsとデプロイのハブは、これらをそれぞれが通常どれくらい早く必要になるかの順に並べています。

演習

1. 計測オブジェクトのJSONリストを受け取り結果のリストを返すPOST /predict_batchエンドポイントを追加し、pydanticで1要求あたり100行の上限を強制してください。スモークテストを拡張して、3行、次に101行を送ってください。
2. 保存するバンドルにcreated_atタイムスタンプとscikit-learnのバージョンを追加し、両方を/healthから返してください。次に、別のscikit-learnバージョンの新しい環境でバンドルを読み込み、どんな警告やエラーが出るか正確に記録してください。
3. irisモデルを、load_breast_cancer（30特徴量）に対するレッスン05の勾配ブースティング分類器に置き換えてください。pydanticモデルで30個のフィールドにどう名前を付けるか、固定長のfeatures: list[float]1つのほうがよいインターフェースかどうかを決めてください。トレードオフを三文で書いてください。

まとめ

• 前処理を含むパイプライン全体を、特徴量名、クラス名、バージョン文字列とともにjoblibで保存します。pickleファイルはライブラリのバージョンに縛られ、信頼できる出所からのみ読み込まなければなりません。
• モデルは起動時に一度だけ読み込み、要求ごとには決して読み込みません。API層がJSONと配列の間を変換するので、モデルのコードはノートブックと同じままです。
• pydanticモデルがすべての要求を検証し、モデルが呼ばれる前にすべての問題を422で返します。フィールドの範囲制限は、モデルが見たことのない入力に対する最初の防衛線です。
• TestClientはサーバーなしでアプリをプロセス内で動かすので、CIで1秒のスモークテストが可能になります。
• 依存関係を固定し、認証とプロキシが整うまではlocalhostにバインドし、READMEが1画面に収まるくらい構成を小さく保ちます。

さらに読む

• FastAPI documentation: 要求ボディ、パスオペレーション、生成されるドキュメント。
• FastAPI documentation, Testing: TestClientと上で使ったpytestの構成。
• scikit-learn user guide, Model persistence: joblib、そのセキュリティとバージョンに関する注意、ONNXとskopsという代替。
• Pydantic documentation, Fields: 要求フィールドに使える制約。
• Uvicorn documentation, Deployment: ワーカー、プロセスマネージャ、プロキシの背後での実行。