146 lines
5.4 KiB
Python
146 lines
5.4 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 tools.evaluation.evaluate_verified_rag import DATASET, FIELDS, binding_matches, digest
|
|
from tools.answers.rag_evidence import rag_evidence_context
|
|
from tools.answers.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()
|