97 lines
4.7 KiB
Python
97 lines
4.7 KiB
Python
"""Adversarial guard replay on real RAG contexts; no generative model calls.
|
|
|
|
These are acceptance/rejection checks with deliberately constructed answers,
|
|
not measured LLM accuracy. Known benchmark gold never enters production code.
|
|
"""
|
|
import argparse
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from .evaluate_verified_rag import DATASET, FIELDS, binding_matches, digest
|
|
from .rag_evidence import rag_evidence_context
|
|
from .verified_rag import generate_verified_rag, rag_answer_issues
|
|
from tools.runtime.report_files import report_output, resolve_report_input
|
|
|
|
|
|
def check_cases(dataset: dict, saved: dict) -> list[dict]:
|
|
results = []
|
|
for case in dataset['cases']:
|
|
payload = saved[case['id']]['retrieval_context']
|
|
context = rag_evidence_context(case['question'], payload)
|
|
negative = {'status':'no_answer', 'answer':'', 'claims':[], 'evidence_quotes':[], 'source_urls':[]}
|
|
|
|
def check(name, answer, should_accept):
|
|
issues = rag_answer_issues(answer, context)
|
|
# Exercise the production generator's two-attempt rejection and
|
|
# deterministic renderer, with a controlled local completion stub.
|
|
calls = []
|
|
def complete(messages):
|
|
calls.append(messages)
|
|
return {'choices':[{'message':{'content':json.dumps(answer, ensure_ascii=False)}}]}
|
|
delivered = generate_verified_rag(case['question'], payload, complete)
|
|
actual = delivered['accepted_for_delivery']
|
|
passed = actual == should_accept and bool(not issues) == should_accept
|
|
if not should_accept:
|
|
passed = passed and len(calls) == 2 and delivered['status'] == 'error'
|
|
results.append({'id':case['id'], 'check':name, 'passed':passed,
|
|
'expected_acceptance':should_accept, 'issues':issues})
|
|
|
|
if not case['should_answer']:
|
|
check('accept_supported_missing_fact_refusal', negative, True)
|
|
continue
|
|
bindings = case['expected_bindings']
|
|
if not bindings or not all(binding_matches(b, context['facts']) for b in bindings):
|
|
continue
|
|
claims = [{k:b[k] for k in FIELDS if k in b} for b in bindings]
|
|
for claim in claims:
|
|
matching = [row for row in context['facts'] if binding_matches(claim, [row])]
|
|
for field in ['title', 'year']:
|
|
values = {row[field] for row in matching if field in row}
|
|
if context['requirements'][field] and field not in claim and len(values) == 1:
|
|
claim[field] = next(iter(values))
|
|
urls = sorted({row['source_url'] for row in context['facts'] if any(
|
|
row['person'] == b['person'] for b in bindings)})
|
|
answer = {'status':'ok', 'answer':'', 'claims':claims, 'evidence_quotes':[], 'source_urls':urls}
|
|
check('accept_supported_tuple', answer, True)
|
|
check('reject_false_refusal', negative, False)
|
|
for field, value in [('year',9876), ('title','Nepodložený názov z cudzieho dokumentu'),
|
|
('person','Neexistujúca Osoba'), ('work_type','doctoral_thesis')]:
|
|
bad = copy.deepcopy(answer)
|
|
if field == 'work_type' and bad['claims'][0].get(field) == value:
|
|
value = 'bachelor_thesis'
|
|
bad['claims'][0][field] = value
|
|
check('reject_wrong_'+field, bad, False)
|
|
bad = copy.deepcopy(answer)
|
|
bad['source_urls'] = ['https://example.invalid/not-provided']
|
|
check('reject_wrong_citation', bad, False)
|
|
return results
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--input', required=True, type=Path)
|
|
parser.add_argument('--dataset', type=Path, default=DATASET)
|
|
parser.add_argument('--output', type=Path)
|
|
args = parser.parse_args()
|
|
source = resolve_report_input(args.input)
|
|
saved_report = json.loads(source.read_text())
|
|
dataset = json.loads(args.dataset.read_text())
|
|
assert saved_report['configuration']['dataset_sha256'] == digest(args.dataset)
|
|
saved = {r['id']:r for r in saved_report['results']}
|
|
results = check_cases(dataset, saved)
|
|
output = report_output(args.output, 'rag_guards')
|
|
summary = {'checks':len(results), 'passed':sum(r['passed'] for r in results),
|
|
'contexts':len(saved), 'llm_calls':0,
|
|
'note':'Controlled fault injection, not an LLM accuracy measurement.'}
|
|
output.write_text(json.dumps({'method':__doc__, 'input':str(source), 'input_sha256':digest(source),
|
|
'summary':summary, 'results':results}, ensure_ascii=False, indent=2)+'\n')
|
|
print(json.dumps(summary))
|
|
print('Report: '+str(output))
|
|
if any(not r['passed'] for r in results):
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|