2026年7月9日 – AIに仕事を任せたい人のためのPython塾 ― Manage AI ―

はじめに — つまずきに寄り添う一言

本番で複数モデルを比べたい、でも「どのようにトラフィックを割り振るか」「指標は何を見ればよいか」「勝者をどう決めるか」「失敗したときにどう戻すか」で悩んでいませんか。この記事では、実務で安全に回せるA/B実験（モデル比較）と段階的ロールアウト（カナリア／フェーズ展開）を、Pythonコード例と運用チェックリストを交えて具体的に示します。第105回（モデルカタログ）、第104回（オーケストレーション）、第95回（SLO監視）とつなぐ実務的な手順に重点を置きます。

目次（記事構成）

目的設定
トラフィック戦略（ランダム・セグメント・ユーザ単位）
計測指標設計（一次／二次／リスク指標）
実装例（FastAPI/Flaskルータ、Redis/DB割当）
評価方法（統計的検定・ベイズ・監視）
自動昇格・ロールバックのオーケストレーション
運用チェックリスト
まとめ

目的設定

まずは実験の目的を明確にします。目的が曖昧だと判断基準もぶれます。

項目	例
一次目的指標	コンバージョン率、CTR、課金率
検出したい差の大きさ	最低で絶対差 0.5%（ビジネスで意味のある差）
期間	最低 2 週間（ユーザ挙動の週次性を考慮）
リスク許容度	SLO の 1 日違反があれば即ロールバック

トラフィック戦略

割当方法は目的とリスクに応じて選びます。代表的な3つを示します。

1) ランダム（ユーザ単位が推奨）

ユーザ単位でハッシュにより決めると、再帰性が保てて計測が安定します。

2) セグメント分割（属性ベース）

地域、デバイス、新規/既存ユーザなどにより偏りがある場合はこちらを併用します。特定セグメントでのみ効果が期待されるとき有効です。

3) セッション／リクエスト単位（限定的に）

短期検証やUIレイアウトのテストで使いますが、学習モデルの比較ではユーザ単位での保持が好まれます。

方式	利点	注意点
ユーザ単位ランダム	安定した計測、バイアス小	匿名ユーザには割当難
セグメント	特定層の効果検証に有効	交絡に注意、分割数は限定
セッション	すばやい反復	ユーザ内相関を無視できない

計測指標設計

指標は一次指標（決定基準）、二次指標（補助）、リスク指標（安全性）に分けます。

種別	例	用途
一次指標	コンバージョン率、平均課金	勝者判定の主軸
二次指標	CTR、滞在時間、リテンション	補助的な解釈や因果の確認
リスク指標	エラーレート、レイテンシ、SLO違反件数	安全性判断と即時ロールバックのトリガー

ログは構造化JSONで出力し、以下のタグは必須にします: experiment_id, model_id（第105回の管理IDを使用）、cohort（A/B）、user_id（可能な限り）、event_time, metric_values, request_id。

実装例（トラフィック割当とルーティング）

ここではユーザ単位ハッシュ割当の簡易実装と、FastAPI のルータ例、Redis を用いた割当保持の例を示します。

1) 決定的な割当関数（ハッシュ）

import hashlib

def assign_variant(user_id: str, experiment_key: str, allocations: dict) -> str:
    """allocations: {'control': 0.5, 'treatment': 0.5} のように合計1.0
    戻り値: variant name"""
    key = f"{experiment_key}:{user_id}".encode('utf-8')
    h = int(hashlib.sha256(key).hexdigest(), 16)
    r = (h % 10000) / 10000.0
    cum = 0.0
    for name, prob in allocations.items():
        cum += prob
        if r < cum:
            return name
    return list(allocations.keys())[-1]

2) FastAPI のエンドポイント例（簡略）

from fastapi import FastAPI, Request
import time

app = FastAPI()

EXPERIMENTS = {
    'exp_v1': {'allocations': {'control': 0.9, 'candidate': 0.1}}
}

@app.post('/predict')
async def predict(request: Request):
    body = await request.json()
    user_id = body.get('user_id', 'anon')
    exp = EXPERIMENTS['exp_v1']
    variant = assign_variant(user_id, 'exp_v1', exp['allocations'])
    # ログは構造化JSONで出力
    log = {
        'event_time': time.time(),
        'experiment_id': 'exp_v1',
        'model_id': 'model:2026-07-01:abc123',
        'cohort': variant,
        'user_id': user_id,
        'request_id': body.get('request_id')
    }
    print(log)
    # 実際のリクエストはモデルルータへフォワード
    return {'variant': variant}

3) Redis による割当保持サンプル（任意）

import redis
r = redis.Redis(host='localhost', port=6379)

# 初回割当を保存して再利用する例
def assign_and_persist(user_id, experiment_key, allocations):
    key = f"assign:{experiment_key}:{user_id}"
    val = r.get(key)
    if val:
        return val.decode('utf-8')
    v = assign_variant(user_id, experiment_key, allocations)
    r.set(key, v, ex=60*60*24*30)  # 30日保持
    return v

注: 第105回モデルカタログの model_id をそのままログに含め、監査ログからモデルのバージョンに遡れるようにします。

評価方法

評価は統計的検定とベイズ推定の両方を示します。実務では両者を組み合わせると解釈が安定します。

頻度主義的な差の検定（例：二項比率の差）

