generated from kgod/ai-review-template
215 lines
9.1 KiB
Python
215 lines
9.1 KiB
Python
"""Classify narrative facts and validate only objective anchors."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import TypedDict
|
|
|
|
from .experience_optimizer import normalize_fact_ledger
|
|
|
|
|
|
class FactRequirement(TypedDict, total=False):
|
|
id: str
|
|
text: str
|
|
reason: str
|
|
kind: str
|
|
|
|
|
|
_LATIN_TOKEN = re.compile(r"[A-Za-z][A-Za-z0-9+#._-]{1,}")
|
|
_COUNTED_OBJECT = re.compile(
|
|
r"(?P<number>\d+(?:\.\d+)?(?:\s*\u4e07)?\+?)\s*"
|
|
r"(?P<unit>\u540d|\u4f4d|\u4eba|\u4e2a|\u9879|\u6b21|\u53f0|\u6761|\u4efd|\u5b57|\u5bb6|\u5929|\u6708|\u5e74|"
|
|
r"\u5b66\u751f|\u7528\u6237|\u5ba2\u6237|\u8bf7\u6c42|\u670d\u52a1|\u6a21\u5757|\u529f\u80fd|"
|
|
r"students?|classmates?|users?|customers?|features?|services?|projects?|requests?)\s*"
|
|
r"(?P<object>[\u4e00-\u9fff]{0,10}|[A-Za-z][A-Za-z -]{0,24})",
|
|
re.I,
|
|
)
|
|
_RATIO = re.compile(r"(?:gpa\s*[:\uff1a]?\s*)?\d+(?:\.\d+)?\s*/\s*\d+(?:\.\d+)?", re.I)
|
|
_RANKING = re.compile(
|
|
r"(?:(?:\u4e13\u4e1a|\u5e74\u7ea7|\u73ed\u7ea7)?\u6392\u540d|\u4f4d\u5217|top)\s*"
|
|
r"(?:\u524d)?\s*(?:\u767e\u5206\u4e4b)?\s*(?P<value>\d+(?:\.\d+)?)\s*%?",
|
|
re.I,
|
|
)
|
|
_PERCENT_METRIC = re.compile(
|
|
r"(?P<object>[\u4e00-\u9fff]{2,10})\s*"
|
|
r"(?P<verb>\u63d0\u5347|\u589e\u957f|\u964d\u4f4e|\u51cf\u5c11|\u7f29\u77ed|\u4f18\u5316)\s*"
|
|
r"(?P<number>\d+(?:\.\d+)?%)"
|
|
)
|
|
_GENERIC_TERMS = frozenset({"api", "docx", "pdf"})
|
|
_COMMON_TECH_TERMS = frozenset({
|
|
"api", "aws", "azure", "docker", "docx", "elasticsearch", "fastapi", "figma",
|
|
"flask", "git", "golang", "java", "javascript", "kafka", "kubernetes", "langchain",
|
|
"langgraph", "linux", "mongodb", "mysql", "next.js", "nextjs", "node.js", "nodejs",
|
|
"numpy", "openai", "pandas", "pdf", "postgresql", "python", "pytorch", "rabbitmq",
|
|
"react", "redis", "spring", "sql", "tensorflow", "typescript", "vue", "vue3",
|
|
})
|
|
_LOW_INFORMATION_FACT = re.compile(
|
|
r"^(?:\u53c2\u4e0e|\u534f\u52a9|\u8d1f\u8d23|\u5b8c\u6210)?"
|
|
r"(?:\u65e5\u5e38|\u76f8\u5173|\u90e8\u5206|\u4e00\u4e9b)?"
|
|
r"(?:\u5de5\u4f5c|\u4efb\u52a1|\u4e8b\u9879|\u9879\u76ee)[\u3002\uff0c,;\uff1b\s]*$"
|
|
)
|
|
_LEAD_RESPONSIBILITY = re.compile(r"(?:\u4e3b\u5bfc|\u7275\u5934|\u72ec\u7acb\u8d1f\u8d23)")
|
|
_OWN_RESPONSIBILITY = re.compile(r"\u8d1f\u8d23")
|
|
_ASSIST_RESPONSIBILITY = re.compile(r"(?:\u534f\u52a9|\u914d\u5408|\u53c2\u4e0e)")
|
|
|
|
|
|
def classify_fact_requirements(
|
|
facts: list[dict[str, str]],
|
|
) -> tuple[list[FactRequirement], list[FactRequirement]]:
|
|
"""Return objective repair anchors and semantic first-pass coverage targets."""
|
|
ledger = normalize_fact_ledger(facts)
|
|
split_parents = {
|
|
fact["id"].rsplit("_part_", 1)[0]
|
|
for fact in ledger
|
|
if fact.get("field") == "description_part"
|
|
}
|
|
candidates = [
|
|
fact
|
|
for fact in ledger
|
|
if fact["id"] not in split_parents
|
|
and (
|
|
fact.get("field") in {"description", "description_part", "highlight"}
|
|
or fact.get("source") == "user_answer"
|
|
)
|
|
]
|
|
hard: list[FactRequirement] = []
|
|
coverage: list[FactRequirement] = []
|
|
seen_hard: set[tuple[str, str]] = set()
|
|
for fact in candidates:
|
|
coverage.append({"id": fact["id"], "text": fact["text"]})
|
|
hard.extend(_objective_anchors(fact, seen_hard))
|
|
return hard, coverage
|
|
|
|
|
|
def missing_hard_facts(
|
|
hard_facts: list[FactRequirement], narrative: str
|
|
) -> list[str]:
|
|
return [fact["text"] for fact in hard_facts if not hard_fact_is_preserved(fact, narrative)]
|
|
|
|
|
|
def semantic_coverage_is_low(
|
|
coverage_targets: list[FactRequirement], covered_fact_ids: list[str] | None
|
|
) -> bool:
|
|
"""Repair only when the model declares widespread semantic omission."""
|
|
target_ids = {fact["id"] for fact in coverage_targets}
|
|
if covered_fact_ids is None or len(target_ids) < 3:
|
|
return False
|
|
covered = target_ids.intersection(str(item).strip() for item in (covered_fact_ids or []))
|
|
return len(covered) / len(target_ids) < 0.70
|
|
|
|
|
|
def missing_semantic_fact_ids(
|
|
coverage_targets: list[FactRequirement], covered_fact_ids: list[str] | None
|
|
) -> list[str]:
|
|
covered = {str(item).strip() for item in (covered_fact_ids or [])}
|
|
return [fact["id"] for fact in coverage_targets if fact["id"] not in covered]
|
|
|
|
|
|
def hard_fact_is_preserved(fact: FactRequirement, narrative: str) -> bool:
|
|
"""Validate deterministic anchors while allowing prose to be freely rewritten."""
|
|
kind = str(fact.get("kind") or "")
|
|
source = str(fact.get("text") or "").strip()
|
|
if kind == "named_term":
|
|
return source.casefold() in {
|
|
term.casefold().rstrip(".,;:!?") for term in _LATIN_TOKEN.findall(narrative)
|
|
}
|
|
if kind == "responsibility":
|
|
return _responsibility_level(narrative) == source
|
|
if kind == "quantity":
|
|
return _quantity_anchor_is_preserved(source, narrative)
|
|
if kind == "percent_metric":
|
|
return _normalize_literal(source) in _normalize_literal(narrative)
|
|
if kind == "literal":
|
|
return _normalize_literal(source) in _normalize_literal(narrative)
|
|
return False
|
|
|
|
|
|
def _objective_anchors(
|
|
fact: dict[str, str], seen: set[tuple[str, str]] | None = None
|
|
) -> list[FactRequirement]:
|
|
text = str(fact.get("text") or "").strip()
|
|
if not text or _LOW_INFORMATION_FACT.fullmatch(text):
|
|
return []
|
|
prefix = str(fact["id"])
|
|
anchors: list[FactRequirement] = []
|
|
seen = seen if seen is not None else set()
|
|
for index, match in enumerate(_COUNTED_OBJECT.finditer(text), start=1):
|
|
_append_anchor(anchors, seen, f"{prefix}:quantity:{index}", match.group(0).strip(), "quantified_fact", "quantity")
|
|
for index, match in enumerate(_RATIO.finditer(text), start=1):
|
|
_append_anchor(anchors, seen, f"{prefix}:ratio:{index}", match.group(0).strip(), "ratio_or_gpa", "literal")
|
|
for index, match in enumerate(_RANKING.finditer(text), start=1):
|
|
_append_anchor(anchors, seen, f"{prefix}:ranking:{index}", f"top{match.group('value')}", "ranking", "literal")
|
|
for index, match in enumerate(_PERCENT_METRIC.finditer(text), start=1):
|
|
_append_anchor(anchors, seen, f"{prefix}:percent:{index}", match.group(0).strip(), "percent_metric", "percent_metric")
|
|
for index, term in enumerate(sorted(_named_terms(text)), start=1):
|
|
_append_anchor(anchors, seen, f"{prefix}:term:{index}", term, "named_tool_or_term", "named_term")
|
|
level = _responsibility_level(text)
|
|
if level:
|
|
_append_anchor(anchors, seen, f"{prefix}:responsibility", level, "responsibility_level", "responsibility")
|
|
return anchors
|
|
|
|
|
|
def _append_anchor(
|
|
anchors: list[FactRequirement], seen: set[tuple[str, str]], identifier: str,
|
|
text: str, reason: str, kind: str,
|
|
) -> None:
|
|
key = (kind, text.casefold())
|
|
if text and key not in seen:
|
|
seen.add(key)
|
|
anchors.append({"id": identifier, "text": text, "reason": reason, "kind": kind})
|
|
|
|
|
|
def _named_terms(text: str) -> set[str]:
|
|
terms: set[str] = set()
|
|
for token in _LATIN_TOKEN.findall(text):
|
|
normalized = token.casefold().rstrip(".,;:!?")
|
|
if normalized in _GENERIC_TERMS:
|
|
continue
|
|
if (
|
|
normalized in _COMMON_TECH_TERMS
|
|
or any(character.isdigit() or character in "+#._/-" for character in normalized)
|
|
or any(character.isupper() for character in token[1:])
|
|
):
|
|
terms.add(normalized)
|
|
return terms
|
|
|
|
|
|
def _responsibility_level(text: str) -> str | None:
|
|
if _LEAD_RESPONSIBILITY.search(text):
|
|
return "lead"
|
|
if _ASSIST_RESPONSIBILITY.search(text):
|
|
return "assist"
|
|
if _OWN_RESPONSIBILITY.search(text):
|
|
return "own"
|
|
return None
|
|
|
|
|
|
def _quantity_anchor_is_preserved(source: str, narrative: str) -> bool:
|
|
source_match = _COUNTED_OBJECT.search(source)
|
|
if source_match is None:
|
|
return False
|
|
source_number, source_unit, source_object = _normalized_binding(source_match)
|
|
for target_match in _COUNTED_OBJECT.finditer(narrative):
|
|
target_number, target_unit, target_object = _normalized_binding(target_match)
|
|
if (source_number, source_unit) != (target_number, target_unit):
|
|
continue
|
|
if not source_object or not target_object:
|
|
return True
|
|
if source_object in target_object or target_object in source_object:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _normalized_binding(match: re.Match[str]) -> tuple[str, str, str]:
|
|
unit = match.group("unit").casefold()
|
|
people_units = {"\u540d", "\u4f4d", "\u4eba", "\u5b66\u751f", "\u7528\u6237", "\u5ba2\u6237", "student", "students", "classmate", "classmates", "user", "users", "customer", "customers"}
|
|
if unit in people_units:
|
|
unit = "people"
|
|
return match.group("number").casefold().replace(" ", ""), unit, match.group("object").strip()
|
|
|
|
|
|
def _normalize_literal(value: str) -> str:
|
|
normalized = value.casefold().replace("\u767e\u5206\u4e4b", "").replace("top", "top")
|
|
normalized = re.sub(r"(?:\u6392\u540d|\u4e13\u4e1a\u6392\u540d|\u5e74\u7ea7\u6392\u540d|\u73ed\u7ea7\u6392\u540d|\u4f4d\u5217)?\s*\u524d\s*(\d+(?:\.\d+)?)\s*%?", r"top\1", normalized)
|
|
normalized = re.sub(r"top\s*(\d+(?:\.\d+)?)\s*%?", r"top\1", normalized)
|
|
return re.sub(r"[\s:\uff1a,\uff0c\u3002\uff1b;]", "", normalized) |