dp-zp-agent/test/test_evidence_selection.py
2026-11-04 00:55:54 +01:00

239 lines
13 KiB
Python

"""Bound excerpts to works, preserve their phase and share comparison space."""
import copy
import json
import sqlite3
import pytest
from test_quality_block import corpus, prepare, complete
from tools.answer_evaluation.content_selection import administrative_only, evidence_units
from tools.answer_evaluation.rag_evidence import rag_evidence_context, content_sections
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.answer_evaluation.verified_rag import overview_excerpts
from tools.rag.llm_context import compact_rag_payload
from tools.rag.rag_utils import build_context_text
def update_eva(corpus,text,headings=None):
db,raw=corpus
with sqlite3.connect(db) as conn:
conn.execute('UPDATE chunks SET text=?,heading_paths_json=? WHERE document_path=?',
(text,json.dumps(headings or [['Eva Nová','Diplomová práca 2024']]),raw['sources'][0]['document_path']))
def selection(context):
return {'status':'ok','answer':'','claims':[],
'evidence_quotes':[{key:c[key] for key in ['source_id','passage_id'] if c.get(key)}|{'quote':c['text']}
for c in context.get('content_candidates',[])],
'source_urls':[s['source_url'] for s in context['sources']]}
def test_assignment_plan_progress_and_recorded_result_are_separate(corpus):
update_eva(corpus,'Zadanie:\n- Implementujte vyhľadávač.\nPriebežný stav:\n'
'- Rozpracovaná integrácia vyhľadávania.\n- Bude implementovaný klasifikátor.\n'
'Výsledky:\n- Natrénovaný model BERT.\nPlánované úlohy:\n- Pripravte experiment s presnosťou.')
q='Zhrň zadanie, plánované úlohy, priebežný stav a výsledky diplomovej práce Evy Novej.'
payload=prepare(corpus,q);context=rag_evidence_context(q,payload)
units=evidence_units(context)
assert {u['phase'] for u in units}>={'assignment','planned','progress','result'}
future=next(u for u in units if 'Bude implementovaný' in u['text'])
assert future['phase']=='planned' and 'results' not in future['facets']
result=generate_verified_rag(q,payload,complete(selection(context)))
assert result['accepted_for_delivery']
assert all(label in result['answer'] for label in ['Zadanie a ciele','Plánované úlohy','Priebežný stav','Zaznamenané výsledky'])
assert 'Bude implementovaný' not in result['answer'].split('**Zaznamenané výsledky:**')[1]
@pytest.mark.parametrize('heading',['Zadanie','Zadanie diplomovej práce'])
def test_phase_from_chunk_heading_ancestry_is_not_lost(corpus,heading):
update_eva(corpus,'- Pripravte vyhodnotenie modelu.',
[['Eva Nová','Diplomová práca 2024',heading]])
q='Aké je zadanie a plánované úlohy diplomovej práce Evy Novej?'
c=rag_evidence_context(q,prepare(corpus,q))
assert any(u['kind']=='assignment' and u['phase']=='assignment' for u in evidence_units(c))
assert not any(u['kind']=='task' for u in evidence_units(c))
assert all(u['year']==2024 and u['work_label']=='Diplomová práca 2024' for u in evidence_units(c))
def test_negated_completion_is_progress_not_a_completed_result(corpus):
update_eva(corpus,'Stav:\n- Parser zatiaľ nie je implementovaný.')
q='Aký je priebežný stav diplomovej práce Evy Novej?'
c=rag_evidence_context(q,prepare(corpus,q))
row=next(u for u in evidence_units(c) if 'Parser' in u['text'])
assert row['phase']=='progress' and 'results' not in row['facets']
q='Aké výsledky dosiahla Eva Nová pri diplomovej práci?'
r=generate_verified_rag(q,prepare(corpus,q),lambda _: (_ for _ in ()).throw(AssertionError('No results model call')))
assert r['status']=='no_answer' and r['accepted_for_delivery']
def test_administration_alone_is_not_a_technical_result(corpus):
update_eva(corpus,'Stav:\n- Konzultácia štruktúry práce.\n- Dohodnutý termín odovzdania.')
q='Aké výsledky dosiahla Eva Nová pri diplomovej práci?'
c=rag_evidence_context(q,prepare(corpus,q))
assert not c['content_candidates']
assert administrative_only('Konzultácia štruktúry práce.')
assert not administrative_only('Vyriešený problém s GPU po konzultácii.')
assert not administrative_only('Odovzdaný zdrojový kód indexovania do SQLite.')
def test_recorded_working_system_is_a_result_not_just_administration(corpus):
update_eva(corpus,'Stav:\n- Funguje RAG systém, zostáva overiť presnosť.')
q='Aké výsledky má diplomová práca Evy Novej?'
c=rag_evidence_context(q,prepare(corpus,q))
assert any(u['phase']=='result' and 'zostáva' in u['text'] for u in c['content_candidates'])
def test_completed_experiments_are_not_confused_with_planned_tasks(corpus):
update_eva(corpus,'Výsledky:\n- Vyhodnotená presnosť modelu v experimente.\n'
'Úlohy:\n- Pripravte experiment s novým tokenizérom.')
q='Aké vykonané experimenty má Eva Nová a aké plánované úlohy jej zostali?'
c=rag_evidence_context(q,prepare(corpus,q))
planned=next(u for u in evidence_units(c) if 'Pripravte' in u['text'])
assert 'plans' in planned['facets'] and 'experiments' not in planned['facets']
def test_a_planned_method_does_not_prove_an_actually_used_method(corpus):
update_eva(corpus,'Úlohy:\n- Použiť metódu BERT.')
q='Aké použité metódy má diplomová práca Evy Novej?'
c=rag_evidence_context(q,prepare(corpus,q))
assert not c['content_candidates']
def test_explicit_negative_finding_remains_a_result(corpus):
update_eva(corpus,'Výsledky:\n- Model nebol úspešný pri teste presnosti.')
q='Aké výsledky má diplomová práca Evy Novej?'
p=prepare(corpus,q);c=rag_evidence_context(q,p)
assert c['content_candidates'][0]['phase']=='result'
r=generate_verified_rag(q,p,complete(selection(c)))
assert r['accepted_for_delivery'] and 'nebol úspešný' in r['answer']
def test_identical_text_in_two_work_years_needs_passage_identity(corpus):
_,raw=corpus;source=raw['sources'][0]
source['text']='Sekcia: Diplomová práca 2024\nStav:\nImplementovaný parser.\n\n'
source['text']+='Sekcia: Diplomová práca 2025\nStav:\nImplementovaný parser.'
raw['sources']=[source];raw['context']=build_context_text(raw['sources'])
q='Aké výsledky má Eva Nová pri diplomovej práci?';c=rag_evidence_context(q,raw)
a={'status':'ok','answer':'','claims':[],'source_urls':[source['source_url']],
'evidence_quotes':[{'source_id':'S1','quote':'Implementovaný parser.'}]}
assert 'ambiguous_evidence_scope' in rag_answer_issues(a,c)
p=c['content_passages'][0];a['evidence_quotes'][0]['passage_id']=p['passage_id']
assert 'ambiguous_evidence_scope' not in rag_answer_issues(a,c)
assert len({p['passage_id'] for p in c['content_passages']})==2
def test_identical_text_cannot_reuse_another_persons_passage_id(corpus):
for person_name in ['eva_nova','boris_novy']:
db,_=corpus
with sqlite3.connect(db) as conn:
conn.execute('UPDATE chunks SET text=? WHERE document_path LIKE ?',
('Stav:\nImplementovaný parser.',f'%{person_name}%'))
q='Porovnaj výsledky Evy Novej a Borisa Nového.';p=prepare(corpus,q);c=rag_evidence_context(q,p)
a=selection(c)
assert not rag_answer_issues(a,c)
wrong=copy.deepcopy(a)
wrong['evidence_quotes'][0]['passage_id']=wrong['evidence_quotes'][1]['passage_id']
assert 'unsupported_evidence_passage' in rag_answer_issues(wrong,c)
def test_overlapping_excerpt_preserves_qualification_without_repetition(corpus):
update_eva(corpus,'Stav:\n- Natrénovaný model BERT.\n'
'- Natrénovaný model BERT. Výsledok platí iba pre malý korpus.')
q='Aké výsledky dosiahla Eva Nová?';p=prepare(corpus,q);c=rag_evidence_context(q,p)
r=generate_verified_rag(q,p,complete(selection(c)))
assert r['accepted_for_delivery']
assert r['answer'].count('Natrénovaný model BERT')==1 and 'iba pre malý korpus' in r['answer']
def add_same_year_projects(corpus):
db,raw=corpus
with sqlite3.connect(db) as conn:
for i,method in [(1,'BERT'),(2,'Cassandra')]:
conn.execute('INSERT INTO chunks VALUES(?,?,?,?,?)',
(raw['sources'][0]['document_path'],f'Stav:\nImplementovaná metóda {method}.',
json.dumps([['Eva Nová',f'Diplomový projekt {i} 2022']]),1,i))
def test_same_year_project_records_are_not_merged(corpus):
add_same_year_projects(corpus)
p=prepare(corpus,'Aké metódy používala Eva Nová pri diplomovom projekte v roku 2022?')
assert p['status']=='ambiguous' and 'projekt 1' in p['clarification'] and 'projekt 2' in p['clarification']
def test_explicit_project_number_binds_the_excerpt_to_that_project(corpus):
add_same_year_projects(corpus)
q='Aké metódy používala Eva Nová pri diplomovom projekte 2 v roku 2022?'
p=prepare(corpus,q);c=rag_evidence_context(q,p)
assert p['status']=='ok' and 'Cassandra' in p['context'] and 'BERT' not in p['context']
assert all('projekt 2' in u['work_label'] for u in evidence_units(c))
a=selection(c)
result=generate_verified_rag(q,p,complete(a))
assert result['accepted_for_delivery'] and 'projekt 2' in result['answer']
def test_large_first_profile_cannot_starve_the_second_person(corpus):
db,raw=corpus;q='Porovnaj výsledky Evy Novej a Borisa Nového.'
p=prepare_named_content(db,q,raw)
p['sources'][0]['text']+='\n'+('Dlhé administratívne poznámky. '*3000)
compact=compact_rag_payload(p,max_tokens=2200)
assert compact['source_count']==2 and compact['context_budget']['used_tokens']<=2200
assert compact['truncated'] and 'llm_context_budget' in compact['truncation_reasons']
c=rag_evidence_context(q,compact)
assert {row['person'] for row in c['content_candidates']}=={'Eva Nová','Boris Nový'}
assert all(row['passage_id'] and row['work_label'] for row in c['content_candidates'])
def test_clipping_cannot_drop_a_wrapped_qualification_from_an_excerpt(corpus):
db,raw=corpus;q='Porovnaj výsledky Evy Novej a Borisa Nového.'
p=prepare_named_content(db,q,raw)
p['sources'][0]['text']='Sekcia: Diplomová práca 2024\nStav:\n- Natrénovaný model BERT.\n '
p['sources'][0]['text']+='Výsledok platí iba pre malý korpus. '*2000
compact=compact_rag_payload(p,max_tokens=1700)
assert compact['truncated'] and 'BERT' not in compact['context']
c=rag_evidence_context(q,compact)
assert any(row['person']=='Boris Nový' for row in c['content_candidates'])
assert not any(row['person']=='Eva Nová' for row in c['content_candidates'])
def test_single_profile_clipping_preserves_the_same_qualification_rule(corpus):
_,raw=corpus;raw['sources']=raw['sources'][:1]
raw['sources'][0]['text']='Sekcia: Diplomová práca 2024\nStav:\n- Natrénovaný model BERT.\n '
raw['sources'][0]['text']+='Výsledok platí iba pre malý testovací korpus. '*2000
compact=compact_rag_payload(raw,max_tokens=1700)
assert compact['truncated'] and 'BERT' not in compact['context']
@pytest.mark.parametrize('with_title',[False,True])
def test_relevant_results_are_prioritized_before_context_is_clipped(corpus,with_title):
update_eva(corpus,('Názov: Skutočný názov práce.\n' if with_title else '')+'Zadanie:\n- Implementujte parser.')
db,raw=corpus;path=raw['sources'][0]['document_path']
with sqlite3.connect(db) as conn:
for i in range(1,12):
conn.execute('INSERT INTO chunks VALUES(?,?,?,?,?)',(path,
'Stav:\n- Konzultácia štruktúry práce. '+('Administratívne poznámky. '*100),
json.dumps([['Eva Nová','Diplomová práca 2024']]),1,i))
conn.execute('INSERT INTO chunks VALUES(?,?,?,?,?)',(path,
'Výsledky:\n- Vyhodnotená presnosť parsera.',
json.dumps([['Eva Nová','Diplomová práca 2024']]),1,12))
q='Porovnaj '+('názvy a ' if with_title else '')+'výsledky Evy Novej a Borisa Nového.'
compact=compact_rag_payload(prepare_named_content(db,q,raw),max_tokens=1700)
assert compact['truncated'] and 'Vyhodnotená presnosť parsera.' in compact['context']
c=rag_evidence_context(q,compact)
assert {u['person'] for u in c['content_candidates']}=={'Eva Nová','Boris Nový'}
if with_title:
assert 'Skutočný názov práce' in compact['context'] and 'title' in c['requirements']['available_fields']['Eva Nová']
def test_overview_cannot_cut_off_a_qualification_for_brevity():
text='Systém vyhľadáva dokumenty. '+('Opis implementácie a dát. '*30)+'Platí iba pre malý testovací korpus.'
quote={'source_id':'S1','quote':text}
c={'content_passages':[{'source_id':'S1','person':'Eva Nová','text':text}],'facts':[]}
assert overview_excerpts({'evidence_quotes':[quote]},c)==[text]
def test_markdown_role_headings_preserve_explicit_source_phase():
sections=content_sections('## Zadanie\nImplementujte systém.\n## Plánované úlohy\nPripravte dáta.\n'
'## Priebežný stav\nRozpracovaná integrácia.\n## Výsledky\nVyhodnotená presnosť.')
assert [(s['kind'],s['phase']) for s in sections]==[
('assignment','assignment'),('task','planned'),('progress','progress'),('progress','result')]