| """Word-aware span decoding for the schema encoder's sigmoid span head (PII long-doc fix, 26 Sep). |
| |
| The v1 decoder (span_data.decode) thresholds every token x label independently and turns each above-threshold |
| run into a span. With sub-word tokens (the Vela tokenizer splits digits one by one) a single below-threshold |
| piece fragments one entity into several false positives ('ois', '0', 'example'), and every label is decoded |
| separately, so overlapping labels each cost a false positive. |
| |
| decode_v2(probs, offs, names, text, thr, ...) -> [dict(start, end, label)] (code-point offsets) |
| 1. label resolution: each token takes its argmax label if that probability > thr, else O |
| (overlapping labels resolved by max score; upstream Vela PII argmax-BIO likewise gives one label per token). |
| 2. word units from the text (GLiNER2 word splitter, see _UNIT): URL | run of letters/digits/@._-+ | any other |
| single non-space char; every CJK / kana / hangul / full-width character is its own unit (character level). |
| A token belongs to the unit that holds its first non-space character. |
| 3. within a unit, one label: the label with the largest summed probability over the unit's labelled tokens |
| wins and the other labelled tokens of the unit are relabelled to it. |
| 4. fill: unlabelled tokens BETWEEN two tokens of the same label inside one unit take that label; |
| extension (extend="unit", default): to the outermost letter/digit tokens of the unit, through inner |
| punctuation ('ois' -> 'lois123@x.example.org'), never onto a bare edge '[' / ']' / '.'; |
| extend="alnum": only through directly adjacent letter/digit tokens; extend=False: off. |
| 5. contiguous tokens of one label form a span (never joins different labels: upstream decode_entities rule); |
| whitespace is trimmed (upstream), and so are enclosing bracket / quote characters at the span edges. |
| 6. optional: bridge=True joins two same-label spans separated by one unit whose probability for that |
| label is > thr * bridge_frac; min_chars={label: n} drops shorter spans. |
| Upstream semantics kept: exact code-point spans, whitespace trim, different types never joined. Not |
| reproducible from a sigmoid head: the B- tag that splits two adjacent same-type entities (it has no B/I). |
| """ |
| import re |
| import numpy as np |
|
|
| _CJK = ("⺀-⿿ -〿-ヿ-ㇿ㐀-䶿一-鿿ꀀ-" |
| "가-豈-︰-﹏-") |
| |
| |
| |
| _UNIT = re.compile(rf"(?:https?://|www\.)[^\s\]\)\}}>「」『』\"'“”{_CJK}]+" |
| rf"|(?:(?![{_CJK}])[^\W_]|[@._\-+])+" |
| rf"|\S") |
| _ALNUM = re.compile(r"[^\W_]") |
| EDGE = set("[](){}<>「」『』【】《》〈〉\"'“”‘’«»‹›") |
|
|
|
|
| def units(text): |
| """char index -> unit id (-1 for whitespace).""" |
| u = np.full(len(text) + 1, -1, dtype=np.int64) |
| for k, m in enumerate(_UNIT.finditer(text)): |
| u[m.start():m.end()] = k |
| return u |
|
|
|
|
| def _first_nonspace(text, a, b): |
| while a < b and text[a].isspace(): |
| a += 1 |
| return a |
|
|
|
|
| def trim(text, s, e): |
| while s < e and (text[s].isspace() or text[s] in EDGE): |
| s += 1 |
| while e > s and (text[e - 1].isspace() or text[e - 1] in EDGE): |
| e -= 1 |
| return s, e |
|
|
|
|
| def decode_v2(probs, offs, names, text, thr, fill=True, extend="unit", unit_vote=True, trim_edges=True, |
| bridge=False, bridge_frac=0.5, min_chars=None, u=None): |
| probs = np.asarray(probs, dtype=np.float32); offs = np.asarray(offs) |
| T = len(offs) |
| if T == 0: |
| return [] |
| valid = offs[:, 1] > offs[:, 0] |
| k = probs.argmax(1); pk = probs[np.arange(T), k] |
| lab = np.where(valid & (pk > thr), k, -1) |
| u = units(text) if u is None else u |
| tu = np.full(T, -1, dtype=np.int64) |
| alnum = np.zeros(T, dtype=bool) |
| for t in range(T): |
| if not valid[t]: |
| continue |
| a, b = int(offs[t, 0]), int(offs[t, 1]) |
| f = _first_nonspace(text, a, b) |
| tu[t] = u[f] if f < b and f < len(text) else -1 |
| alnum[t] = bool(_ALNUM.search(text[a:b])) |
| |
| if unit_vote or fill or extend: |
| t = 0 |
| while t < T: |
| if tu[t] < 0: |
| t += 1; continue |
| t2 = t |
| while t2 + 1 < T and (tu[t2 + 1] == tu[t] or not valid[t2 + 1]): |
| t2 += 1 |
| idx = [i for i in range(t, t2 + 1) if valid[i]] |
| L = [i for i in idx if lab[i] >= 0] |
| if L: |
| if unit_vote and len({int(lab[i]) for i in L}) > 1: |
| sc = {} |
| for i in L: |
| sc[int(lab[i])] = sc.get(int(lab[i]), 0.0) + float(pk[i]) |
| w = max(sc, key=sc.get) |
| for i in L: |
| lab[i] = w |
| if fill: |
| for c in set(int(lab[i]) for i in L): |
| pos = [i for i in L if lab[i] == c] |
| for i in idx: |
| if pos[0] < i < pos[-1] and lab[i] < 0: |
| lab[i] = c |
| if extend: |
| for c in set(int(lab[i]) for i in idx if lab[i] >= 0): |
| pos = [idx.index(i) for i in idx if lab[i] == c] |
| if extend == "unit": |
| an = [j for j, i in enumerate(idx) if alnum[i]] |
| lo, hi = min([pos[0]] + an[:1]), max([pos[-1]] + an[-1:]) |
| if all(lab[idx[j]] < 0 or lab[idx[j]] == c for j in range(lo, hi + 1)): |
| for j in range(lo, hi + 1): |
| lab[idx[j]] = c |
| else: |
| j = pos[0] - 1 |
| while j >= 0 and lab[idx[j]] < 0 and alnum[idx[j]]: |
| lab[idx[j]] = c; j -= 1 |
| j = pos[-1] + 1 |
| while j < len(idx) and lab[idx[j]] < 0 and alnum[idx[j]]: |
| lab[idx[j]] = c; j += 1 |
| t = t2 + 1 |
| |
| spans = [] |
| cur = None |
| for t in range(T): |
| if not valid[t]: |
| continue |
| c = int(lab[t]) |
| if cur is not None and c == cur[0]: |
| cur[2] = int(offs[t, 1]); cur[3].append(t); continue |
| if cur is not None: |
| spans.append(cur) |
| cur = [c, int(offs[t, 0]), int(offs[t, 1]), [t]] if c >= 0 else None |
| if cur is not None: |
| spans.append(cur) |
| if bridge and len(spans) > 1: |
| out = [spans[0]] |
| for sp in spans[1:]: |
| pv = out[-1] |
| if sp[0] == pv[0]: |
| gap = [t for t in range(pv[3][-1] + 1, sp[3][0]) if valid[t]] |
| gu = {int(tu[t]) for t in gap if tu[t] >= 0} |
| if gap and len(gu) <= 1 and all(probs[t, sp[0]] > thr * bridge_frac for t in gap if tu[t] >= 0): |
| pv[2] = sp[2]; pv[3] = pv[3] + gap + sp[3]; continue |
| out.append(sp) |
| spans = out |
| res = [] |
| for c, s, e, _ in spans: |
| s, e = trim(text, s, e) if trim_edges else trim_ws(text, s, e) |
| if e <= s: |
| continue |
| if min_chars and e - s < min_chars.get(names[c], 0): |
| continue |
| res.append(dict(start=s, end=e, label=names[c])) |
| return res |
|
|
|
|
| def trim_ws(text, s, e): |
| while s < e and text[s].isspace(): |
| s += 1 |
| while e > s and text[e - 1].isspace(): |
| e -= 1 |
| return s, e |
|
|
|
|
| def decode_v1_trimmed(probs, offs, names, text, thr): |
| """The scorer's current rule (span_data.decode + whitespace trim), for before/after tables.""" |
| from span_data import decode |
| out = [] |
| for s in decode(probs, offs, names, thr): |
| a, b = trim_ws(text, s["start"], s["end"]) |
| if b > a: |
| out.append(dict(start=a, end=b, label=s["label"])) |
| return out |
|
|
|
|
| |
| |
| |
| GRID = [round(x, 2) for x in np.arange(0.05, 0.96, 0.05)] |
| DECODERS = { |
| "v1": lambda r, thr: decode_v1_trimmed(r["probs"], r["offs"], r["names"], r["text"], thr), |
| "v2": lambda r, thr: decode_v2(r["probs"], r["offs"], r["names"], r["text"], thr), |
| "v2_noext": lambda r, thr: decode_v2(r["probs"], r["offs"], r["names"], r["text"], thr, extend=False), |
| "v2_alnum": lambda r, thr: decode_v2(r["probs"], r["offs"], r["names"], r["text"], thr, extend="alnum"), |
| "v2_notrim": lambda r, thr: decode_v2(r["probs"], r["offs"], r["names"], r["text"], thr, trim_edges=False), |
| "v2_bridge": lambda r, thr: decode_v2(r["probs"], r["offs"], r["names"], r["text"], thr, bridge=True), |
| "v2_argmax_only": lambda r, thr: decode_v2(r["probs"], r["offs"], r["names"], r["text"], thr, fill=False, |
| extend=False, unit_vote=False, trim_edges=False), |
| } |
|
|
|
|
| def decode_rows(rows, thr, decoder="v2"): |
| f = DECODERS[decoder] if isinstance(decoder, str) else decoder |
| return [f(r, thr) for r in rows] |
|
|
|
|
| def _psd(span_data_path): |
| """Load the upstream semantic-router span_data.py (src/training/model_classifier/pii_model_fine_tuning_lora/span_data.py) |
| from span_data_path. Eval helper only (dev threshold picks); inference never calls it.""" |
| import importlib.util |
| from pathlib import Path |
| if span_data_path is None: |
| raise ValueError("pii_metrics needs span_data_path: the path of the upstream span_data.py") |
| spec = importlib.util.spec_from_file_location("pii_span_data_up", Path(span_data_path)) |
| m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m) |
| return m |
|
|
|
|
| PII_TYPES = ["AGE", "CREDIT_CARD", "DATE_TIME", "DOMAIN_NAME", "EMAIL_ADDRESS", "GPE", "IBAN_CODE", "IP_ADDRESS", "NRP", |
| "ORGANIZATION", "PERSON", "PHONE_NUMBER", "STREET_ADDRESS", "TITLE", "US_DRIVER_LICENSE", "US_SSN", "ZIP_CODE"] |
|
|
|
|
| def pii_metrics(rows, preds, span_data_path): |
| """Upstream Vela PII exact-entity metric (span_data.entity_metrics, whitespace-trimmed gold); span_data_path: see _psd.""" |
| P = _psd(span_data_path) |
| recs = [dict(full_text=r["text"], spans=[dict(start_position=s["start"], end_position=s["end"], entity_type=s["label"], |
| entity_value=r["text"][s["start"]:s["end"]]) for s in r["spans"]]) for r in rows] |
| return P.entity_metrics(recs, [{(s["label"], s["start"], s["end"]) for s in p} for p in preds], PII_TYPES) |
|
|
|
|
| def toxic_score(rows, preds): |
| from span_data import char_f1 |
| return float(np.mean([char_f1(p, r["spans"]) for r, p in zip(rows, preds)])) |
|
|
|
|
| def halu_metrics(rows, preds): |
| from train_schema_span import ragtruth_metrics_spans |
| return ragtruth_metrics_spans([dict(text=r["text"], spans=r["spans"]) for r in rows], preds) |
|
|
|
|
| def family_objective(family, span_data_path=None): |
| """Dev objective per span label family (what the threshold is chosen to maximise); span_data_path: PII only (see _psd).""" |
| if family == "pii": |
| return lambda rows, preds: pii_metrics(rows, preds, span_data_path)["micro"]["f1"] |
| if family == "toxic": |
| return toxic_score |
| if family == "halu": |
| return lambda rows, preds: halu_metrics(rows, preds)["example_f1"] |
| raise KeyError(family) |
|
|
|
|
| def pick_threshold(dev_sets, family, decoder="v2", grid=GRID, span_data_path=None): |
| """Choose one threshold per label family on DEV rows only. |
| dev_sets: list of row lists; the objective is the unweighted mean over the sets (for PII: dev short, dev 4096, |
| dev 8192 -> upstream's length-macro selection, so the dense long rows cannot dominate). Returns (thr, curve).""" |
| obj = family_objective(family, span_data_path) |
| curve = {t: float(np.mean([obj(rows, decode_rows(rows, t, decoder)) for rows in dev_sets])) for t in grid} |
| best = max(curve.values()) |
| thr = min(t for t, v in curve.items() if v >= best - 1e-12) |
| return thr, curve |
|
|
|
|
| if __name__ == "__main__": |
| text = "Le formulaire indique [pavel_200015@summit15.example.org] dans adresse courriel." |
| names = ["EMAIL_ADDRESS", "PERSON"] |
| import unicodedata |
| |
| offs = [(m.start(), m.end()) for m in re.finditer(r"\s?\w+|\s?[^\w\s]", text)] |
| p = np.zeros((len(offs), 2), dtype=np.float32) |
| s0 = text.index("pavel"); e0 = text.index("]") |
| for i, (a, b) in enumerate(offs): |
| if a >= s0 - 1 and b <= e0 + 1: |
| p[i, 0] = 0.9 if i % 3 else 0.2 |
| p[[i for i, (a, b) in enumerate(offs) if text[a:b].strip() == "["][0], 0] = 0.8 |
| out = decode_v2(p, np.array(offs), names, text, 0.5) |
| assert [(o["label"], text[o["start"]:o["end"]]) for o in out] == [("EMAIL_ADDRESS", "pavel_200015@summit15.example.org")], out |
| t2 = "表格在日期一栏填写了「2015-06-06」。" |
| print(units(t2)[:14], "ok") |
|
|
