Spaces:
Sleeping
Sleeping
File size: 3,490 Bytes
57f6176 | 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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """Human-in-the-loop review logic for the Live Parser output.
Pure functions over the entity list (no Streamlit imports) so the corrected
highlighted text, the structured summary, the JSON export and the override
stats all derive from one editable source of truth.
A "work" item mirrors a model entity plus review bookkeeping:
{"_id", "type", "start", "end", "text", "conf",
"origin": "model"|"added", "orig_type": <type or None>}
"""
from __future__ import annotations
import re
def seed_review(entities):
"""Copy model entities into an editable working list with stable ids."""
work = []
for i, e in enumerate(entities):
work.append({
"_id": i,
"type": e["type"],
"start": e["start"],
"end": e["end"],
"text": e["text"],
"conf": e.get("conf", 1.0),
"origin": "model",
"orig_type": e["type"], # frozen original label, for the relabel diff
})
return work
def _next_id(work):
return max((e["_id"] for e in work), default=-1) + 1
def relabel(work, _id, new_type):
"""Change the label of the entity with id ``_id`` (in place)."""
for e in work:
if e["_id"] == _id:
e["type"] = new_type
return work
def delete(work, _id):
"""Remove the entity with id ``_id``."""
return [e for e in work if e["_id"] != _id]
def add_entity(work, phrase, full_text, etype, hint=0):
"""Tag ``phrase`` as a new entity, locating it in ``full_text``.
Returns (work, error). ``error`` is None on success. The phrase is located
near ``hint`` (the clicked char offset) first, then anywhere, so repeated
words resolve to the one the reviewer clicked.
"""
phrase = (phrase or "").strip()
if not phrase:
return work, "Nothing to add — the selection was empty."
idx = full_text.find(phrase, max(0, hint - len(phrase)))
if idx < 0:
idx = full_text.find(phrase)
if idx < 0:
return work, f"Couldn't find “{phrase}” in the CV text."
start, end = idx, idx + len(phrase)
for e in work:
if e["start"] == start and e["end"] == end:
return work, "That exact span is already tagged."
work = work + [{
"_id": _next_id(work),
"type": etype, "start": start, "end": end, "text": phrase,
"conf": 1.0, "origin": "added", "orig_type": None,
}]
return work, None
def diff_stats(original, corrected):
"""Compare model output (``original``) with human-corrected output.
``original`` is the seeded snapshot; ``corrected`` is the edited work list.
"""
corr_spans = {(e["start"], e["end"]) for e in corrected}
relabeled, added = [], []
for e in corrected:
if e.get("origin") == "added":
added.append(e)
elif e.get("orig_type") and e["type"] != e["orig_type"]:
relabeled.append(e)
deleted = [e for e in original if (e["start"], e["end"]) not in corr_spans]
n_model = len(original)
n_touched = len(relabeled) + len(deleted) # model entities the human changed
return {
"n_model": n_model,
"n_corrected": len(corrected),
"relabeled": relabeled,
"deleted": deleted,
"added": added,
"n_relabeled": len(relabeled),
"n_deleted": len(deleted),
"n_added": len(added),
"n_touched": n_touched,
"override_rate": (n_touched / n_model) if n_model else 0.0,
}
|