model-deployment

モデルのデプロイ

このスキルを使うと、AI エージェントはトレーニング済みの機械学習モデルを本番環境にデプロイできます。FastAPI または Flask を使ったサービング API へのモデルのパッケージ化、Docker によるコンテナ化、Kubernetes によるオーケストレーション、サーバーレスプラットフォームへのデプロイを扱います。エージェントは、モデルのバージョン管理、ヘルスチェック、入力検証、ログ記録、モニタリングを行い、本番環境で信頼性とスケーラビリティの高い推論を実現します。

ワークフロー

1. モデルをシリアライズしてパッケージ化する： トレーニング済みモデルを、ONNX、TorchScript、SavedModel、joblib pickle などの移植可能な形式にエクスポートします。モデルのアーティファクトを前処理パイプラインや必要な設定ファイルとともにまとめ、推論を単独で実行できるようにします。

2. サービング API を構築する： FastAPI または Flask を使って REST API を作成します。起動時にモデルを読み込み、予測エンドポイントを公開します。ヘルスチェックエンドポイント、入力検証付きのリクエスト／レスポンススキーマ（Pydantic モデル）、構造化ログ、適切な HTTP ステータスコードを返すエラー処理を追加します。

3. Docker でコンテナ化する： 固定された requirements.txt から依存関係をインストールし、モデルのアーティファクトとサービングコードをコピーして、API サーバーをエントリポイントに設定する Dockerfile を作成します。マルチステージビルドを使ってイメージサイズを最小限に抑え、トレーニング専用の依存関係が含まれないようにします。

4. オーケストレーションとスケーリングを設定する： リソースのリクエスト／上限、ヘルスチェックエンドポイントを参照するレディネスプローブとライブネスプローブ、CPU、メモリ、リクエストレイテンシなどのカスタムメトリクスに基づいてスケーリングする Horizontal Pod Autoscaler を含む Kubernetes の Deployment マニフェストと Service マニフェスト（または利用するプラットフォームに相当するもの）を定義します。

5. デプロイして検証する： コンテナイメージをレジストリにプッシュし、Kubernetes のマニフェストを適用するか、サーバーレスプラットフォームにデプロイして、稼働中のエンドポイントに対してスモークテストを実行します。既知の入力セットに対するレスポンスが期待される出力と一致することを検証します。

6. モニタリングして改善する： Prometheus や Grafana などのモニタリングツールと連携し、リクエストレイテンシ、エラー率、スループット、予測分布のドリフトなどのモデル固有のメトリクスを追跡します。異常を検知するアラートを設定し、ブルーグリーンまたはカナリア戦略を使って更新されたモデルバージョンを再デプロイするワークフローを確立します。

対応テクノロジー

• API フレームワーク： FastAPI、Flask、TorchServe、TensorFlow Serving、Triton Inference Server
• コンテナ化： Docker、Podman
• オーケストレーション： Kubernetes、Docker Compose、AWS ECS、Google Cloud Run
• サーバーレス： AWS Lambda、Google Cloud Functions、Azure Functions
• モニタリング： Prometheus、Grafana、Datadog、AWS CloudWatch
• モデルレジストリ： MLflow Model Registry、AWS SageMaker Model Registry、Weights & Biases

使い方

トレーニング済みモデルのアーティファクト、その依存関係、デプロイ先の環境（ローカルの Docker、Kubernetes クラスター、サーバーレス）をエージェントに提供します。エージェントは必要なサービングコード、コンテナ設定、デプロイマニフェストを生成し、デプロイの手順を案内します。

例

例 1：FastAPI を使ったモデルのデプロイ

# app.py
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from contextlib import asynccontextmanager
from typing import List

model = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global model
    model = joblib.load("model.pkl")
    yield

app = FastAPI(title="ML Model API", version="1.0.0", lifespan=lifespan)

class PredictionRequest(BaseModel):
    features: List[float]

    @validator("features")
    def validate_features(cls, v):
        if len(v) != 4:
            raise ValueError("Expected exactly 4 features")
        return v

