"""Opt-in live semantic detect comparison. Not part of production generate. Usage from backend/: python entity_detect_eval.py python entity_detect_eval.py --live Synthetic sentences only. No personal data. Live quality stays unconfirmed until an explicit --live run succeeds against gold spans. Fake mode proves the scoring contract, not live semantic quality. """ from __future__ import annotations import argparse import json import os import sys import time from pathlib import Path ROOT = Path(__file__).resolve().parent if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) def _span(text: str, needle: str, entity_type: str, *, occurrence: int = 0) -> dict: start = -1 found = -1 while True: start = text.find(needle, start + 1) if start < 0: raise ValueError(f"needle {needle!r} occurrence {occurrence} missing in {text!r}") found += 1 if found == occurrence: return { "start": start, "end": start + len(needle), "text": needle, "entity_type": entity_type, } CASES = ( { "id": "person_vs_food", "text": "Ich aß Sushi. Sushi kam später.", "expected": [_span("Ich aß Sushi. Sushi kam später.", "Sushi", "PERSON", occurrence=1)], "note": "Person versus Lebensmittel: nur die identifizierende Nennung.", }, { "id": "project_vs_activity", "text": "Ich arbeitete am privaten Projekt Aurora. Danach arbeitete ich.", "expected": [_span("Ich arbeitete am privaten Projekt Aurora. Danach arbeitete ich.", "Aurora", "PROJECT")], "note": "Privates Projekt versus allgemeine Tätigkeit.", }, { "id": "place_vs_room", "text": "Ich war in Hamburg. Später saß ich im Wohnzimmer.", "expected": [_span("Ich war in Hamburg. Später saß ich im Wohnzimmer.", "Hamburg", "PLACE")], "note": "Genauer Eigenort versus allgemeiner Raum.", }, { "id": "org_vs_noun", "text": "Ich sprach mit der Firma Nordwerk. Das Notizbuch blieb liegen.", "expected": [_span("Ich sprach mit der Firma Nordwerk. Das Notizbuch blieb liegen.", "Nordwerk", "ORG")], "note": "Organisation versus allgemeines Substantiv.", }, { "id": "person_vs_group", "text": "Anna kam vorbei. Die Nachbarn spielten draußen.", "expected": [_span("Anna kam vorbei. Die Nachbarn spielten draußen.", "Anna", "PERSON")], "note": "Personenname versus generische Personengruppe.", }, { "id": "animal_vs_org", "text": "Die Seehunde schwammen nah am Ufer.", "expected": [], "note": "Tierbezeichnung versus Organisation oder Person.", }, { "id": "typo_common", "text": "Ich gieng zum Laden.", "expected": [], "note": "Tippfehler eines Allgemeinbegriffs ist keine Identität.", }, { "id": "same_word_two_roles", "text": "Ich aß Sushi. Sushi kam ins Zimmer.", "expected": [_span("Ich aß Sushi. Sushi kam ins Zimmer.", "Sushi", "PERSON", occurrence=1)], "note": "Identischer Wortlaut in zwei semantischen Rollen.", }, { "id": "multipart_place", "text": "Ich war in Hamburg und sprach mit der Organisation Nordwerk.", "expected": [ _span("Ich war in Hamburg und sprach mit der Organisation Nordwerk.", "Hamburg", "PLACE"), _span("Ich war in Hamburg und sprach mit der Organisation Nordwerk.", "Nordwerk", "ORG"), ], "note": "Mehrteiliger Ort plus Organisation, keine Ganzsatz-Spans.", }, { "id": "no_sentence_span", "text": "Ich traf Anna am Nachmittag.", "expected": [_span("Ich traf Anna am Nachmittag.", "Anna", "PERSON")], "note": "Nur der Name, kein Ganzsatz- oder Satzfragment-Span.", }, ) def _key(item: dict) -> tuple[int, int, str, str]: return ( int(item.get("start") or -1), int(item.get("end") or -1), str(item.get("text") or ""), str(item.get("entity_type") or "").upper(), ) def _offset_key(item: dict) -> tuple[int, int, str]: return (int(item.get("start") or -1), int(item.get("end") or -1), str(item.get("text") or "")) def score_spans(predicted: list[dict], expected: list[dict]) -> dict: pred = [_key(item) for item in predicted] gold = [_key(item) for item in expected] pred_off = [_offset_key(item) for item in predicted] gold_off = [_offset_key(item) for item in expected] found = [item for item in gold if item in pred] unexpected = [item for item in pred if item not in gold] missing = [item for item in gold if item not in pred] wrong_type = [] wrong_offset = [] for item in predicted: matches_text = [ gold_item for gold_item in expected if (gold_item.get("text") or "") == (item.get("text") or "") ] if not matches_text: continue if _key(item) in gold: continue same_offsets = any(_offset_key(item) == _offset_key(gold_item) for gold_item in matches_text) if same_offsets: wrong_type.append(_key(item)) else: wrong_offset.append(_key(item)) tp = len(found) fp = len(unexpected) fn = len(missing) precision = tp / (tp + fp) if (tp + fp) else 1.0 recall = tp / (tp + fn) if (tp + fn) else 1.0 return { "expected_found": tp, "expected_count": len(gold), "unexpected": fp, "missing": fn, "wrong_type": len(wrong_type), "wrong_offset": len(wrong_offset), "precision": round(precision, 4), "recall": round(recall, 4), "sentence_or_fragment_span": any( (item.get("end") or 0) - (item.get("start") or 0) > max(len(item.get("text") or ""), 0) and " " in (item.get("text") or "") for item in predicted ), } def _summarize(entities: list[dict], expected: list[dict]) -> dict: types = sorted({(item.get("entity_type") or "").upper() for item in entities}) return { "count": len(entities), "types": types, "has_labels": False, "metrics": score_spans(entities, expected), } def run_fake() -> dict: from entity_detect import _contract_fake_spans rows = [] for case in CASES: entities = _contract_fake_spans(case["text"]) rows.append( { "id": case["id"], "note": case["note"], "entities": _summarize(entities, case["expected"]), "mode": "fake", } ) return { "mode": "fake", "live_quality": "unconfirmed", "detect_model_unconfirmed": "openai/gpt-4.1-nano", "cases": rows, } def run_live() -> dict: if os.environ.get("KANSHO_FAKE_DETECT"): raise SystemExit("Live-Detect verweigert, solange KANSHO_FAKE_DETECT gesetzt ist.") from entity_detect import detect_personal_egress rows = [] for case in CASES: started = time.perf_counter() outcome = detect_personal_egress(None, case["text"]) elapsed = int((time.perf_counter() - started) * 1000) predicted = [ { "start": item.get("start"), "end": item.get("end"), "text": item.get("local_label") or item.get("text"), "entity_type": item.get("entity_type"), } for item in outcome.mappings ] rows.append( { "id": case["id"], "note": case["note"], "metrics": score_spans(predicted, case["expected"]), "request_local_hits": outcome.stats.request_local_hits, "detect_calls": outcome.stats.detect_calls, "detect_ms": elapsed, "coverage": outcome.stats.full_detection_coverage, "detect_tokens": outcome.stats.total_tokens, "detect_cost": outcome.stats.cost, "detect_model": outcome.stats.detect_model, } ) return { "mode": "live", "live_quality": "unconfirmed", "detect_model_unconfirmed": "openai/gpt-4.1-nano", "cases": rows, } def main() -> None: parser = argparse.ArgumentParser(description="Synthetic detect evaluation. No personal data.") parser.add_argument("--live", action="store_true", help="Call the configured detect provider.") args = parser.parse_args() payload = run_live() if args.live else run_fake() print(json.dumps(payload, ensure_ascii=False, indent=2)) print( "Live-Qualität: unbestätigt. Das konfigurierte openai/gpt-4.1-nano " "gilt durch reale False-Positive-Vorschläge nicht als zuverlässig bestätigt." ) if payload["mode"] != "live": print("Explizit live: python entity_detect_eval.py --live") if __name__ == "__main__": main()