#!/usr/bin/env python3 """ Did the shorter answers survive? Anything can make output shorter. The only interesting question is whether the answer is still complete. This grades every arm A (no system prompt) against the matching arm C (org block with the LENGTH line) response for the same prompt and repetition. Blinding, stated honestly: the judge is never told which response came from which condition, and the order is swapped on odd-numbered pairs so position bias does not line up with condition. It is NOT possible to hide that one answer is shorter than the other. That is inherent to the thing being measured, and the rubric is written to judge completeness rather than length so the visible difference is not itself the answer. Verdicts: complete the shorter answer covers everything material in the longer one minor_loss loses detail a reader could live without, still answers the question material_loss loses something a reader actually needed, or fails to answer Usage: python3 grade_quality.py [--in results.csv] [--out grades.csv] """ import csv import json import os import subprocess import time import sys MODEL = "claude-sonnet-5" API = "https://api.anthropic.com/v1/messages" KEY_FILE = os.path.expanduser("~/Documents/GitHub/anthropic_credentials.json") RUBRIC = """You are grading two answers to the same question for COMPLETENESS, not for style and not for length. QUESTION: {prompt} RESPONSE 1: {r1} RESPONSE 2: {r2} Compare them. Identify which response is shorter. Then judge ONLY this: does the shorter response omit anything material that the longer response contains, where "material" means a fact, caveat, step, or number that a competent reader would actually need to act on the answer? Do not reward length. A shorter answer that covers everything material is COMPLETE. Padding, restatement of the question, throat-clearing, and offers of further help are not material. Reply with JSON only, no other text: {{"shorter": 1 or 2, "verdict": "complete" | "minor_loss" | "material_loss", "missing": "one short sentence naming what is missing, or empty string if nothing"}}""" def load_key(): with open(KEY_FILE) as f: return json.load(f)["api_key"] def _first_json_object(text): """Pull the first balanced {...} out of arbitrary text. The judge sometimes writes a sentence before the JSON despite being told not to.""" start = text.find("{") if start < 0: raise ValueError(f"no JSON object in: {text[:120]!r}") depth, in_str, esc = 0, False, False for i in range(start, len(text)): ch = text[i] if in_str: if esc: esc = False elif ch == "\\": esc = True elif ch == '"': in_str = False continue if ch == '"': in_str = True elif ch == "{": depth += 1 elif ch == "}": depth -= 1 if depth == 0: return json.loads(text[start:i + 1]) raise ValueError(f"unbalanced JSON in: {text[:120]!r}") def judge(api_key, prompt, r1, r2, attempts=4): """Ask for JSON and dig it out of whatever comes back. Assistant prefill would be the clean way to force this, but claude-sonnet-5 rejects it ("This model does not support assistant message prefill"), so the budget is set generously instead and _first_json_object does the work. """ payload = { "model": MODEL, "max_tokens": 900, "messages": [ {"role": "user", "content": RUBRIC.format(prompt=prompt, r1=r1, r2=r2)}, ], } cmd = ["curl", "-sS", "-X", "POST", API, "-H", "content-type: application/json", "-H", "anthropic-version: 2023-06-01", "-H", f"x-api-key: {api_key}", "--data-binary", "@-"] last = None for n in range(attempts): proc = subprocess.run(cmd, input=json.dumps(payload), capture_output=True, text=True) try: data = json.loads(proc.stdout) except json.JSONDecodeError: last = RuntimeError(f"non-JSON HTTP body: {proc.stdout[:200]!r}") time.sleep(2 ** n) continue if "error" in data: last = RuntimeError(data["error"]) time.sleep(2 ** n) # overloaded_error is transient, back off continue text = "".join(b.get("text", "") for b in data.get("content", [])) try: return _first_json_object(text) except Exception as e: last = e time.sleep(1) raise last def main(): in_path, out_path = "token-measure-results.csv", "quality-grades.csv" left, right = "A", "C" args = sys.argv[1:] if "--in" in args: in_path = args[args.index("--in") + 1] if "--out" in args: out_path = args[args.index("--out") + 1] if "--arms" in args: # e.g. --arms BC isolates the LENGTH line left, right = args[args.index("--arms") + 1][0], args[args.index("--arms") + 1][1] rows = list(csv.DictReader(open(in_path))) by_key = {(r["arm"], r["rep"], r["prompt_id"]): r for r in rows} pairs = [] for (arm, rep, pid), r in by_key.items(): if arm != left: continue c = by_key.get((right, rep, pid)) if c: pairs.append((r, c)) pairs.sort(key=lambda p: (int(p[0]["rep"]), int(p[0]["prompt_id"]))) api_key = load_key() print(f"grading {len(pairs)} {left}/{right} pairs, blind, order swapped on odd pairs\n") fields = ["rep", "prompt_id", "category", "prompt", "left_arm", "right_arm", "a_tokens", "c_tokens", "swapped", "judge_said_shorter", "shorter_was", "verdict", "missing"] counts = {} with open(out_path, "w", newline="") as f: w = csv.DictWriter(f, fieldnames=fields) w.writeheader() for i, (a, c) in enumerate(pairs): swapped = (i % 2 == 1) r1, r2 = (c["response"], a["response"]) if swapped else (a["response"], c["response"]) try: v = judge(api_key, a["prompt"], r1, r2) except Exception as e: print(f" pair {i} FAILED: {e}") continue # which position held arm C c_pos = 1 if swapped else 2 shorter_was = right if v.get("shorter") == c_pos else left w.writerow({ "rep": a["rep"], "prompt_id": a["prompt_id"], "category": a["category"], "prompt": a["prompt"], "left_arm": left, "right_arm": right, "a_tokens": a["output_tokens"], "c_tokens": c["output_tokens"], "swapped": swapped, "judge_said_shorter": v.get("shorter"), "shorter_was": shorter_was, "verdict": v.get("verdict", ""), "missing": v.get("missing", ""), }) f.flush() counts[v.get("verdict")] = counts.get(v.get("verdict"), 0) + 1 print(f" [{i+1}/{len(pairs)}] p{a['prompt_id']:<2} rep{a['rep']} " f"A={a['output_tokens']:<5} C={c['output_tokens']:<5} -> {v.get('verdict')}" + (f" ({v.get('missing')[:60]})" if v.get("missing") else "")) print("\nverdict counts:", counts) print("->", out_path) if __name__ == "__main__": main()