Spaces:
Runtime error
Runtime error
File size: 3,045 Bytes
aad7814 | 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 | """Validate and sanitize LLM narrative section output."""
from __future__ import annotations
import re
from typing import Literal
InterferenceLevel = Literal["medium", "maximum"]
_UNMATCHED_TAG_RE = re.compile(
r"\[UNMATCHED_OBSERVATION:\s*[^\]]*\]",
re.IGNORECASE,
)
_UNMATCHED_LINE_RE = re.compile(
r"^\s*UNMATCHED_OBSERVATION:\s*.+$",
re.IGNORECASE | re.MULTILINE,
)
_UNMATCHED_HEADING_BLOCK_RE = re.compile(
r"\n*#{1,6}\s*UNMATCHED_OBSERVATION\b.*",
re.IGNORECASE | re.DOTALL,
)
_GROUNDING_TAG_RE = re.compile(
r"\[Grounding\s+review\s+required[^\]]*\]",
re.IGNORECASE,
)
_SOURCE_TAG_RE = re.compile(
r"\[Source:\s[^\]]+\]",
re.IGNORECASE,
)
def _observation_signal(text: str, observations: list[str]) -> bool:
"""True when at least one note contributes a recognizable token to the output."""
lower = text.lower()
for obs in observations:
obs = (obs or "").strip()
if not obs:
continue
tokens = [t for t in re.findall(r"[a-z]{4,}", obs.lower()) if len(t) >= 4]
if any(t in lower for t in tokens[:6]):
return True
if len(obs) >= 12 and obs.lower()[:12] in lower:
return True
return False
def sanitize_section_prose(text: str) -> str:
"""Remove internal audit tags that must never appear in user-facing prose."""
cleaned = (text or "").strip()
if not cleaned:
return ""
cleaned = _UNMATCHED_TAG_RE.sub("", cleaned)
cleaned = _UNMATCHED_LINE_RE.sub("", cleaned)
cleaned = _GROUNDING_TAG_RE.sub("", cleaned)
cleaned = _SOURCE_TAG_RE.sub("", cleaned)
cleaned = _UNMATCHED_HEADING_BLOCK_RE.sub("", cleaned)
cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
return cleaned.strip()
def accept_narrative_section_output(
text: str,
observations: list[str],
) -> bool:
"""Accept polished narrative prose that integrates notes without internal tags."""
cleaned = sanitize_section_prose(text)
if not cleaned:
return False
if "•" in cleaned or "·" in cleaned:
return False
if re.search(r"UNMATCHED_OBSERVATION", cleaned, re.IGNORECASE):
return False
if re.search(r"\[Source:", cleaned, re.IGNORECASE):
return False
if re.search(r"\[Grounding\s+review", cleaned, re.IGNORECASE):
return False
if observations and not _observation_signal(cleaned, observations):
return False
return True
def accept_inplace_baseline_output(
text: str,
baseline: str,
observations: list[str],
) -> bool:
"""Backward-compatible alias — narrative validation (baseline retained for callers)."""
_ = baseline
return accept_narrative_section_output(text, observations)
def accept_llm_compose_output(
text: str,
reference: str,
observations: list[str],
*,
level: InterferenceLevel,
) -> bool:
"""Backward-compatible alias — all levels use narrative rules."""
_ = reference, level
return accept_narrative_section_output(text, observations)
|