Spaces:
Sleeping
Sleeping
samsadar236
feat: deploy ControlPlane.ai — updated backend, frontend redesign, Responsible AI Gate
54f341c | """Composite uncertainty for the review queue (T1.4, Lane A). | |
| Under Lane A the pipeline is deterministic and runs a single pass, so | |
| variance across passes is ~0 and is not a useful confidence signal. This | |
| module derives an uncertainty score in [0, 1] from the grade's EXTERNAL, | |
| deterministic signals instead: | |
| - grounding faithfulness (HHEM score, or the band if no score) | |
| - the deterministic rule verdict (citation grounding) | |
| - the LLM Critic verdict (only present when the gate escalated) | |
| - justifier flags (hallucination_risk, ambiguous_evidence, etc.) | |
| Higher uncertainty sorts to the FRONT of the review queue. Verbalised | |
| self-confidence is deliberately NOT used: it is poorly calibrated. | |
| `aggregate_passes` is a drop-in superset of grader.aggregate.aggregate_scores: | |
| it returns the same keys (median, max_score, min_score, std_dev, n_passes) | |
| plus `uncertainty`, so the review queue can sort on `uncertainty` while the | |
| rest of the code keeps working. For the Lane B option (num_passes > 1) it | |
| folds normalised score spread across passes into the uncertainty. | |
| """ | |
| from __future__ import annotations | |
| import statistics | |
| # Weights. Tuned so a fully-bad answer approaches 1.0; adjust on the eval set. | |
| W_GROUNDING = 0.40 # weight on the faithfulness gap | |
| W_RULE = 0.30 # deterministic rule check failed | |
| W_CRITIC = 0.20 # LLM Critic failed (if it ran) | |
| W_FLAG = 0.05 # per justifier flag | |
| W_FLAG_CAP = 0.15 # cap on the flag contribution | |
| W_SPREAD = 0.50 # weight on normalised score spread across passes (Lane B) | |
| # Fallback grounding uncertainty when there is no numeric score. | |
| _BAND_UNCERTAINTY = {"low": 1.0, "borderline": 0.6, "unknown": 0.5, "high": 0.0} | |
| def _clamp(x: float, lo: float = 0.0, hi: float = 1.0) -> float: | |
| return max(lo, min(hi, x)) | |
| def answer_uncertainty(grade: dict) -> float: | |
| """Uncertainty in [0, 1] for one graded answer. Higher = review first.""" | |
| gs = grade.get("grounding_score") | |
| if gs is not None: | |
| try: | |
| g_component = 1.0 - _clamp(float(gs)) | |
| except (TypeError, ValueError): | |
| g_component = _BAND_UNCERTAINTY.get(grade.get("grounding_band"), 0.5) | |
| else: | |
| g_component = _BAND_UNCERTAINTY.get(grade.get("grounding_band"), 0.5) | |
| u = W_GROUNDING * g_component | |
| if not grade.get("deterministic_passed", True): | |
| u += W_RULE | |
| if not grade.get("critic_passed", True): | |
| u += W_CRITIC | |
| n_flags = len([f for f in (grade.get("flags") or []) if f]) | |
| u += min(n_flags * W_FLAG, W_FLAG_CAP) | |
| return round(_clamp(u), 4) | |
| def uncertainty_components(grade: dict) -> dict: | |
| """Explainable breakdown for the audit view.""" | |
| gs = grade.get("grounding_score") | |
| if gs is not None: | |
| try: | |
| g_component = 1.0 - _clamp(float(gs)) | |
| except (TypeError, ValueError): | |
| g_component = _BAND_UNCERTAINTY.get(grade.get("grounding_band"), 0.5) | |
| else: | |
| g_component = _BAND_UNCERTAINTY.get(grade.get("grounding_band"), 0.5) | |
| n_flags = len([f for f in (grade.get("flags") or []) if f]) | |
| return { | |
| "grounding": round(W_GROUNDING * g_component, 4), | |
| "rule": W_RULE if not grade.get("deterministic_passed", True) else 0.0, | |
| "critic": W_CRITIC if not grade.get("critic_passed", True) else 0.0, | |
| "flags": round(min(n_flags * W_FLAG, W_FLAG_CAP), 4), | |
| } | |
| def aggregate_passes(passes: list[dict]) -> dict: | |
| """Drop-in superset of aggregate_scores, adding `uncertainty`. | |
| passes: list of grade_one_pass dicts (each has at least 'score', and | |
| ideally grounding_score / grounding_band / deterministic_passed / | |
| critic_passed / flags). | |
| """ | |
| scores = [float(p.get("score", 0.0)) for p in passes if "score" in p] | |
| if not scores: | |
| return { | |
| "median": 0.0, "max_score": 0.0, "min_score": 0.0, | |
| "std_dev": 0.0, "n_passes": 0, "uncertainty": 1.0, | |
| } | |
| std = float(statistics.pstdev(scores)) if len(scores) > 1 else 0.0 | |
| max_score = float(max(scores)) | |
| per_pass_u = [answer_uncertainty(p) for p in passes if "score" in p] | |
| mean_u = statistics.mean(per_pass_u) if per_pass_u else 1.0 | |
| # Lane B: disagreement across passes adds uncertainty (0 for a single pass). | |
| denom = max((float(p.get("max_score", 0.0)) for p in passes), default=0.0) | |
| norm_spread = (std / denom) if denom > 0 else 0.0 | |
| uncertainty = round(_clamp(mean_u + W_SPREAD * norm_spread), 4) | |
| return { | |
| "median": float(statistics.median(scores)), | |
| "max_score": max_score, | |
| "min_score": float(min(scores)), | |
| "std_dev": std, | |
| "n_passes": len(scores), | |
| "uncertainty": uncertainty, | |
| } | |
| # --------------------------------------------------------------------------- | |
| # Self-test: `python confidence.py` (pure stdlib, no relative imports). | |
| # --------------------------------------------------------------------------- | |
| if __name__ == "__main__": | |
| failures = 0 | |
| def check(label, cond): | |
| global failures | |
| print(f"[{'PASS' if cond else 'FAIL'}] {label}") | |
| if not cond: | |
| failures += 1 | |
| clean = {"score": 2, "max_score": 2, "grounding_score": 0.95, | |
| "deterministic_passed": True, "critic_passed": True, "flags": []} | |
| bad = {"score": 2, "max_score": 2, "grounding_score": 0.1, | |
| "deterministic_passed": False, "critic_passed": False, | |
| "flags": ["hallucination_risk", "ambiguous_evidence"]} | |
| mid = {"score": 1, "max_score": 2, "grounding_score": None, | |
| "grounding_band": "borderline", "deterministic_passed": True, | |
| "critic_passed": True, "flags": []} | |
| uc, ub, um = answer_uncertainty(clean), answer_uncertainty(bad), answer_uncertainty(mid) | |
| print(f" clean={uc} mid={um} bad={ub}") | |
| check("clean answer is low uncertainty", uc < 0.15) | |
| check("bad answer is high uncertainty", ub > 0.85) | |
| check("mid (borderline, unknown score) is in between", uc < um < ub) | |
| check("uncertainty stays in [0,1]", all(0.0 <= x <= 1.0 for x in (uc, ub, um))) | |
| # aggregate_passes: single pass -> uncertainty == answer_uncertainty | |
| agg1 = aggregate_passes([clean]) | |
| check("single-pass aggregate keeps std_dev 0", agg1["std_dev"] == 0.0) | |
| check("single-pass uncertainty matches answer_uncertainty", | |
| abs(agg1["uncertainty"] - uc) < 1e-6) | |
| check("aggregate exposes the legacy keys", | |
| {"median", "max_score", "min_score", "std_dev", "n_passes"}.issubset(agg1)) | |
| # Lane B: two passes that disagree add spread | |
| p1 = dict(clean, score=2) | |
| p2 = dict(clean, score=0) | |
| agg2 = aggregate_passes([p1, p2]) | |
| check("multi-pass disagreement raises uncertainty above the per-pass mean", | |
| agg2["uncertainty"] > answer_uncertainty(p1)) | |
| # empty | |
| check("empty passes -> uncertainty 1.0", aggregate_passes([])["uncertainty"] == 1.0) | |
| print() | |
| if failures: | |
| print(f"SELF-TEST FAILED: {failures}") | |
| raise SystemExit(1) | |
| print("SELF-TEST PASSED: all cases correct") | |