| """Opt-in synthetic evaluation; records aggregate metrics, never document payloads.""" |
|
|
| import importlib.metadata |
| import json |
| import resource |
| import statistics |
| import time |
| from collections import Counter |
| from pathlib import Path |
|
|
| from ocr.schemas import SafeError |
|
|
| from .detectors import detect |
| from .model import MODEL_ID, MODEL_REVISION |
| from .replacement import replace, resolve_overlaps, validate_entities |
| from .schemas import LABELS |
|
|
|
|
| def metrics(counts): |
| output = {} |
| for label in LABELS: |
| tp, fp, fn = (counts[(label, name)] for name in ("tp", "fp", "fn")) |
| p = tp / (tp + fp) if tp + fp else 0.0 |
| r = tp / (tp + fn) if tp + fn else 0.0 |
| output[label] = dict( |
| tp=tp, fp=fp, fn=fn, precision=p, recall=r, f1=2 * p * r / (p + r) if p + r else 0.0 |
| ) |
| return output |
|
|
|
|
| def score_cases(recognizer, cases, *, detector=detect, case_reports=None): |
| ner_counts, combined_counts, leaks, failures = Counter(), Counter(), Counter(), Counter() |
| literal_leaks = Counter() |
| partial_canaries, uncovered_canaries, coverage = Counter(), Counter(), Counter() |
| durations, exact, blocked = [], 0, 0 |
| groups = {} |
| for case in cases: |
| text = case["text"] |
| gold = {(s["start"], s["end"], s["label"]) for s in case["spans"]} |
| for s in case["spans"]: |
| if text[s["start"] : s["end"]] != s["text"]: |
| raise SafeError("invalid_evaluation_fixture") |
| start = time.perf_counter() |
| actual, predicted, ner = None, set(), set() |
| raw, combined, error_code = [], [], None |
| try: |
| raw = validate_entities(text, recognizer.recognize(text)) |
| ner = {(s.start, s.end, s.label) for s in raw} |
| combined = resolve_overlaps(validate_entities(text, raw + detector(text))) |
| predicted = {(s.start, s.end, s.label) for s in combined} |
| actual, _ = replace(text, combined) |
| except SafeError as e: |
| error_code = e.code |
| failures[e.code] += 1 |
| blocked += 1 |
| durations.append((time.perf_counter() - start) * 1000) |
| for predicted_set, counts in ((ner, ner_counts), (predicted, combined_counts)): |
| for _, _, label in predicted_set & gold: |
| counts[(label, "tp")] += 1 |
| for _, _, label in predicted_set - gold: |
| counts[(label, "fp")] += 1 |
| for _, _, label in gold - predicted_set: |
| counts[(label, "fn")] += 1 |
| exact += actual == case["expected"] |
| for s in case["spans"]: |
| if actual is None or s["text"].casefold() in actual.casefold(): |
| literal_leaks[s["label"]] += 1 |
| sensitive_positions = [i for i in range(s["start"], s["end"]) if text[i].isalnum()] |
| covered = sum( |
| actual is not None and any(left <= i < right for left, right, _ in predicted) |
| for i in sensitive_positions |
| ) |
| coverage[(s["label"], "total")] += len(sensitive_positions) |
| coverage[(s["label"], "covered")] += covered |
| if covered < len(sensitive_positions): |
| leaks[s["label"]] += 1 |
| if covered: |
| partial_canaries[s["label"]] += 1 |
| else: |
| uncovered_canaries[s["label"]] += 1 |
| if case_reports is not None: |
| case_reports.append( |
| dict( |
| case_id=case["id"], |
| scenario=case.get("scenario"), |
| predictions=[s.model_dump() for s in raw], |
| combined_predictions=[s.model_dump() for s in combined], |
| false_positives=sorted(predicted - gold), |
| false_negatives=sorted(gold - predicted), |
| exact_match=actual == case["expected"], |
| error_code=error_code, |
| canaries=[ |
| dict( |
| **s, |
| uncovered_positions=[ |
| i |
| for i in range(s["start"], s["end"]) |
| if text[i].isalnum() |
| and ( |
| actual is None |
| or not any(left <= i < right for left, right, _ in predicted) |
| ) |
| ], |
| ) |
| for s in case["spans"] |
| ], |
| ) |
| ) |
| group = case.get("language", "unspecified") |
| stats = groups.setdefault(group, dict(documents=0, exact_matches=0, false_negatives=0)) |
| stats["documents"] += 1 |
| stats["exact_matches"] += actual == case["expected"] |
| stats["false_negatives"] += len(gold - predicted) |
| if case.get("ocr_noise"): |
| stats = groups.setdefault( |
| "ocr_noise", dict(documents=0, exact_matches=0, false_negatives=0) |
| ) |
| stats["documents"] += 1 |
| stats["exact_matches"] += actual == case["expected"] |
| stats["false_negatives"] += len(gold - predicted) |
| return dict( |
| documents=len(cases), |
| ner_entity_metrics=metrics(ner_counts), |
| combined_entity_metrics=metrics(combined_counts), |
| unredacted_canaries={k: leaks[k] for k in LABELS}, |
| literal_unredacted_canaries={k: literal_leaks[k] for k in LABELS}, |
| partial_canaries={k: partial_canaries[k] for k in LABELS}, |
| fully_uncovered_canaries={k: uncovered_canaries[k] for k in LABELS}, |
| alphanumeric_canary_coverage={ |
| k: dict(covered=coverage[(k, "covered")], total=coverage[(k, "total")]) for k in LABELS |
| }, |
| document_exact_match_rate=exact / len(cases), |
| groups=groups, |
| blocked_documents=blocked, |
| failure_codes=dict(failures), |
| warm_latency_ms=dict(median=statistics.median(durations), max=max(durations)), |
| automated_release_eligible=False, |
| qualification_passed=not any(leaks.values()) and blocked == 0, |
| ) |
|
|
|
|
| def run(cache, output): |
| from .recognizer import GLiNERRecognizer |
|
|
| start = time.perf_counter() |
| recognizer = GLiNERRecognizer(cache) |
| load = (time.perf_counter() - start) * 1000 |
| |
| recognizer.recognize("The cat sat on the mat.") |
| report = dict( |
| model=MODEL_ID, |
| revision=MODEL_REVISION, |
| threshold=recognizer.threshold, |
| prompts=recognizer.prompts, |
| runtime_versions={ |
| k: importlib.metadata.version(k) |
| for k in ("gliner", "torch", "transformers", "numpy", "pydantic") |
| }, |
| device="cpu", |
| threads=2, |
| loading_ms=load, |
| suites={}, |
| ) |
| root = Path(__file__).parent / "fixtures" |
| for name in ("qualification", "validation"): |
| report["suites"][name] = score_cases( |
| recognizer, json.loads((root / f"{name}.json").read_text()) |
| ) |
| report["peak_rss_mib"] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 |
| report["automated_release_eligible"] = False |
| report["recognition_qualification_passed"] = all( |
| s["qualification_passed"] for s in report["suites"].values() |
| ) |
| Path(output).write_text(json.dumps(report, indent=2) + "\n") |
| return report |
|
|
|
|
| if __name__ == "__main__": |
| import argparse |
|
|
| parser = argparse.ArgumentParser(description="Pinned local synthetic model evaluation") |
| parser.add_argument("--cache", required=True) |
| parser.add_argument("--output", required=True) |
| args = parser.parse_args() |
| try: |
| report = run(args.cache, args.output) |
| except Exception: |
| raise SystemExit("anonymization_evaluation_failed") from None |
| print( |
| json.dumps( |
| { |
| "recognition_qualification_passed": report["recognition_qualification_passed"], |
| "automated_release_eligible": False, |
| } |
| ) |
| ) |
|
|
