274 lines
13 KiB
Python
274 lines
13 KiB
Python
"""Cover the real comparison failures without relaxing factual evidence checks."""
|
|
|
|
import copy
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from test_rag_work_content import payload, QUESTION, URL
|
|
from test_quality_block import corpus, prepare, complete
|
|
from tools.answers.evidence_selection import resolve_selection, EvidenceSelectionError
|
|
from tools.answers.rag_evidence import rag_evidence_context
|
|
from tools.answers.verified_rag import generate_verified_rag, rag_answer_issues
|
|
from tools.answers.content_selection import fit_candidates
|
|
from tools.answers.question_intent import graph_needs_text
|
|
from tools.answers import answer_delivery
|
|
from tools.rag.rag_utils import build_context_text
|
|
|
|
GOALS = [
|
|
"1. Vypracujte prehľad metód ohodnotenia dokumentov.",
|
|
"2. Implementujte systém získavania informácií v slovenskom jazyku.",
|
|
"3. Navrhnite a vypracujte experimenty, v ktorých vyhodnotíte vybrané metódy.",
|
|
"4. Navrhnite možné zlepšenia presnosti vyhľadávania.",
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def four_goals(payload):
|
|
payload["sources"] = payload["sources"][:1]
|
|
payload["sources"][0]["text"] = (
|
|
"Sekcia: Diplomová práca 2021\nNázov: Vyhľadávanie informácií.\nZadanie:\n"
|
|
+ "\n".join(GOALS)
|
|
)
|
|
payload["context"] = build_context_text(payload["sources"])
|
|
return payload
|
|
|
|
|
|
def id_answer(context):
|
|
return {
|
|
"status": "ok",
|
|
"answer": "",
|
|
"claims": [],
|
|
"evidence_quotes": [],
|
|
"evidence_ids": [row["evidence_id"] for row in context["content_candidates"]],
|
|
"source_urls": [],
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("question", [QUESTION, "Aké je celé zadanie diplomovej práce Jána Holpa?"])
|
|
def test_all_four_goals_are_delivered_with_exact_evidence_and_automatic_citation(
|
|
four_goals, question
|
|
):
|
|
c = rag_evidence_context(question, four_goals)
|
|
a = id_answer(c)
|
|
assert len(a["evidence_ids"]) == 4
|
|
before = copy.deepcopy((a, c))
|
|
expanded = resolve_selection(a, c)
|
|
assert (a, c) == before and expanded["source_urls"] == [URL]
|
|
assert [q["quote"] for q in expanded["evidence_quotes"]] == GOALS
|
|
r = generate_verified_rag(question, four_goals, complete(a))
|
|
assert r["accepted_for_delivery"] and len(r["attempts"]) == 1
|
|
assert "3. Návrh a vypracovanie experimentov" in r["answer"]
|
|
assert "4. Návrh možných zlepšení presnosti vyhľadávania." in r["answer"]
|
|
if r["answer_coverage"]["scope"] == "assignment":
|
|
assert r["answer_coverage"]["status"] == "partial"
|
|
assert "assignment_completeness_unknown" in r["answer_coverage"]["warnings"]
|
|
else:
|
|
assert r["answer_coverage"]["status"] == "supported"
|
|
assert "Výpis je neúplný" not in r["answer"]
|
|
assert r["attempts"][0]["selected_answer"]["evidence_ids"] == a["evidence_ids"]
|
|
|
|
|
|
def test_two_of_four_goals_cannot_be_silently_accepted_as_an_overview(four_goals):
|
|
c = rag_evidence_context(QUESTION, four_goals)
|
|
a = id_answer(c)
|
|
a["evidence_ids"] = a["evidence_ids"][:2]
|
|
r = generate_verified_rag(QUESTION, four_goals, complete(a))
|
|
assert not r["accepted_for_delivery"] and len(r["attempts"]) == 2
|
|
assert all("missing_assignment_evidence" in attempt["issues"] for attempt in r["attempts"])
|
|
|
|
|
|
def test_unknown_id_is_rejected_in_both_attempts(four_goals):
|
|
a = id_answer(rag_evidence_context(QUESTION, four_goals))
|
|
a["evidence_ids"] = ["Eunknown"]
|
|
r = generate_verified_rag(QUESTION, four_goals, complete(a))
|
|
assert not r["accepted_for_delivery"]
|
|
assert all(attempt["issues"] == ["unknown_evidence_id"] for attempt in r["attempts"])
|
|
|
|
|
|
def test_ids_cannot_smuggle_additional_quotes_or_urls(four_goals):
|
|
c = rag_evidence_context(QUESTION, four_goals)
|
|
a = id_answer(c)
|
|
both = {**a, "evidence_quotes": [{"source_id": "S1", "quote": GOALS[0]}]}
|
|
with pytest.raises(EvidenceSelectionError, match="mixed_evidence_selection"):
|
|
resolve_selection(both, c)
|
|
a["source_urls"] = ["https://invented.test/result"]
|
|
r = generate_verified_rag(QUESTION, four_goals, complete(a))
|
|
assert not r["accepted_for_delivery"] and "unsupported_citation" in r["attempts"][-1]["issues"]
|
|
|
|
|
|
def test_duplicate_ids_do_not_duplicate_facts(four_goals):
|
|
c = rag_evidence_context(QUESTION, four_goals)
|
|
a = id_answer(c)
|
|
a["evidence_ids"] *= 2
|
|
resolved = resolve_selection(a, c)
|
|
assert len(resolved["evidence_quotes"]) == 4 and resolved["source_urls"] == [URL]
|
|
a["evidence_ids"].reverse()
|
|
assert [q["quote"] for q in resolve_selection(a, c)["evidence_quotes"]] == GOALS
|
|
|
|
|
|
def test_model_does_not_receive_duplicate_full_sources_or_need_to_copy_markdown(four_goals):
|
|
def select(messages):
|
|
c = json.loads(messages[1]["content"])["context"]
|
|
assert "content_passages" not in c and "context" not in c
|
|
assert all("text" not in row for row in c["sources"])
|
|
assert all(row["required"] for row in c["content_candidates"])
|
|
return complete(id_answer(c))(messages)
|
|
|
|
assert generate_verified_rag(QUESTION, four_goals, select)["accepted_for_delivery"]
|
|
|
|
|
|
def test_primary_goals_are_not_extended_with_the_task_backlog(four_goals):
|
|
four_goals["sources"][0][
|
|
"text"
|
|
] += "\nZásobník úloh:\n- Implementujte neskorší doplnkový modul."
|
|
four_goals["context"] = build_context_text(four_goals["sources"])
|
|
c = rag_evidence_context(QUESTION, four_goals)
|
|
assert len(c["content_candidates"]) == 4
|
|
r = generate_verified_rag(QUESTION, four_goals, complete(id_answer(c)))
|
|
assert r["accepted_for_delivery"] and "doplnkový modul" not in r["answer"]
|
|
|
|
|
|
@pytest.mark.parametrize("heading", ["Napr.", "Príklady:"])
|
|
def test_illustrative_examples_after_assignment_are_not_extra_mandatory_goals(four_goals, heading):
|
|
four_goals["sources"][0]["text"] += (
|
|
"\n\n" + heading + "\n\n- osnovy pre výuku na základnej škole."
|
|
)
|
|
four_goals["context"] = build_context_text(four_goals["sources"])
|
|
c = rag_evidence_context(QUESTION, four_goals)
|
|
assert len(c["content_candidates"]) == 4
|
|
r = generate_verified_rag(QUESTION, four_goals, complete(id_answer(c)))
|
|
assert r["accepted_for_delivery"] and "osnovy pre výuku" not in r["answer"]
|
|
|
|
|
|
def test_qualified_goal_is_kept_whole(four_goals):
|
|
text = "- Implementujte parser. Výsledok sa smie použiť iba na testovacom korpuse; nejde o overenie produkcie."
|
|
four_goals["sources"][0]["text"] += "\n" + text
|
|
four_goals["context"] = build_context_text(four_goals["sources"])
|
|
c = rag_evidence_context(QUESTION, four_goals)
|
|
r = generate_verified_rag(QUESTION, four_goals, complete(id_answer(c)))
|
|
assert r["accepted_for_delivery"]
|
|
assert "iba na testovacom korpuse; nejde o overenie produkcie" in r["answer"]
|
|
|
|
|
|
def test_budgeted_out_id_cannot_be_resolved_and_limited_overview_has_a_specific_notice(four_goals):
|
|
c = rag_evidence_context(QUESTION, four_goals)
|
|
original = c["content_candidates"][:]
|
|
sent = {"requirements": c["requirements"], "content_candidates": original}
|
|
fit_candidates(sent, c, 450, "cl100k_base")
|
|
dropped = [row for row in original if row not in c["content_candidates"]]
|
|
assert dropped and c["truncated"] and "answer_candidate_budget" in c["truncation_reasons"]
|
|
with pytest.raises(EvidenceSelectionError, match="unknown_evidence_id"):
|
|
resolve_selection({**id_answer(c), "evidence_ids": [dropped[0]["evidence_id"]]}, c)
|
|
from tools.answers.answer_coverage import answer_coverage, coverage_notices
|
|
|
|
a = resolve_selection(id_answer(c), c)
|
|
coverage = answer_coverage(a, c)
|
|
assert coverage["status"] == "partial" and "overview_aspects_clipped" in coverage["warnings"]
|
|
assert any(
|
|
"prehľad preto nemusí pokrývať všetky oblasti" in n for n in coverage_notices(coverage)
|
|
)
|
|
|
|
|
|
def test_an_indivisible_overlimit_goal_does_not_force_an_answer_the_model_never_received(
|
|
four_goals,
|
|
):
|
|
four_goals["sources"][0]["text"] = (
|
|
"Sekcia: Diplomová práca 2021\nZadanie:\n- Implementujte parser. "
|
|
+ ("Platí iba pre testovací korpus. " * 100)
|
|
)
|
|
four_goals["context"] = build_context_text(four_goals["sources"])
|
|
c = rag_evidence_context(QUESTION, four_goals)
|
|
assert not c["content_candidates"] and c["truncated"]
|
|
a = {"status": "no_answer", "claims": [], "evidence_quotes": [], "source_urls": []}
|
|
r = generate_verified_rag(QUESTION, four_goals, complete(a))
|
|
assert r["accepted_for_delivery"] and r["status"] == "no_answer"
|
|
assert "Časť podkladov bola pri príprave kontextu skrátená." in r["answer"]
|
|
|
|
|
|
@pytest.mark.parametrize("tail,want_year", [("", False), (" Uveď aj rok práce.", True)])
|
|
def test_year_filter_is_preserved_without_forcing_a_year_claim(corpus, tail, want_year):
|
|
q = "Aké problémy riešila Eva Nová pri diplomovej práci v roku 2024?" + tail
|
|
p = prepare(corpus, q)
|
|
c = rag_evidence_context(q, p)
|
|
assert c["requirements"]["years"] == [2024] and c["requirements"]["year"] == want_year
|
|
assert all(row["year"] == 2024 for row in c["content_candidates"])
|
|
if not want_year:
|
|
r = generate_verified_rag(q, p, complete(id_answer(c)))
|
|
assert r["accepted_for_delivery"] and r["claims"] == []
|
|
|
|
|
|
def test_comparison_cannot_replace_one_persons_evidence_with_another(corpus):
|
|
q = "Porovnaj výsledky Evy Novej a Borisa Nového."
|
|
p = prepare(corpus, q)
|
|
c = rag_evidence_context(q, p)
|
|
a = id_answer(c)
|
|
a["evidence_ids"] = [
|
|
row["evidence_id"] for row in c["content_candidates"] if row["person"] == "Eva Nová"
|
|
]
|
|
r = generate_verified_rag(q, p, complete(a))
|
|
assert (
|
|
not r["accepted_for_delivery"]
|
|
and "missing_progress_evidence" in r["attempts"][-1]["issues"]
|
|
)
|
|
|
|
|
|
def test_partial_refusal_gets_corrected_without_inventing_the_missing_persons_evidence(corpus):
|
|
db, _ = corpus
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute(
|
|
"UPDATE chunks SET text='Zadanie:\n- Implementujte parser.' WHERE document_path LIKE '%boris_novy%'"
|
|
)
|
|
q = "Aké problémy riešili Eva Nová a Boris Nový a aké experimenty mali pripraviť?"
|
|
p = prepare(corpus, q)
|
|
c = rag_evidence_context(q, p)
|
|
calls = []
|
|
|
|
def select(messages):
|
|
calls.append(copy.deepcopy(messages))
|
|
assert "Doložené časti po osobách" in messages[0]["content"]
|
|
assert (
|
|
"Eva Nová" not in messages[0]["content"] and "Boris Nový" not in messages[0]["content"]
|
|
)
|
|
return complete(
|
|
{"status": "no_answer", "claims": [], "evidence_quotes": [], "source_urls": []}
|
|
if len(calls) == 1
|
|
else id_answer(c)
|
|
)(messages)
|
|
|
|
r = generate_verified_rag(q, p, select)
|
|
assert (
|
|
r["accepted_for_delivery"]
|
|
and r["answer_coverage"]["status"] == "partial"
|
|
and len(calls) == 2
|
|
)
|
|
assert "rejected_supported_content_comparison" in r["attempts"][0]["issues"]
|
|
assert "Boris Nový" in r["answer"] and "chýbajú údaje o riešených problémoch" in r["answer"]
|
|
assert len(r["source_urls"]) == 2 and r["claims"] == []
|
|
|
|
|
|
def test_graph_narrative_request_uses_shared_plan_and_content_evidence(monkeypatch):
|
|
from contextlib import nullcontext
|
|
from tools.answers import question_plan
|
|
from tools.knowledge import query
|
|
from tools.knowledge.sources import profile_units
|
|
from tools.knowledge.planned_answers import answer_facets
|
|
names=['Ján Pták','Ján Malinovský']
|
|
catalog=[{'name':name,'kind':'Person','source_document':f'pages/students/2022/p{i}/README.md'} for i,name in enumerate(names)]
|
|
rows=[r for person in catalog for r in profile_units({'title':person['name'],'path':person['source_document'],
|
|
'content':'# Diplomový projekt 2026\nStav:\n- Vznikol problém s modelom.\nÚlohy:\n- Pripravte experiment.'},include_records=True)]
|
|
def forbidden(*args,**kwargs):pytest.fail('No metadata retrieval/model needed for directly planned content')
|
|
monkeypatch.setattr(question_plan,'planning_catalog',lambda db:catalog)
|
|
monkeypatch.setattr(answer_delivery,'index_snapshot',lambda *a,**kw:nullcontext())
|
|
monkeypatch.setattr(answer_delivery,'search_graph',forbidden)
|
|
def execute(q,*,engine,limit,plan,**kw):
|
|
assert plan.spec.intent=='facets' and engine=='graphrag'
|
|
return answer_facets(plan,{'index_version':{},'version':'v'},rows,engine=engine,limit=limit)
|
|
monkeypatch.setattr(query,'deliver_content',execute)
|
|
q='Aké problémy riešili Ján Pták a Ján Malinovský pri diplomovom projekte v roku 2026 a aké experimenty mali pripraviť?'
|
|
result=answer_delivery.deliver_answer(q,engine='graphrag',complete=forbidden)
|
|
assert result['status']=='ok' and result['accepted_for_delivery']
|
|
assert all(name in result['answer'] for name in names)
|
|
assert 'problém s modelom' in result['answer'] and 'experiment' in result['answer']
|