Rifqi Hafizuddin
[NOTICKET] feat(knowledge_extraction): paid extraction stage + validate, diff, queue
ab5ea78 | """Verbatim span validation — the primary anti-hallucination control. | |
| Three rules that must not be relaxed: | |
| 1. **Normalise whitespace only.** Not case, not punctuation, not diacritics. | |
| Every additional normalisation is a hole a fabrication can fit through. | |
| 2. **On failure the FIELD becomes `None`** and the rejection is recorded, so a | |
| reviewer can see what the control caught rather than only what it let past. | |
| 3. **Never repair a failed span** by rewriting it to something that does match. | |
| A repaired span is an unfalsifiable claim, which is precisely what this | |
| control exists to prevent. | |
| If the provenance span itself is not verbatim, *every* guarded field on the | |
| entry is rejected: the entry's only link to evidence is broken, so nothing on it | |
| can be trusted. | |
| """ | |
| from __future__ import annotations | |
| import re | |
| import unicodedata | |
| from ..models import Branch, Chunk, RejectedField | |
| # Fields carrying a factual claim, and therefore span-guarded. | |
| GUARDED_FIELDS: dict[str, tuple[str, ...]] = { | |
| "glossary": ("definition", "full_name", "source_wording", "formula_latex", "interpretation"), | |
| "rule": ("statement", "condition", "consequence"), | |
| "formula": ("formula_latex",), | |
| # Generation, not extraction — it cannot be span-checked at all. | |
| "summary": (), | |
| } | |
| def normalise_ws(text: str) -> str: | |
| return re.sub(r"\s+", " ", unicodedata.normalize("NFKC", text)).strip() | |
| def span_present(span: str, source: str) -> bool: | |
| if not span or not span.strip(): | |
| return False | |
| return normalise_ws(span) in normalise_ws(source) | |
| def evidence_text(chunk_ids: list[str], chunks: list[Chunk]) -> str: | |
| """The text a span is validated against. | |
| Includes each chunk's HEADING as well as its body. The heading is part of | |
| the source document and is often where a term is formally named — the | |
| reference standard heads a section "Physical of Availability (PA)" while | |
| the body never repeats the phrase. Excluding it would reject a correct, | |
| verbatim quotation of the document's own section title, which is precisely | |
| the wording we are required to preserve. | |
| """ | |
| by_id = {c.chunk_id: c for c in chunks} | |
| parts: list[str] = [] | |
| for chunk_id in chunk_ids: | |
| chunk = by_id.get(chunk_id) | |
| if chunk is None: | |
| continue | |
| if chunk.heading: | |
| parts.append(chunk.heading) | |
| parts.append(chunk.text) | |
| return "\n".join(parts) | |
| def validate_entry( | |
| entry, branch: Branch, source_text: str, label: str | |
| ) -> tuple[object, list[RejectedField]]: | |
| """Returns `(entry, rejections)`; the entry is mutated in place. | |
| Also checks each guarded field's own value against the source where the | |
| field is expected to be quoted: `source_wording` and `full_name` are | |
| literal transcriptions, so a value that cannot be located is a silent | |
| normalisation — exactly the failure this pipeline is required to surface. | |
| """ | |
| rejections: list[RejectedField] = [] | |
| guarded = GUARDED_FIELDS.get(branch, ()) | |
| span_ok = span_present(entry.provenance.span, source_text) | |
| for field in guarded: | |
| value = getattr(entry, field, None) | |
| if value is None: | |
| continue | |
| if not span_ok: | |
| reason = "provenance.span not found verbatim in evidence" | |
| elif field in _TRANSCRIBED and not span_present(str(value), source_text): | |
| reason = f"{field} is not a verbatim transcription of the source" | |
| else: | |
| continue | |
| rejections.append( | |
| RejectedField( | |
| entry_term=label, | |
| field=field, | |
| offending_value=str(value)[:300], | |
| reason=reason, | |
| branch=branch, | |
| ) | |
| ) | |
| setattr(entry, field, None) | |
| return entry, rejections | |
| # Fields that claim to be copied from the document word for word. A definition | |
| # may legitimately be assembled across sentences; a "full name" may not. | |
| _TRANSCRIBED = frozenset({"full_name", "source_wording"}) | |