258 lines
8.9 KiB
Python
258 lines
8.9 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 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()
|