Grading: code first, judges second

What the case asks for                       Grader            Cost per case

one right string, from a fixed set           exact match       nothing
one right string, free-form wording          exact + normalise nothing
a number, a date, an id                      parse and compare nothing
valid structure: JSON, fields, types         schema check      nothing
an internal rule: totals, ranges, citations  constraint check  nothing
generated code                               run the tests     a sandbox
the right documents used                     set overlap       nothing
wording that could be phrased many ways      judge             a model call
faithful, on-topic, right tone               judge             a model call
which of two answers is better               pairwise judge    two model calls

import re, unicodedata

def normalise(s):
    s = unicodedata.normalize("NFKC", s).strip().lower()
    s = s.replace("’", "'")                   # the curly apostrophe
    s = re.sub(r"[.!?;:]+$", "", s)           # trailing punctuation only
    s = re.sub(r"\s+", " ", s)
    return s

CASES = [
    ("refund window, annual plan", "14 days", "14 days"),
    ("refund window, monthly plan", "30 days", "30 days."),
    ("can I pause my plan", "no", "No"),
    ("support hours", "9am to 5pm", "9am  to 5pm"),
    ("price of the pro plan", "$1,200", "1200"),
    ("who signs the contract", "the account owner", "the owner of the account"),
]

exact = sum(exp == got for _, exp, got in CASES)
norm  = sum(normalise(exp) == normalise(got) for _, exp, got in CASES)
print("raw exact match  ", exact, "/", len(CASES))
print("after normalising", norm, "/", len(CASES))
for name, exp, got in CASES:
    if normalise(exp) != normalise(got):
        print("  still fails:", name, "|", exp, "->", got)

# raw exact match   1 / 6
# after normalising 4 / 6
#   still fails: price of the pro plan | $1,200 -> 1200
#   still fails: who signs the contract | the account owner -> the owner of the account

from difflib import SequenceMatcher

def ratio(a, b):
    """Character-level similarity, in the standard library."""
    return SequenceMatcher(None, a.lower(), b.lower()).ratio()

def token_f1(a, b):
    """Word overlap, the metric behind most question-answering scores."""
    ta, tb = a.lower().split(), b.lower().split()
    common = sum(min(ta.count(w), tb.count(w)) for w in set(ta))
    if common == 0:
        return 0.0
    p, r = common / len(tb), common / len(ta)
    return 2 * p * r / (p + r)

PAIRS = [
    ("the account owner", "the owner of the account"),
    ("refunds take 5 working days", "refunds are processed in five working days"),
    ("the refund is available", "the refund is not available"),
    ("cancel any time", "you can cancel whenever you like"),
]
print(f"{'ratio':>7}{'tokenF1':>9}  pair")
for a, b in PAIRS:
    print(f"{ratio(a, b):7.2f}{token_f1(a, b):9.2f}  {a!r} / {b!r}")

#   ratio  tokenF1  pair
#    0.54     0.75  'the account owner' / 'the owner of the account'
#    0.70     0.50  'refunds take 5 working days' / 'refunds are processed in five working days'
#    0.92     0.89  'the refund is available' / 'the refund is not available'
#    0.51     0.22  'cancel any time' / 'you can cancel whenever you like'

import json

SCHEMA = {
    "order_id": str,
    "items": list,
    "total_cents": int,
    "refundable": bool,
}

def check_structure(raw):
    """Returns a list of reasons the output is unusable. Empty means it passed."""
    problems = []
    try:
        obj = json.loads(raw)
    except json.JSONDecodeError as e:
        return ["not valid JSON: " + str(e)]
    if not isinstance(obj, dict):
        return ["top level is " + type(obj).__name__ + ", expected object"]
    for field, want in SCHEMA.items():
        if field not in obj:
            problems.append("missing field " + field)
        elif not isinstance(obj[field], want) or isinstance(obj[field], bool) != (want is bool):
            problems.append(field + " is " + type(obj[field]).__name__ + ", expected " + want.__name__)
    for extra in set(obj) - set(SCHEMA):
        problems.append("unexpected field " + extra)
    if not problems:
        line_total = sum(i.get("cents", 0) for i in obj["items"])
        if line_total != obj["total_cents"]:
            problems.append(f"total_cents {obj['total_cents']} but line items sum to {line_total}")
        if obj["total_cents"] < 0:
            problems.append("total_cents is negative")
    return problems

OUTPUTS = [
    '{"order_id": "A-101", "items": [{"cents": 900}, {"cents": 2000}], "total_cents": 2900, "refundable": true}',
    'Sure! Here is the JSON:\n{"order_id": "A-101", "items": [], "total_cents": 0, "refundable": true}',
    '{"order_id": "A-102", "items": [{"cents": 4900}], "total_cents": 4900, "refundable": "yes"}',
    '{"order_id": "A-103", "items": [{"cents": 1000}, {"cents": 1000}], "total_cents": 1000, "refundable": false}',
]
for raw in OUTPUTS:
    print(check_structure(raw) or "pass")

