新しいモデルを現場に出すとき、不安やつまずきは誰にでもあります。例えば「学習はうまくいったがデプロイでレスポンスが遅い」「差し替えたら誤判定が増えた」「ロールバック手順が複雑で時間がかかる」──こうした現場の悩みに寄り添い、1日〜数日で試せる実践手順を提示します。第35回（運用自動化）・第36回（監視／再学習）の成果物を受けて、次にやるべき“安全に差し替える”工程にフォーカスします。
対象読者と到達目標
対象：AIを仕事に活かしたい実務担当者、個人事業主、中小企業の担当者。前提は「学習済みモデルのアーティファクトがあること」です。到達目標は次のとおりです。
- モデルアーティファクトの格納とバージョン管理の基本を理解する
- スモーク・契約・回帰テストを自動実行する方法を試せる
- カナリア／トラフィック分割で段階的にロールアウトし、自動ロールバックを設計できる
- 現場で使える「チェックリスト」と「ロールバック手順書」を持ち帰る
導入の全体像
ここでは全体フローを俯瞰します。実際の手順は後節でコードやCI例と合わせて示します。
- アーティファクト保存（モデルファイル＋メタデータ）→ アーティファクトレジストリ（S3/ファイルサーバ/MLflow）
- CIでテスト（スモーク／契約／性能回帰）→ ステージングで検証
- 段階的ロールアウト（カナリア／トラフィック分割）＋監視
- 自動／手動でのロールバック手順
前準備：アーティファクトとメタデータの整備
モデルの差し替えで失敗しやすい点は、「依存ライブラリの違い」「入力フォーマットの変化」「モデルのサイズ」が多いです。まずはモデル＋メタデータを一つのバージョンドキュメントとして保存します。
モデル保存の簡易例（Python）
以下は学習済みオブジェクトをjoblibで保存し、メタデータをJSONで保存する最小例です。
import joblib
import json
from datetime import datetime
model = ... # 学習済みオブジェクト
version = "v1.2.0"
model_path = f"models/model-{version}.pkl"
meta_path = f"models/model-{version}.meta.json"
joblib.dump(model, model_path)
metadata = {
"version": version,
"created_at": datetime.utcnow().isoformat() + "Z",
"framework": "scikit-learn",
"framework_version": "1.2.0",
"input_schema": {"columns": ["age","income","score"], "dtypes": {"age":"int","income":"float","score":"float"}},
"baseline_rmse": 0.85
}
with open(meta_path, "w") as f:
json.dump(metadata, f)
このペアをS3やMLflowに登録します。S3ならキーにバージョンを含めて保管してください。
テストと検証（スモーク・契約・性能）
テストは最低でも次の3つを自動化します：
- スモークテスト：代表入力でレスポンスとステータスを確認
- データ契約テスト：列名・型・欠損率など入力フォーマットの整合性
- モデル回帰テスト：既知の評価指標（RMSEなど）で前バージョンと比較
スモークテスト（Python）
import requests
url = "http://staging.example.com/predict"
payload = {"age": 30, "income": 45000.0, "score": 0.5}
resp = requests.post(url, json=payload, timeout=5)
resp.raise_for_status()
print(resp.json())
データ契約テスト（Python）
簡易チェック：期待する列があるか、欠損率が閾値以下かを確認します。
import pandas as pd
def check_contract(df: pd.DataFrame, schema: dict, max_missing=0.1):
# schema: {"columns": [...], "dtypes": {...}}
missing_ok = True
for col in schema["columns"]:
if col not in df.columns:
raise ValueError(f"Missing column: {col}")
if df[col].isna().mean() > max_missing:
missing_ok = False
return missing_ok
モデル回帰テスト（簡易、Python）
from sklearn.metrics import mean_squared_error
import numpy as np
def regression_test(y_true, y_pred, baseline_rmse, tol=0.05):
rmse = np.sqrt(mean_squared_error(y_true, y_pred))
if rmse > baseline_rmse * (1 + tol):
raise AssertionError(f"RMSE increased: {rmse:.3f} > {baseline_rmse:.3f}")
return rmse
デプロイ戦略（ブルー/グリーン、カナリア、トラフィック分割）
小さなサービスではカナリア方式（段階的にトラフィックを新バージョンへ移す）が実践的です。比率の決め方や監視KPIは次表を参照してください。
| フェーズ | 説明 | 推奨比率（例） |
|---|---|---|
| 初期カナリア | 1〜5%を新バージョンへ流す。問題が出れば即ロールバック。 | 1%→5% |
| 拡張カナリア | 問題がなければ段階的に10%→30%へ。 | 10%→30% |
| 全量切替 | 指標が安定すれば全トラフィックへ移行。 | 100% |
カナリア比率決め方（実務的な指針）
- 最大リスク許容度を考える：データ感度が高ければ小さく（1%から）
- トラフィック量で調整：1%のトラフィックで統計的に有意な指標が取れない場合は段階を増やす
- 応答性・影響範囲で決定：遅延が業務に直結するなら短時間で段階を進める
CI/CDの実装例（簡易GitHub Actions）
以下は典型的な流れのYAML断片です（省略を含む）。目的は「ビルド→テスト→アーティファクト保存→ステージングデプロイ→カナリア開始」です。
name: model-deploy
on:
push:
paths:
- 'models/**'
jobs:
build-and-test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Setup Python
uses: actions/setup-python@v4
with:
python-version: '3.10'
- name: Install deps
run: pip install -r requirements.txt
- name: Run tests (smoke/contract/regression)
run: pytest tests/
- name: Upload artifact
uses: actions/upload-artifact@v3
with:
name: model-artifact
path: models/model-*.pkl
deploy-staging:
needs: build-and-test
runs-on: ubuntu-latest
steps:
- name: Download artifact
uses: actions/download-artifact@v3
with:
name: model-artifact
- name: Deploy to staging
run: |
python scripts/deploy_to_env.py --env staging --artifact models/model-*.pkl
canary-release:
needs: deploy-staging
runs-on: ubuntu-latest
steps:
- name: Start canary
run: |
python scripts/start_canary.py --version v1.2.0 --percent 1
deploy_to_env.pyやstart_canary.pyは、自社のデプロイAPIやロードバランサ管理API（例：NGINX、ALB、Istio）に合わせて作成します。小規模なら単純にリバースプロキシの設定を差し替えるスクリプトでも十分です。
監視・自動ロールバックの設計
監視は以下のKPIを中心に行います。しきい値を超えた場合、自動でロールバックを走らせる設計が重要です。
| KPI | 目的 | 簡易しきい値（例） |
|---|---|---|
| エラー率 | 致命的な不具合を素早く検出 | 通常+3σまたは >1% |
| レイテンシ（P95） | 性能劣化の検出 | 通常の1.5倍以上 |
| 入力分布の逸脱 | 入力データの仕様変更検出 | 主要カテゴリのシェア差 >10% |
自動ロールバックの基本トリガーと手順
トリガー例：
- 5分間でエラー率が閾値を超えた
- 30分間でP95レイテンシが閾値を超えた
- 入力分布の変化が継続（人手で確認が必要な場合はアラートで止める）
自動ロールバック手順（概要）：
- アラート発生 → 通知（Slack/メール）
- 自動でトラフィックを旧バージョンに戻す（例：カナリアを0%へ）
- ロールバックを実行（旧アーティファクトを再度デプロイ）
- インシデント対応（ログ収集、根本原因分析）
実装例（簡易、Pythonでトラフィック操作のスクリプト）：
import requests
LB_API = "https://deploy-api.example.com/traffic"
def set_traffic(version, percent):
resp = requests.post(LB_API, json={"version": version, "percent": percent}, timeout=5)
resp.raise_for_status()
# 自動ロールバック呼び出し例
set_traffic("v1.1.9", 100)
実運用チェックリストとトラブルシュート
まずは「小さなサービスで試すための最短チェックリスト」と「現場で使えるロールバック手順書」を用意します。
| チェック項目 | 説明 | 完了 |
|---|---|---|
| モデルファイルとメタデータをバージョン管理 | モデル+meta.jsonをS3/レジストリに保存 | |
| 自動テストが通る | スモーク・契約・回帰テストをCIで実行 | |
| ステージングでスモーク通過 | 代表入力でレスポンス・スキーマ確認 | |
| カナリア開始の監視準備 | アラートとダッシュボードを設定 | |
| ロールバック手順の検証 | 手動でロールバックを一度実行しておく |
トラブルシュートの注意点（よくある失敗）
- 依存ライブラリのバージョン差：virtualenv/コンテナで再現性を確保する
- モデルサイズで冷スタートが遅い：ロード時間を計測し、スケール戦略を立てる
- 入力スキーマの微妙な差：契約テストで早期に検出する
すぐ使える：ロールバック手順書（テンプレ）
以下は現場でそのまま使える簡易テンプレートです。
| 手順 | 実行コマンド/文言 |
|---|---|
| 1. アラート確認 | Slack通知チャンネルの内容を確認し、アラートIDを控える |
| 2. 一時的トラフィック切り戻し | python scripts/set_traffic.py --version v1.1.9 --percent 100 |
| 3. ロールバック実行 | python scripts/deploy_to_env.py --env production --artifact models/model-v1.1.9.pkl |
| 4. 状況共有 | Slackテンプレを使い、関係者へ報告（下段参照） |
| 5. Postmortem | ログ・メトリクスを収集して原因解析 |
Slack報告テンプレ例：
[ALERT] モデルカナリアで閾値超過
- アラートID: {alert_id}
- 発生時間: {time}
- 対象バージョン: {version}
- 実施: 自動でトラフィックを旧バージョンに戻しました
- 次の対応: ログ収集＆原因調査（@SRE @ML担当）
付録：サービングの簡易サンプル（FastAPI）
そのまま貼って動く最小限のサービング例です。起動後、/health と /metrics、/predict を確認できます。
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
app = FastAPI()
model = joblib.load("models/model-v1.2.0.pkl")
class Input(BaseModel):
age: int
income: float
score: float
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/metrics")
def metrics():
# 簡易メトリクス
return {"model_version": "v1.2.0", "uptime": 12345}
@app.post("/predict")
def predict(inp: Input):
x = [[inp.age, inp.income, inp.score]]
y = model.predict(x)
return {"prediction": float(y[0])}
付録：想定作業時間・難易度とサンプルリポジトリ
| 項目 | 所要時間（目安） | 難易度 |
|---|---|---|
| モデル保存＋メタデータ整備 | 0.5〜1日 | 初心者可 |
| スモーク／契約テスト作成 | 0.5〜1日 | 初心者〜中級 |
| CI設定（簡易） | 1日 | 中級 |
| カナリア運用の検証 | 0.5〜2日 | 中級 |
サンプルリポジトリ（例）: https://github.com/manageai/sample-deploy（READMEに手順を記載）
まとめ
本記事では、第35回・第36回で整えた運用自動化／監視の土台を前提に、実際にモデルを安全にデプロイ・差し替えするための手順を示しました。ポイントは次の3つです。
- モデルは「ファイル＋メタデータ」で一つのバージョンとして扱う（再現性と比較が容易になる）
- スモーク・契約・回帰テストをCIに組み込み、ステージング→カナリア→全量の段階を踏む
- 監視KPIを定義し、自動ロールバックのトリガー・手順を用意しておく（手動テンプレも準備）
付録で示したPythonスニペット、GitHub Actions断片、チェックリストは小さなサービスで試すために最小限に絞っています。まずはステージングで一度フルフロー（デプロイ→カナリア→ロールバック）を演習して、手順を社内で文書化してください。次回は監視データを用いた自動再学習の実務的つなぎ方（第39回）を想定しています。