何本作ったかを指標にするのをやめた — 生成エージェントに「棄権」という結果を持たせる設計

from collections import Counter
 
 
class RunOutcome(str, Enum):
    PRODUCED = "produced"            # ACCEPT が1本以上。純増あり
    HEALTHY_ABSTAIN = "healthy_abstain"  # 全部 ABSTAIN。正しく見送った成功
    DEGRADED = "degraded"           # REJECT が支配的。入力かモデルの異常
    BLOCKED = "blocked"             # そもそも生成に入れなかった（後述）
 
 
def classify_run(results: list[GateResult]) -> RunOutcome:
    if not results:
        return RunOutcome.BLOCKED
 
    tally = Counter(r.verdict for r in results)
    accepts = tally[Verdict.ACCEPT]
    rejects = tally[Verdict.REJECT]
 
    if accepts >= 1:
        return RunOutcome.PRODUCED
 
    # ACCEPT ゼロでも、内訳で意味が変わる
    if rejects > len(results) // 2:
        # 半分超が要件違反 = ゲートの手柄ではなく入力側の異常
        return RunOutcome.DEGRADED
 
    # 大半が ABSTAIN = 出すに値するものが今回は無かった、という健全な結論
    return RunOutcome.HEALTHY_ABSTAIN

@dataclass
class AbstentionWindow:
    """直近Nラン分の棄権率を追い、傾向で警告を出す。"""
    window: int = 14
    history: list[float] = field(default_factory=list)
 
    def record(self, results: list[GateResult]) -> None:
        if not results:
            return
        abstains = sum(1 for r in results if r.verdict is Verdict.ABSTAIN)
        self.history.append(abstains / len(results))
        self.history = self.history[-self.window:]
 
    def rate(self) -> float:
        return sum(self.history) / len(self.history) if self.history else 0.0
 
    def trend_signal(self) -> str:
        if len(self.history) < self.window:
            return "collecting"
        recent = sum(self.history[-3:]) / 3
        base = sum(self.history[:-3]) / (len(self.history) - 3)
        if recent > base + 0.20:
            # 棄権率の上昇は「品質低下」ではなく多くの場合「題材の飽和」
            return "input_saturation"   # → 入力を変える。ゲートは触らない
        if recent < base - 0.20 and self.rate() < 0.10:
            return "gate_too_loose"     # → 通り過ぎ。ゲートを見直す
        return "stable"

from __future__ import annotations
from dataclasses import dataclass
import statistics
 
 
class VacuousThreshold(RuntimeError):
    """観測分布の外にあるしきい値。起動時に落とす。"""
 
 
@dataclass(frozen=True)
class CalibratedFloor:
    """しきい値を「どの指標で測ったか」と束ねて持つ。
 
    metric_id が変われば floor も必ず引き直す。samples は自分のコーパスで
    測った最近傍スコアの実測列（novelty = 1 - similarity に変換済み）。
    """
    metric_id: str
    floor: float
    samples: tuple[float, ...]
 
    def __post_init__(self) -> None:
        if len(self.samples) < 30:
            raise VacuousThreshold(
                f"{self.metric_id}: 実測サンプルが {len(self.samples)} 本。30 本以上で校正する"
            )
        lo, hi = min(self.samples), max(self.samples)
        if not (lo <= self.floor <= hi):
            side = "全候補が常に通過" if self.floor < lo else "全候補が常に棄権"
            raise VacuousThreshold(
                f"{self.metric_id}: floor={self.floor} が観測レンジ "
                f"[{lo:.3f}, {hi:.3f}] の外です（{side}）"
            )
 
    def fires_on(self) -> float:
        """このしきい値が実測サンプルのうち何割を棄権させるか。"""
        return sum(1 for v in self.samples if v < self.floor) / len(self.samples)
 
    def describe(self) -> str:
        return (
            f"{self.metric_id}: floor={self.floor:.3f} "
            f"median={statistics.median(self.samples):.3f} "
            f"棄権率見込み={self.fires_on():.0%}"
        )
 
 
def load_novelty_samples(path: str) -> tuple[float, ...]:
    """校正用の実測列を読む。similarity を 1 から引いて novelty にする。"""
    with open(path, encoding="utf-8") as fp:
        sims = [float(line) for line in fp if line.strip()]
    return tuple(1.0 - s for s in sims)

samples: 120
OK   : jaccard3gram-novelty-v1: floor=0.900 median=0.909 棄権率見込み=25%
BLOCK: jaccard3gram-novelty-v1: floor=0.35 が観測レンジ [0.850, 0.943] の外です（全候補が常に通過）
OK   : jaccard3gram-novelty-v1: floor=0.860 median=0.909 棄権率見込み=1%

@dataclass
class GroundingCheck:
    required_paths: list[str]
    min_bytes: int = 512
    max_age_hours: float = 48.0
 
    def ready(self, fs, now_ts: float) -> tuple[bool, list[str]]:
        problems: list[str] = []
        for path in self.required_paths:
            meta = fs.stat(path)
            if meta is None:
                problems.append(f"欠落: {path}")            # 無音失敗を許さない
            elif meta.size < self.min_bytes:
                problems.append(f"実体なし({meta.size}B): {path}")
            elif (now_ts - meta.mtime) / 3600 > self.max_age_hours:
                age = (now_ts - meta.mtime) / 3600
                problems.append(f"陳腐化({age:.0f}h): {path}")
        return (len(problems) == 0, problems)
 
 
def run_generation(candidates_fn, gate, grounding, fs, now_ts, log):
    ok, problems = grounding.ready(fs, now_ts)
    if not ok:
        # 生成に入らない。純増できる保証がない入力で走らせない
        log.warn("BLOCKED: grounding not ready", extra={"problems": problems})
        return RunOutcome.BLOCKED, []
 
    results = [gate.evaluate(c) for c in candidates_fn()]
    outcome = classify_run(results)
    log.info("run complete", extra={"outcome": outcome, "n": len(results)})
    return outcome, results