# pass
# ['not valid JSON: Expecting value: line 1 column 1 (char 0)']
# ['refundable is str, expected bool']
# ['total_cents 1000 but line items sum to 2000']

TESTS = [
    ((["a", "b", "a"],), {"a": 2, "b": 1}),
    (([],), {}),
    ((["x"],), {"x": 1}),
    ((["a", "a", "a"],), {"a": 3}),
]

CANDIDATES = {
    "v1": "def tally(xs):\n    out = {}\n    for x in xs:\n        out[x] = out.get(x, 0) + 1\n    return out\n",
    "v2": "def tally(xs):\n    out = {}\n    for x in xs:\n        out[x] = 1\n    return out\n",
    "v3": "def tally(xs):\n    return {x: xs.count(x) for x in xs[1:]}\n",
    "v4": "def tally(xs)\n    return {}\n",
}

def run_tests(source, tests):
    ns = {}
    try:
        exec(source, ns)            # sandbox this properly before running model output
    except SyntaxError as e:
        return 0, len(tests), "SyntaxError: " + e.msg
    fn = ns.get("tally")
    if fn is None:
        return 0, len(tests), "no function named tally"
    passed, first_bad = 0, ""
    for args, want in tests:
        try:
            got = fn(*args)
        except Exception as e:
            got = type(e).__name__
        if got == want:
            passed += 1
        elif not first_bad:
            first_bad = f"tally({args[0]!r}) -> {got!r}, wanted {want!r}"
    return passed, len(tests), first_bad

for name, src in CANDIDATES.items():
    passed, total, why = run_tests(src, TESTS)
    print(f"{name}  {passed}/{total}  {why}")

# v1  4/4
# v2  2/4  tally(['a', 'b', 'a']) -> {'a': 1, 'b': 1}, wanted {'a': 2, 'b': 1}
# v3  3/4  tally(['x']) -> {}, wanted {'x': 1}
# v4  0/4  SyntaxError: expected ':'

Weak rubric:   "Is this a good summary?"

Usable rubric:
  Check one thing: does the summary claim anything the source
  does not support? Ignore style and length.

  FAIL if any claim is missing from the source or contradicts it.
  PASS if every claim can be found in the source.

  One sentence of critique, then PASS or FAIL on its own line.

import numpy as np
rng = np.random.default_rng(7)

N = 400
human_fail = rng.random(N) < 0.12       # your own labels: 12% of outputs are bad

def stub_judge(human_fail, catch, false_alarm, rng):
    """Stands in for a hosted judge so this runs offline. catch is how often it
    flags a real failure, false_alarm how often it flags a good output."""
    r = rng.random(len(human_fail))
    return np.where(human_fail, r < catch, r < false_alarm)

naive = stub_judge(human_fail, catch=0.45, false_alarm=0.03, rng=rng)

human = ["FAIL" if x else "PASS" for x in human_fail]
judge = ["FAIL" if x else "PASS" for x in naive]
print(human.count("FAIL"), "failures in", len(human), "labelled outputs")

# 44 failures in 400 labelled outputs

fails  = [j for h, j in zip(human, judge) if h == "FAIL"]
passes = [j for h, j in zip(human, judge) if h == "PASS"]
print(f"caught {fails.count('FAIL')}/{len(fails)} failures, "
      f"accepted {passes.count('PASS')}/{len(passes)} passes")

# caught 20/44 failures, accepted 350/356 passes

def agreement(a, b):
    return float((a == b).mean())

def kappa(a, b):
    """Agreement above what two graders would reach by chance."""
    po = agreement(a, b)
    pa1, pb1 = a.mean(), b.mean()
    pe = pa1 * pb1 + (1 - pa1) * (1 - pb1)
    return float("nan") if pe == 1 else float((po - pe) / (1 - pe))

print("raw agreement", round(agreement(human_fail, naive), 3))
print("kappa        ", round(kappa(human_fail, naive), 3))

lazy = np.zeros(N, dtype=bool)          # a judge that passes everything
print("passes-everything judge: agreement", round(agreement(human_fail, lazy), 3),
      " kappa", round(kappa(human_fail, lazy), 3))

from sklearn.metrics import cohen_kappa_score
print("sklearn check", round(cohen_kappa_score(human_fail, naive), 3))

print()
print("base rate   agreement of a judge that passes everything")
for base in (0.02, 0.05, 0.12, 0.30):
    print(f"{base:>8.0%}   {1 - base:>5.0%}")

