モデルを置き換えるとき、不安になるのは当然です。ユーザ影響、法務要件、サービスの安定性――どれを優先するか迷った経験は多いはずです。本記事では、現場で使える「退役→互換性検証→段階的切替→完全退役」までの実務ワークフローを、Pythonで自動化する具体例を交えて整理します。第79回のローンチ後、長期運用で必ず出てくる課題に焦点を当てます。
なぜモデル退役・置換が必要か（狙いと前提）
以下の観点で退役・置換計画を考えます。実務ではこれらが混在するため、優先順位を明確にすることが重要です。
| 観点 | 主要な懸念点 | 実務上の対策例 |
|---|---|---|
| ビジネス影響 | ユーザ体験の低下、売上機会の損失 | 段階的切替、ABテスト、SLO設計 |
| コンプライアンス | 保存要件、説明責任、データ保持ポリシー | ログ保持設計、法務への事前確認 |
| コスト | 推論コスト、運用負荷、アセット維持費 | インスタンスタイプ見直し、古いモデルの段階的削減 |
退役計画の作り方（チェックリスト）
まずは影響範囲を洗い出し、関係者への通知とフェーズ設計を行います。期限や役割を明確にすることがポイントです。
| フェーズ | 主要タスク | 関係者 |
|---|---|---|
| 通知 | 内部外部への告知、保存要件の確認 | プロダクト、法務、CS |
| Shadow（非公開検証） | トラフィックを複製して新旧比較 | 開発、SRE、QA |
| Canary（割合配信） | 少量ユーザで新モデルを試行、KPI監視 | 開発、SRE、プロダクト |
| 全切替 | スケール確認、コスト検証、最終通知 | 全関係者 |
| 完全退役 | 古いモデルのアーティファクト削除、ドキュメント更新 | 運用、法務 |
互換性＆検証の自動化
入力/出力スキーマチェック、ベースラインとの差分テスト、性能・レイテンシ比較を自動化します。ここではpytestベースの契約テストと差分可視化の骨子を示します。
ポイント
- スキーマ検査：型と必須フィールドを厳密に検証する。
- 差分テスト：代表的なリクエストセットで旧モデルと比較する。
- 性能試験：レイテンシとスループットを短期負荷で測定する。
pytestベースの簡易例
下はベースライン出力と新モデル出力を比較するテストのスニペットです（実運用ではログやメトリクス収集を追加）。
import pytest
import requests
BASE_URL_OLD = "https://api.example.com/v1/old_model"
BASE_URL_NEW = "https://api.example.com/v1/new_model"
SAMPLE_PAYLOADS = [
{"text":"請求書の支払い期限を教えて"},
{"text":"商品の返品ポリシーは？"},
]
@pytest.mark.parametrize("payload", SAMPLE_PAYLOADS)
def test_contract_and_output_similarity(payload):
r_old = requests.post(BASE_URL_OLD, json=payload, timeout=5)
r_new = requests.post(BASE_URL_NEW, json=payload, timeout=5)
assert r_old.status_code == 200
assert r_new.status_code == 200
out_old = r_old.json()
out_new = r_new.json()
# スキーマ検査（例）
assert "answer" in out_old and "answer" in out_new
# 差分ルール（業務要件に合わせて閾値を決める）
assert similarity_score(out_old["answer"], out_new["answer"]) > 0.85
def similarity_score(a, b):
# 単純な例：文字列類似度を計算（実運用では語彙や意図の比較を推奨）
return 1.0 - (levenshtein_distance(a, b) / max(len(a), len(b), 1))
段階的切替パターンとPythonでの実装
代表的なパターンは shadow、canary、gradual（traffic-splitting）です。実運用ではFeature FlagやTraffic Routerを組み合わせます。
| パターン | 用途 | 期待される安全性 |
|---|---|---|
| Shadow | 本番トラフィックを複製して比較（応答はユーザへ返さない） | 高 |
| Canary | 一部ユーザへ新モデルを直接適用（AB的運用） | 中 |
| Gradual | 割合を徐々に増やす。自動スケールと監視が前提 | 中〜高 |
簡易Traffic-split制御スクリプト（例）
これはルーティングサービスに対して割合を更新する想定のサンプルです。実際はFeature FlagサービスやAPI GatewayのAPIを使います。
import requests
ROUTER_API = "https://traffic-router.example.com/api/split"
API_KEY = "${ROUTER_API_KEY}"
def set_split(new_model_pct):
payload = {"routes": {"old_model": 100 - new_model_pct, "new_model": new_model_pct}}
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
r = requests.post(ROUTER_API, json=payload, headers=headers, timeout=5)
r.raise_for_status()
return r.json()
# canaryで5%から開始
set_split(5)
データと学習資産の移行
モデルは学習済みパラメータだけでなく、メタデータ、特徴量定義、前処理コードに依存します。移行計画を作って自動化テストを用意してください。
| 資産 | 移行方法 | 検証ポイント |
|---|---|---|
| 特徴量定義 | バージョン管理と変換スクリプト | 同一入力での特徴量一致、欠損処理挙動 |
| メタデータ（モデル説明、署名） | JSON/YAMLでの移植とスキーマ検査 | 必須フィールドの有無、説明責任の要件充足 |
| アーティファクト（チェックポイント） | ストレージに移し、ACLと保持ポリシーを設定 | アクセス権限、整合性ハッシュの確認 |
互換性がない場合の変換スクリプト例
import json
# 古い特徴量を新仕様にマッピングする簡易例
MAPPING = {
"old_age": "age_years",
"old_income": "income_k"]
def transform_record(old):
new = {}
for k, v in old.items():
if k in MAPPING:
new[MAPPING[k]] = v
# 追加の正規化処理
return new
# バッチ変換
with open('old_data.json') as f:
records = json.load(f)
new_records = [transform_record(r) for r in records]
with open('new_data.json','w') as f:
json.dump(new_records, f, ensure_ascii=False, indent=2)
監視・アラートとロールバック基準
切替後に注視すべきKPI/SLIとアラート条件、即時ロールバックトリガーを事前に決めておきます。
| カテゴリ | 指標 | アラート条件（例） |
|---|---|---|
| 品質 | リジェクト率、正答率、意図一致率 | 基準値より5%以上悪化 |
| 性能 | P95レイテンシ、エラー率 | P95が2倍、エラー率が0.5%以上増加 |
| ビジネス | コンバージョン、解約率、CS受信数 | 重要KPIが10%以上悪化 |
自動ロールバックの簡易API呼び出し例
import requests
ROLLBACK_API = "https://deploy.example.com/api/rollback"
API_KEY = "${DEPLOY_API_KEY}"
def trigger_rollback(reason):
payload = {"target":"old_model","reason":reason}
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
r = requests.post(ROLLBACK_API, json=payload, headers=headers, timeout=10)
r.raise_for_status()
return r.json()
# 例：監視で閾値超過時に呼ぶ
# trigger_rollback("p95_latency_exceeded")
ドキュメントと運用ランブック更新
退役手順はRunbookへ必ず反映し、運用担当が1人で対応できるチェックリストに落とし込みます。FAQを用意して想定される問い合わせに即座に答えられる形にします。
| 項目 | テンプレート例 |
|---|---|
| 緊急手順 | traffic-splitを旧モデル100%に戻すAPIコマンド、ログの収集場所 |
| 定常作業 | アーティファクト削除、保持ポリシー更新、ドキュメント更新手順 |
落とし穴と実務Tips
- 依存サービスの見落とし：前処理や特徴量生成を別サービスが担っているケースは多い。連携箇所を一覧化する。
- バージョン相互運用の問題：新旧フォーマット混在時の互換性レイヤーを用意する。
- ユーザ影響の最小化：重要ユーザは初期canaryから除外する、十分な通知期間を設ける。
- 法務・保存要件：古いモデルのログやアーティファクト保持期間を法務と合意する。
付録（成果物）
以下は現場でそのまま使えるチェックリスト（HTMLテーブル）と、主要スニペットのまとめです。必要に応じてコピーしてRunbookへ貼り付けてください。
退役チェックリスト（コピー用）
| 項目 | 担当 | 期限 | 完了 |
|---|---|---|---|
| 影響範囲の洗い出し | プロダクト | ||
| ステークホルダー通知 | PM | ||
| Shadowテスト実施 | 開発/SRE | ||
| Canary開始（%設定） | SRE | ||
| 監視閾値の確認 | 開発/プロダクト | ||
| 完全切替と古いモデルの退役 | 運用/法務 |
まとめ
モデルの退役・置換は技術だけでなく、関係者調整、法務対応、運用手順整備が鍵です。ポイントは「段階的に進めること」と「自動化された検証と明確なロールバック基準」を用意すること。この記事で示したチェックリストとPythonスニペットを基に、まずは小さなcanaryから始め、運用ランブックに落とし込んでいってください。次回は実際の運用ログから異常を検知するモニタリング例を紹介します。