190 lines
9.6 KiB
Python
190 lines
9.6 KiB
Python
"""Answer guards using only actual retrieval context; no benchmark or oracle."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from html import unescape
|
|
from html.parser import HTMLParser
|
|
import unicodedata
|
|
|
|
|
|
def normalized(value) -> str:
|
|
text=unicodedata.normalize('NFKD',str(value or ''))
|
|
return ' '.join(re.findall(r'[a-z0-9]+',''.join(c for c in text if not unicodedata.combining(c)).casefold()))
|
|
|
|
|
|
class _HTMLLinks(HTMLParser):
|
|
def __init__(self):
|
|
super().__init__(convert_charrefs=True)
|
|
self.links = set()
|
|
|
|
def handle_starttag(self, tag, attrs):
|
|
self.links.update(value for key, value in attrs if key in {'href', 'src'} and value)
|
|
|
|
|
|
def text_urls(text: str) -> set[str]:
|
|
"""Inspect absolute, inline, reference-style Markdown and HTML links."""
|
|
text = unescape(text)
|
|
urls = set()
|
|
for url in re.findall(r'(?:[a-zA-Z][a-zA-Z0-9+.-]*://|www\.)[^\s<>"\x27`“”„«»‘’\]]+', text):
|
|
url = url.rstrip('.,;:!?')
|
|
while url.endswith(')') and url.count(')') > url.count('('):
|
|
url = url[:-1]
|
|
urls.add(url)
|
|
for match in re.finditer(r'\]\(\s*', text):
|
|
tail = text[match.end():]
|
|
if tail.startswith('<'):
|
|
urls.add(tail[1:].split('>', 1)[0]); continue
|
|
level = 0; end = 0
|
|
for end, char in enumerate(tail):
|
|
if char.isspace() or (char == ')' and level == 0): break
|
|
if char == '(': level += 1
|
|
if char == ')': level -= 1
|
|
else:
|
|
end = len(tail)
|
|
if tail[:end]: urls.add(tail[:end])
|
|
for match in re.finditer(r'^\s{0,3}\[[^]\n]+\]:\s*(?:<([^>]+)>|(\S+))', text, flags=re.M):
|
|
urls.add(match[1] or match[2])
|
|
parser = _HTMLLinks(); parser.feed(text)
|
|
return urls | parser.links
|
|
|
|
|
|
def citation_check(answer: dict, allowed: set[str]) -> dict:
|
|
declared=answer.get('source_urls',[])
|
|
if not isinstance(declared,list) or any(not isinstance(url,str) for url in declared):
|
|
return {'citations_ok':False,'unsupported_urls':[],'reason':'invalid_source_urls'}
|
|
used=set(declared)|text_urls(str(answer.get('answer','')))
|
|
unsupported=sorted(used-allowed)
|
|
return {'citations_ok':not unsupported,'unsupported_urls':unsupported}
|
|
|
|
|
|
def allowed_urls(context: dict) -> set[str]:
|
|
return {row['source_url'] for row in context.get('sources',[]) if row.get('source_url')}
|
|
|
|
|
|
def joined_candidates(facts: list[dict], intent: dict) -> list[dict]:
|
|
"""Join requested academic fields and tag by person, retaining real sources."""
|
|
needs_title = intent.get('mentions_title')
|
|
needs_year = intent.get('mentions_academic_year') or intent.get('years')
|
|
if not (needs_title or needs_year) or not (intent.get('mentions_keyword') or intent.get('keyword_phrase')):
|
|
return []
|
|
result=[]
|
|
for record in facts:
|
|
if (needs_title and not record.get('title')) or (not needs_title and record.get('year') is None):
|
|
continue
|
|
for tag in facts:
|
|
if normalized(tag.get('person'))!=normalized(record.get('person')) or not tag.get('keyword'):
|
|
continue
|
|
fields = ['person', 'title', 'year'] if needs_title else ['person', 'year']
|
|
candidate={k:record[k] for k in fields if record.get(k) is not None}
|
|
candidate['keyword']=tag['keyword']
|
|
candidate['evidence_documents']=sorted({p for p in [record.get('document_path'),tag.get('document_path')] if p})
|
|
candidate['source_urls']=sorted({u for u in [record.get('source_url'),tag.get('source_url')] if u})
|
|
if candidate not in result:result.append(candidate)
|
|
return result
|
|
|
|
|
|
def coverage_issues(answer: dict, context: dict) -> list[str]:
|
|
incomplete = context.get('truncated') is True or context.get('complete') is False
|
|
unknown = context.get('complete') is None
|
|
clauses = [normalized(part) for part in re.split(r'[.!?;\n]+', answer.get('answer', ''))]
|
|
caveat = any(re.search(r'(?:vypis|zoznam|vysled\w*|dokazy) (?:je |su |nie je |nie su )?(?:neupln\w*|orezan\w*)'
|
|
r'|nie (?:je|su) (?:to )?upln\w*|nemozno zarucit uplnost|nemusi byt upln', clause)
|
|
for clause in clauses)
|
|
issues = []
|
|
if incomplete and answer.get('status') == 'ok' and not caveat:
|
|
issues.append('missing_incompleteness_notice')
|
|
if incomplete or unknown:
|
|
for clause in clauses:
|
|
for match in re.finditer(r'(?:upln\w*|kompletn\w*) (?:zoznam|vypis|prehlad)'
|
|
r'|vset(?:ci|ky|kych) (?:student\w*|osob\w*|prac\w*)', clause):
|
|
prefix = clause[:match.start()]
|
|
# Negation must qualify this particular assertion. A caveat in
|
|
# another sentence never licenses a positive exhaustive claim.
|
|
if not re.search(r'(?:nie (?:je|su)(?: to)?|nejde o|neuvadzam|'
|
|
r'nemozno (?:zarucit|uviest)|nemozem (?:zarucit|poskytnut))\s*$', prefix):
|
|
issues.append('unsupported_exhaustive_statement')
|
|
return list(dict.fromkeys(issues))
|
|
|
|
|
|
def prose_issues(answer: dict) -> list[str]:
|
|
"""Conservative numeric consistency check; final prose is rendered from claims."""
|
|
text = str(answer.get('answer', ''))
|
|
for url in sorted(text_urls(text), key=len, reverse=True):
|
|
text = text.replace(url, '')
|
|
allowed_numbers = set()
|
|
for claim in answer.get('claims', []):
|
|
for key, value in claim.items():
|
|
if value is None: continue
|
|
if key in {'year', 'start_year'}: allowed_numbers.add(str(value))
|
|
else: text = re.sub(re.escape(str(value)), '', text, flags=re.I)
|
|
if answer.get('count') is not None: allowed_numbers.add(str(answer['count']))
|
|
unsupported = set(re.findall(r'\b\d+\b', text)) - allowed_numbers
|
|
return ['unsupported_prose_number'] if unsupported else []
|
|
|
|
|
|
def text_contains_value(text: str, value: str, *, person: bool = False) -> bool:
|
|
value = normalized(value); text = normalized(text)
|
|
if not person: return value in text
|
|
suffix = r'(?:a|u|ovi|om|e|ou|y|ovej|oveho|ym|emu)?'
|
|
pattern = r'\b' + r'\s+'.join(re.escape(token) + (suffix if token.isalpha() else '')
|
|
for token in value.split()) + r'\b'
|
|
return bool(re.search(pattern, text))
|
|
|
|
|
|
def answer_context_issues(answer: dict, context: dict, *, check_text: bool = True) -> list[str]:
|
|
issues=[]
|
|
if not citation_check(answer,allowed_urls(context))['citations_ok']:
|
|
issues.append('unsupported_citation')
|
|
facts=context.get('facts',[])
|
|
aggregation=context.get('aggregation') or {}
|
|
if answer.get('count') is not None:
|
|
if aggregation.get('complete') is not True:
|
|
issues.append('unsupported_total_count')
|
|
elif answer['count']!=aggregation.get('count'):
|
|
issues.append('incorrect_context_count')
|
|
claims = answer.get('claims', [])
|
|
cited = set(answer.get('source_urls', []))
|
|
if answer.get('status') == 'ok' and facts and not claims:
|
|
issues.append('missing_required_claims')
|
|
if claims and not cited:
|
|
issues.append('missing_required_citations')
|
|
if answer.get('status') == 'no_answer' and claims:
|
|
issues.append('claims_on_no_answer')
|
|
if check_text and not str(answer.get('answer', '')).strip():
|
|
issues.append('empty_answer_text')
|
|
for claim in answer.get('claims',[]):
|
|
person=normalized(claim.get('person'))
|
|
rows=[f for f in facts if normalized(f.get('person'))==person]
|
|
if not rows:
|
|
issues.append('unsupported_person');continue
|
|
for key in ['keyword','category','document_author','start_year']:
|
|
if claim.get(key) is not None and not any(normalized(row.get(key))==normalized(claim[key]) for row in rows):
|
|
issues.append('unsupported_'+key)
|
|
for key in ['title', 'year', 'start_year', 'keyword', 'category', 'document_author']:
|
|
if claim.get(key) is None: continue
|
|
evidence = {row.get('source_url') for row in rows
|
|
if normalized(row.get(key)) == normalized(claim[key]) and row.get('source_url')}
|
|
if evidence and not evidence & cited:
|
|
issues.append('missing_claim_source')
|
|
academic={k:claim[k] for k in ['title','year'] if claim.get(k) is not None}
|
|
if academic and not any(all(normalized(row.get(k))==normalized(v) for k,v in academic.items()) for row in rows):
|
|
issues.append('unsupported_academic_binding')
|
|
if claim.get('category') is not None and claim.get('document_author') is not None:
|
|
if not any(a.get('document_path') and a.get('document_path')==b.get('document_path')
|
|
for a in rows if normalized(a.get('category'))==normalized(claim['category'])
|
|
for b in rows if normalized(b.get('document_author'))==normalized(claim['document_author'])):
|
|
issues.append('disconnected_author_category')
|
|
candidates=context.get('answer_candidates',[])
|
|
if context.get('status')=='ok' and candidates:
|
|
if answer.get('status')=='no_answer':
|
|
issues.append('rejected_supported_person_join')
|
|
elif not all(any(all(normalized(claim.get(k))==normalized(candidate[k]) for k in ['person','title','year','keyword'] if k in candidate)
|
|
for claim in answer.get('claims',[])) for candidate in candidates):
|
|
issues.append('split_or_missing_person_join')
|
|
if context.get('status') in {'no_answer','ambiguous','error'} and answer.get('status')=='ok':
|
|
issues.append('unsupported_positive_status')
|
|
issues.extend(prose_issues(answer))
|
|
if check_text or str(answer.get('answer', '')).strip():
|
|
issues.extend(coverage_issues(answer,context))
|
|
return list(dict.fromkeys(issues))
|