diff --git a/backend/app/engines/adversary.py b/backend/app/engines/adversary.py index 4d465b8..ff43b81 100644 --- a/backend/app/engines/adversary.py +++ b/backend/app/engines/adversary.py @@ -3,6 +3,7 @@ """ from __future__ import annotations from .. import llm, config +from . import grounding PERSONA_DESC = { "Hardball": "Direct, aggressive. Uses blunt pressure lines ('standard clause', 'non-negotiable'). Rarely volunteers concessions. Every position is stated as final until genuinely earned.", @@ -181,17 +182,38 @@ def respond(session: dict, user_msg: str) -> dict: history.append({"role": "user", "content": user_msg}) out = llm.chat_json(history, model=config.ADVERSARY_MODEL, temperature=0.75, max_tokens=800) + out = grounding.validate_adversary_output(out) # schema guard reply = out.get("reply", "") conceded = bool(out.get("conceded")) gate = None if conceded: - gate = _referee(sc, user_msg, out.get("trigger_claimed", ""), difficulty) - if not gate.get("earned"): + claimed = out.get("trigger_claimed", "") + triggers = sc["adversary"].get("concession_triggers", []) + + # trigger validation: claimed trigger must match a known trigger + if not grounding.trigger_is_valid(claimed, triggers): conceded = False reply = (reply.rstrip() + - "\n\n[That said — you haven't given me a compelling enough reason to move on this. My position stands.]") + "\n\n[That said — I need a stronger basis than that. My position stands.]") out["concession_detail"] = "" + gate = {"earned": False, "reason": "Claimed trigger did not match any valid concession condition.", + "_grounding": "trigger_validation_failed"} + else: + # self-consistency referee: 3 independent calls, majority vote + gate = grounding.majority_vote( + _referee, + args=(sc, user_msg, claimed, difficulty), + kwargs={}, + n=3, + key="earned", + ) + gate = grounding.validate_referee_output(gate) + if not gate.get("earned"): + conceded = False + reply = (reply.rstrip() + + "\n\n[That said — you haven't given me a compelling enough reason to move on this. My position stands.]") + out["concession_detail"] = "" session["turn"] += 1 session["transcript"].append({"role": "user", "content": user_msg}) diff --git a/backend/app/engines/grounding.py b/backend/app/engines/grounding.py new file mode 100644 index 0000000..9308959 --- /dev/null +++ b/backend/app/engines/grounding.py @@ -0,0 +1,206 @@ +"""Anti-hallucination grounding layer for Whetstone. + +Five mechanisms: + 1. Quote verification — evidence_quote must appear in the actual transcript + 2. Trigger validation — claimed trigger must match a known trigger (fuzzy) + 3. Self-consistency — referee called N times; majority vote wins + 4. Score sanity — numeric scores clamped to valid ranges + 5. Schema guard — required fields and types enforced on all JSON output +""" +from __future__ import annotations +import re +from difflib import SequenceMatcher +from typing import Any, Callable + + +# ── 1. Quote verification ───────────────────────────────────────────────────── + +def _normalise(s: str) -> str: + return re.sub(r"\s+", " ", s.lower().strip()) + + +def quote_in_transcript(quote: str, transcript: list[dict], threshold: float = 0.65) -> bool: + """Return True if quote appears verbatim or near-verbatim in any transcript turn.""" + if not quote or not quote.strip(): + return True # empty quote — nothing to verify + q = _normalise(quote) + for turn in transcript: + body = _normalise(turn.get("content", "")) + # exact substring + if q in body: + return True + # sliding-window fuzzy match (handles minor whitespace / ellipsis diffs) + words_q = q.split() + words_b = body.split() + window = len(words_q) + for i in range(max(1, len(words_b) - window + 1)): + chunk = " ".join(words_b[i : i + window]) + ratio = SequenceMatcher(None, q, chunk).ratio() + if ratio >= threshold: + return True + return False + + +def verify_scoring_quotes(result: dict, transcript: list[dict]) -> dict: + """ + Check every rubric_result evidence_quote against the actual transcript. + If a quote cannot be found, mark it as unverified and zero the score + to prevent evidence-hallucinated verdicts from standing. + """ + rubric_results = result.get("rubric_results", []) + any_fixed = False + for item in rubric_results: + quote = item.get("evidence_quote", "") + if quote and not quote_in_transcript(quote, transcript): + item["_quote_unverified"] = True + item["comment"] = ( + f"[GROUNDING: quote not found in transcript] {item.get('comment', '')}" + ) + # downgrade verdict — unverified evidence ≠ earned point + if item.get("verdict") in ("MET", "1", "2"): + item["verdict"] = "UNMET" + item["_auto_corrected"] = True + any_fixed = True + if any_fixed: + result["_grounding_corrections"] = True + return result + + +# ── 2. Trigger validation ───────────────────────────────────────────────────── + +def trigger_is_valid(claimed: str, valid_triggers: list[str], threshold: float = 0.45) -> bool: + """ + Return True if claimed trigger is semantically related to any valid trigger. + Uses token-overlap ratio — good enough for 1-2 sentence trigger descriptions. + """ + if not claimed: + return False + c = _normalise(claimed) + for t in valid_triggers: + t_norm = _normalise(t) + ratio = SequenceMatcher(None, c, t_norm).ratio() + if ratio >= threshold: + return True + # keyword overlap fallback + c_words = set(c.split()) - {"the", "a", "an", "of", "in", "and", "or", "to", "is"} + t_words = set(t_norm.split()) - {"the", "a", "an", "of", "in", "and", "or", "to", "is"} + if c_words and t_words: + overlap = len(c_words & t_words) / min(len(c_words), len(t_words)) + if overlap >= 0.40: + return True + return False + + +# ── 3. Self-consistency referee ─────────────────────────────────────────────── + +def majority_vote( + fn: Callable[..., dict], + args: tuple, + kwargs: dict, + n: int = 3, + key: str = "earned", +) -> dict: + """ + Call fn(*args, **kwargs) n times and return the majority result. + Ties default to the conservative answer (False for 'earned'). + """ + results: list[dict] = [] + for _ in range(n): + try: + results.append(fn(*args, **kwargs)) + except Exception: # noqa: BLE001 + results.append({key: False, "reason": "call failed"}) + + votes_true = sum(1 for r in results if r.get(key) is True) + votes_false = n - votes_true + winner = votes_true > votes_false # ties → False (conservative) + + # Return the result whose vote matched the majority + matching = [r for r in results if r.get(key) is winner] + best = matching[0] if matching else results[0] + best["_consistency_votes"] = {"true": votes_true, "false": votes_false, "n": n} + return best + + +# ── 4. Score sanity ─────────────────────────────────────────────────────────── + +_DIMENSION_NAMES = { + "Position Defense", "Concession Strategy", + "Anchoring & Offers", "Pressure Resistance", "Legal Reasoning", +} + + +def clamp_scores(result: dict) -> dict: + """Ensure dimension scores are non-negative integers within [0, max].""" + dim_scores = result.get("dimension_scores", {}) + for dim, data in dim_scores.items(): + if not isinstance(data, dict): + continue + raw_score = data.get("score", 0) + raw_max = data.get("max", 0) + try: + score = int(round(float(raw_score))) + max_ = int(round(float(raw_max))) + except (TypeError, ValueError): + score, max_ = 0, 0 + data["score"] = max(0, min(score, max(max_, 0))) + data["max"] = max(0, max_) + + # overall_score must be 0-100 + ov = result.get("overall_score") + try: + result["overall_score"] = max(0, min(100, int(round(float(ov))))) + except (TypeError, ValueError): + result["overall_score"] = 0 + + return result + + +# ── 5. Schema guard ─────────────────────────────────────────────────────────── + +def _ensure_str(d: dict, key: str, default: str = "") -> None: + if not isinstance(d.get(key), str): + d[key] = default + + +def _ensure_bool(d: dict, key: str, default: bool = False) -> None: + v = d.get(key) + if not isinstance(v, bool): + # accept "true"/"false" strings from LLM + if isinstance(v, str): + d[key] = v.lower() in ("true", "1", "yes") + else: + d[key] = default + + +def validate_adversary_output(out: dict) -> dict: + """Enforce schema on adversary JSON: required fields, correct types.""" + _ensure_str(out, "reply", "[No response generated]") + _ensure_bool(out, "conceded", False) + _ensure_str(out, "concession_detail", "") + _ensure_str(out, "trigger_claimed", "") + # reply must be non-empty + if not out["reply"].strip(): + out["reply"] = "[No response generated]" + return out + + +def validate_referee_output(out: dict) -> dict: + """Enforce schema on referee JSON.""" + _ensure_bool(out, "earned", False) + _ensure_str(out, "reason", "No reason provided.") + return out + + +def validate_score_output(out: dict) -> dict: + """Enforce schema on scoring JSON and clamp numeric fields.""" + if not isinstance(out.get("rubric_results"), list): + out["rubric_results"] = [] + if not isinstance(out.get("dimension_scores"), dict): + out["dimension_scores"] = {d: {"score": 0, "max": 0} for d in _DIMENSION_NAMES} + _ensure_str(out, "summary", "Scoring unavailable.") + _ensure_str(out, "attribution", "") + if not isinstance(out.get("turning_point"), dict): + out["turning_point"] = {"turn": 1, "what_happened": "", "stronger_play": ""} + out = clamp_scores(out) + return out diff --git a/backend/app/engines/scoring.py b/backend/app/engines/scoring.py index ee3180d..9d02d82 100644 --- a/backend/app/engines/scoring.py +++ b/backend/app/engines/scoring.py @@ -3,6 +3,7 @@ """ from __future__ import annotations from .. import llm, config +from . import grounding def _transcript_text(session: dict) -> str: @@ -93,6 +94,8 @@ def score(session: dict) -> dict: ] result = llm.chat_json(msg, model=config.JUDGE_MODEL, temperature=0.15, max_tokens=3000, retries=2) + result = grounding.validate_score_output(result) # schema guard + score clamp + result = grounding.verify_scoring_quotes(result, session["transcript"]) # quote grounding session["finished"] = True session["score"] = result return result