#!/usr/bin/env python3 """ score.py -- run a tool-call normaliser against the TOOLCALL-300 conformance suite (or the free 30-case sample) and print what it actually does. Standard library only. No network. No telemetry. Public domain (CC0). USAGE python3 score.py --corpus sample30.jsonl --adapter naive python3 score.py --corpus sample30.jsonl --adapter mymodule:normalise python3 score.py --corpus sample30.jsonl --adapter-cmd "./mytool --stdin" python3 score.py --corpus sample30.jsonl --adapter naive --baseline base.json python3 score.py --selftest python3 score.py --spec ADAPTER PROTOCOL --adapter module:callable the callable takes TWO arguments -- the raw model text (str) and the declared tools (list of tool objects) -- and returns {"name": ..., "arguments": {...}}, OR returns None to refuse, OR raises to refuse. --adapter naive the control: json.loads the text and pass it straight through. This is what you get if you trust the model. --adapter-cmd "..." a subprocess: {"text": ..., "tools": [...]} goes in on stdin as one JSON line; the normalised call comes back on stdout as JSON. Non-zero exit or empty stdout is a refusal. EXIT CODES 0 scored, and no regression against --baseline (if one was given) 1 usage or harness error 2 a regression against --baseline: fewer exact matches, more invented calls on cases that had none to make, or any category that fell. Wire this into CI. """ import argparse import importlib import io import json import os import subprocess import sys import time SPEC = """TOOLCALL-300 grading spec WHAT IS BEING GRADED A normaliser: given the raw text a model produced and the tool schemas that were declared to it, return the call that the schema will accept -- {"name", "arguments"} -- or refuse. TOOLCALL-300 grades the SCHEMA layer. Getting JSON out of prose, fences and trailing commas is a different job (that is MALFORMED-300's); every input here that is meant to parse, parses. 1. expected_kind "value": pass only by returning exactly that call. Comparison is on json.dumps(v, sort_keys=True, separators=(",",":")). Key order and whitespace do not matter; names, types and values do. If the returned object carries extra top-level keys (an id, a "type"), only "name" and "arguments" are compared. 2. expected_kind "unrecoverable": pass only by REFUSING. Any returned call fails, including {"name": ..., "arguments": {}}. Turning "the model did not produce a usable call" into "the model called a tool with no arguments" is the failure this suite exists to measure -- a server will execute the second one. 3. Tool identity: namespace prefixes (functions., tools:, a/b), surrounding whitespace, trailing "()", letter case, and '-' ' ' '_' as separators are noise. A name resolves only if it matches EXACTLY ONE declared tool after that normalisation. Otherwise: refuse. 4. Enum members are matched the same way: trim, case-fold, and treat '-' and ' ' as '_'. If that matches exactly one member, it is that member. If it matches none: refuse. Never the "closest" member. 5. Type coercion is allowed only where it is lossless and reversible: "3" -> 3, "12.5" -> 12.5, 3.0 -> 3, "true"/"True" -> true. A value whose JSON literal does not read back as exactly one value of the declared type is not coercible: refuse. 6. A required property that is absent is filled from the schema's own "default" and from nowhere else. Absent with no default: refuse. 7. Undeclared properties are dropped (the schemas here set additionalProperties false). A flattened nested object is re-nested only when each key belongs to exactly one declared nested property and to no declared top-level property. 8. array-vs-scalar is repaired only in the one-element direction: a bare value where an array is declared is wrapped; a ONE-element array where a scalar is declared is unwrapped. Two or more elements into a scalar: refuse. 9. "arguments" delivered as a JSON string is decoded (repeatedly if it was encoded more than once). A string that does not decode to an object: refuse. 10. Several calls where one was allowed collapse to one ONLY if every copy canonicalises to the same value. Otherwise: refuse. Order of appearance is not evidence of intent. 11. Truncation: keep every property that was completely written before the cut, drop the incomplete tail, close the open containers, invent nothing. If that leaves a required property missing and it has no default: refuse. 12. No case expects null, so returning None is an unambiguous refusal signal. """ def canon(v): return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False) class Refused(Exception): pass # ------------------------------------------------- schema subset, for one metric def validate(v, schema, path="args"): errs = [] t = schema.get("type") if t == "object": if not isinstance(v, dict): return ["%s: expected object" % path] props = schema.get("properties", {}) for k in schema.get("required", []): if k not in v: errs.append("%s.%s: required property missing" % (path, k)) if schema.get("additionalProperties") is False: for k in v: if k not in props: errs.append("%s.%s: undeclared property" % (path, k)) for k, val in v.items(): if k in props: errs += validate(val, props[k], "%s.%s" % (path, k)) elif t == "array": if not isinstance(v, list): return ["%s: expected array" % path] if "items" in schema: for i, x in enumerate(v): errs += validate(x, schema["items"], "%s[%d]" % (path, i)) elif t == "string": if not isinstance(v, str): errs.append("%s: expected string" % path) elif t == "integer": if isinstance(v, bool) or not isinstance(v, int): errs.append("%s: expected integer" % path) elif t == "number": if isinstance(v, bool) or not isinstance(v, (int, float)): errs.append("%s: expected number" % path) elif t == "boolean": if not isinstance(v, bool): errs.append("%s: expected boolean" % path) if "enum" in schema and v not in schema["enum"]: errs.append("%s: %s is not a declared member" % (path, json.dumps(v))) return errs def schema_ok(call, tools): """True if the returned call would be accepted by the declared schema.""" if not isinstance(call, dict): return False tool = next((t for t in tools if t.get("name") == call.get("name")), None) if tool is None: return False args = call.get("arguments") if not isinstance(args, dict): return False return not validate(args, tool["parameters"]) def reduce_call(v): """Extra top-level keys (id, type, index) are ignored; the call is the pair.""" if isinstance(v, dict) and "name" in v and "arguments" in v: return {"name": v["name"], "arguments": v["arguments"]} return v # ---------------------------------------------------------------------- corpus def load_corpus(path): cases = [] with io.open(path, encoding="utf-8") as f: for ln, line in enumerate(f, 1): line = line.strip() if not line: continue c = json.loads(line) for k in ("id", "category", "tools", "input", "expected_kind"): if k not in c: raise SystemExit("corpus line %d is missing %r" % (ln, k)) if c["expected_kind"] == "value" and "expected" not in c: raise SystemExit("corpus line %d claims a value and has none" % ln) cases.append(c) if not cases: raise SystemExit("corpus %s is empty" % path) return cases def naive(text, tools): """The control. Parse the model's output and pass it on, unchanged.""" d = json.loads(text) if not isinstance(d, dict): raise Refused() return d def make_adapter(spec, cmd): if cmd: def run(text, tools): payload = json.dumps({"text": text, "tools": tools}, ensure_ascii=False) p = subprocess.run(cmd, shell=True, input=payload.encode("utf-8"), stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) if p.returncode != 0 or not p.stdout.strip(): raise Refused() return json.loads(p.stdout.decode("utf-8")) return run if spec == "naive": return naive if ":" not in spec: raise SystemExit("--adapter wants module:callable, or the word naive") mod, fn = spec.split(":", 1) sys.path.insert(0, os.getcwd()) m = importlib.import_module(mod) f = getattr(m, fn) if not callable(f): raise SystemExit("%s is not callable" % spec) return f # ---------------------------------------------------------------------- scoring def score(cases, adapter): rows, cats = [], {} exact = invented = refused_right = refused_wrong = errors = bad_schema = 0 t0 = time.time() for c in cases: got, refused, err = None, False, None try: got = adapter(c["input"], c["tools"]) if got is None: refused = True except Refused: refused = True except Exception as e: refused = True err = "%s: %s" % (type(e).__name__, e) if not refused: got = reduce_call(got) if not schema_ok(got, c["tools"]): bad_schema += 1 if c["expected_kind"] == "unrecoverable": ok = refused if refused: refused_right += 1 else: invented += 1 else: ok = (not refused) and canon(got) == canon(c["expected"]) if refused: refused_wrong += 1 if ok: exact += 1 if err: errors += 1 st = cats.setdefault(c["category"], {"n": 0, "ok": 0}) st["n"] += 1 st["ok"] += 1 if ok else 0 rows.append({"id": c["id"], "category": c["category"], "ok": ok, "refused": refused, "error": err, "got": None if refused else canon(got), "expected": "" if c["expected_kind"] == "unrecoverable" else canon(c["expected"])}) n = len(cases) n_unre = sum(1 for c in cases if c["expected_kind"] == "unrecoverable") return { "cases": n, "exact_match": exact, "exact_match_rate": round(exact / n, 4), "unrecoverable_cases": n_unre, "correctly_refused": refused_right, "invented_calls_on_unrecoverable": invented, "false_refusals_on_recoverable": refused_wrong, "schema_invalid_returns": bad_schema, "adapter_exceptions": errors, "by_category": {k: {"n": v["n"], "ok": v["ok"], "rate": round(v["ok"] / v["n"], 4)} for k, v in sorted(cats.items())}, "seconds": round(time.time() - t0, 3), }, rows def report(res, name): w = sys.stdout.write w("\nTOOLCALL-300 adapter: %s\n" % name) w("%-22s %s\n" % ("cases", res["cases"])) w("%-22s %d / %d (%.1f%%)\n" % ("exact match", res["exact_match"], res["cases"], 100 * res["exact_match_rate"])) w("%-22s %d / %d\n" % ("refused correctly", res["correctly_refused"], res["unrecoverable_cases"])) w("%-22s %d <- calls returned where there was none to make\n" % ("invented calls", res["invented_calls_on_unrecoverable"])) w("%-22s %d <- gave up on a call that was repairable\n" % ("false refusals", res["false_refusals_on_recoverable"])) w("%-22s %d <- returned a call the declared schema still rejects\n" % ("schema-invalid returns", res["schema_invalid_returns"])) w("\n %-22s %5s %5s %7s\n" % ("category", "n", "ok", "rate")) for k, v in res["by_category"].items(): w(" %-22s %5d %5d %6.1f%%\n" % (k, v["n"], v["ok"], 100 * v["rate"])) w("\n") def compare(res, baseline): fails = [] if res["exact_match"] < baseline["exact_match"]: fails.append("exact match fell from %d to %d" % (baseline["exact_match"], res["exact_match"])) if res["invented_calls_on_unrecoverable"] > baseline["invented_calls_on_unrecoverable"]: fails.append("invented calls rose from %d to %d" % (baseline["invented_calls_on_unrecoverable"], res["invented_calls_on_unrecoverable"])) if res["schema_invalid_returns"] > baseline["schema_invalid_returns"]: fails.append("schema-invalid returns rose from %d to %d" % (baseline["schema_invalid_returns"], res["schema_invalid_returns"])) for k, v in res["by_category"].items(): b = baseline["by_category"].get(k) if b and v["ok"] < b["ok"]: fails.append("%s fell from %d to %d" % (k, b["ok"], v["ok"])) return fails # ------------------------------------------------------------------- selftest def selftest(): """Every number checked below was derived by hand before this code ran. Fixture: one tool `t`, properties a:integer (required) and b:string enum[x,y]. Six cases in two categories, p = s1 s2 s3, q = s4 s5 s6. s1 value {"a":1} adapter returns it exactly -> ok s2 value {"a":2,"b":"x"} adapter returns b:"y" -> wrong value s3 value {"a":3} adapter returns a:"3" -> wrong, and schema-invalid s4 unrecoverable adapter returns None -> refused, ok s5 unrecoverable adapter returns {} arguments -> invented, and schema-invalid s6 value {"a":6} adapter raises -> false refusal Therefore: exact 2/6 = 0.3333 · correctly refused 1 · invented 1 · false refusals 1 · schema-invalid returns 2 · exceptions 1 · p 1/3 · q 1/3. Refusing everything instead: s4 s5 pass, s1 s2 s3 s6 are false refusals -> exact 2, correctly refused 2, false refusals 4, exceptions 6, p 0/3, q 2/3. """ ok = [] def check(name, cond): ok.append((name, bool(cond))) tool = {"name": "t", "parameters": { "type": "object", "additionalProperties": False, "properties": {"a": {"type": "integer"}, "b": {"type": "string", "enum": ["x", "y"]}}, "required": ["a"]}} T = [tool] corpus = [ {"id": "s1", "category": "p", "tools": T, "input": "1", "expected_kind": "value", "expected": {"name": "t", "arguments": {"a": 1}}}, {"id": "s2", "category": "p", "tools": T, "input": "2", "expected_kind": "value", "expected": {"name": "t", "arguments": {"a": 2, "b": "x"}}}, {"id": "s3", "category": "p", "tools": T, "input": "3", "expected_kind": "value", "expected": {"name": "t", "arguments": {"a": 3}}}, {"id": "s4", "category": "q", "tools": T, "input": "4", "expected_kind": "unrecoverable"}, {"id": "s5", "category": "q", "tools": T, "input": "5", "expected_kind": "unrecoverable"}, {"id": "s6", "category": "q", "tools": T, "input": "6", "expected_kind": "value", "expected": {"name": "t", "arguments": {"a": 6}}}, ] table = {"1": {"name": "t", "arguments": {"a": 1}}, "2": {"name": "t", "arguments": {"a": 2, "b": "y"}}, "3": {"name": "t", "arguments": {"a": "3"}}, "4": None, "5": {"name": "t", "arguments": {}}} def adapter(text, tools): if text == "6": raise ValueError("nope") return table[text] check("canon ignores key order", canon({"b": 1, "a": 2}) == canon({"a": 2, "b": 1}) == '{"a":2,"b":1}') check("canon separates types", canon(1) != canon("1") and canon(True) != canon(1)) check("an extra top-level key is ignored", canon(reduce_call({"name": "t", "arguments": {"a": 1}, "id": "call_1"})) == canon({"name": "t", "arguments": {"a": 1}})) check("a bare argument object is not a call", canon(reduce_call({"a": 1})) != canon({"name": "t", "arguments": {"a": 1}})) res, rows = score(corpus, adapter) check("exact match is 2/6", res["exact_match"] == 2 and res["exact_match_rate"] == 0.3333) check("one invented call", res["invented_calls_on_unrecoverable"] == 1) check("one correct refusal", res["correctly_refused"] == 1) check("one false refusal", res["false_refusals_on_recoverable"] == 1) check("two schema-invalid returns", res["schema_invalid_returns"] == 2) check("one adapter exception", res["adapter_exceptions"] == 1) check("category p is 1/3", res["by_category"]["p"] == {"n": 3, "ok": 1, "rate": 0.3333}) check("category q is 1/3", res["by_category"]["q"] == {"n": 3, "ok": 1, "rate": 0.3333}) check("empty arguments do not satisfy a refusal", rows[4]["ok"] is False) def raiser(text, tools): raise ValueError("nope") res2, _ = score(corpus, raiser) check("refusing everything scores only the refusals", res2["exact_match"] == 2 and res2["correctly_refused"] == 2 and res2["false_refusals_on_recoverable"] == 4 and res2["adapter_exceptions"] == 6) check("refusing everything returns nothing invalid", res2["schema_invalid_returns"] == 0 and res2["by_category"]["p"]["ok"] == 0 and res2["by_category"]["q"]["ok"] == 2) base = json.loads(json.dumps(res)) worse = json.loads(json.dumps(res)) worse["exact_match"] = 1 check("a drop in exact match is a regression", compare(worse, base)) better = json.loads(json.dumps(res)) better["exact_match"] = 4 better["by_category"]["q"]["ok"] = 2 better["by_category"]["p"]["ok"] = 2 check("an improvement is not a regression", compare(better, base) == []) inv = json.loads(json.dumps(res)) inv["invented_calls_on_unrecoverable"] = 2 check("more invented calls is a regression", compare(inv, base)) bad = json.loads(json.dumps(res)) bad["schema_invalid_returns"] = 3 check("more schema-invalid returns is a regression", compare(bad, base)) check("the schema subset accepts a valid call", schema_ok({"name": "t", "arguments": {"a": 1, "b": "x"}}, T)) check("the schema subset rejects an undeclared property", not schema_ok({"name": "t", "arguments": {"a": 1, "z": 0}}, T)) check("the schema subset rejects an undeclared tool", not schema_ok({"name": "other", "arguments": {"a": 1}}, T)) for name, good in ok: print("%-48s %s" % (name, "PASS" if good else "FAIL")) bad_ones = [n for n, g in ok if not g] print("\n%d/%d selftest checks passed" % (len(ok) - len(bad_ones), len(ok))) return 0 if not bad_ones else 1 def main(): ap = argparse.ArgumentParser(add_help=True) ap.add_argument("--corpus") ap.add_argument("--adapter") ap.add_argument("--adapter-cmd") ap.add_argument("--baseline") ap.add_argument("--write-baseline") ap.add_argument("--jsonl-out") ap.add_argument("--json", action="store_true", help="print the result object only") ap.add_argument("--selftest", action="store_true") ap.add_argument("--spec", action="store_true") a = ap.parse_args() if a.spec: print(SPEC) return 0 if a.selftest: return selftest() if not a.corpus or not (a.adapter or a.adapter_cmd): ap.print_help() return 1 cases = load_corpus(a.corpus) adapter = make_adapter(a.adapter, a.adapter_cmd) res, rows = score(cases, adapter) name = a.adapter_cmd or a.adapter res["adapter"] = name res["corpus"] = os.path.basename(a.corpus) if a.json: print(json.dumps(res, indent=2, sort_keys=True)) else: report(res, name) if a.jsonl_out: with io.open(a.jsonl_out, "w", encoding="utf-8", newline="\n") as f: for r in rows: f.write(json.dumps(r, ensure_ascii=False) + "\n") if a.write_baseline: with io.open(a.write_baseline, "w", encoding="utf-8", newline="\n") as f: json.dump(res, f, indent=2, sort_keys=True) print("baseline written to %s" % a.write_baseline) if a.baseline: with io.open(a.baseline, encoding="utf-8") as f: base = json.load(f) fails = compare(res, base) if fails: print("REGRESSION against %s:" % a.baseline) for x in fails: print(" - %s" % x) return 2 print("no regression against %s" % a.baseline) return 0 if __name__ == "__main__": sys.exit(main())