263 lines
13 KiB
Python
263 lines
13 KiB
Python
"""Behavioral checks for scope, relevant evidence, partial answers and failures."""
|
|
import copy
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from tools.answer_evaluation.content_selection import candidates, facets_for
|
|
from tools.answer_evaluation.rag_evidence import rag_evidence_context
|
|
from tools.answer_evaluation.rag_preparation import prepare_named_content
|
|
from tools.answer_evaluation.verified_rag import generate_verified_rag, rag_answer_issues
|
|
from tools.answers.answer_errors import public_failure
|
|
from tools.graphrag import graph_search as gs
|
|
from tools.graphrag.tests.test_graph_search import FakeSession
|
|
from tools.rag.llm_context import compact_rag_payload
|
|
from tools.rag.rag_utils import build_context_text
|
|
from tools.retrieval.question_constraints import unsupported_constraint, requested_years
|
|
|
|
|
|
@pytest.fixture
|
|
def corpus(tmp_path):
|
|
db=tmp_path/'index.sqlite'; sources=[]
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute('CREATE TABLE documents(path,title,author,published)')
|
|
conn.execute('CREATE TABLE chunks(document_path,text,heading_paths_json,published,chunk_index)')
|
|
for i,(person,slug,kind,year,text) in enumerate([
|
|
('Eva Nová','eva_nova','Diplomová práca',2024,
|
|
'Zadanie:\n1. Implementujte systém vyhľadávania.\nStav:\n- Konzultácia štruktúry práce.\n'
|
|
'- Natrénovaný model BERT a vykonané vyhodnotenie presnosti.\n'
|
|
'- Vznikol problém s tokenizérom.\nÚlohy:\n- Pripravte experiment s rôznymi tokenizérmi.'),
|
|
('Boris Nový','boris_novy','Bakalárska práca',2023,
|
|
'Zadanie:\n- Implementujte parser.\nStav:\n- Implementovaný parser pomocou Python.\n'
|
|
'- Vznikla chyba pri načítaní PDF.\nÚlohy:\n- Pripravte experiment s presnosťou extrakcie.')],1):
|
|
path=f'pages/students/2020/{slug}/README.md'
|
|
conn.execute('INSERT INTO documents VALUES(?,?,?,?)',(path,person,'Daniel Hladek',1))
|
|
conn.execute('INSERT INTO chunks VALUES(?,?,?,?,?)',(path,text,json.dumps([[person,f'{kind} {year}']]),1,0))
|
|
sources.append({'source_id':f'S{i}','title':person,'document_path':path,'published':True,
|
|
'source_url':f'https://example.test/{slug}',
|
|
'text':f'Sekcia: {kind} {year}\nNázov: Ukážkový názov.'})
|
|
raw={'status':'ok','reason':None,'sources':sources,'context':build_context_text(sources),
|
|
'complete':False,'truncated':True,'truncation_reasons':['result_limit'],'index_version':{'version':'fixed'}}
|
|
return db,raw
|
|
|
|
|
|
def prepare(corpus,question):
|
|
db,raw=corpus
|
|
return compact_rag_payload(prepare_named_content(db,question,raw))
|
|
|
|
|
|
def model_selection(question,payload):
|
|
c=rag_evidence_context(question,payload)
|
|
return {'status':'ok','answer':'','claims':[],
|
|
'evidence_quotes':[{'source_id':x['source_id'],'quote':x['text']} for x in c.get('content_candidates',[])],
|
|
'source_urls':[s['source_url'] for s in c['sources']]}
|
|
|
|
|
|
def complete(answer):
|
|
return lambda _: {'choices':[{'message':{'content':json.dumps(answer)}}]}
|
|
|
|
|
|
@pytest.mark.parametrize('question,facets',[
|
|
('Aké problémy riešili Eva Nová a Boris Nový a aké experimenty mali pripraviť?',{'problems','experiments'}),
|
|
('Aké metódy používa Eva Nová?',{'methods'}),
|
|
('Aké sú dosiahnuté výsledky Evy Novej?',{'results'}),
|
|
])
|
|
def test_content_without_explicit_work_type(corpus,question,facets):
|
|
payload=prepare(corpus,question); c=rag_evidence_context(question,payload)
|
|
assert c['requirements']['content'] and set(c['requirements']['facets'])==facets
|
|
assert c['content_candidates']
|
|
|
|
|
|
def test_multiple_people_and_aspects_remain_bound(corpus):
|
|
q='Aké problémy riešili Eva Nová a Boris Nový a aké experimenty mali pripraviť?'
|
|
payload=prepare(corpus,q); a=model_selection(q,payload)
|
|
r=generate_verified_rag(q,payload,complete(a))
|
|
assert r['accepted_for_delivery'] and r['status']=='ok'
|
|
text=r['answer']; eva,boris=text.split('Boris Nový — bakalárska práca')
|
|
assert 'tokenizérom' in eva and 'PDF' not in eva
|
|
assert 'PDF' in boris and 'tokenizérmi' not in boris
|
|
assert text.count('**Riešené problémy:**')==2
|
|
assert text.count('**Plánované experimenty a vyhodnotenie:**')==2
|
|
|
|
|
|
def test_ambiguous_work_is_a_clarification_without_model(corpus):
|
|
db,raw=corpus; source=raw['sources'][0]
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute('INSERT INTO chunks VALUES(?,?,?,?,?)',(source['document_path'],
|
|
'Stav:\nVznikol iný problém.',json.dumps([['Eva Nová','Bakalárska práca 2022']]),1,1))
|
|
q='Aké problémy mala Eva Nová počas práce?'
|
|
p=prepare(corpus,q)
|
|
r=generate_verified_rag(q,p,lambda _:pytest.fail('No model call for a clarification'))
|
|
assert r['status']=='ambiguous' and r['reason']=='ambiguous_work'
|
|
assert '2022' in r['answer'] and '2024' in r['answer']
|
|
|
|
|
|
def test_explicit_type_and_year_resolve_scope(corpus):
|
|
db,raw=corpus
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute('INSERT INTO chunks VALUES(?,?,?,?,?)',(raw['sources'][0]['document_path'],
|
|
'Stav:\nChyba v inom projekte Cassandra.',json.dumps([['Eva Nová','Diplomový projekt 2022']]),1,1))
|
|
q='Aké problémy mala Eva Nová pri diplomovej práci v roku 2024?'
|
|
p=prepare(corpus,q)
|
|
assert p['status']=='ok' and 'Cassandra' not in p['context']
|
|
|
|
|
|
def test_year_condition_without_the_words_v_roku_is_preserved(corpus):
|
|
q='Aké problémy mala Eva Nová pri diplomovej práci 1999?'
|
|
p=prepare(corpus,q); c=rag_evidence_context(q,p)
|
|
assert c['requirements']['years']==[1999] and not c.get('content_candidates')
|
|
r=generate_verified_rag(q,p,lambda _:pytest.fail('No model for missing scoped evidence'))
|
|
assert r['accepted_for_delivery'] and r['status']=='no_answer' and 'tokenizérom' not in r['answer']
|
|
assert requested_years('Práca „Model 2020“ z roku 2024')==[2024]
|
|
|
|
|
|
def test_named_source_can_be_loaded_when_absent_from_top_results(corpus):
|
|
db,raw=corpus; raw['sources']=raw['sources'][:1];raw['context']=build_context_text(raw['sources'])
|
|
p=prepare(corpus,'Aké problémy riešili Eva Nová a Boris Nový?')
|
|
assert {s['title'] for s in p['sources']}=={'Eva Nová','Boris Nový'}
|
|
|
|
|
|
def test_preparation_does_not_change_the_original_retrieval(corpus):
|
|
db,raw=corpus; before=copy.deepcopy(raw)
|
|
prepare_named_content(db,'Aké problémy riešili Eva Nová a Boris Nový?',raw)
|
|
assert raw==before
|
|
|
|
|
|
def test_source_budget_cannot_silently_remove_a_requested_person(corpus):
|
|
db,raw=corpus; q='Aké problémy riešili Eva Nová a Boris Nový?'
|
|
p=prepare_named_content(db,q,raw)
|
|
p['sources'][0]['text']+='\n'+'Veľmi dlhý zdrojový text. '*1000
|
|
compact=compact_rag_payload(p,max_tokens=1100)
|
|
c=rag_evidence_context(q,compact)
|
|
assert compact['truncated']
|
|
assert set(c['requirements']['people'])=={'Eva Nová','Boris Nový'}
|
|
|
|
|
|
def test_candidates_are_actual_bound_literal_evidence(corpus):
|
|
q='Aké problémy riešili Eva Nová a Boris Nový?'; c=rag_evidence_context(q,prepare(corpus,q))
|
|
for candidate in candidates(c):
|
|
assert any(all(candidate[k]==p[k] for k in ['person','work_type','year','source_id'])
|
|
and candidate['text'] in p['text'] for p in c['content_passages'])
|
|
|
|
|
|
def test_unknown_second_subject_is_not_silently_ignored(corpus):
|
|
p=prepare(corpus,'Aké problémy riešili študenti Eva Nová a Boris Noyy?')
|
|
assert p['status']=='ambiguous' and p['reason']=='unresolved_subject'
|
|
|
|
|
|
@pytest.mark.parametrize('question',[
|
|
'Aké problémy riešili Eva Nová a Xaver Neexistujúci?',
|
|
'Aké problémy riešil Boris Noyy?',
|
|
])
|
|
def test_unresolved_subject_cannot_be_replaced_by_available_profiles(corpus,question):
|
|
p=prepare(corpus,question)
|
|
assert p['status']=='ambiguous' and p['reason']=='unresolved_subject'
|
|
|
|
|
|
def test_partial_answer_names_the_missing_aspect(corpus):
|
|
db,raw=corpus
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute("UPDATE chunks SET text='Stav:\nVznikol problém s parserom.' 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); r=generate_verified_rag(q,p,complete(model_selection(q,p)))
|
|
assert r['accepted_for_delivery'] and r['answer_coverage']['status']=='partial'
|
|
assert 'Táto časť odpovede nie je v poskytnutých podkladoch doložená.' in r['answer']
|
|
assert 'parserom' in r['answer']
|
|
|
|
|
|
def test_all_aspects_absent_is_no_answer_not_rejection(corpus):
|
|
db,raw=corpus
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute("UPDATE chunks SET text='Zadanie:\n- Implementujte parser.'")
|
|
q='Aké problémy riešili Eva Nová a Boris Nový?'
|
|
r=generate_verified_rag(q,prepare(corpus,q),lambda _:pytest.fail('No model for a bounded empty result'))
|
|
assert r['status']=='no_answer' and r['accepted_for_delivery'] and not r['claims']
|
|
|
|
|
|
def test_known_year_is_returned_when_requested_title_is_missing(corpus):
|
|
_,raw=corpus; source=raw['sources'][0]
|
|
source['text']='Sekcia: Diplomová práca 2024\nStav:\nImplementovaný parser.'
|
|
raw['sources']=[source];raw['context']=build_context_text(raw['sources'])
|
|
q='Uveď názov a rok diplomovej práce Evy Novej.'
|
|
a={'status':'ok','answer':'','claims':[{'person':'Eva Nová','work_type':'diploma_thesis','year':2024}],
|
|
'evidence_quotes':[],'source_urls':[source['source_url']]}
|
|
r=generate_verified_rag(q,raw,complete(a))
|
|
assert r['accepted_for_delivery'] and '2024' in r['answer']
|
|
assert 'názov práce nie je v poskytnutých podkladoch doložený' in r['answer']
|
|
assert r['answer_coverage']['status']=='partial' and not r['claims'][0].get('title')
|
|
|
|
|
|
def test_concrete_results_win_over_administrative_notes(corpus):
|
|
q='Aké výsledky dosiahla Eva Nová?'; p=prepare(corpus,q); c=rag_evidence_context(q,p)
|
|
text='\n'.join(x['text'] for x in c['content_candidates'])
|
|
assert 'Natrénovaný model BERT' in text
|
|
r=generate_verified_rag(q,p,complete(model_selection(q,p)))
|
|
assert r['accepted_for_delivery'] and 'Natrénovaný model BERT' in r['answer']
|
|
assert 'Konzultácia' not in r['answer']
|
|
|
|
|
|
def test_duplicate_quotes_and_overlapping_chunks_do_not_duplicate_output(corpus):
|
|
q='Aké výsledky dosiahla Eva Nová?'; p=prepare(corpus,q); a=model_selection(q,p)
|
|
a['evidence_quotes']*=2
|
|
r=generate_verified_rag(q,p,complete(a))
|
|
assert r['accepted_for_delivery'] and r['answer'].count('Natrénovaný model BERT')==1
|
|
assert 'Ďalšie podklady:' not in r['answer'] and 'Ďalšie doložené podklady' not in r['answer']
|
|
|
|
|
|
def test_plans_cannot_answer_actual_use_or_completed_experiments():
|
|
assert 'methods' not in facets_for('Použite metódu BERT.','task','Aké metódy skutočne používal?')
|
|
assert 'experiments' not in facets_for('Pripravte experiment.','task','Aké experimenty vykonal?')
|
|
assert 'results' not in facets_for('Cieľom je implementovať systém.','progress','Aké dosiahol výsledky?')
|
|
|
|
|
|
def test_method_answer_does_not_add_valid_but_unrequested_profile_metadata(corpus):
|
|
_,raw=corpus
|
|
raw['sources'][0]['author']='Daniel Hladek'
|
|
q='Aké metódy používala Eva Nová pri diplomovej práci v roku 2024?'
|
|
p=prepare(corpus,q); a=model_selection(q,p)
|
|
a['claims']=[{'person':'Eva Nová','work_type':'diploma_thesis','year':2024,
|
|
'document_author':'Daniel Hladek'}]
|
|
r=generate_verified_rag(q,p,complete(a))
|
|
assert r['accepted_for_delivery'] and 'BERT' in r['answer']
|
|
assert 'Daniel Hladek' not in r['answer'] and 'autor dokumentu' not in r['answer']
|
|
assert r['answer'].count('2024')==1
|
|
|
|
|
|
@pytest.mark.parametrize('question',[
|
|
'Práce od roku 2020 do roku 2024?', 'Práce pred rokom 2020?',
|
|
'Študenti s aspoň 3 prácami?', 'Práce okrem diplomových?', 'Študenti bez tagu RAG?',
|
|
'V ktorom roku Ján Holp začal pracovať na diplomovej práci?',
|
|
'Kedy Ján Holp obhájil diplomovú prácu?',
|
|
])
|
|
def test_unsupported_operators_are_not_ignored(question):
|
|
assert unsupported_constraint(question)
|
|
result=gs.search_graph_with_session(FakeSession({}),question)
|
|
assert result['status']=='ambiguous' and result['reason']=='unsupported_constraint' and not result['facts']
|
|
|
|
|
|
@pytest.mark.parametrize('question',[
|
|
'Práca v roku 1999?', 'Aký je názov práce „Od 2020 do 2024“?',
|
|
'Aké práce alebo projekty má Eva Nová?',
|
|
])
|
|
def test_supported_year_and_work_alternative_are_not_blocked(question):
|
|
assert unsupported_constraint(question) is None
|
|
|
|
|
|
@pytest.mark.parametrize('reason',['model_unavailable','retrieval_unavailable','answer_rejected','answer_capacity_exceeded'])
|
|
def test_public_errors_are_distinct_and_do_not_leak_rejected_text(reason):
|
|
r=public_failure({'reason':reason,'answer':'SECRET YEAR 2099','claims':[{'person':'Eva'}],'source_urls':['SECRET']})
|
|
assert r['reason']==reason and not r['accepted_for_delivery']
|
|
assert 'SECRET' not in r['answer'] and '2099' not in r['answer']
|
|
assert not r['claims'] and not r['source_urls']
|
|
|
|
|
|
def test_retrieval_failure_is_not_misreported_as_bad_model_answer(monkeypatch):
|
|
import tools.answers.answer_delivery as delivery
|
|
def unavailable(_): raise RuntimeError('SECRET backend address')
|
|
monkeypatch.setattr(delivery,'search_graph',unavailable)
|
|
r=delivery.deliver_answer('Aké práce má Eva Nová?',engine='graphrag',
|
|
complete=lambda _:pytest.fail('No model call after failed retrieval'))
|
|
assert r['status']=='error' and r['reason']=='retrieval_unavailable'
|
|
assert 'SECRET' not in r['answer']
|