File size: 4,090 Bytes
f8dbde1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """CPU tests for the bench scoring rule and the baseline schema adapters."""
from __future__ import annotations
from deg.bench.score import Prediction, aggregate, drop_generate, normalize, rouge_l, score_record
from deg.bench.systems.clef import _set
from deg.bench.systems.frontier import strict_schema, unstrict
from deg.bench.systems.nuextract import template, unmap
SCHEMA = {
"type": "object",
"properties": {
"dept": {"enum": ["billing", "tech"]},
"urgent": {"type": "boolean"},
"order_id": {"type": "string"},
"note": {"type": "string"},
"reply": {"type": "string", "x-generate": True, "maxLength": 40},
},
"required": ["dept", "urgent", "order_id", "reply"],
}
REC = {"uid": "r1", "context": "Order AB-1 charged twice.", "schema": SCHEMA,
"value": {"dept": "billing", "urgent": True, "order_id": "AB-1", "reply": "We will refund."}}
def test_perfect_prediction_scores_one():
pred = Prediction("r1", value=dict(REC["value"]), generated_tokens=5, latency_s=0.1,
confidence={"/dept": 0.9, "/urgent": 0.8})
s = score_record(REC, pred)
assert s["valid"] is True
assert all(d["ok"] for d in s["decide"]) and len(s["decide"]) == 2
assert all(e["em"] for e in s["extract"]) and len(s["extract"]) == 1
assert s["generate"][0]["bounded"] and s["generate"][0]["rouge_l"] == 1.0
m = aggregate([s], [pred])
assert m["decide_acc"] == 1.0 and m["decide_ece"] is not None
def test_failed_prediction_scores_fields_wrong():
pred = Prediction("r1", error="boom")
s = score_record(REC, pred)
assert s["valid"] is False
assert not any(d["ok"] for d in s["decide"]) and len(s["decide"]) == 2
assert not s["extract"][0]["em"] and not s["generate"][0]["bounded"]
def test_bool_is_not_int_and_omitted_extract_scores_null():
value = {**REC["value"], "urgent": 1, "note": None}
s = score_record(REC, Prediction("r1", value=value))
assert [d["ok"] for d in s["decide"] if d["ptr"] == "/urgent"] == [False]
assert {"ptr": "/note", "em": True, "norm": True} in s["extract"]
def test_scope_limits_validity_and_fields():
value = {k: v for k, v in REC["value"].items() if k != "reply"}
s = score_record(REC, Prediction("r1", value=value), ("decide", "extract"))
assert s["valid"] is True and s["generate"] == []
s = score_record(REC, Prediction("r1", value={"dept": "billing"}), ("decide",))
assert s["valid"] is None and s["extract"] == []
def test_drop_generate_and_normalize_and_rouge():
assert "reply" not in drop_generate(SCHEMA)["properties"]
assert "reply" not in drop_generate(SCHEMA)["required"]
assert normalize(" AB-1. ") == normalize("ab-1") and normalize("1,200") == 1200.0
assert rouge_l("a b c", "a c") == 0.8 and rouge_l("", "x") == 0.0
def test_strict_schema_roundtrip():
s = strict_schema(SCHEMA)
assert s["additionalProperties"] is False and set(s["required"]) == set(SCHEMA["properties"])
assert s["properties"]["note"]["type"] == ["string", "null"]
assert "x-generate" not in s["properties"]["reply"]
out = unstrict({"dept": "billing", "urgent": True, "order_id": "AB-1", "note": None,
"reply": "x"}, SCHEMA)
assert "note" not in out
def test_nuextract_template_and_unmap():
t = template({"type": "object", "properties": {
**SCHEMA["properties"], "level": {"enum": [1, 2, None]},
"tags": {"type": "array", "items": {"enum": ["a", "b"]}},
"when": {"type": "string", "format": "date"}}})
assert t["dept"] == ["billing", "tech"] and t["urgent"] == "boolean"
assert t["order_id"] == "verbatim-string" and "reply" not in t
assert t["level"] == ["1", "2"] and t["tags"] == [["a", "b"]] and t["when"] == "date"
assert unmap({"level": "2"}, {"type": "object", "properties": {"level": {"enum": [1, 2]}}}) \
== {"level": 2}
def test_clef_set_pointer():
root: dict = {}
_set(root, "/items/1/flag", True)
_set(root, "/a", "x")
assert root == {"items": [None, {"flag": True}], "a": "x"}
|