Rifqi Hafizuddin
[NOTICKET] feat(knowledge_extraction): paid extraction stage + validate, diff, queue
ab5ea78
Raw
History Blame
4.03 kB
"""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"})