| """CPU tests for the JevBench converter, scorer and training guard (deg.bench.jevbench).""" |
|
|
| from __future__ import annotations |
|
|
| import json |
| import math |
| from pathlib import Path |
|
|
| import pytest |
|
|
| from deg.api import parse_systemone, systemone_response |
| from deg.bench import jevbench as jb |
| from deg.schema import compile_schema |
|
|
| NOUL = {"id": "original-policy-01-0", "family": "policy", "group": "original-policy-01", |
| "split": "public", "expected": "no", "labels": ["no", "yes"], |
| "question": {"type": "noul", "instructions": "Is the refund permitted?", |
| "criteria": {"false": "A condition is missing.", |
| "true": "Every condition holds."}}, |
| "provenance": {"license": "MIT"}, |
| "state": "Policy: refunds require a receipt. The customer has no receipt."} |
| NOUL2 = {**NOUL, "id": "original-policy-01-1", |
| "state": "A receipt is required for any refund; this buyer lost theirs. Refund?"} |
| CHOICE = {"id": "easy-intent-00", "family": "intent", "group": None, "split": "public", |
| "expected": "track_order", "labels": ["track_order", "cancel_order"], |
| "question": {"type": "choice", "instructions": "Which intent?", |
| "criteria": {"cancel_order": "Wants to cancel", |
| "track_order": "Wants to know where an order is"}}, |
| "provenance": {}, "state": "Where is my package? I ordered it last week and it " |
| "still has not arrived at my house."} |
| SCORE = {"id": "original-ordinal-01-0", "family": "ordinal", "group": None, "split": "public", |
| "expected": 2, "labels": ["0", "1", "2"], |
| "question": {"type": "score", "instructions": "How urgent?", |
| "criteria": ["not urgent", "somewhat", "very urgent"]}, |
| "provenance": {}, "state": "The server room is on fire right now."} |
| PROB = {"id": "hard-opus-b-probability-02", "family": "probability", "group": "g-p", |
| "split": "public", "expected": "late", "labels": ["late", "on_time", "lost"], |
| "question": {"type": "choice", "instructions": "Likely outcome?", |
| "criteria": {"late": "arrives late", "lost": "never arrives", |
| "on_time": "arrives on time"}}, |
| "provenance": {"gold_probs": {"late": 0.6, "on_time": 0.3, "lost": 0.1}}, |
| "state": {"log": "Of 40 comparable shipments 24 were late, 12 on time and 4 lost."}} |
|
|
|
|
| @pytest.fixture() |
| def repo(tmp_path: Path) -> Path: |
| pub = tmp_path / "datasets" / "public" |
| pub.mkdir(parents=True) |
| files = {"easy.jsonl": [CHOICE], "original.jsonl": [NOUL, NOUL2, SCORE], |
| "hard.jsonl": [PROB]} |
| for name, rows in files.items(): |
| (pub / name).write_text("".join(json.dumps(r) + "\n" for r in rows)) |
| res = tmp_path / "results" / "v1.2" |
| res.mkdir(parents=True) |
| (res / "jevbench-v1.2-per-task.json").write_text(json.dumps({"systems": {"x": { |
| "public_tasks": {"easy-intent-00": ["c", 0.5], "original-policy-01-0": ["w", 0.5], |
| "original-policy-01-1": ["c", 0.5], |
| "original-ordinal-01-0": ["c", 0.5], |
| "hard-opus-b-probability-02": ["f", 0.5]}}}})) |
| (res / "jevbench-v1.2-results.json").write_text(json.dumps({"systems": [ |
| {"key": "x", "display": "X", "jevbench_score": 70.0, "axes": {}, "tiers": {}, |
| "cost": {"usd_per_1000": 0.01}}]})) |
| return tmp_path |
|
|
|
|
| def test_load_tasks_tiers_and_validation(repo: Path) -> None: |
| tasks = jb.load_tasks(repo) |
| assert [t["tier"] for t in tasks] == ["easy", "standard", "standard", "standard", "hard"] |
| bad = {**CHOICE, "expected": "refund"} |
| with pytest.raises(ValueError): |
| jb.validate_task(bad) |
| with pytest.raises(ValueError): |
| jb.validate_task({**NOUL, "state": {"expected": "no"}}) |
|
|
|
|
| def test_payload_is_the_typesafe_adapter_request() -> None: |
| p = jb.systemone_payload(CHOICE, "jev-latest") |
| assert p == {"model": "jev-latest", "state": CHOICE["state"], |
| "questions": {"decision": {"type": "choice", "instructions": "Which intent?", |
| "criteria": CHOICE["question"]["criteria"]}}} |
| |
| assert list(p["questions"]["decision"]["criteria"]) == ["cancel_order", "track_order"] |
|
|
|
|
| @pytest.mark.parametrize("task,gold", [(NOUL, False), (CHOICE, "track_order"), (SCORE, 2), |
| (PROB, "late")]) |
| def test_record_is_deg_systemone_schema(task: dict, gold: object) -> None: |
| rec = jb.record({**task, "tier": "x"}) |
| plan = compile_schema(rec["schema"]) |
| (fld,) = plan.fields |
| assert fld.pointer == "/decision" and fld.mode.startswith("decide.") |
| assert rec["value"] == {"decision": gold} |
| assert rec["schema"] == parse_systemone(jb.systemone_payload(task)).schema |
|
|
|
|
| def _fake_fill(task: dict, probs: list[float]) -> tuple: |
| job = parse_systemone(jb.systemone_payload(task)) |
| kind = task["question"]["type"] |
| meta = ({"mode": "decide.noul", "probability": probs[0]} if kind == "noul" else |
| {"mode": f"decide.{kind}", "probabilities": probs, "confidence": max(probs)}) |
| value = (probs[0] > 0.5 if kind == "noul" else |
| (list(task["question"]["criteria"])[probs.index(max(probs))] if kind == "choice" |
| else probs.index(max(probs)))) |
| resp = {"value": {"decision": value}, "meta": {"/decision": meta}, |
| "usage": {"prompt_tokens": 50, "generated_tokens": 0}} |
| return job, systemone_response(job, resp) |
|
|
|
|
| def test_probs_from_deg_systemone_answers() -> None: |
| _, body = _fake_fill(NOUL, [0.2]) |
| assert jb.probs_from_answer(NOUL, body["answers"]["decision"]) == pytest.approx( |
| {"yes": 0.2, "no": 0.8}) |
| _, body = _fake_fill(CHOICE, [0.1, 0.9]) |
| assert jb.probs_from_answer(CHOICE, body["answers"]["decision"]) == { |
| "cancel_order": 0.1, "track_order": 0.9} |
| _, body = _fake_fill(SCORE, [0.1, 0.2, 0.7]) |
| assert jb.probs_from_answer(SCORE, body["answers"]["decision"]) == { |
| "0": 0.1, "1": 0.2, "2": 0.7} |
| with pytest.raises(ValueError): |
| jb.probs_from_answer(NOUL, {"noul": True}) |
|
|
|
|
| def test_score_item_rules() -> None: |
| t = {**CHOICE, "tier": "easy"} |
| s = jb.score_item(t, {"track_order": 0.7, "cancel_order": 0.3}) |
| assert s["valid"] and s["strict_valid"] and s["correct"] and s["confidence"] == 0.7 |
| s = jb.score_item(t, {"track_order": 0.705, "cancel_order": 0.3}) |
| assert s["valid"] and not s["strict_valid"] and s["correct"] |
| assert sum(s["probs"].values()) == pytest.approx(1.0) |
| s = jb.score_item(t, {"track_order": 0.9, "cancel_order": 0.3}) |
| assert not s["valid"] and s["correct"] is False |
| s = jb.score_item(t, {"track_order": 1.0}) |
| assert not s["valid"] |
| s = jb.score_item(t, {"track_order": 0.5, "cancel_order": 0.5}) |
| assert s["predicted"] == "cancel_order" and s["correct"] is False |
| s = jb.score_item({**SCORE, "tier": "standard"}, {"0": 0.4, "1": 0.2, "2": 0.4}) |
| assert s["predicted"] == "0" and s["ordinal_ev"] == pytest.approx(1.0) |
| s = jb.score_item({**PROB, "tier": "hard"}, {"late": 0.5, "on_time": 0.4, "lost": 0.1}) |
| assert s["tvd"] == pytest.approx(0.1) |
| assert jb.score_item(t, None, "boom")["correct"] is False |
|
|
|
|
| def test_published_formulas_reproduce_jev_1_13_row() -> None: |
| """Jev 1.13.0's v1.3.0 row (results/v1.2/jevbench-v1.2-results.json).""" |
| hist = {"easy": {2: 18, 4: 13, 5: 41}, "standard": {2: 32, 4: 40, 5: 12, 6: 12}, |
| "judge": {2: 68, 9: 78}, "hard": {2: 77, 3: 26, 4: 73, 5: 38, 6: 6}} |
| chances = {k: sum(c / o for o, c in v.items()) / sum(v.values()) for k, v in hist.items()} |
| tiers = {"easy": 1.0, "standard": 0.9895833333333334, "judge": 0.9452054794520548, |
| "hard": 0.740909090909091} |
| assert jb.intelligence(tiers, chances) == pytest.approx(85.69176285262259) |
| assert jb.speed(0.6524335257709026, 0.7221905551850795, "api") == pytest.approx( |
| 83.26811926100174) |
| assert jb.cost(0.03991423595505617) == pytest.approx(51.96616538951724) |
| assert jb.calibration(0.06061111111111118, 1 - 77.42800000000001 / 100) == pytest.approx( |
| 82.6528888888889) |
| axes = {"intelligence": 85.69176285262259, "calibration": 82.6528888888889, |
| "speed": 83.26811926100174, "cost": 51.96616538951724} |
| assert jb.jevbench_score(axes) == pytest.approx(74.40448849535433) |
|
|
|
|
| def test_near_chance_penalty_and_self_hosted_latency() -> None: |
| axes = {"intelligence": 25.0, "calibration": 80.0, "speed": 80.0, "cost": 80.0} |
| base = math.exp(0.25 * sum(math.log(v) for v in axes.values())) |
| assert jb.jevbench_score(axes) == pytest.approx(base * 0.25) |
| assert jb.adjusted_latency(0.5, "gpu") == pytest.approx(1.15) |
| assert jb.adjusted_latency(0.5, "demo") == pytest.approx(1.0) |
| assert jb.harmonic_view({**axes, "intelligence": 100.0}) == pytest.approx( |
| 4 / (1 / 100 + 3 / 80)) |
|
|
|
|
| def test_summarize_axes(repo: Path) -> None: |
| tasks = jb.load_tasks(repo) |
| preds = { |
| "easy-intent-00": {"probs": {"track_order": 0.9, "cancel_order": 0.1}, |
| "latency_s": 0.1, "input_tokens": 100}, |
| "original-policy-01-0": {"probs": {"yes": 0.2, "no": 0.8}, "latency_s": 0.2, |
| "input_tokens": 100}, |
| "original-policy-01-1": {"probs": {"yes": 0.6, "no": 0.4}, "latency_s": 0.3, |
| "input_tokens": 100}, |
| "original-ordinal-01-0": {"error": "FillError: x", "latency_s": 0.4}, |
| "hard-opus-b-probability-02": {"probs": {"late": 0.6, "on_time": 0.3, "lost": 0.1}, |
| "latency_s": 1.0, "input_tokens": 400}, |
| } |
| s = jb.summarize(tasks, preds, usd_per_m_input=0.02) |
| assert s["tiers"] == {"easy": 1.0, "standard": pytest.approx(1 / 3), "hard": 1.0} |
| ch = s["tier_chance"] |
| assert ch["standard"] == pytest.approx((0.5 + 0.5 + 1 / 3) / 3) |
| want_i = (0.14 * 100 + 0.28 * jb.chance_corrected(1 / 3, ch["standard"]) + 0.30 * 100) / 0.72 |
| assert s["axes"]["intelligence"] == pytest.approx(want_i) |
| |
| assert s["calibration_detail"]["ece_hard"] == pytest.approx(0.4) |
| assert s["axes"]["calibration"] == pytest.approx((100 * (1 - 0.4 / 0.5) + 100) / 2) |
| assert s["latency"]["standard"]["p50_s"] == pytest.approx(0.3) |
| assert s["cost"]["usd_per_1000"] == pytest.approx(175 * 0.02 / 1000) |
| assert s["paraphrase"] == {"pairs": 1, "both_answered_valid": 1, "both_correct": 0, |
| "consistency": 0.0} |
| assert s["valid"] == 4 and s["attempted"] == 5 |
|
|
|
|
| def test_leaderboard_public_intelligence(repo: Path) -> None: |
| (row,) = jb.leaderboard(repo) |
| assert row["public_tiers"] == {"easy": 1.0, "standard": pytest.approx(2 / 3), "hard": 0.0} |
| assert row["published_score"] == 70.0 and row["public_intelligence"] is not None |
|
|
|
|
| def test_contamination_guard(repo: Path) -> None: |
| g = jb.ContaminationGuard(repo) |
| assert g.loaded == 5 |
| assert g.match(NOUL["state"]) |
| long_state = PROB["state"] |
| assert g.match("Intro text. " + json.dumps(long_state, sort_keys=True) + " trailing") |
| assert g.match("Header\n" + CHOICE["state"].upper() + "\nfooter") |
| assert g.match("Where is my package? It was ordered yesterday.") is None |
| short = {**SCORE, "id": "easy-intent-11", "state": "Set an alarm for 7 am."} |
| (repo / "datasets" / "public" / "easy.jsonl").write_text( |
| json.dumps(CHOICE) + "\n" + json.dumps(short) + "\n") |
| g2 = jb.ContaminationGuard(repo) |
| assert g2.match("set an alarm for 7 AM please") |
| assert g2.match("hey, i need to set an alarm for 7 am tomorrow but also can you find " |
| "the best pizza place near me for lunch?") is None |
| assert g.match("A completely unrelated support ticket about printer toner.") is None |
| empty = jb.ContaminationGuard(repo / "missing") |
| assert empty.loaded == 0 and empty.match(NOUL["state"]) is None |
|
|
|
|
| def test_guard_path_refuses_jevbench() -> None: |
| with pytest.raises(ValueError): |
| jb.guard_path("/data/deg/external/jevbench/repo/datasets/public/hard.jsonl") |
| jb.guard_path("/data/deg/datagen/full-off/records.jsonl") |
|
|
