Claude Agent SDK でシャドウモードを実装する — 本番影響ゼロでエージェント精度を測る段階的ロールアウト設計

// shadow-runner.ts
// 目的: 既存の分類関数 legacyClassify を正としつつ、
//      Claude Agent SDK のエージェントを同じ入力で並走させ、
//      結果だけを記録する（本番出力には影響を与えない）
import Anthropic from "@anthropic-ai/sdk";
import { recordShadowResult } from "./shadow-logger";
 
const client = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY });
 
type Ticket = { id: string; subject: string; body: string };
type Category = "billing" | "technical" | "account" | "other";
 
export async function classifyWithShadow(
  ticket: Ticket,
  legacyClassify: (t: Ticket) => Promise<Category>,
): Promise<Category> {
  // 1. 本番の結果は legacy を正とする(最優先・失敗させない)
  const legacyPromise = legacyClassify(ticket);
 
  // 2. シャドウは非同期に走らせ、本番のレイテンシには影響させない
  //    Promise.allSettled にせず、fire-and-forget で切り離す
  runShadow(ticket, legacyPromise).catch((err) => {
    // シャドウ側のエラーは本番に伝播させない
    console.warn("[shadow] suppressed error:", err?.message);
  });
 
  return legacyPromise;
}
 
async function runShadow(ticket: Ticket, legacyPromise: Promise<Category>) {
  const startedAt = Date.now();
  try {
    const response = await client.messages.create({
      model: "claude-haiku-4-5-20251001",
      max_tokens: 200,
      system:
        "あなたはチケット分類エージェントです。" +
        "billing / technical / account / other のいずれか一語のみ返してください。",
      messages: [
        {
          role: "user",
          content: `件名: ${ticket.subject}\n本文: ${ticket.body}`,
        },
      ],
    });
    const shadow = normalizeCategory(
      response.content[0].type === "text" ? response.content[0].text : "",
    );
    const legacy = await legacyPromise;
 
    await recordShadowResult({
      ticketId: ticket.id,
      legacy,
      shadow,
      match: legacy === shadow,
      latencyMs: Date.now() - startedAt,
      usage: response.usage, // input_tokens / output_tokens
    });
  } catch (err) {
    // タイムアウト・429・5xx などを分類して記録する
    await recordShadowResult({
      ticketId: ticket.id,
      legacy: await legacyPromise.catch(() => "other"),
      shadow: null,
      match: null,
      latencyMs: Date.now() - startedAt,
      error: normalizeError(err),
    });
  }
}
 
function normalizeCategory(raw: string): Category {
  const lower = raw.trim().toLowerCase();
  if (["billing", "technical", "account", "other"].includes(lower)) {
    return lower as Category;
  }
  // 予想外の出力を other に丸めるのは安全策だが、集計側で「フォーマット違反」
  // を別カウントするために、もとの文字列もログに残す設計が望ましい
  return "other";
}
 
function normalizeError(err: unknown) {
  if (err instanceof Error) {
    return { name: err.name, message: err.message };
  }
  return { name: "Unknown", message: String(err) };
}

# shadow_compare.py
# 目的: 生成系エージェントの出力と既存処理の出力を比較する指標を計算する
#      完全一致ではなく、複数のシグナルを組み合わせて乖離を検知する
from dataclasses import dataclass
from typing import Optional
import difflib
 
@dataclass
class ComparisonResult:
    text_similarity: float           # 0.0-1.0
    length_ratio: float              # 長さの比率(1.0で同じ)
    sentiment_match: Optional[bool]  # 感情極性の一致(分析できた場合のみ)
    structural_match: bool           # 見出し・リスト構造の一致
    flags: list[str]                 # 人間確認が必要と判断した理由
 
