"""Verifier Budgeter: Selective verification for high-risk outputs only.""" from typing import Dict, Optional, Tuple from dataclasses import dataclass @dataclass class VerifierDecision: should_verify: bool verifier_type: str # "full", "spot_check", "skip" confidence: float reasoning: str estimated_cost: float RISK_VERIFIER_MAP = { "critical": {"min_confidence": 0.95, "verifier_type": "full"}, "high": {"min_confidence": 0.85, "verifier_type": "full"}, "medium": {"min_confidence": 0.70, "verifier_type": "spot_check"}, "low": {"min_confidence": 0.50, "verifier_type": "spot_check"}, } class VerifierBudgeter: def __init__(self, verifier_cost: float = 0.02, max_verifications_per_run: int = 3): self.verifier_cost = verifier_cost self.max_per_run = max_verifications_per_run self.verifications_this_run = 0 self.stats = {"verified":0,"skipped":0,"spot_checked":0,"false_passes":0,"false_rejects":0} def should_verify(self, task_type: str, risk: str, model_confidence: float, is_irreversible: bool = False, has_prior_failures: bool = False, model_tier: int = 4, step_num: int = 0, total_steps: int = 1) -> VerifierDecision: if self.verifications_this_run >= self.max_per_run: self.stats["skipped"] += 1 return VerifierDecision(False, "skip", model_confidence, "max verifications reached", 0.0) # Check if this is the final answer is_final = (step_num == total_steps or step_num == -1) # Risk-based threshold risk_config = RISK_VERIFIER_MAP.get(risk, RISK_VERIFIER_MAP["medium"]) min_conf = risk_config["min_confidence"] default_type = risk_config["verifier_type"] # Adjust for context should = False reasons = [] if model_confidence < min_conf: should = True reasons.append(f"low confidence ({model_confidence:.2f} < {min_conf})") if is_irreversible and risk in ("high", "critical"): should = True reasons.append("irreversible + high risk") if has_prior_failures: should = True reasons.append("prior failures exist") if model_tier <= 2 and risk in ("high", "critical"): should = True reasons.append("cheap model on high-risk task") if is_final and risk in ("high", "critical"): should = True reasons.append("final answer on high-risk task") if not should: self.stats["skipped"] += 1 return VerifierDecision(False, "skip", model_confidence, "no verification needed", 0.0) self.verifications_this_run += 1 vtype = default_type if model_confidence > min_conf and not is_irreversible: vtype = "spot_check" self.stats["spot_checked"] += 1 else: self.stats["verified"] += 1 return VerifierDecision(True, vtype, model_confidence, "; ".join(reasons), self.verifier_cost) def reset_run(self): self.verifications_this_run = 0