自動採用された出力こそ抜き取る — 限られたレビュー予算で静かな品質劣化に気づく設計

# detection_power.py
from math import log, ceil
 
def detect_prob(p: float, n: int, days: int) -> float:
    """不良率 p の母集団から毎日 n 件抜いたとき、days 日以内に1件以上捉える確率"""
    return 1.0 - (1.0 - p) ** (n * days)
 
def days_to_detect(p: float, n: int, confidence: float = 0.95) -> int:
    """confidence の確率で捉えるまでに要する日数"""
    if p <= 0 or n <= 0:
        return 10**9
    return ceil(log(1 - confidence) / (n * log(1 - p)))
 
def daily_n_for(p: float, days: int, confidence: float = 0.95) -> int:
    """days 日以内・confidence で捉えるために必要な1日あたり抜き取り件数"""
    return ceil(log(1 - confidence) / (days * log(1 - p)))
 
if __name__ == "__main__":
    for p in (0.01, 0.02, 0.05, 0.10):
        row = [f"p={p:.0%}"]
        for n in (10, 20, 30):
            row.append(f"n={n}: {days_to_detect(p, n)}日")
        print("  ".join(row))

# stratified_sampler.py
import random
from collections import defaultdict
 
# 層ごとのリスク重み。運用しながら手で調整する値
RISK_WEIGHTS = {
    "new_model": 4.0,      # 解決後モデルIDが直近7日以内に初出
    "new_prompt": 3.0,     # プロンプトリビジョンが直近7日以内に初出
    "changed_input": 2.0,  # 入力に前回からの差分あり
    "steady": 1.0,         # 上記いずれにも当たらない定常層
}
 
def stratum_of(rec: dict, fresh_models: set[str], fresh_prompts: set[str]) -> str:
    if rec["model_id"] in fresh_models:
        return "new_model"
    if rec["prompt_rev"] in fresh_prompts:
        return "new_prompt"
    if rec["changed_input"]:
        return "changed_input"
    return "steady"
 
def allocate(pool: list[dict], budget: int,
             fresh_models: set[str], fresh_prompts: set[str],
             min_per_stratum: int = 2, seed: int | None = None) -> list[dict]:
    rng = random.Random(seed)
    strata: dict[str, list[dict]] = defaultdict(list)
    for rec in pool:
        strata[stratum_of(rec, fresh_models, fresh_prompts)].append(rec)
 
    present = {k: v for k, v in strata.items() if v}
    # 定常層にも必ず最低枠を残す。ここを削ると「安全なはずの層」が永久に見られなくなる
    reserved = {k: min(min_per_stratum, len(v)) for k, v in present.items()}
    remaining = max(budget - sum(reserved.values()), 0)
 
    total_w = sum(RISK_WEIGHTS[k] * len(v) for k, v in present.items())
    picked: list[dict] = []
    for k, v in present.items():
        share = 0 if total_w == 0 else remaining * (RISK_WEIGHTS[k] * len(v)) / total_w
        n = min(len(v), reserved[k] + int(round(share)))
        picked.extend(rng.sample(v, n))
    return picked[:budget]

# sequential_monitor.py
from dataclasses import dataclass
 
@dataclass
class CusumState:
    s: float = 0.0          # 累積和
    days_above: int = 0     # しきい値超過が続いた日数
 
def update(state: CusumState, defects: int, sampled: int,
           p0: float = 0.01, slack: float = 0.005, limit: float = 0.06) -> tuple[CusumState, bool]:
    """
    p0    : 許容する定常不良率
    slack : この幅の上振れは日常のゆらぎとして吸収する
    limit : 累積和がこれを超えたら劣化とみなす
    """
    if sampled == 0:
        return state, False
    observed = defects / sampled
    state.s = max(0.0, state.s + (observed - p0 - slack))
    fired = state.s > limit
    state.days_above = state.days_above + 1 if fired else 0
    return state, fired
 
def reset_after_action(state: CusumState) -> CusumState:
    """原因に手を入れたら累積をリセットする。放置したままだと鳴り続ける"""
    return CusumState()
 
if __name__ == "__main__":
    # 前半は定常（1%相当）、後半に3%へ劣化したときの挙動
    import random
    rng = random.Random(7)
    st = CusumState()
    for day in range(1, 21):
        p = 0.01 if day <= 10 else 0.03
        defects = sum(1 for _ in range(20) if rng.random() < p)
        st, fired = update(st, defects, 20)
        print(f"day{day:2d} p={p:.0%} defects={defects} S={st.s:.3f} {'ALERT' if fired else ''}")