def compare_shadow_vs_legacy(shadow: str, legacy: str) -> ComparisonResult:
    """
    シャドウ出力と既存出力を多角的に比較する。
    単一スコアではなく、複数の軸で差分を取ることで
    後のレビュー担当者が「どこが違うか」を絞り込みやすくなる。
    """
    flags: list[str] = []
 
    # 1. 表層の類似度(テキスト差分率)
    text_sim = difflib.SequenceMatcher(None, shadow, legacy).ratio()
    if text_sim < 0.3:
        flags.append("very_different_text")
 
    # 2. 長さの比率 — 極端に短い/長い応答は要注意
    shadow_len = max(len(shadow), 1)
    legacy_len = max(len(legacy), 1)
    length_ratio = shadow_len / legacy_len
    if length_ratio < 0.3 or length_ratio > 3.0:
        flags.append("length_anomaly")
 
    # 3. 構造の一致 — 見出しや箇条書きの数が合うか
    structural_match = (
        shadow.count("\n-") == legacy.count("\n-")
        and shadow.count("\n#") == legacy.count("\n#")
    )
    if not structural_match:
        flags.append("structural_diff")
 
    # 4. 感情極性の一致 — 返信の「お詫び/案内」など方向が違うと致命的
    sentiment_match = None
    # 実装は外部ライブラリやモデル呼び出しに委ねる(この関数外で行う)
 
    return ComparisonResult(
        text_similarity=text_sim,
        length_ratio=length_ratio,
        sentiment_match=sentiment_match,
        structural_match=structural_match,
        flags=flags,
    )
 
if __name__ == "__main__":
    # 期待する出力例:
    # ComparisonResult(text_similarity=0.72, length_ratio=1.1, ...)
    r = compare_shadow_vs_legacy(
        shadow="## ご案内\nご請求について、次の通り対応いたします。\n- 返金可能です",
        legacy="## ご案内\nご請求について、以下の通り対応します。\n- 返金いたします",
    )
    print(r)

const sampleRate = Number(process.env.SHADOW_SAMPLE_RATE ?? "0.05");
if (Math.random() < sampleRate) {
  runShadow(ticket, legacyPromise).catch(() => {});
}

# shadow_budget.py — シャドウ並走の月額コストを、実行前に見積もる
# 単価は仮置きです。公式の料金ページで現行値を確認して差し替えてください。
def monthly_cost_usd(requests_per_month, sample_rate, in_tokens, out_tokens,
                     usd_per_mtok_in, usd_per_mtok_out):
    n = requests_per_month * sample_rate
    return n * (in_tokens * usd_per_mtok_in + out_tokens * usd_per_mtok_out) / 1_000_000
 
if __name__ == "__main__":
    for rate in (1.0, 0.25, 0.05):
        c = monthly_cost_usd(10_000_000, rate, 800, 20, 1.0, 5.0)  # 仮単価 $1 / $5
        print(f"sample_rate={rate:>4}: ${c:>9,.0f} / 月 （{10_000_000*rate:,.0f} 件）")

sample_rate= 1.0: $    9,000 / 月 （10,000,000 件）
sample_rate=0.25: $    2,250 / 月 （2,500,000 件）
sample_rate=0.05: $      450 / 月 （500,000 件）

// producer.ts — 本番サービス内では投入のみ
await env.SHADOW_QUEUE.send({
  ticket,
  legacyResult,
  enqueuedAt: Date.now(),
});
 
// consumer.ts — 別ワーカーで実処理
export default {
  async queue(batch: MessageBatch, env: Env) {
    for (const message of batch.messages) {
      try {
        // legacyResult は確定済みの値。runShadow は Promise も値も受けられるようにしておく
        await runShadow(message.body.ticket, Promise.resolve(message.body.legacyResult));
        message.ack();
      } catch (err) {
        message.retry({ delaySeconds: 60 });
      }
    }
  },
};

# shadow_sample_size.py — 「一致率 92%」を信じてよい件数を、Wilson 下限で確かめる
from math import sqrt
 
def wilson_lower(matches, n, z=1.96):
    if n == 0:
        return 0.0
    p = matches / n
    denom = 1 + z * z / n
    centre = p + z * z / (2 * n)
    margin = z * sqrt(p * (1 - p) / n + z * z / (4 * n * n))
    return (centre - margin) / denom
 
if __name__ == "__main__":
    # 観測した一致率が 94% のとき、n 件で「真の一致率は少なくとも」どこまで言えるか
    for n in (50, 200, 1000, 5000):
        m = round(n * 0.94)
        print(f"n={n:>5}  観測 94%  95%下限 = {wilson_lower(m, n):.1%}")
    # 下限が 92% を超えるのに必要な最小件数（観測 94% 固定）
    n = 50
    while wilson_lower(round(n * 0.94), n) < 0.92:
        n += 10
    print("下限が 92% を超える最小件数 ≒", n)

n=   50  観測 94%  95%下限 = 83.8%
n=  200  観測 94%  95%下限 = 89.8%
n= 1000  観測 94%  95%下限 = 92.4%
n= 5000  観測 94%  95%下限 = 93.3%
下限が 92% を超える最小件数 ≒ 690