FastAPIエンドポイント付きの表形式データ分類器
乳がんデータセットでscikit-learnのパイプラインを学習し、joblibで保存し、/predictと/healthを持つFastAPIアプリから提供します。サーバーを起動する前に、アプリをインプロセスでテストします。
| 実行環境 | ノートPCのCPU、任意のOS、5分以内 |
|---|---|
| 所要時間 | 約2時間 |
| 使用ライブラリ | sklearn, python |
| 追加パッケージ | fastapi、uvicorn、joblib |
作るもの
このプロジェクトは、学習済みモデルから動作するHTTPサービスまでの、最短で誠実な道筋をたどります。同梱の乳がんデータセットでscikit-learnのパイプライン（スケーラーの後にロジスティック回帰）を学習し、パイプライン全体をjoblibで1つのファイルに保存し、2つのルートを持つFastAPIアプリケーションでラップします。GET /healthはモデルが読み込まれているかを返します。POST /predictは30個の数値を受け取り、クラスラベルと確率を返します。次に、FastAPIのTestClientでアプリケーションをインプロセスでテストします。これはポートを開かずに、実際のルーティング・バリデーション・シリアライズのコードを動かします。その後ではじめてuvicornを起動し、curlで呼び出します。
意図的に小さなプロジェクトにしています。モデルはあえて退屈なものにして、サービング側に注意を集中させます。入門レッスンの評価とノートブックからサービスへを終えているなら、それらを使って最初に作るべきものがこれです。このパターン（パイプラインファイル、メタデータファイル、lifespanでのローダー、型付きのリクエストとレスポンス、インプロセスのテスト）は、勾配ブースティング木、PyTorchモデル、そして一度読み込んで何度も呼び出せるあらゆるものに、そのまま持ち越せます。
必要なもの
- Python 3.10以降と
scikit-learn、numpy、joblib。最近のバージョンなら何でも動きます。当サイトではscikit-learn 1.9.1でテストしました。 fastapi、uvicorn、httpx。pip install fastapi uvicorn httpxでインストールします。FastAPIはpydanticとstarletteを一緒に入れます。httpxはfastapi.testclientがインプロセスのリクエストを送るのに使うもので、素のfastapiパッケージではインストールされません（fastapi[standard]ならされます）。- 最後の確認用に
curlか任意のHTTPクライアント。 - GPUは不要、インストール後のネットワークも不要、ダウンロードも不要です。
load_breast_cancerはscikit-learnに同梱されています。
以下のすべてのPythonブロックは、Apple SiliconのMac mini上で、Field Guideのパイプライン環境を使い、空のディレクトリから実行しました。所要時間は各ブロックに記載しています。
設計
このサービスは3つの部分と、それらの間の明確な境界からなります。
学習はサービングとは別のプログラムです。 train.pyがパイプラインを学習し、2つのファイルを書き出します。model.joblibにはスケーラーと分類器が一緒に入っているので、サーバーが誤ったスケーリングを適用することはありえません。model_meta.jsonには順序どおりの特徴量名、クラス名、テスト精度が入っているので、サーバーは学習データをインポートせずに入力を検証し、出力にラベルを付けられます。サーバーは何も学習しません。
モデルは起動時に一度だけ読み込みます。 FastAPIのlifespanフックは、最初のリクエストの前と最後のリクエストの後に実行されます。ルートの中ではなくそこで読み込めば、最初のリクエストも1000番目のリクエストと同じ速さになり、モデルファイルが欠けていれば負荷の下でエラーを返すのではなく起動時にプロセスが失敗します。
pydanticモデルが契約を定義します。 list[float]フィールドを持つPredictRequestは、文字列やネストしたオブジェクト、キーの欠落を、コードが実行される前にHTTP 422で拒否します。特徴量の個数はpydanticにはできない意味的なチェックなので、ルートがこれを行い、平易なメッセージ付きで同じ422を送出します。
テストはサーバーではなく、アプリオブジェクトに対して実行します。 TestClientはASGIアプリケーションを直接駆動します。コンテキストマネージャとして使えばlifespanも実行されるので、テストは起動処理もカバーします。継続的インテグレーションで実行するのはこれです。最後のuvicornのステップは、実際に動いているところを見るためのものです。
知っておくべきトレードオフが1つあります。joblibファイルはPythonのpickleです。信頼できるソースからしか安全に読み込めず、書き出したscikit-learnのバージョンに縛られます。1時間前に自分で学習したモデルなら、どちらも問題ありません。長期間運用するサービスでは、requirementsでライブラリのバージョンを固定し、メタデータファイルにも記録してください。
コード
パイプラインの学習と保存
これをtrain.pyとして保存します。データセットは細胞画像から得た30個の計測値を持つ569行で、悪性か良性かのラベルが付いています。層化80/20分割により、両方の半分でクラスの比率が同じに保たれます。Pipelineが重要です。生の入力に対するpredict_probaが、まず学習時の統計量で標準化を行うことを意味します。
# train.py: fit a scikit-learn pipeline on the breast cancer data and save it.
import json
import joblib
import numpy as np
from sklearn.datasets import load_breast_cancer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
SEED = 0
data = load_breast_cancer()
X, y = data.data, data.target
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=SEED, stratify=y
)
pipe = Pipeline([
("scale", StandardScaler()),
("clf", LogisticRegression(max_iter=1000, random_state=SEED)),
])
pipe.fit(X_train, y_train)
acc = pipe.score(X_test, y_test)
joblib.dump(pipe, "model.joblib")
# The API needs the feature names and the class labels, so save them too.
meta = {
"features": list(data.feature_names),
"classes": list(data.target_names),
"test_accuracy": round(float(acc), 4),
}
with open("model_meta.json", "w") as f:
json.dump(meta, f, indent=2)
print(f"train rows: {len(X_train)}, test rows: {len(X_test)}")
print(f"test accuracy: {acc:.4f}")
print(f"saved model.joblib and model_meta.json ({len(meta['features'])} features)")期待される出力です。計算は約1秒です（Mac miniでの初回実行は10秒かかりましたが、ほぼすべてがネットワークボリュームからのscikit-learnのインポートでした）。
train rows: 455, test rows: 114
test accuracy: 0.9825
saved model.joblib and model_meta.json (30 features)モデルファイルは約2 kBです。ロジスティック回帰は特徴量ごとに1つの重みとバイアスを保持し、スケーラーは特徴量ごとに平均と標準偏差を保持します。
アプリケーション
これを同じディレクトリにapp.pyとして保存します。何かを出力するスクリプトではなく、uvicornで実行するファイルなので、スニペットとして示しています。
# snippet
# app.py: serve the saved pipeline over HTTP. Run with: uvicorn app:app --port 8000
import json
from contextlib import asynccontextmanager
from pathlib import Path
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
MODEL_PATH = Path("model.joblib")
META_PATH = Path("model_meta.json")
state: dict = {"model": None, "meta": None}
@asynccontextmanager
async def lifespan(app: FastAPI):
# Runs once at startup, before the first request is served.
state["model"] = joblib.load(MODEL_PATH)
state["meta"] = json.loads(META_PATH.read_text())
yield
state["model"] = None
app = FastAPI(title="Tumour classifier", version="1.0", lifespan=lifespan)
class PredictRequest(BaseModel):
features: list[float] = Field(..., description="30 values in model_meta.json order")
class PredictResponse(BaseModel):
label: str
probability: float
model_version: str
@app.get("/health")
def health() -> dict:
ready = state["model"] is not None
return {"status": "ok" if ready else "loading", "model_loaded": ready}
@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest) -> PredictResponse:
meta = state["meta"]
expected = len(meta["features"])
if len(req.features) != expected:
raise HTTPException(
status_code=422,
detail=f"expected {expected} features, got {len(req.features)}",
)
x = np.asarray(req.features, dtype=float).reshape(1, -1)
proba = state["model"].predict_proba(x)[0]
idx = int(np.argmax(proba))
return PredictResponse(
label=meta["classes"][idx],
probability=round(float(proba[idx]), 4),
model_version=app.version,
)説明に値する選択がいくつかあります。ルートはasync defではなく、素のdefです。scikit-learnの推論は同期的なCPU処理であり、FastAPIは同期ルートをスレッドプールで実行するので、遅い予測がイベントループを止めることはありません。state辞書は、lifespanからルートへオブジェクトを渡す最も単純な方法です。response_modelにより、FastAPIは返す値を検証し、/docsに生成されるOpenAPIドキュメントにその形を公開します。レスポンスにはmodel_versionが含まれるので、クライアントはどのモデルが応答したかを判別できます。これは2つ目のモデルを初めてデプロイするときに重要になります。
インプロセスでテストする
このブロックは単独で実行できます。モデルファイルがなければ学習ステップを繰り返し、app.pyと同じアプリケーションを定義し、TestClientを通じて呼び出します。実際のプロジェクトでは、テストファイルはコードを繰り返すのではなくfrom app import appとします。ここで繰り返しているのは、ブロックが空のディレクトリから実行できるようにするためです。
# Self-contained test: trains the model if needed, defines the same app, calls it in-process.
import json
from contextlib import asynccontextmanager
from pathlib import Path
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field
from sklearn.datasets import load_breast_cancer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
MODEL_PATH = Path("model.joblib")
META_PATH = Path("model_meta.json")
# --- Step 1 again, so this block works in an empty directory ---
data = load_breast_cancer()
if not MODEL_PATH.exists():
X_train, X_test, y_train, y_test = train_test_split(
data.data, data.target, test_size=0.2, random_state=0, stratify=data.target
)
pipe = Pipeline([("scale", StandardScaler()),
("clf", LogisticRegression(max_iter=1000, random_state=0))])
pipe.fit(X_train, y_train)
joblib.dump(pipe, MODEL_PATH)
META_PATH.write_text(json.dumps({
"features": list(data.feature_names),
"classes": list(data.target_names),
"test_accuracy": round(float(pipe.score(X_test, y_test)), 4),
}))
# --- The app, identical to app.py ---
state: dict = {"model": None, "meta": None}
@asynccontextmanager
async def lifespan(app: FastAPI):
state["model"] = joblib.load(MODEL_PATH)
state["meta"] = json.loads(META_PATH.read_text())
yield
state["model"] = None
app = FastAPI(title="Tumour classifier", version="1.0", lifespan=lifespan)
class PredictRequest(BaseModel):
features: list[float] = Field(..., description="30 values in model_meta.json order")
class PredictResponse(BaseModel):
label: str
probability: float
model_version: str
@app.get("/health")
def health() -> dict:
ready = state["model"] is not None
return {"status": "ok" if ready else "loading", "model_loaded": ready}
@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest) -> PredictResponse:
meta = state["meta"]
expected = len(meta["features"])
if len(req.features) != expected:
raise HTTPException(status_code=422,
detail=f"expected {expected} features, got {len(req.features)}")
x = np.asarray(req.features, dtype=float).reshape(1, -1)
proba = state["model"].predict_proba(x)[0]
idx = int(np.argmax(proba))
return PredictResponse(label=meta["classes"][idx],
probability=round(float(proba[idx]), 4),
model_version=app.version)
# --- Tests. The `with` runs the lifespan startup, as uvicorn would. ---
with TestClient(app) as client:
r = client.get("/health")
assert r.status_code == 200 and r.json()["model_loaded"], r.text
print("GET /health ->", r.json())
# Row 0 of the dataset is a known malignant case, row 19 a known benign one.
for row in (0, 19):
payload = {"features": data.data[row].tolist()}
r = client.post("/predict", json=payload)
assert r.status_code == 200, r.text
truth = data.target_names[data.target[row]]
print(f"POST /predict row {row:2d} -> {r.json()} (truth: {truth})")
r = client.post("/predict", json={"features": [1.0, 2.0, 3.0]})
print("POST /predict with 3 features ->", r.status_code, r.json()["detail"])
r = client.post("/predict", json={"features": "not a list"})
print("POST /predict with bad JSON type ->", r.status_code)
print("all checks passed")期待される出力です。Mac miniで約1.4秒です。
GET /health -> {'status': 'ok', 'model_loaded': True}
POST /predict row 0 -> {'label': 'malignant', 'probability': 1.0, 'model_version': '1.0'} (truth: malignant)
POST /predict row 19 -> {'label': 'benign', 'probability': 0.9328, 'model_version': '1.0'} (truth: benign)
POST /predict with 3 features -> 422 expected 30 features, got 3
POST /predict with bad JSON type -> 422
all checks passedここでは行0と行19はどちらも学習セットに含まれているので、自信のある答えが証明しているのは配管であって汎化ではありません。汎化を物語る数字は、train.pyが出力した98%のテスト精度です。2つの422のケースは、実務で最も重要なものです。1つはあなたのコードが捕捉し、もう1つはコードが実行される前にpydanticが捕捉します。
実行
train.pyとapp.pyがあるディレクトリから実行します。
pip install scikit-learn joblib fastapi uvicorn httpx
python train.py
uvicorn app:app --host 127.0.0.1 --port 8000lifespanがモデルを読み込むと、uvicornはApplication startup completeと表示します。そのまま起動しておき、以下の確認は別のターミナルで行います。app.pyを編集している間は--reloadを付けると、保存のたびにサーバーが再起動します。本番では--reloadを使わないでください。また、0.0.0.0にバインドするのは、他のマシンからの接続を受け付けるつもりがあるときだけにしてください。
テスト
まずhealthです。
curl -s http://127.0.0.1:8000/health{"status":"ok","model_loaded":true}次に予測です。以下の30個の数値は、データセットの行19を小数点以下4桁に丸めたもので、良性のケースです。
curl -s -X POST http://127.0.0.1:8000/predict \
-H 'Content-Type: application/json' \
-d '{"features": [13.54, 14.36, 87.46, 566.3, 0.0978, 0.0813, 0.0666, 0.0478, 0.1885, 0.0577, 0.2699, 0.7886, 2.058, 23.56, 0.0085, 0.0146, 0.0239, 0.0132, 0.0198, 0.0023, 15.11, 19.26, 99.7, 711.2, 0.144, 0.1773, 0.239, 0.1288, 0.2977, 0.0726]}'{"label":"benign","probability":0.9326,"model_version":"1.0"}入力を丸めたため、確率はインプロセスのテストと小数点以下4桁目で異なります。短すぎるリクエストはきれいに失敗するはずです。
curl -s -X POST http://127.0.0.1:8000/predict \
-H 'Content-Type: application/json' -d '{"features": [1, 2, 3]}'{"detail":"expected 30 features, got 3"}ブラウザでhttp://127.0.0.1:8000/docsを開くと、FastAPIがpydanticモデルから生成する対話的なOpenAPIページが表示されます。そこからリクエストを送ることもできます。
試してみること
- 分類器を
sklearn.ensembleのHistGradientBoostingClassifierに入れ替えてみてください。app.pyは何も変わりません。それがパイプライン全体を保存する意味です。テスト精度とファイルサイズを比べてください。 model_meta.jsonから読んだ確率ベクトル全体と特徴量名を、新しいGET /schemaルートで返すようにしてください。クライアントがソースを読まずにリクエストを組み立てられるようになります。- 行のリストを受け付ける
POST /predict/batchルートを追加してください。TestClientとtime.perf_counterを使って、1,000行の1リクエストと、1行の1,000リクエストのスループットを測ってください。 - 学習時に
sklearn.__version__をメタデータに記録し、サービング側のバージョンが異なれば起動を拒否するようにしてください。1行のチェックで、紛らわしいエラーの一群を防げます。 from app import appしてクライアントをフィクスチャにする、きちんとしたpytest関数としてテストを書いてください。そしてpushのたびにGitHub Actionsのワークフローで実行してください。python:3.12-slimベースイメージに2つのモデルファイルをコピーして、アプリをコンテナに入れてください。Dockerfileはノートブックからサービスへのレッスンで扱っています。
まとめ
- 素の分類器ではなくパイプライン全体を保存し、前処理がモデルからずれないようにします。
- 特徴量名、クラス名、スコアを記した小さなメタデータファイルをモデルの横に置き、サーバーにそれを基準に検証させます。
- モデルはルートの中ではなく、
lifespanフックで一度だけ読み込みます。 - 型付きのpydanticモデルは、バリデーション、ドキュメント、明確な422エラーを無償で提供します。pydanticにできない意味的なチェックを追加します。
- まず
TestClientでインプロセスのテストをします。uvicornとcurlは、テストがすでに証明したことを確認するためのものです。