# raw agreement 0.925
# kappa         0.533
# passes-everything judge: agreement 0.89  kappa 0.0
# sklearn check 0.533
#
# base rate   agreement of a judge that passes everything
#       2%     98%
#       5%     95%
#      12%     88%
#      30%     70%

rubric = stub_judge(human_fail, catch=0.85, false_alarm=0.04, rng=rng)

print("naive judge : agreement", round(agreement(human_fail, naive), 3),
      " kappa", round(kappa(human_fail, naive), 3))
print("with rubric : agreement", round(agreement(human_fail, rubric), 3),
      " kappa", round(kappa(human_fail, rubric), 3))
fails_n = int(human_fail.sum())
print(f"caught {int((rubric & human_fail).sum())}/{fails_n} real failures, "
      f"false alarms on {int((rubric & ~human_fail).sum())}/{int((~human_fail).sum())} good outputs")

# naive judge : agreement 0.925  kappa 0.533
# with rubric : agreement 0.948  kappa 0.763
# caught 40/44 real failures, false alarms on 17/356 good outputs

def kappa_interval(human, judge, n_labels, boot=2000, seed=1):
    r = np.random.default_rng(seed)
    ks = []
    for _ in range(boot):
        idx = r.integers(0, n_labels, n_labels)
        ks.append(kappa(human[:n_labels][idx], judge[:n_labels][idx]))
    ks = np.array(ks)                   # nan where a resample drew no failures
    lo, hi = np.nanpercentile(ks, [2.5, 97.5])
    return lo, hi

for n in (50, 100, 400):
    lo, hi = kappa_interval(human_fail, rubric, n)
    print(f"{n:>4} labels ({int(human_fail[:n].sum()):>2} of them failures): "
          f"kappa {kappa(human_fail[:n], rubric[:n]):.2f}, 95% interval {lo:.2f} to {hi:.2f}")

#   50 labels ( 6 of them failures): kappa 0.81, 95% interval 0.39 to 1.00
#  100 labels (11 of them failures): kappa 0.67, 95% interval 0.43 to 0.86
#  400 labels (44 of them failures): kappa 0.76, 95% interval 0.66 to 0.85

import numpy as np
rng = np.random.default_rng(11)

def pairwise_stub(n, p_b_on_merit, position_pull, rng):
    """Each pair judged in both orders. Returns whether B won when it was
    shown second, and whether it won when it was shown first."""
    b_second = rng.random(n) < (p_b_on_merit - position_pull)
    b_first  = rng.random(n) < (p_b_on_merit + position_pull)
    return b_second, b_first

N = 600
b_second, b_first = pairwise_stub(N, p_b_on_merit=0.58, position_pull=0.15, rng=rng)

print("B wins when shown second:", f"{b_second.mean():.1%}")
print("B wins when shown first :", f"{b_first.mean():.1%}")
print("position gap            :", f"{b_first.mean() - b_second.mean():.1%}")
print("swap and average        :", f"{(b_first.mean() + b_second.mean()) / 2:.1%}")

both = (b_first & b_second).mean()
neither = (~b_first & ~b_second).mean()
print(f"consistent verdicts: B twice {both:.1%}, A twice {neither:.1%}, "
      f"flipped {1 - both - neither:.1%}")

# B wins when shown second: 44.5%
# B wins when shown first : 75.5%
# position gap            : 31.0%
# swap and average        : 60.0%
# consistent verdicts: B twice 32.8%, A twice 12.8%, flipped 54.3%

import numpy as np

PADDING = (" It is worth noting that policies of this kind are designed with "
           "the customer in mind, and we are always happy to help further.")

def scoring_stub(answers, quality, length_weight, rng):
    """Pass or fail on one answer at a time. length_weight is how much the
    judge is swayed by how long the answer is."""
    words = np.array([len(a.split()) for a in answers])
    pull = length_weight * (words - 25) / 25
    return (quality + pull + rng.normal(0, 0.15, len(answers))) > 0.5

base = ["The annual plan can be refunded within 14 days of purchase, "
        "and the refund reaches your card in about 5 working days."] * 300
quality = np.random.default_rng(3).normal(0.55, 0.2, 300)   # same answers, same quality

for weight, label in ((0.25, "judge as written"), (0.02, "after 'ignore length'")):
    plain  = scoring_stub(base, quality, weight, np.random.default_rng(5))
    padded = scoring_stub([a + PADDING for a in base], quality, weight,
                          np.random.default_rng(5))
    print(f"{label:<24} plain {plain.mean():>5.0%}   padded {padded.mean():>5.0%}   "
          f"gap {padded.mean() - plain.mean():>+5.0%}")
print("words:", len(base[0].split()), "plain,", len((base[0] + PADDING).split()), "padded")

# judge as written         plain   52%   padded   87%   gap  +35%
# after 'ignore length'    plain   56%   padded   61%   gap   +4%
# words: 22 plain, 46 padded