from statsmodels.stats.proportion import proportions_ztest

# successes = [succ_A, succ_B]
# nobs = [n_A, n_B]
stat, pvalue = proportions_ztest(successes, nobs)
print('z=', stat, 'p=', pvalue)

pvalue が事前に定めた閾値（例 0.01 か 0.05）を下回り、かつ実際の差がビジネス上意味がある大きさであれば勝者判定の条件を満たします。ただし多重検定や途中停止によるバイアスに注意します。

ベイズ的判定（Beta-Bernoulli の例）

import numpy as np
from scipy.stats import beta

# 観測: success_A, n_A, success_B, n_B
alpha0, beta0 = 1, 1  # 澄明な事前
posterior_A = beta(alpha0 + success_A, beta0 + n_A - success_A)
posterior_B = beta(alpha0 + success_B, beta0 + n_B - success_B)
# サンプリングで優位確率を推定
samps = 10000
pa = posterior_A.rvs(samps)
pb = posterior_B.rvs(samps)
prob_B_better = (pb > pa).mean()
print('P(B > A) =', prob_B_better)

実務ルール例: P(B > A) > 0.95 かつ期待差がビジネス閾値以上なら昇格候補とする。

監視とSLO連携

SLO（第95回）と連動し、リスク指標が閾値を超えたら即時ロールバックします。SLOの監視はリアルタイムに近い形で短いウィンドウ（例 5 分・1 時間）と長期ウィンドウ（1 日）で評価します。

自動昇格・ロールバックのオーケストレーション

自動化は段階的に進めます。最初は半自動（人の承認を挟む）から始め、安全が確認できたら自動化を拡張します。

段階的ロールアウトの例

フェーズ	比率	期間／条件
カナリア	1%（内部ユーザ）	24 時間、SLO違反なしで次へ
フェーズ1	10%	3 日、メトリクス安定で次へ
フェーズ2	50%	1 週間、定量基準合格で全体展開

自動化フロー（概念コード）

# オーケストレーター（簡略）
# 1) 定期ジョブで評価スクリプトを実行（第104回のジョブ化）
# 2) 結果が閾値を満たせば昇格 API を呼ぶ
# 3) SLO 監視が閾値超過ならロールバック webhook を呼ぶ

def evaluation_job(experiment_id):
    metrics = fetch_metrics(experiment_id)
    result = analyze(metrics)
    if result['auto_promote']:
        orchestrator.api.promote(experiment_id)
    if result['slo_violated']:
        orchestrator.api.rollback(experiment_id)

Webhook 例: SLO モニタが Slack とオーケストレーターの rollback エンドポイントに通知する形を想定します。ロールバックはモデルカタログの previous_version に差し戻す運用が確実です。

実務的判断ルール（テンプレート）

状況	ルール（例）
必要サンプル数未達	昇格せず、期間延長を提案。ノイズが大きければ分割を減らす。
SLO 違反（短期）	即時ロールバック、自動チケット発行
短期変動あり	週次・日次のトレンドを確認し、単日での判断は避ける
複数指標の矛盾	一次指標優先。ただしリスク指標悪化は停止

失敗しやすいポイントと対策

トラフィック漏れ: ルーティングの網羅テストと監査ログで検出する。テストユーザを使った end-to-end 検証を自動化する。
計測バイアス: 新規ユーザ偏りはセグメント別集計を必須にする。
メトリクススキーマ不一致: ログスキーマを Schema Registry 的に管理し、型チェックを入れる。
実験期間の誤り: 週次性や祝日を考慮して期間を設定する。短すぎる判断は避ける。

運用チェックリスト（実務に落とし込む）

チェック項目	合格ライン/備考
experiment_id と model_id の紐付け	モデルカタログに登録済み（第105回）
ログに必要タグが含まれているか	experiment_id, model_id, cohort, user_id, request_id
SLO 監視連携	監視が Webhook でオーケストレーターに通知できる
オーケストレーションのジョブ化	評価ジョブが第104回の仕組みで定期実行されている
ロールバック手順の文書化	即時対応フローと担当者が明示されている

まとめ

本記事では、実務で回すモデルのA/Bテストと段階的ロールアウトについて、目的設定からトラフィック割当、指標設計、実装例、評価手法、自動昇格・ロールバックまでを一通り示しました。ポイントは次のとおりです。

目的（一次指標）をはっきりさせ、ログに model_id を含めて監査可能にする。
ユーザ単位の決定的割当を基本とし、Redis 等で割当を保持すると安定する。
評価は頻度主義とベイズの併用が実務的に有効。SLO 連携で安全策を自動化する。
段階的なカナリア→フェーズ→全体展開の流れを守り、異常時は即時ロールバックする運用を作る。

次の一歩: 本記事のコードをもとに、A/B テストの自動解析と報告テンプレート（運用レポート自動化）を扱う続編を予定しています。記事内の実装例は現場でそのまま使えるよう簡潔に示していますが、導入時はステージングで十分に検証してから本番へ適用してください。

参考: Manage AI の第105回（モデルカタログ）、第104回（オーケストレーション）、第95回（SLO監視）と組み合わせることで、実務で回る安全なワークフローが構築できます。

月	火	水	木	金	土	日
		1	2	3	4	5
6	7	8	9	10	11	12
13	14	15	16	17	18	19
20	21	22	23	24	25	26
27	28	29	30	31