236 lines
13 KiB
Python
236 lines
13 KiB
Python
"""Real multi-person query regressions, including deliberately wrong evidence."""
|
|
import copy
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from tools.answer_evaluation.context_policy import answer_context_issues, text_contains_value
|
|
from tools.answer_evaluation.graph_context import graph_context
|
|
from tools.answer_evaluation.presentation import content_lines, goal_text
|
|
from tools.answer_evaluation.rag_evidence import (content_sections, expand_comparison_sources,
|
|
rag_evidence_context, requested_types)
|
|
from tools.answer_evaluation.verified_rag import generate_verified_rag, rag_answer_issues
|
|
from tools.answer_evaluation.render_answer import render_graph_answer
|
|
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
|
|
|
|
COMPARE = ('Porovnaj zameranie bakalárskej práce Ondreja Megelu a diplomovej práce Jána Holpa. '
|
|
'Rozlišuj zadanie od dosiahnutých výsledkov a uveď zdroje.')
|
|
SHARED = 'V ktorých rokoch majú Ján Holp a Ondrej Megela obaja evidovanú prácu alebo projekt?'
|
|
|
|
|
|
@pytest.mark.parametrize('canonical,wording', [
|
|
('Ondrej Megela', 'Ondreja Megelu'), ('Martin Šarišský', 'Martina Šarišského'),
|
|
('Ján Malinovský', 'Jána Malinovského'), ('Eva Nová', 'Evy Novej')])
|
|
def test_complete_case_endings(canonical, wording):
|
|
assert text_contains_value(wording, canonical, person=True)
|
|
|
|
|
|
@pytest.mark.parametrize('canonical,other', [
|
|
('Martin Šarišský', 'Martin Jančura'), ('Ján Holp', 'Ján Holpák'),
|
|
('Ondrej Megela', 'Ondrej Megelovec'), ('Martin NaN1', 'Martin NaN2')])
|
|
def test_no_short_prefix_matching(canonical, other):
|
|
assert not text_contains_value(other, canonical, person=True)
|
|
|
|
|
|
def test_each_person_gets_the_requested_work_type():
|
|
assert requested_types(COMPARE, ['Ján Holp', 'Ondrej Megela']) == {
|
|
'Ján Holp': 'diploma_thesis', 'Ondrej Megela': 'bachelor_thesis'}
|
|
|
|
|
|
def test_goal_conversion_is_not_all_or_nothing():
|
|
text = '\n'.join(content_lines(['- Zistite, čo je to RAG.', '- Naučte sa čo je to GraphRAG.',
|
|
'- Zostavte malý znalostný graf.', '- Využite graf pri vyhľadávaní.',
|
|
'- Použite databázu Neo4j.', 'Práca neimplementuje vlastný model.'],
|
|
'Ján Malinovský', 'diploma_project', 'Aké sú ciele diplomového projektu Jána Malinovského?'))
|
|
assert 'Diplomový projekt Jána Malinovského' in text
|
|
assert 'Zistiť, čo je to RAG.' in text and 'Použiť databázu Neo4j.' in text
|
|
assert 'Ďalšie podklady:' in text and 'Práca neimplementuje vlastný model.' in text
|
|
assert not any(word in text for word in ['Zistite', 'Naučte', 'Zostavte', 'Využite', 'Použite'])
|
|
|
|
|
|
def test_prefixed_goal_keeps_other_converted_items():
|
|
text = '\n'.join(content_lines(['1. Vypracujte prehľad systémov generovania prirodzeného jazyka.',
|
|
'2. Na základe tohto prehľadu navrhnite a implementujte agentový systém.'],
|
|
'Martin Šarišský', 'diploma_thesis', 'Čomu sa venuje diplomová práca Martina Šarišského?'))
|
|
assert 'Diplomová práca Martina Šarišského' in text
|
|
assert 'Vypracovanie prehľadu' in text and 'navrhnúť a implementovať' in text
|
|
assert 'Vypracujte' not in text and 'navrhnite' not in text
|
|
|
|
|
|
def test_coordinated_commands_and_objects_keep_grammatical_infinitives():
|
|
text='\n'.join(content_lines(['Vypracujte prehľad aplikácií BERT a spôsoby ich vyhodnotenia.',
|
|
'Vypracujte prehľad metód a implementujte systém.'],'Ondrej Megela','bachelor_thesis'))
|
|
assert 'Vypracovať prehľad aplikácií BERT a spôsoby ich vyhodnotenia.' in text
|
|
assert 'Vypracovať prehľad metód a implementovať systém.' in text
|
|
|
|
|
|
def test_goal_headings_are_not_extra_evidence_and_sentences_keep_capitals():
|
|
text='\n'.join(content_lines(['Ciele na semester.', 'Zlepšite proces. Zlepšite extrakciu.'],
|
|
'Ján Malinovský','diploma_project'))
|
|
assert 'Ďalšie podklady' not in text and 'Zlepšiť proces. Zlepšiť extrakciu.' in text
|
|
|
|
|
|
@pytest.mark.parametrize('literal', ['Použite https://example.test/pouzite/vypracujte.',
|
|
'Implementujte parser pre `navrhnite a implementujte`.'])
|
|
def test_grammar_does_not_change_urls_or_code_literals(literal):
|
|
assert goal_text(literal)==(literal,False)
|
|
|
|
|
|
@pytest.fixture
|
|
def comparison_payload(tmp_path):
|
|
sources = []; rows = []
|
|
for i, (person,slug,kind,year,goal,progress) in enumerate([
|
|
('Ján Holp','jan_holp','Diplomová práca',2021,'Implementujte systém vyhľadávania.',
|
|
'Implementovaný PageRank a indexovanie webových stránok.'),
|
|
('Ondrej Megela','ondrej_megela','Bakalárska práca',2020,'Natrénujte jazykový model metódou BERT.',
|
|
'Natrénovaný model wiki103 a vykonané vyhodnotenie.')],1):
|
|
path=f'pages/students/2018/{slug}/README.md'
|
|
sources.append({'source_id':f'S{i}','title':person,'document_path':path,
|
|
'source_url':f'https://example.test/{slug}','text':f'Sekcia: {kind} {year}\nNázov: Ukážkový názov práce.'})
|
|
rows.append((path,json.dumps([[person,f'{kind} {year}']]),f'Zadanie:\n- {goal}\nStav:\n- {progress}',1,0))
|
|
rows.append((path,json.dumps([[person,'Diplomový projekt 2022']]),'Stav:\nCudzí projekt využíva Cassandra.',1,1))
|
|
db=tmp_path/'index.sqlite'
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute('CREATE TABLE chunks(document_path,heading_paths_json,text,published,chunk_index)')
|
|
conn.executemany('INSERT INTO chunks VALUES(?,?,?,?,?)',rows)
|
|
raw={'status':'ok','reason':None,'sources':sources,'context':build_context_text(sources),
|
|
'complete':False,'truncated':True,'truncation_reasons':['result_limit'],'index_version':{'version':'fixture'}}
|
|
return db,raw
|
|
|
|
|
|
def selected_comparison(context):
|
|
quotes=[{'source_id':p['source_id'],'quote':s['text']} for p in context['content_passages']
|
|
for s in content_sections(p['text']) if s['kind'] in {'assignment','progress'}]
|
|
return {'status':'ok','answer':'','claims':[],'evidence_quotes':quotes,
|
|
'source_urls':[s['source_url'] for s in context['sources']]}
|
|
|
|
|
|
def test_expansion_preserves_binding_versions_and_budget(comparison_payload):
|
|
db,raw=comparison_payload; before=copy.deepcopy(raw)
|
|
payload=compact_rag_payload(expand_comparison_sources(db,COMPARE,raw))
|
|
c=rag_evidence_context(COMPARE,payload)
|
|
assert c['requirements']['work_types']=={'Ján Holp':'diploma_thesis','Ondrej Megela':'bachelor_thesis'}
|
|
assert c['requirements']['content_roles']=={p:['assignment','progress'] for p in c['requirements']['people']}
|
|
assert 'Cassandra' not in payload['context']
|
|
assert payload['index_version']==raw['index_version'] and payload['complete'] is False
|
|
assert payload['context_budget']['used_tokens']<=payload['context_budget']['max_tokens']
|
|
assert raw==before
|
|
|
|
|
|
def test_hidden_chunks_are_not_added_to_comparison_evidence(comparison_payload):
|
|
db,raw=comparison_payload; path=raw['sources'][0]['document_path']
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute('INSERT INTO chunks VALUES(?,?,?,?,?)',(path,
|
|
json.dumps([['Ján Holp','Diplomová práca 2021']]),'Stav:\nNezverejnený tajný výsledok.',0,2))
|
|
expanded=expand_comparison_sources(db,COMPARE,raw)
|
|
assert 'tajný výsledok' not in expanded['context']
|
|
|
|
|
|
def test_relevant_child_document_is_preserved(comparison_payload):
|
|
db,raw=comparison_payload
|
|
child={**raw['sources'][0],'source_id':'S3','title':'Meranie projektu',
|
|
'document_path':raw['sources'][0]['document_path'].replace('README.md','meranie/README.md'),
|
|
'text':'Sekcia: Diplomová práca 2021\nStav:\nVyhodnotená presnosť vyhľadávania.'}
|
|
raw['sources'].append(child); raw['context']=build_context_text(raw['sources'])
|
|
expanded=expand_comparison_sources(db,COMPARE,raw)
|
|
assert 'Vyhodnotená presnosť vyhľadávania.' in expanded['context']
|
|
|
|
|
|
def test_comparison_separates_goals_and_results(comparison_payload):
|
|
db,raw=comparison_payload; payload=compact_rag_payload(expand_comparison_sources(db,COMPARE,raw))
|
|
c=rag_evidence_context(COMPARE,payload); selected=selected_comparison(c)
|
|
assert rag_answer_issues(selected,c)==[]
|
|
result=generate_verified_rag(COMPARE,payload,lambda _: {'choices':[{'message':{'content':json.dumps(selected)}}]})
|
|
assert result['accepted_for_delivery']
|
|
text=result['answer']
|
|
assert text.count('**Zadanie a ciele:**')==text.count('**Zaznamenaný stav a výsledky:**')==2
|
|
assert 'Natrénovať jazykový model metódou BERT.' in text
|
|
assert 'Natrénovaný model wiki103' in text and 'Implementovaný PageRank' in text
|
|
assert 'Cassandra' not in text and len(result['source_urls'])==2
|
|
|
|
|
|
def test_goals_cannot_replace_available_results(comparison_payload):
|
|
db,raw=comparison_payload; c=rag_evidence_context(COMPARE,expand_comparison_sources(db,COMPARE,raw))
|
|
a=selected_comparison(c)
|
|
a['evidence_quotes']=[q for q in a['evidence_quotes'] if not any(
|
|
word in q['quote'] for word in ['Implementovaný','Natrénovaný'])]
|
|
assert 'missing_progress_evidence' in rag_answer_issues(a,c)
|
|
|
|
|
|
def test_missing_results_allow_an_explicit_partial_comparison(comparison_payload):
|
|
db,raw=comparison_payload
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute("UPDATE chunks SET text='Zadanie:\n- Implementujte systém vyhľadávania.' WHERE document_path LIKE '%jan_holp%' AND chunk_index=0")
|
|
payload=expand_comparison_sources(db,COMPARE,raw); c=rag_evidence_context(COMPARE,payload)
|
|
selected=selected_comparison(c)
|
|
result=generate_verified_rag(COMPARE,payload,lambda _: {'choices':[{'message':{'content':json.dumps(selected)}}]})
|
|
assert result['accepted_for_delivery']
|
|
assert 'nie sú dosiahnuté výsledky doložené' in result['answer']
|
|
assert 'Implementácia systému vyhľadávania' in result['answer']
|
|
|
|
|
|
def test_future_tasks_are_not_presented_as_achieved_results(comparison_payload):
|
|
db,raw=comparison_payload
|
|
with sqlite3.connect(db) as conn:
|
|
conn.execute("UPDATE chunks SET text=text || '\nÚlohy:\n- Pripravte experiment s PageRank.' WHERE document_path LIKE '%jan_holp%' AND chunk_index=0")
|
|
payload=expand_comparison_sources(db,COMPARE,raw); c=rag_evidence_context(COMPARE,payload)
|
|
selected=selected_comparison(c)
|
|
p=next(p for p in c['content_passages'] if p['person']=='Ján Holp')
|
|
selected['evidence_quotes'].append({'source_id':p['source_id'],'quote':'- Pripravte experiment s PageRank.'})
|
|
result=generate_verified_rag(COMPARE,payload,lambda _: {'choices':[{'message':{'content':json.dumps(selected)}}]})
|
|
assert result['accepted_for_delivery']
|
|
assert '**Ďalšie doložené úlohy (plán):**' in result['answer']
|
|
assert 'Pripraviť experiment s PageRank.' in result['answer']
|
|
|
|
|
|
@pytest.fixture
|
|
def shared_session():
|
|
catalog=[]; theses=[]; works=[]
|
|
for pid,person,slug in [('p1','Ján Holp','jan'),('p2','Ondrej Megela','ondrej')]:
|
|
path=f'pages/students/2018/{slug}/README.md'; url=f'https://example.test/{slug}'
|
|
catalog.append({'element_id':pid,'kind':'Person','name':person,'source_document':path})
|
|
base={'person_element_id':pid,'person':person,'document_path':path,'source_url':url}
|
|
theses.append({**base,'thesis_element_id':pid+'t','thesis_id':pid+'t','title':'Ukážkový názov',
|
|
'year':2019 if pid=='p1' else 2020,'heading':'Bakalárska práca', 'thesis_type':'bachelor_thesis'})
|
|
works.append({**base,'work_element_id':pid+'w','work_id':pid+'w','year':2020,
|
|
'heading':'Diplomový projekt','work_type':'diploma_project'})
|
|
return FakeSession({gs.CATALOG_QUERY:catalog,gs.THESIS_FACTS_QUERY:theses,gs.WORK_FACTS_QUERY:works})
|
|
|
|
|
|
@pytest.mark.parametrize('word',['obaja','obidvaja'])
|
|
def test_both_means_shared_years(word):
|
|
c=gs.extract_query_constraints(SHARED.replace('obaja',word))
|
|
assert c['mentions_shared'] and c['mentions_academic_year']
|
|
|
|
|
|
def test_shared_year_has_only_intersection_and_bound_work_details(shared_session):
|
|
result=gs.search_graph_with_session(shared_session,SHARED)
|
|
assert result['academic_year_comparison']['shared_years']==[2020]
|
|
c=graph_context(result); records=c['required_academic_records']
|
|
assert len(records)==3 and all(r['year']==2020 for r in records)
|
|
a={'status':'ok','claims':records,'source_urls':[s['source_url'] for s in c['sources']]}
|
|
rendered=render_graph_answer(a,c)
|
|
assert answer_context_issues(rendered,c)==[]
|
|
assert 'Spoločný evidovaný rok je **2020**.' in rendered['answer']
|
|
assert '2019' not in rendered['answer'] and 'Ukážkový názov' in rendered['answer']
|
|
|
|
|
|
def test_no_common_year_is_not_a_union(shared_session):
|
|
for row in shared_session.responses[gs.WORK_FACTS_QUERY]:
|
|
if row['person_element_id']=='p1': row['year']=2021
|
|
result=gs.search_graph_with_session(shared_session,SHARED)
|
|
assert result['status']=='no_answer' and result['academic_year_comparison']['shared_years']==[]
|
|
c=graph_context(result)
|
|
assert 'spoločný rok nenašiel' in render_graph_answer({'status':'no_answer','claims':[],'source_urls':[]},c)['answer']
|
|
|
|
|
|
def test_partial_records_do_not_claim_an_unwitnessed_common_year(shared_session):
|
|
result=gs.search_graph_with_session(shared_session,SHARED,limit=1); c=graph_context(result)
|
|
a={'status':'ok','claims':c['required_academic_records'],'source_urls':[s['source_url'] for s in c['sources']]}
|
|
text=render_graph_answer(a,c)['answer']
|
|
assert 'Výpis je neúplný.' in text and 'Spoločný evidovaný rok je' not in text
|