| """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 |
| 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) |
| |
| is_final = (step_num == total_steps or step_num == -1) |
| |
| risk_config = RISK_VERIFIER_MAP.get(risk, RISK_VERIFIER_MAP["medium"]) |
| min_conf = risk_config["min_confidence"] |
| default_type = risk_config["verifier_type"] |
| |
| 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 |
|
|
