Kansho/backend/entity_detect_eval.py

114 lines
3.6 KiB
Python

"""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.
"""
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))
CASES = (
{
"id": "common_noun",
"text": "Ich ging auf den Balkon und setzte mich.",
"expect_empty_types": True,
"note": "Allgemeines Substantiv, kein Eigenname.",
},
{
"id": "food_vs_person",
"text": "Ich aß Sushi. Sushi kam ins Wohnzimmer.",
"note": "Dasselbe Wort als Gericht und als mögliche Person.",
},
{
"id": "person",
"text": "Ich traf Anna am Nachmittag.",
"expect_types": {"PERSON"},
"note": "Klarer Personenname.",
},
{
"id": "project",
"text": "Ich arbeitete am privaten Projekt Aurora.",
"expect_types": {"PROJECT"},
"note": "Privates Projekt, kein Allerweltsgegenstand.",
},
{
"id": "place_org",
"text": "Ich war in Hamburg und sprach mit der Organisation Nordwerk.",
"expect_types": {"PLACE", "ORG"},
"note": "Ort und Organisation.",
},
)
def _summarize(entities: list[dict]) -> dict:
types = sorted({(item.get("entity_type") or "").upper() for item in entities})
return {
"count": len(entities),
"types": types,
"has_labels": False,
}
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), "mode": "fake"})
return {"mode": "fake", "live_quality": "unconfirmed", "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)
types = sorted({(item.get("entity_type") or "") for item in outcome.mappings})
rows.append(
{
"id": case["id"],
"note": case["note"],
"types": types,
"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,
}
)
return {"mode": "live", "live_quality": "ran", "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))
if payload["mode"] != "live":
print("Live-Qualität: noch nicht bestätigt. Explizit: python entity_detect_eval.py --live")
if __name__ == "__main__":
main()