r"""Score the Level-4 AI author against the reviewer's corrections dataset — the objective gate the author must pass before it can be trusted or taken live. For a race it runs the author over the current feed and lines its suggestions up against the reviewer's graded beats (`reviews/_.corrections.json`, produced by review_commentary.py --ingest), reporting three things per the plan: * COVERAGE — of the beats the reviewer marked wrong/dull, how many the author actually attempted. Split honestly: EDITABLE beats (colour/play — the author's job) vs FACTUAL beats it deliberately never rewrites (a penalty, a lead change). A factual beat the reviewer disliked is NOT an author miss — it needs engine work / a spawned Analyst beat (see the penalty_clear pattern). * SAFETY — the author only emits validator-passing lines; this re-checks and reports any that fail (should be zero) so a regression in the safety gate can't hide. * QUALITY vs GOLD — did the author's line move toward the reviewer's "better" line? A deterministic fact-preservation check (does it keep the boats/teams the gold names — the split-bug class of error) always runs; an optional --judge asks the model whether the author beats the engine line against the gold (the real quality signal; needs a key, off by default). ..\..\..\.venv-demo\Scripts\python.exe export\score_author.py --race gsys06 --lang en ..\..\..\.venv-demo\Scripts\python.exe export\score_author.py --race gsys06 --lang en --judge ..\..\..\.venv-demo\Scripts\python.exe export\score_author.py # every corrections file Deterministic core runs with no key (author = no-op → coverage 0, a truthful baseline). Changes no data. """ from __future__ import annotations import argparse import json import sys from pathlib import Path _HERE = Path(__file__).resolve().parent ROOT = _HERE.parent.parent # v3/broadcast sys.path.insert(0, str(ROOT.parent / "scoring")) REVIEWS = ROOT / "demo" / "data" / "reviews" # reuse the review tool's env + feed + author plumbing verbatim (one source of truth) from ai_author_schema import EDITABLE_TYPES, build_factsheet, validate_line # noqa: E402 from review_commentary import _load_env, author_suggestions, load_feed # noqa: E402 def _entities(feed: dict) -> list[str]: """Team names + boat labels in the race — the names a 'better' line may hinge on (the split bug was naming the wrong one). Longest first so 'Blue Socks 6' is checked before 'Blue Socks'.""" ents = set() for t in feed.get("teams", []): if t.get("name"): ents.add(t["name"]) for b in t.get("boats", []): if b.get("visualId"): ents.add(b["visualId"]) return sorted(ents, key=len, reverse=True) MATCH_WINDOW_S = 12 # a graded beat may drift a few seconds when the feed is re-enriched def _match(corr: dict, events: list[dict]) -> dict | None: """The current beat a graded correction refers to. Re-enrichment (new detectors, cadence shifts) moves a beat's tSec, so match by NEAREST beat of the SAME TYPE within a window — never exact tSec (misses a beat that slid a couple of seconds), and never a different type (a `play` correction must not silently match a neighbouring `margin` call). No same-type beat nearby ⇒ the graded beat is gone from the current feed — a real signal, reported, not papered over.""" t, typ = corr["t_sec"], corr.get("type") same = [e for e in events if e["type"] == typ and abs(e["tSec"] - t) <= MATCH_WINDOW_S] return min(same, key=lambda e: abs(e["tSec"] - t)) if same else None def evaluate(corrections: list[dict], author_lines: dict, events: list[dict], entities: list[str]) -> dict: """Pure: line the author's suggestions up against the graded beats. `author_lines` is {tSec: text} keyed by the CURRENT feed; `events` is the current commentary (for the live engine line + drift- tolerant matching). A beat's editability is the reviewer's graded TYPE (what it was).""" beats = [] for c in corrections: t = c["t_sec"] gold = (c.get("better") or "").strip() editable = c.get("type") in EDITABLE_TYPES m = _match(c, events) present = m is not None mt = m["tSec"] if m else None engine = m["textEn"] if m else c.get("said", "") author = author_lines.get(mt) if mt is not None else None gold_ents = [e for e in entities if gold and e in gold] # don't let a longer label satisfied by a shorter one give a false pass: check each named entity kept = None if author and gold_ents: kept = all(e in author for e in gold_ents) beats.append({ "t_sec": t, "matchedT": mt, "present": present, "type": c.get("type", ""), "verdict": c.get("verdict", ""), "tag": c.get("tag", ""), "editable": editable, "attempted": author is not None, "engine": engine, "author": author, "gold": gold, "goldEntities": gold_ents, "entitiesKept": kept, }) # coverage is only meaningful over EDITABLE beats that STILL EXIST; a graded beat that re-enrichment # removed is neither a hit nor a miss — it's a separate "gone" bucket the report calls out. editable_present = [b for b in beats if b["editable"] and b["present"]] attempted = [b for b in editable_present if b["attempted"]] checked = [b for b in beats if b["entitiesKept"] is not None] return { "total": len(corrections), "editablePresent": len(editable_present), "attempted": len(attempted), "coveragePct": (round(100 * len(attempted) / len(editable_present)) if editable_present else None), "factualPresent": [{"t_sec": b["t_sec"], "type": b["type"]} for b in beats if b["present"] and not b["editable"]], "gone": [{"t_sec": b["t_sec"], "type": b["type"], "verdict": b["verdict"]} for b in beats if not b["present"]], "factPreserved": sum(1 for b in checked if b["entitiesKept"]), "factChecked": len(checked), "beats": beats, } def _safety(scored: dict, author_lines: dict, lang: str) -> tuple[int, list[int]]: """Independently re-validate each attempted line against its beat fact sheet — the author should only have emitted safe lines, so `failed` should be empty. Returns (n_checked, failed_tSecs).""" by_t = {b["tSec"]: b for b in (scored.get("aiBeats") or [])} failed = [] for t, line in author_lines.items(): beat = by_t.get(t) if beat and not validate_line(line, build_factsheet(scored, beat), lang)[0]: failed.append(t) return len(author_lines), failed # ---------- optional LLM judge: does the author beat the engine, against the gold? ---------- _JUDGE_TOOL = { "type": "object", "additionalProperties": False, "properties": {"verdicts": {"type": "array", "items": { "type": "object", "additionalProperties": False, "properties": {"tSec": {"type": "integer"}, "winner": {"type": "string", "enum": ["author", "engine", "tie"]}, "why": {"type": "string"}}, "required": ["tSec", "winner", "why"]}}}, "required": ["verdicts"], } def judge(beats: list[dict], lang: str) -> dict: """Ask the model, per attempted beat, whether the AUTHOR or the ENGINE line is closer to the reviewer's gold in intent + quality. Returns {tSec: {winner, why}}; {} with no key/SDK.""" import os items = [b for b in beats if b["attempted"] and b["gold"]] if not items or not os.environ.get("ANTHROPIC_API_KEY"): return {} try: import anthropic except ImportError: return {} payload = [{"tSec": b["t_sec"], "engine": b["engine"], "author": b["author"], "gold": b["gold"], "reviewerVerdict": b["verdict"]} for b in items] system = ( "You are a team-racing broadcast editor. For each beat you get the ENGINE's line, the AI " "AUTHOR's line, and a GOLD line written by an expert sailing reviewer (the target). Decide " "which of engine/author is closer to the gold in MEANING and quality — accuracy of the tactic, " "specificity, and how a real commentator would say it. 'tie' only if genuinely level. Judge " "against the gold, not your own taste.") client = anthropic.Anthropic() msg = client.messages.create( model="claude-opus-4-8", max_tokens=4000, system=system, messages=[{"role": "user", "content": json.dumps(payload, ensure_ascii=False)}], tools=[{"name": "judge", "description": "Return a winner per beat.", "strict": True, "input_schema": _JUDGE_TOOL}], tool_choice={"type": "tool", "name": "judge"}) for block in msg.content: if getattr(block, "type", None) == "tool_use" and block.name == "judge": return {v["tSec"]: {"winner": v["winner"], "why": v["why"]} for v in block.input.get("verdicts", [])} return {} def _print_report(race: str, lang: str, rep: dict, safety, verdicts: dict) -> None: n_checked, failed = safety print(f"\nAUTHOR SCORE — {race} ({lang})") print(f" corrections in set: {rep['total']} " f"(editable present {rep['editablePresent']}, factual {len(rep['factualPresent'])}, " f"gone {len(rep['gone'])})") cov = rep["coveragePct"] print(f" coverage: {rep['attempted']}/{rep['editablePresent']} present editable beats attempted" f"{f' = {cov}%' if cov is not None else ''}") if rep["factualPresent"]: fl = ", ".join(f"t={b['t_sec']}({b['type']})" for b in rep["factualPresent"]) print(f" factual beats the author never rewrites (need engine / a spawned beat): {fl}") if rep["gone"]: gl = ", ".join(f"t={b['t_sec']}({b['type']},{b['verdict']})" for b in rep["gone"]) print(f" graded beats no longer in the feed (moved/merged out by later changes): {gl}") print(f" safety: {n_checked - len(failed)}/{n_checked} attempted lines re-validated clean" + (f" ** FAILED: {failed}" if failed else "")) if rep["factChecked"]: print(f" fact-keep: {rep['factPreserved']}/{rep['factChecked']} kept every boat/team the gold names") if verdicts: w = sum(1 for v in verdicts.values() if v["winner"] == "author") e = sum(1 for v in verdicts.values() if v["winner"] == "engine") ti = sum(1 for v in verdicts.values() if v["winner"] == "tie") print(f" quality vs gold (judge): author {w} / engine {e} / tie {ti} of {len(verdicts)} judged") print(" per beat:") for b in rep["beats"]: if not b["present"]: a = "-- beat no longer in the feed" elif b["attempted"]: a = b["author"] else: a = "-- not attempted (factual call)" if not b["editable"] else "-- not attempted" vk = "" if b["entitiesKept"] is None else (" names OK" if b["entitiesKept"] else " names MISS") jv = verdicts.get(b["matchedT"]) or verdicts.get(b["t_sec"]) jt = f" [judge: {jv['winner']}]" if jv else "" drift = f"->{b['matchedT']}" if (b["matchedT"] is not None and b["matchedT"] != b["t_sec"]) else "" print(f" t={b['t_sec']}{drift:<6} {b['verdict']:<5} {b['type']:<8}{vk}{jt}") print(f" engine: {b['engine']}") print(f" author: {a}") if b["gold"]: print(f" gold : {b['gold']}") def score_one(race: str, lang: str, do_judge: bool) -> dict | None: corr_path = REVIEWS / f"{race}_{lang}.corrections.json" if not corr_path.exists(): return None corrections = json.loads(corr_path.read_text(encoding="utf-8")).get("corrections", []) feed = load_feed(None if race == "real" else race) author_lines, scored = author_suggestions(feed, lang) events = [{"tSec": e["tSec"], "type": e.get("type", ""), "textEn": (e.get("i18n") or {}).get(lang) or (e.get("text") if lang == "en" else "")} for e in scored["events"]] rep = evaluate(corrections, author_lines, events, _entities(feed)) safety = _safety(scored, author_lines, lang) verdicts = judge(rep["beats"], lang) if do_judge else {} _print_report(race, lang, rep, safety, verdicts) rep["race"], rep["lang"] = race, lang rep["safetyFailed"] = safety[1] rep["judge"] = verdicts return rep def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--race", help="race id (e.g. gsys06) or 'real'; default: every corrections file present") ap.add_argument("--lang", default="en", help="language (default en)") ap.add_argument("--judge", action="store_true", help="add the LLM quality verdict (needs ANTHROPIC_API_KEY)") ap.add_argument("--out", help="write the full report(s) as JSON here") args = ap.parse_args() _load_env() if args.race: races = [args.race] else: races = sorted(p.name[: -len(f"_{args.lang}.corrections.json")] for p in REVIEWS.glob(f"*_{args.lang}.corrections.json")) if not races: raise SystemExit(f"no corrections files for lang '{args.lang}' in {REVIEWS} " f"(grade a CSV with review_commentary.py --ingest first)") reports = [r for race in races if (r := score_one(race, args.lang, args.judge))] if len(reports) > 1: cov = [r["attempted"] for r in reports] ed = [r["editablePresent"] for r in reports] gone = sum(len(r["gone"]) for r in reports) print(f"\nALL RACES: {sum(cov)}/{sum(ed)} present editable beats attempted across " f"{len(reports)} races ({gone} graded beats no longer in feed)") if args.out and reports: Path(args.out).write_text(json.dumps(reports, ensure_ascii=False, indent=1), encoding="utf-8") print(f"wrote {args.out}") return 0 if __name__ == "__main__": raise SystemExit(main())