Claude CodeでFastAPIの推論APIを作る — lifespanで1回ロードしメモリを検証

推論APIのメモリが膨らむのは「リクエストごとのロード」から

学習済みモデルをFastAPIの推論APIにするときの定石は、モデルをアプリの起動時に1回だけ読み込み、全リクエストで共有することです。FastAPIのドキュメントも、共有する機械学習モデルの読み込みをlifespanの代表的な用途に挙げています。

Claude Codeに「推論APIを作って」と頼むと、動くコードはすぐ出てきます。ただ、エンドポイントの中でモデルを読み込む形になっていても、1リクエストずつ試す限りは正しい答えが返るので気づきません。壊れるのは同時リクエストが重なったときで、メモリが一気に跳ね上がります。

この記事では、lifespanで1回だけロードする書き方と、その効果をClaude Codeに数値で検証させる手順を扱います。FastAPIの汎用的な実装フローはClaude CodeでFastAPIバックエンドを構築する手順にあります。

疑似モデルで測ると、ピークメモリは約13倍に違った

推論ライブラリを入れなくても、メモリの挙動は再現できます。100MBのNumPy配列を「モデルの重み」に見立て、同じエンドポイントを2通りで書きました。

• 悪い例: リクエストのたびにload_model()を呼ぶ
• 良い例: lifespanで1回だけload_model()を呼び、辞書に保持する

同時にリクエストを投げた場合のプロセスのRSS(実際に使っている物理メモリ)の最大値は、次のとおりです。

測定はPython 3.12.2・FastAPI 0.162.0・NumPy 2.5.4で、HTTPクライアントからasyncio.gatherで一斉にリクエストを投げ、別スレッドで5ミリ秒ごとにRSSを読んだ1回分の値です。再実行すれば数字は多少ずれますが、構図は変わりません。

読みどころは2点あります。

• 良い例は、起動直後に約100MB増えたあと、100件を捌いても約5MBしか増えません
• 悪い例は、起動時の増加がないぶん一見軽そうに見えますが、リクエストが重なった分だけ重みのコピーが同時に生きます

同時数20で約1.7GB、100でも約2GBで頭打ちに近づくのは、同期関数(def)のエンドポイントがスレッドプールで並列実行され、同時に動く数に上限があるためと考えられます。実際のモデルは重みの大きさが桁違いなので、同じ書き方をすればコンテナのメモリ上限に直接当たります。数字は疑似モデルのものなので、手元のモデルで同じ測定をやり直してください。やり直しをClaude Codeに任せる方法は後の節で扱います。

lifespanでモデルを1回だけロードする書き方

FastAPIのlifespanは、アプリがリクエストを受け付ける前に1回、終了時に1回実行される処理を定義する仕組みです。yieldの前が起動時、後が終了時にあたります。ドキュメントのサンプルを推論API向けに整えると、次の形になります。

from contextlib import asynccontextmanager
 
import numpy as np
from fastapi import FastAPI
 
