はじめに — つまずきに寄り添う一言
「ローカルでモデルは動くが、実際の業務にAPIとして出すと想定外の問題が出る」——こうした悩みはよく聞きます。本稿では、短時間で動く軽量な推論APIをFastAPIで設計・実装し、オンデマンドとバッチの両方を同一サービスで扱うパターンと、運用で必要なチェック項目を手順で示します。実務でそのまま使える観点を優先し、落とし穴と対処法も整理します。
1) 要件定義と設計（オンデマンド vs バッチ）
まず設計方針を明確にしましょう。用途によってSLA・レイテンシ・コスト感が変わります。
| 用途 | 要求 | 実装上の着眼点 |
|---|---|---|
| オンデマンド推論（同期） | 低レイテンシ（数百ms〜数秒） | 非同期I/O、リクエストタイムアウト、スケール方針 |
| バッチ推論（大量一括） | スループット重視、コスト効率 | バッチング、バックグラウンドキュー、バッチサイズ制御 |
| 混在ケース | 同一サービスで両立 | 優先度制御、リソース隔離、レート制限 |
設計時に決めるべき最小項目：
- SLA（最大許容遅延、成功率）
- 一回の推論当たりコスト想定
- 同時実行数とスケール戦略（垂直 vs 水平）
- 冪等性と再試行ポリシー
2) FastAPIの最小セットアップとエンドポイント設計
入力検証にはpydanticを使い、落ちる可能性を早期に検出します。ここではエンドポイント設計をテーブルで示します（コード例はポイントのみ記述）。
| エンドポイント | メソッド | 目的 | 備考（入力検証） |
|---|---|---|---|
| /predict | POST | オンデマンド推論（同期応答） | pydanticモデルで必須フィールド・型チェック |
| /predict_async | POST | 非同期キュー登録（バッチング対象） | リクエストIDとコールバックURLを受け取る設計が実務向け |
| /batch/status/{job_id} | GET | バッチジョブの状態取得 | ステータスとエラー情報を返す |
| /health, /ready | GET | ヘルスとレディネスチェック | 監視用に簡潔な応答を返す |
pydantic例（説明のみ）：入力モデルはサイズ制限、文字列長、必須キーを明示する。大きなペイロードは事前チェックで弾く。
3) バッチング戦略 — バックグラウンドキュー＋集約タイムウィンドウ
バッチ処理は「一定時間で集めて一度に処理する」か「サイズでトリガする」方式が基本です。ここでは簡易ワーカーと非同期キューの考え方を示します。
| 方式 | 特徴 | 向き不向き |
|---|---|---|
| 時間ウィンドウ（例：100msごと） | レイテンシばらつきが小さくバッチ効率が良い | 短めの応答要件を満たしつつスループットを稼ぎたい場合 |
| サイズトリガ（例：batch_size=32） | バッチ効率が最大化されるが待ちが発生しうる | バッチ性能を最優先にする業務 |
| ハイブリッド | どちらのしきい値でもトリガ可能 | 実運用でよく使われる |
実装ヒント（要点のみ、簡潔に）：
- APIは受信時にリクエストを軽くバリデートし、非同期キューに格納する。
- バックグラウンドタスクがタイマーやサイズでバッチを切り出し、推論ワーカーに渡す。
- ワーカーは外部モデルAPI呼び出し／ローカルモデル推論を行い、結果を保存またはコールバック。
- 非同期実装はasyncio.QueueやRedis/RQを利用すると堅牢。単純構成ならuvicornのバックグラウンドタスクで十分。
4) エラー処理と再試行方針
実運用では予期せぬ例外、タイムアウト、外部API障害が起きます。方針を明確にしておきましょう。
- 例外ハンドラでHTTP 500を返す前に、詳細はログに残す。クライアントには説明的なエラーコードとメッセージを返す。
- 再試行は冪等性が担保できる場合のみ行う（重複結果の影響を考慮）。指数バックオフ＋最大試行回数を設定する。
- タイムアウトはAPIゲートウェイとアプリ双方で設定する（例：APIは短め、バッチジョブでは長めに）。
- 長時間処理のバッチはジョブ状態を保存し、失敗時に部分成功を扱えるようにする。
5) ロギング・メトリクス・ヘルスチェック
監視のための最低限のエンドポイントとログ設計を示します。
| 名称 | 用途 | 実装例（返す情報） |
|---|---|---|
| /health | プロセスが稼働しているか（簡易） | {“status”: “ok”} |
| /ready | 依存（モデルロード、外部サービス接続）が整っているか | {“ready”: true, “model_loaded”: true} |
| /metrics | Prometheus互換のメトリクス公開 | リクエスト数、エラーレート、レイテンシヒストグラム |
ログ設計のポイント：
- 構造化ログ（JSON）でリクエストID、ジョブID、処理時間、エラー詳細を出力する。
- ログレベルは運用と開発で分ける。デバッグはファイル/外部でのみ残す。
- 例外はスタックトレースを追えるように保存しつつ、機密情報はマスクする。
6) Docker化と簡易CI
軽量APIはコンテナ化してデプロイするのが運用しやすいです。CIは次の流れを推奨します。
- pytestでFastAPI TestClientを使ったエンドポイントテストを用意する（/predictの成功系・異常系、バッチ登録の統合テスト）。
- Dockerfileは小さなベースイメージ（python:3.x-slim）を使い、依存は最小化する。
- CI（例：GitHub Actions）でテスト→イメージビルド→（必要なら）イメージのスキャン→registryへpushの流れを作る。
CIに含める最小ステップ（例）:
| ステップ | 目的 |
|---|---|
| pytest実行 | 機能回帰チェック |
| lint（optional） | コード品質チェック |
| docker build & push | イメージ化とレジストリ反映 |
7) デプロイと運用チェックリスト
以下は現場ですぐチェックできる実用的な項目です。最後に表で整理します。
- リソース制限（CPU/メモリ）の設定とOOM対策
- レートリミットと優先度制御（オンデマンド優先／バッチはスロット確保）
- セキュリティ：TLS、認証、シークレット管理
- 監視：メトリクス収集、アラートルール（エラーレート、レイテンシ、スループット）
| レベル | 項目 | チェック内容（そのまま使える） |
|---|---|---|
| 必須 | ヘルス/レディネス | /health と /ready を監視に登録し、復旧不能時は自動再起動 |
| 必須 | リクエストタイムアウト | APIゲートウェイとアプリに最大タイムアウトを設定（例：30s/300s） |
| 必須 | ログレベル・フォーマット | JSONログでリクエストIDを必ず出力 |
| 推奨 | メトリクス | Prometheus互換でリクエスト数/エラー/レイテンシを収集 |
| 推奨 | リソース制限 | コンテナごとにCPU/メモリ上限を設ける |
| 発展 | 認証・シークレット管理 | VaultやクラウドKMSでキー管理、APIはBearerトークンで保護 |
8) よくある落とし穴とトラブルシューティング
実務で遭遇しやすい問題と対処法を短くまとめます。
- 問題：バッチの遅延が増える。対処：ウィンドウ幅を短縮、優先度でオンデマンド優先にする。
- 問題：メモリ不足でOOM。対処：バッチサイズ制限、ワーカー数削減、モデルを軽量化。
- 問題：外部APIの不安定。対処：タイムアウト、回路遮断（circuit breaker）、キャッシュ導入。
- 問題：再試行で重複データ。対処：リクエストIDによる冪等性チェック、または重複排除ロジック。
まとめ
本稿では、FastAPIを使った軽量推論APIの設計と実装の考え方を、オンデマンドとバッチを一つのサービスで扱う観点から整理しました。重要なポイントは次の通りです：
- 設計段階でSLAとコスト感を明確にすること
- 入力検証（pydantic）と構造化ログで問題の早期把握を行うこと
- バッチは時間ウィンドウとサイズトリガのハイブリッドがおすすめで、非同期キューを使って安全に処理すること
- ヘルス/レディネス/メトリクスを整備し、CIでテスト・コンテナ化を自動化すること
最後に、運用チェックリストを実運用で回しながら改善を続けることが、安定稼働への近道です。次回は第138回で触れたテスト方針を踏まえ、具体的なpytestによるエンドツーエンドテストの例を紹介します。