class PredictionResponse(BaseModel):
    prediction: int
    probability: List[float]

@app.get("/health")
def health_check():
    return {"status": "healthy", "model_loaded": model is not None}

@app.post("/predict", response_model=PredictionResponse)
def predict(request: PredictionRequest):
    try:
        features = np.array(request.features).reshape(1, -1)
        prediction = int(model.predict(features)[0])
        probability = model.predict_proba(features)[0].tolist()
        return PredictionResponse(prediction=prediction, probability=probability)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

例 2：Docker + Kubernetes によるデプロイ

Dockerfile：

FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /usr/local/bin/uvicorn /usr/local/bin/uvicorn
COPY app.py model.pkl ./
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

k8s-deployment.yaml：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ml-model-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ml-model-api
  template:
    metadata:
      labels:
        app: ml-model-api
    spec:
      containers:
        - name: api
          image: registry.example.com/ml-model-api:v1.0.0
          ports:
            - containerPort: 8000
          resources:
            requests: { cpu: "250m", memory: "512Mi" }
            limits: { cpu: "1000m", memory: "1Gi" }
          readinessProbe:
            httpGet: { path: /health, port: 8000 }
            initialDelaySeconds: 10
            periodSeconds: 5
          livenessProbe:
            httpGet: { path: /health, port: 8000 }
            initialDelaySeconds: 15
            periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: ml-model-api
spec:
  selector:
    app: ml-model-api
  ports:
    - port: 80
      targetPort: 8000
  type: LoadBalancer

ベストプラクティス

• requirements.txt で依存関係のバージョンをすべて固定し、決定的な Docker ビルドを使用して、環境間での再現性を確保します。
• モデルをコンテナイメージ全体の再ビルドなしで更新できるよう、モデルのアーティファクトをコードから分離します。バージョン管理されたパスを持つモデルレジストリまたはクラウドストレージを使用します。
• Pydantic または JSON Schema で入力検証を実装し、不正なリクエストがモデルに到達して、原因の分かりにくいエラーを引き起こす前に拒否します。
• Kubernetes ではレディネスプローブを使用し、モデルの読み込みが完了していない Pod にトラフィックが到達しないようにします。大規模なモデルでは、読み込みにかなりの時間がかかることがあります。
• 新しいモデルバージョンをリリースする際は、カナリアデプロイを採用します。少量のトラフィックを新バージョンに振り分け、全面展開の前にメトリクスを比較します。
• 本番環境でのデバッグ、監査、データドリフトの検出に役立つよう、予測と入力をログに記録します（PII はマスキングしてください）。

エッジケース

• 大きなモデルファイル（> 1 GB）： Docker イメージへの組み込みは避けてください。代わりに、起動時にクラウドストレージ（S3、GCS）からダウンロードするか、永続ボリュームをマウントします。モデルの初期化に時間がかかる場合は、遅延読み込みを使用します。
• サーバーレスでのコールドスタート遅延： サーバーレス関数では、大きなモデルの読み込みに 10～30 秒かかることがあります。プロビジョンドコンカレンシー（AWS Lambda）、最小インスタンス数（Cloud Run）、または ONNX Runtime などの最適化された形式を使用して軽減します。
• 推論時の前処理の不一致： 学習で使用した前処理パイプラインは、推論時に実行するものと完全に一致している必要があります。変換を個別に再実装するのではなく、パイプライン全体（例：scikit-learn Pipeline + joblib）をシリアライズします。
• グレースフルシャットダウンと処理中のリクエスト： SIGTERM シグナルを処理し、シャットダウン前に処理中のリクエストを完了させます。保留中のリクエストを完了するのに十分な時間を確保できるよう、Kubernetes の terminationGracePeriodSeconds を設定します。
• GPU と CPU での推論結果の不一致： GPU で学習したモデルは、CPU のみの環境にデプロイすると失敗することがあります。読み込み時にモデルのテンソルを CPU に明示的にマッピングし（torch.load(path, map_location="cpu")）、デプロイ前に対象のハードウェアで推論をテストします。