110 lines
5.9 KiB
Python
110 lines
5.9 KiB
Python
"""Answer guards using only actual retrieval context; no benchmark or oracle."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
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()))
|
|
|
|
|
|
def text_urls(text: str) -> set[str]:
|
|
urls = re.findall(r'(?:[a-zA-Z][a-zA-Z0-9+.-]*://|www\.)[^\s<>"\x27`“”„«»‘’\]\)]+', text)
|
|
# Relative Markdown targets are citations too, and must be supplied by
|
|
# the tool just like absolute links. Never resolve/guess another URL.
|
|
urls += re.findall(r'\[[^\]]*\]\(\s*<?([^\s)>]+)>?', text)
|
|
return {url.rstrip('.,;:!?') for url in urls}
|
|
|
|
|
|
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
|
|
text=normalized(answer.get('answer',''))
|
|
caveat=any(phrase in text for phrase in ['vysledok je neuplny','vysledky su neuplne','vypis je neuplny',
|
|
'zoznam je neuplny','dokazy su orezane','vysledok je orezany','vysledky su orezane',
|
|
'nemusi byt uplny','nemusi byt uplna','nie je uplny','nie su uplne','nemozno zarucit uplnost'])
|
|
issues=[]
|
|
if incomplete and answer.get('status')=='ok' and not caveat:
|
|
issues.append('missing_incompleteness_notice')
|
|
exhaustive=any(phrase in text for phrase in ['uplny zoznam','kompletny zoznam','vsetci studenti','vsetky osoby','vsetky prace'])
|
|
if (incomplete or unknown) and exhaustive and not caveat:
|
|
issues.append('unsupported_exhaustive_statement')
|
|
return issues
|
|
|
|
|
|
def answer_context_issues(answer: dict, context: dict) -> 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 and aggregation:
|
|
if aggregation.get('complete') is not True:
|
|
issues.append('unsupported_total_count')
|
|
elif answer['count']!=aggregation.get('count'):
|
|
issues.append('incorrect_context_count')
|
|
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)
|
|
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(coverage_issues(answer,context))
|
|
return list(dict.fromkeys(issues))
|