models: dict = {}
 
 
def load_model():
    # 本物ではここでtorch.loadやonnxruntimeのセッション生成などを行う
    return np.ones(100 * 1024 * 1024 // 8, dtype=np.float64)
 
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    models["m"] = load_model()  # 起動時に1回だけ
    yield
    models.clear()  # 終了時に解放
 
 
good = FastAPI(lifespan=lifespan)
 
 
@good.get("/predict")
def predict(x: float):
    return {"y": float(models["m"][0]) * x}
 
 
@good.get("/healthz")
def healthz():
    return {"loaded": "m" in models}

比較用の悪い例は、同じapp.pyにbadとして並べます。lifespanを渡さず、エンドポイントの中で毎回load_model()を呼ぶ形です。

bad = FastAPI()
 
 
@bad.get("/predict")
def predict_bad(x: float):
    m = load_model()  # リクエストのたびに100MBを確保する
    return {"y": float(m[0]) * x}

以降の検証スクリプトとテストは、このgoodとbadをimport app as Aで読み込んで使います。

ドキュメントは、モデルをモジュールのトップレベルで読み込む方法も取れるとしたうえで、その場合は単純な自動テストを走らせるだけでもモデルの読み込みを待つことになると書いています。lifespanなら、読み込みはアプリがリクエストを受け始める直前まで遅らせられます。

古いon_eventと混ぜない

旧来の@app.on_event("startup")も動きますが、ドキュメント上は非推奨の扱いで、lifespanを渡すとstartup・shutdownのイベントハンドラは呼ばれなくなります。「全部lifespan」か「全部イベント」のどちらかです。Claude Codeが学習データの古い書き方でon_eventを混ぜてきたら、指示で止めます。

落とし穴が2つある

1つ目はテストです。TestClient(app)をwithなしで使うと、lifespanが走らずmodelsが空のままになります。上のコードで試すと、KeyError: 'm'で落ちました。with TestClient(app) as client:と書くとlifespanが動き、/predictが{"y": 2.0}を返します。「Testing Events」のページも、lifespanをテストで動かすにはwith文を使うよう案内しています。

2つ目はサブアプリです。mountしたサブアプリでは、lifespanのstartup・shutdownが実行されるのはメインアプリだけです。モデルをサブアプリ側で持たせる構成にすると、ロードされないまま動き出します。

Claude Codeに検証させる——CLAUDE.mdには「確かめる方法」を書く

Claude Codeのベストプラクティスは、合否が返るチェックを用意すると、Claudeが自分で変更・実行・修正を回せるようになると説明しています。メモリの肥大は、正しい答えが返るため見た目では気づかない不具合です。数値で合否が出る検証を用意するのが、この種の問題に向いています。

ベストプラクティスの例にならい、CLAUDE.mdには検証のコマンドを短く置きます。

# Commands
- 起動: uv run uvicorn app:good --reload
- テスト: uv run pytest -q
- メモリ検証: uv run python -m scripts.check_memory good 100
 
# Conventions
- モデルの読み込みはlifespanの中だけで行う。エンドポイントやDependsの中でload_modelを呼ばない
- テストでlifespanが必要なときは with TestClient(good) を使う
 
# Gotchas
- 同期の def エンドポイントはスレッドプールで並列実行される。モデルは共有オブジェクトとして読み取り専用に扱う

同ガイドは、CLAUDE.mdには「このリポジトリで作業の完了を確かめる方法」を置き、短く保つよう勧めています。規約を守らせたい行がClaudeに無視されるなら、強調語を足すより次節のフックに移すのが筋です。

検証スクリプトとプロンプト

検証の本体は、先ほどの測定と同じ内容のスクリプトです。同時リクエスト数を引数にして、起動直後とピークのRSSを出力します。ルートのapp.pyを読み込むため、scripts/__init__.py(空ファイル)を置き、ルートからpython -m scripts.check_memoryで実行します。python scripts/check_memory.pyと直接呼ぶと、先頭の探索パスがscripts/になりimport appが通りません。

# scripts/check_memory.py(抜粋)  実行: uv run python -m scripts.check_memory good 100
import asyncio, sys, threading, time
 
import httpx
import psutil
 
import app as A  # FastAPIインスタンスgood / bad を持つモジュール
 
proc = psutil.Process()
peak, stop = 0, False
 
 
def sampler():
    global peak
    while not stop:
        peak = max(peak, proc.memory_info().rss)
        time.sleep(0.005)
 
 
async def main(which: str, n: int):
    global stop
    target = getattr(A, which)
    t = threading.Thread(target=sampler)
    t.start()
    async with target.router.lifespan_context(target):
        baseline = proc.memory_info().rss  # 起動直後(ロード済み、リクエスト前)
        transport = httpx.ASGITransport(app=target)
        async with httpx.AsyncClient(transport=transport, base_url="http://t") as c:
            rs = await asyncio.gather(
                *[c.get("/predict", params={"x": 2}) for _ in range(n)]
            )
    stop = True
    t.join()
    assert all(r.status_code == 200 for r in rs)
    mb = lambda b: round(b / 1024**2)
    print(f"{which} n={n} start={mb(baseline)}MB peak={mb(peak)}MB")
 
 
asyncio.run(main(sys.argv[1], int(sys.argv[2])))

あとはClaude Codeへの依頼文に、数字で答えさせる条件を入れます。

推論APIのメモリ使用量を検証して。python -m scripts.check_memory を同時数20と100で実行し、
起動直後とピークのRSSを表で報告して。ピークが起動直後+モデルサイズの1.5倍を
超えたら、モデルの読み込み箇所を探して直し、もう一度測って。

ベストプラクティスは、完了の報告に加えて「実行したコマンドとその出力」を求めるよう勧めています。「メモリは問題ありません」だけでは、読み手が確かめようがないためです。上のようにしきい値を決めておくと、Claudeは測定値を見て直すかどうかを自分で判断できます。しきい値の1.5倍は例示の値なので、モデルの形式に合わせて調整してください。

フックで「モデルをリクエスト内で読み込まない」を強制する

CLAUDE.mdの規約は、Claudeが従う可能性が高い指示にとどまり、毎回守られる保証はありません。ベストプラクティスは、例外を許さないルールはフックにするよう整理しています。Stopフックなら、検証コマンドが失敗している間はClaudeが作業を終えられません。

{
  "hooks": {
    "Stop": [
      {
        "hooks": [
          {
            "type": "command",
            "command": "uv run pytest -q 1>&2 || exit 2"
          }
        ]
      }
    ]
  }
}

終了コード2がターンの終了を止め、標準エラーに出た内容が「まだ終われない理由」としてClaudeに返ります。連続8回ブロックしてもツール呼び出しが挟まらないと、次のブロックは上書きされて終了します。上限は環境変数CLAUDE_CODE_STOP_HOOK_BLOCK_CAPで変えられます。

フックに載せるテストは、メモリの絶対値でなく「リクエストの中でモデルを読み込んでいない」ことを見る形にすると安定します。たとえば、load_modelの呼び出し回数を数えるテストです。

from unittest import mock
 
from fastapi.testclient import TestClient
 
import app as A
 
 
def test_load_model_called_once():
    with mock.patch.object(A, "load_model", wraps=A.load_model) as spy:
        with TestClient(A.good) as client:
            for _ in range(30):
                assert client.get("/predict", params={"x": 2}).status_code == 200
        assert spy.call_count == 1

RSSの値は実行環境で揺れますが、呼び出し回数は揺れません。メモリそのものの検証はスクリプトに、構造の退行防止はフックのテストに、と役割を分けると、CIでも壊れにくくなります。

ワーカー数を増やすと、メモリはその分だけ増える

lifespanで1回ロードしても、それが守るのは「1プロセスの中」です。FastAPIのドキュメント(デプロイの概念)は、複数のプロセスは通常メモリを共有せず、コード内でモデルのような大きなものを持てば、各プロセスが同じ量のメモリを消費すると説明しています。

つまり--workers 4なら、lifespanは4つのワーカーそれぞれで1回ずつ走り、重みのコピーは4つ存在します。2GBのモデルなら8GB前後の見積もりが要ります。メモリが逼迫するときの選択肢は次の2つです。

ドキュメントはKubernetesで動かす場合、ワーカーは使わず、コンテナごとにUvicornのプロセスを1つ走らせたいことが多いと説明しています。複製はKubernetesの層に任せる構成です。この場合も、各コンテナがモデルを1つ持つので、Podのメモリ上限の見積もりにはモデルサイズを含めます。

Claude Codeには、検証スクリプトを--workers付きで実際に起動する条件を追加で頼めます。たとえば「uvicorn app:good --workers 2で起動し、親と子のRSSをpsutilで合算して報告して」と指示します。プロセス単位のRSSの合算は共有ページの二重計上に注意が要るので、数値の意味まで報告に含めさせると読み違いを防げます。

ヘルスチェックはロード完了を反映する

起動に時間のかかるモデルでは、ロード中にロードバランサーやオーケストレーターがリクエストを流してはいけません。lifespanはリクエストを受け付ける前に実行されるので、/healthzが応答できる時点ではモデルは載っています。上のコードの/healthzは、それを"m" in modelsで返す最小の形です。

実際のモデルでは、ロード後に1回ダミー入力で推論を走らせるウォームアップをlifespanに加えるのも手です。初回の推論だけが遅い、というモデルがあるためです。ウォームアップを入れた場合は、/healthzが「ロード済み」でなく「推論できる」状態を返すかどうかを、Claude Codeへのレビュー依頼に含めておくと取りこぼしません。

検証を依頼するときのつまずき

• 良い例のつもりでDependsの中にモデルのロードを書かせてしまう。Dependsの関数はリクエストごとに呼ばれるため、キャッシュしない限り悪い例と同じです
• 測定をシェルでcurlを並べて行う。逐次実行になり、同時リクエストのピークが出ません。asyncio.gatherやhttpxの並列実行で投げさせます
• 起動直後のRSSを測らずにピークだけを見る。モデルの常駐分と、リクエストで増えた分が分けられません

まとめ

推論APIのメモリ問題は、正しい答えが返るぶん気づきにくい不具合です。lifespanで1回だけロードする書き方に直し、同時リクエストでのピークを数値で出させ、構造の退行は呼び出し回数のテストで止める、という分業が現実的です。ワーカーを増やすと各プロセスがモデルを持つため、検証は--workersの構成でもやり直す必要があります。

同じ「検証を先に書かせる」型は、FastifyのAPIをinjectで検証する手順やHonoのapp.requestでの検証にもあります。検証コマンドを置く場所の考え方は、Claude Code完全ガイドの設定の節と共通です。