画像分類の信頼度を鵜呑みにしない — クラス別しきい値の較正と棄却ルーティング

# before.py — 一律しきい値
THRESHOLD = 0.80
 
def route(prediction: dict) -> str:
    """prediction = {"category": "animal", "confidence": 0.87}"""
    if prediction["confidence"] >= THRESHOLD:
        return "auto_accept"
    return "manual_review"

# calibrate.py — クラス別しきい値の探索
from collections import defaultdict
from dataclasses import dataclass
 
@dataclass
class Sample:
    predicted: str
    truth: str
    confidence: float
 
MIN_SUPPORT = 20          # 厚いクラスの下限
FALLBACK_THRESHOLD = 0.93  # 薄いクラスの共有しきい値
GRID = [round(0.50 + 0.01 * i, 2) for i in range(50)]  # 0.50〜0.99
 
def _precision_at(samples: list[Sample], t: float) -> tuple[float, int]:
    accepted = [s for s in samples if s.confidence >= t]
    if not accepted:
        return 0.0, 0
    hits = sum(1 for s in accepted if s.predicted == s.truth)
    return hits / len(accepted), len(accepted)
 
def calibrate(
    samples: list[Sample],
    target_precision: float = 0.95,
    min_coverage: int = 5,
) -> dict[str, float]:
    """予測クラスごとに、目標 precision を満たす最小しきい値を返す。"""
    by_class: dict[str, list[Sample]] = defaultdict(list)
    for s in samples:
        by_class[s.predicted].append(s)
 
    thresholds: dict[str, float] = {}
    for cls, group in by_class.items():
        if len(group) < MIN_SUPPORT:
            thresholds[cls] = FALLBACK_THRESHOLD
            continue
 
        chosen = None
        for t in GRID:  # 低い順に走査し、最初に条件を満たした点で止める
            prec, n = _precision_at(group, t)
            if n >= min_coverage and prec >= target_precision:
                chosen = t
                break
        # どの点でも目標に届かないクラスは、自動採用を諦める
        thresholds[cls] = chosen if chosen is not None else 1.01
 
    return thresholds

# route.py — 棄却理由つきルーティング
from typing import Literal, TypedDict
 
class Prediction(TypedDict):
    category: str
    confidence: float
 
class Decision(TypedDict):
    action: Literal["auto_accept", "manual_review"]
    reason: str
    priority: int  # 小さいほど先に処理する
 
def route(pred: Prediction, thresholds: dict[str, float]) -> Decision:
    cls = pred["category"]
    t = thresholds.get(cls, FALLBACK_THRESHOLD)
    c = pred["confidence"]
 
    if c >= t:
        return {"action": "auto_accept", "reason": "above_threshold", "priority": 9}
 
    if t > 1.0:
        # 到達不能クラス: 較正で救えない。定義の見直し対象
        return {"action": "manual_review", "reason": "class_not_calibratable", "priority": 1}
 
    if t - c <= 0.05:
        # 惜しい: 人が見れば数秒で決まる
        return {"action": "manual_review", "reason": "near_threshold", "priority": 2}
 
    return {"action": "manual_review", "reason": "low_confidence", "priority": 5}

# drift_check.py — 較正の劣化検知
def check_drift(
    samples: list[Sample],
    thresholds: dict[str, float],
    target_precision: float = 0.95,
    tolerance: float = 0.02,
) -> list[str]:
    """しきい値を固定したまま、現在の precision を測り直す。"""
    alerts = []
    by_class: dict[str, list[Sample]] = defaultdict(list)
    for s in samples:
        by_class[s.predicted].append(s)
 
    for cls, group in by_class.items():
        t = thresholds.get(cls, FALLBACK_THRESHOLD)
        prec, n = _precision_at(group, t)
        if n == 0:
            alerts.append(f"{cls}: 自動採用が 0 件。しきい値が高すぎる可能性")
            continue
        if prec < target_precision - tolerance:
            alerts.append(
                f"{cls}: precision {prec:.3f} < {target_precision - tolerance:.3f} "
                f"(n={n}, threshold={t}) → 再較正が必要"
            )
    return alerts