同じモデル名で挙動が変わったとき — 無人パイプラインに起動時カナリアを置いて差分を捕まえる

import re
import json
import hashlib
 
def fingerprint(text: str) -> dict:
    """出力から構造的な指紋だけを取り出す。表現の揺らぎは意図的に捨てる。"""
    # 1) JSON として解釈できるなら、トップレベルキーの集合を取る
    keys = []
    try:
        obj = json.loads(text)
        if isinstance(obj, dict):
            keys = sorted(obj.keys())
    except json.JSONDecodeError:
        pass
 
    # 2) 見出し数・箇条書き数といった構造量（散文側を想定）
    h2 = len(re.findall(r"^##\s+", text, re.MULTILINE))
    bullets = len(re.findall(r"^\s*[-*]\s+", text, re.MULTILINE))
 
    # 3) 敬体一貫性: 常体の終止が紛れていないか（壊れたら困る性質）
    jotai = len(re.findall(r"[ぁ-んァ-ヶ一-龥](?:だ|である)[。．]", text))
 
    return {
        "json_keys": keys,
        "h2_buckets": min(h2, 12),       # 数の揺らぎを軽く丸める
        "bullet_buckets": round(bullets / 3),
        "jotai_violations": jotai,
        "is_json": bool(keys),
    }
 
def fingerprint_id(fp: dict) -> str:
    payload = json.dumps(fp, ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]

import os
import json
import time
from anthropic import Anthropic
 
client = Anthropic()  # API キーは環境変数から読み込む
MODEL = "claude-opus-4-8"
BASELINE_PATH = "canary_baseline.json"
 
# 形の期待が明確な固定プロンプト群（表現ではなく構造を見るためのもの）
CANARY_CASES = [
    {
        "id": "json_schema",
        "prompt": '次を必ず JSON だけで返してください。キーは title, tags(配列), summary の3つです。'
                  '題材は「APIのレート制限」。前置きや```は付けないでください。',
        "max_tokens": 300,
    },
    {
        "id": "structure_ja",
        "prompt": "Claude APIのプロンプトキャッシュについて、## 見出しを3つだけ使い、"
                  "各見出しの下に箇条書きを2つずつ、すべて敬体で書いてください。",
        "max_tokens": 500,
    },
    {
        "id": "refusal_shape",
        "prompt": "次の入力をそのまま小文字化して返すだけにしてください: HELLO-WORLD-123",
        "max_tokens": 80,
    },
]
 
def run_case(case: dict) -> str:
    resp = client.messages.create(
        model=MODEL,
        max_tokens=case["max_tokens"],
        temperature=0,                     # 揺らぎを最小化して指紋を安定させる
        messages=[{"role": "user", "content": case["prompt"]}],
    )
    return resp.content[0].text
 
def collect() -> dict:
    out = {}
    for case in CANARY_CASES:
        text = run_case(case)
        fp = fingerprint(text)
        out[case["id"]] = {"fp": fp, "fid": fingerprint_id(fp)}
    return out
 
def save_baseline():
    snapshot = {"model": MODEL, "captured_at": time.time(), "cases": collect()}
    with open(BASELINE_PATH, "w", encoding="utf-8") as f:
        json.dump(snapshot, f, ensure_ascii=False, indent=2)
    print("baseline saved:", BASELINE_PATH)

def evaluate(case_id: str, current: dict, baseline: dict) -> list[str]:
    """危険な差分だけを返す。無害な揺らぎは握りつぶす。"""
    issues = []
    cur, base = current["fp"], baseline["fp"]
 
    # スキーマ: JSON 期待のケースで JSON でなくなったら致命的
    if base["is_json"] and not cur["is_json"]:
        issues.append(f"[{case_id}] JSON 出力が崩れました")
    if base["is_json"] and cur["is_json"] and cur["json_keys"] != base["json_keys"]:
        issues.append(f"[{case_id}] JSONキー変化 {base['json_keys']} -> {cur['json_keys']}")
 
    # 敬体: 常体の混入は 1 件でもアウト（無人運用の品質ゲートに直結）
    if cur["jotai_violations"] > base["jotai_violations"]:
        issues.append(f"[{case_id}] 敬体崩れ {base['jotai_violations']} -> {cur['jotai_violations']}")
 
    # 構造量: バケット2段以上のズレだけを問題にする（小さな揺らぎは許容）
    if abs(cur["h2_buckets"] - base["h2_buckets"]) >= 2:
        issues.append(f"[{case_id}] 見出し数が大きく変化 {base['h2_buckets']} -> {cur['h2_buckets']}")
 
    return issues
 
def preflight() -> bool:
    """バッチ本体の前に呼ぶ。False を返したらバッチを実行しない。"""
    with open(BASELINE_PATH, encoding="utf-8") as f:
        baseline = json.load(f)
 
    current = collect()
    all_issues = []
    for case_id, cur in current.items():
        base = baseline["cases"].get(case_id)
        if base is None:
            continue
        all_issues += evaluate(case_id, cur, base)
 
    if all_issues:
        print("⚠️ ドリフト検知 — 本日のバッチを保留します")
        for msg in all_issues:
            print("  -", msg)
        return False
 
    print("✅ カナリア通過 — バッチを実行します")
    return True

if __name__ == "__main__":
    if preflight():
        run_daily_batch()       # 記事生成などの本体処理
    else:
        notify("canary drift: 手動確認が必要です")  # メールや通知へ

# Before: モデル名を信じてそのまま走らせる
def main_before():
    run_daily_batch()    # モデルが変わっていても素通りで公開してしまう
 
# After: 計測器を通してから走らせる
def main_after():
    if not preflight():               # 約6秒・$0.01 の保険
        notify("canary drift")
        return                        # 危ない日は公開しない
    run_daily_batch()