100 lines
6.2 KiB
Python
100 lines
6.2 KiB
Python
"""Live synthetic functional checks; separate from corpus accuracy metrics."""
|
|
import argparse
|
|
import concurrent.futures
|
|
import copy
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from tools.rag.evaluation.rag_runner import request_json, OPENWEBUI_URL, DEFAULT_MODEL
|
|
from tools.graphrag.evaluation.reproducibility import sha256_file
|
|
from .checked_generation import generate_checked_answer
|
|
from .context_policy import joined_candidates
|
|
|
|
HERE = Path(__file__).parent
|
|
|
|
|
|
def scenarios():
|
|
base = {'status': 'ok', 'reason': None, 'complete': True, 'truncated': False,
|
|
'truncation_reasons': [], 'index_version': {'version': 'synthetic-functional-fixture'},
|
|
'sources': [{'source_id': 'S1', 'document_path': 'a.md', 'source_url': 'https://example.test/a'}],
|
|
'facts': [{'person': 'Eva Vzorová', 'title': 'Vyhľadávanie informácií', 'year': 2026,
|
|
'document_path': 'a.md', 'source_url': 'https://example.test/a'}],
|
|
'aggregation': None, 'answer_candidates': []}
|
|
query = 'Uveď názov práce osoby Eva Vzorová a rok práce.'
|
|
result = []
|
|
for name, fields in [
|
|
('complete', {}), ('incomplete_without_truncated', {'complete': False}),
|
|
('truncated', {'complete': False, 'truncated': True, 'truncation_reasons': ['limit']}),
|
|
('unknown_completeness', {'complete': None})]:
|
|
context = copy.deepcopy(base); context.update(fields)
|
|
result.append({'id': name, 'question': query, 'context': context,
|
|
'expected': {'status': 'ok', 'claims': [{'person': 'Eva Vzorová',
|
|
'title': 'Vyhľadávanie informácií', 'year': 2026}]}})
|
|
context = copy.deepcopy(base)
|
|
context.update(complete=False, truncated=True, truncation_reasons=['evidence_limit'],
|
|
aggregation={'count': 2, 'complete': True})
|
|
result.append({'id': 'exact_count_partial_evidence', 'question': 'Aký je počet vyhovujúcich osôb?',
|
|
'context': context, 'expected': {'status': 'ok', 'count': 2}})
|
|
context = copy.deepcopy(base)
|
|
context['sources'].append({'source_id': 'S2', 'document_path': 'child.md', 'source_url': 'https://example.test/child'})
|
|
context['facts'].append({'person': 'Eva Vzorová', 'keyword': 'python',
|
|
'document_path': 'child.md', 'source_url': 'https://example.test/child'})
|
|
context['answer_candidates'] = joined_candidates(context['facts'], {'mentions_title': True, 'mentions_keyword': True})
|
|
result.append({'id': 'same_person_child_tag',
|
|
'question': 'Uveď názov práce osoby Eva Vzorová v roku 2026 s kľúčovým slovom python.',
|
|
'context': context, 'expected': {'status': 'ok', 'claims': [
|
|
{'person': 'Eva Vzorová', 'title': 'Vyhľadávanie informácií', 'year': 2026, 'keyword': 'python'}]}})
|
|
context = copy.deepcopy(context)
|
|
context['facts'][0].pop('title')
|
|
context['answer_candidates'] = joined_candidates(context['facts'], {'years': [2026], 'mentions_keyword': True})
|
|
result.append({'id': 'same_person_year_child_tag',
|
|
'question': 'Ktorá osoba spĺňa kľúčové slovo python a rok práce 2026?',
|
|
'context': context, 'expected': {'status': 'ok', 'claims': [
|
|
{'person': 'Eva Vzorová', 'year': 2026, 'keyword': 'python'}]}})
|
|
result.append({'id': 'type_is_not_keyword', 'question': 'Uveď názov diplomového projektu osoby Eva Vzorová a rok.',
|
|
'context': copy.deepcopy(base), 'expected': {'status': 'ok', 'no_keyword': True}})
|
|
context = copy.deepcopy(base)
|
|
context.update(status='no_answer', reason='unknown_person', facts=[], sources=[])
|
|
result.append({'id': 'no_answer', 'question': 'Akú prácu má Neexistujúca Osoba?',
|
|
'context': context, 'expected': {'status': 'no_answer'}})
|
|
return result
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--output', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
if args.output.exists():
|
|
parser.error('Refusing to overwrite results.')
|
|
files = [Path(__file__), HERE/'graph_answer_prompt.txt', HERE/'context_policy.py', HERE/'checked_generation.py', HERE/'answer_format.py', HERE/'render_answer.py']
|
|
hashes = {str(p): sha256_file(p) for p in files}
|
|
declared = scenarios()
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.with_suffix('.declaration.json').write_text(json.dumps(
|
|
{'method': __doc__, 'model': DEFAULT_MODEL, 'hashes': hashes, 'scenarios': declared}, ensure_ascii=False, indent=2)+'\n')
|
|
prompt = (HERE/'graph_answer_prompt.txt').read_text()
|
|
def run(case):
|
|
def complete(messages):
|
|
return request_json(OPENWEBUI_URL, method='POST', timeout=90, max_attempts=2,
|
|
headers={'Authorization': 'Bearer '+os.environ['OPENWEBUI_API_KEY'], 'Content-Type': 'application/json'},
|
|
payload={'model': DEFAULT_MODEL, 'temperature': 0, 'stream': False,
|
|
'response_format': {'type': 'json_object'}, 'messages': messages})
|
|
out = generate_checked_answer(case['question'], case['context'], prompt, complete)
|
|
answer = out.get('answer', {}); expected = case['expected']
|
|
ok = answer.get('status') == expected['status']
|
|
for binding in expected.get('claims', []):
|
|
ok &= any(all(claim.get(k) == v for k, v in binding.items()) for claim in answer.get('claims', []))
|
|
if 'count' in expected: ok &= answer.get('count') == expected['count']
|
|
if expected.get('no_keyword'): ok &= not any(c.get('keyword') for c in answer.get('claims', []))
|
|
return {'id': case['id'], **out, 'passed': bool(ok and out['accepted_for_delivery'])}
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
|
|
rows = list(executor.map(run, declared))
|
|
assert hashes == {str(p): sha256_file(p) for p in files}
|
|
output = {'method': __doc__, 'questions': rows, 'summary': {'total': len(rows), 'passed': sum(r['passed'] for r in rows)}}
|
|
args.output.write_text(json.dumps(output, ensure_ascii=False, indent=2)+'\n')
|
|
print(json.dumps(output['summary']))
|
|
|
|
|
|
if __name__ == '__main__': main()
|