dp-zp-agent/tools/answers/answer_coverage.py

136 lines
5.8 KiB
Python

"""Separate supported answer scope from the search's technical selection limits.
Supported means the delivered facts/excerpts have evidence; it never promises
exhaustive retrieval, a complete source document or absence of other records.
"""
import re
from tools.answers.context_policy import normalized
def response_scope(question: str, *, content: bool = False) -> str:
query = normalized(re.sub(r'[„“"«][^„“”"»]+[“”"»]', " ", question))
# Record lists differ from the numbered assignment of one selected work.
records = r"(?:prac\w*|projekt\w*|student\w*|osob\w*|autor\w*|tag\w*|kategori\w*|zdroj\w*)"
modifiers = (
r"(?:(?:vsetk\w*|diplomov\w*|bakalarsk\w*|zaverecn\w*|timov\w*|vyskumn\w*|a|aj)\s+)*"
)
if re.search(r"\b(?:kolko|pocet|spocitaj)\b", query):
return "count"
if re.search(
rf"\b(?:vsetk\w*|upln\w*|kompletn\w*)\s+{modifiers}{records}\b"
rf"|\b(?:ake|ktore|vypis|zoznam|prehlad)\s+{modifiers}{records}\b",
query,
):
return "list"
if content:
if re.search(r"\b(?:zadani\w*|ciel\w*)\b", query) and not re.search(
r"\b(?:strucn\w*|zhrn\w*)\b", query
):
return "assignment"
if re.search(
r"\b(?:comu|o com|opis\w*|popis\w*|obsah\w*|zamer\w*|zhrn\w*|strucn\w*)\b", query
):
return "overview"
return "fact"
def answer_coverage(answer: dict, context: dict) -> dict:
"""Describe only the scope of an already validated RAG selection."""
requirements = context.get("requirements", {})
scope = response_scope(context.get("question", ""), content=requirements.get("content", False))
reasons = list(context.get("truncation_reasons") or [])
selected = {row.get("source_id") for row in answer.get("evidence_quotes", [])}
cited = set(answer.get("source_urls", []))
clipped = any(
"[OBSAH ZDROJA OREZANÝ]" in row.get("body", "")
for row in context.get("source_blocks", [])
if row.get("source_id") in selected
or (
not selected and (row.get("source_url") in cited or answer.get("status") == "no_answer")
)
)
# A context budget may drop only unrelated sources. A clipping marker in
# the actual cited block is stronger evidence than this global reason.
context_limited = any(
reason in reasons
for reason in ["llm_context_budget", "answer_candidate_budget", "evidence_unit_limit"]
)
limited = context.get("complete") is not True or context.get("truncated") is True
limitations = [*reasons]
if clipped:
limitations.append("cited_source_text_clipped")
warnings = []
status = answer.get("status")
if status == "ok":
coverage = "supported"
if any(
requirements.get(key) and key not in fields
for fields in requirements.get("available_fields", {}).values()
for key in ["title", "year", "start_year"]
):
coverage = "partial"
warnings.append("partial_requested_fields")
if requirements.get("facets"):
from tools.answers.content_selection import missing_facets
if missing_facets(answer, context):
coverage = "partial"
warnings.append("partial_requested_aspects")
elif requirements.get("content"):
from tools.answers.rag_evidence import missing_content_people
if missing_content_people(answer, context):
coverage = "partial"
warnings.append("partial_requested_aspects")
if scope == "overview" and (clipped or context_limited):
coverage = "partial"
warnings.append("overview_aspects_clipped")
if scope == "list" and (limited or clipped):
coverage = "partial"
warnings.append("list_may_be_incomplete")
elif scope == "count" and (context.get("aggregation") or {}).get("complete") is not True:
coverage = "unknown"
warnings.append("count_not_established")
elif scope == "assignment" and clipped:
coverage = "partial"
warnings.append("assignment_text_clipped")
elif (
scope == "assignment"
and limited
and re.search(
r"\b(?:vsetk\w*|upln\w*|kompletn\w*|cele)\b",
normalized(context.get("question", "")),
)
):
coverage = "partial"
warnings.append("assignment_completeness_unknown")
elif status == "no_answer":
coverage = "not_found"
if clipped or context_limited:
warnings.append("missing_information_with_clipped_context")
else:
coverage = "unavailable"
return {
"scope": scope,
"status": coverage,
"warnings": warnings,
"limitations": list(dict.fromkeys(limitations)),
}
def coverage_notices(coverage: dict) -> list[str]:
messages = {
"overview_aspects_clipped": "Časť podkladov bola skrátená; prehľad preto nemusí pokrývať všetky oblasti práce.",
"list_may_be_incomplete": "Uvedený zoznam nemusí obsahovať všetky záznamy.",
"assignment_text_clipped": "Časť podkladov k zadaniu bola skrátená; uvedené body nemusia tvoriť celé zadanie.",
"assignment_completeness_unknown": "Z dostupných podkladov nemožno zaručiť všetky body zadania.",
"missing_information_with_clipped_context": "Časť podkladov bola pri príprave kontextu skrátená.",
"count_not_established": "Celkový počet sa z dostupných podkladov nedá spoľahlivo určiť.",
}
notices = [messages[reason] for reason in coverage["warnings"] if reason in messages]
if "list_may_be_incomplete" in coverage["warnings"]:
notices.append("Výpis môžeš zúžiť uvedením osoby, typu práce alebo konkrétneho roku.")
return notices