File size: 3,707 Bytes
4e12f7c | 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 | """Compare the pinned GLiNER2 native classifier with its Core ML export."""
import argparse
import json
import statistics
import time
from pathlib import Path
import coremltools as ct
import numpy as np
import torch
from gliner2 import AutoExtractor
from preprocessing import prepare_classification
MODEL_ID = "fastino/gliner2.5-small-v1"
MODEL_REVISION = "7e6f537f10337497069276892a5ef435028252ce"
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--package", default="build/gliner2_small_classification_fp16_L128_K8.mlpackage")
parser.add_argument("--suite", required=True)
parser.add_argument("--limit", type=int, default=100)
parser.add_argument("--length", type=int, default=128)
parser.add_argument("--max-options", type=int, default=8)
parser.add_argument("--out", default="build/verify.json")
args = parser.parse_args()
torch.set_num_threads(4)
source = Path.home() / ".cache/huggingface/hub/models--fastino--gliner2.5-small-v1/snapshots" / MODEL_REVISION
native = AutoExtractor.from_pretrained(str(source), map_location="cpu").eval()
model = ct.models.MLModel(args.package, compute_units=ct.ComputeUnit.ALL)
counts = {"checked": 0, "too_many_options": 0, "too_long": 0, "duplicate_labels": 0, "mismatches": 0}
errors = []
latencies = []
failures = []
for line in Path(args.suite).open():
row = json.loads(line)
options = row.get("options") or []
if len(options) > args.max_options or not options:
counts["too_many_options"] += 1
continue
labels = [(description or key).strip() for key, description in options]
if len(set(labels)) != len(labels):
counts["duplicate_labels"] += 1
continue
text = row["state"]
task = "decision"
try:
arrays = prepare_classification(native, text, task, labels, args.length, args.max_options)
except ValueError:
counts["too_long"] += 1
continue
expected = native.classify_text(text, {task: labels}, include_confidence=True, max_len=args.length)[task]
start = time.perf_counter()
prediction = model.predict(arrays)
latencies.append((time.perf_counter() - start) * 1000)
probabilities = np.asarray(prediction["probabilities"])[0, : len(labels)]
chosen = labels[int(probabilities.argmax())]
error = abs(float(probabilities.max()) - float(expected["confidence"]))
errors.append(error)
counts["checked"] += 1
if chosen != expected["label"]:
counts["mismatches"] += 1
failures.append({"suite": row["suite"], "index": row["index"], "native": expected, "coreml": chosen})
if counts["checked"] >= args.limit:
break
result = {
"model": MODEL_ID, "revision": MODEL_REVISION, "package": args.package,
"selected_manifest": "first eligible rows in source suite order; no gold labels used",
"counts": counts, "maximum_confidence_error": max(errors, default=None),
"mean_confidence_error": statistics.mean(errors) if errors else None,
"model_call_p50_ms": statistics.median(latencies[1:]) if len(latencies) > 1 else None,
"model_call_p95_ms": sorted(latencies[1:])[int(0.95 * (len(latencies) - 1))] if len(latencies) > 1 else None,
"failures": failures[:20],
}
Path(args.out).parent.mkdir(parents=True, exist_ok=True)
Path(args.out).write_text(json.dumps(result, indent=2) + "\n")
print(json.dumps(result, indent=2))
if counts["checked"] == 0 or counts["mismatches"]:
raise SystemExit(1)
if __name__ == "__main__":
main()
|