dp-zp-agent/test/test_synthesis.py

355 lines
20 KiB
Python

"""Prose fidelity, rejection mutations and deterministic fallback, without network."""
import copy
import json
import pytest
from tools.answers import synthesis as s
from tools.answers.question_plan import QuestionPlan
from tools.knowledge.sources import profile_units
def fixture(phase='result', text='Implementovaný PageRank a indexovanie webových stránok.'):
rows = profile_units({'title':'Ján Holp', 'path':'pages/students/2016/jan_holp/README.md',
'content':'# Diplomová práca 2021\nStretnutie 14.5.2020\nStav:\n- '+text}, include_records=True)
row = next(r for r in rows if text in r['text'] and r['kind'] not in {'meeting','work_record'})
row.update(phase=phase, annotation_status='knowledge')
result = {'status':'ok','accepted_for_delivery':True,'answer':'ORIGINAL EXACT TEXT',
'source_urls':[row['source_url']], 'index_version':{'version':'v'},'knowledge_version':'k',
'complete':False,'truncated':False,'truncation_reasons':[], 'evidence':[row],
'presentation_notes':['Konkrétne upozornenie na obmedzený výber.']}
plan=QuestionPlan(question='Čomu sa venuje diplomová práca Jána Holpa?', intent='facets',
facets=['assignment'],presentation='brief')
return result, plan
def draft_for(result, text='Súčasťou riešenia je PageRank a indexovanie webových stránok.'):
return {'claims':[{'kind':'description','text':text,'evidence_ids':[r['id'] for r in result['evidence']]}]}
def scripted(draft, *, support='supported', critic='supported', coverage='supported'):
calls=[]
def complete(messages):
calls.append(messages)
payload=json.loads(messages[-1]['content'])
if 'claims' not in payload:
value=draft
else:
decision=critic if messages[0]['content'].startswith(s.CRITIC) else support
value={'coverage':coverage,'verdicts':[{'claim_id':c['claim_id'],'decision':decision,'reason':'entailed' if decision=='supported' else 'unsupported_fact'} for c in payload['claims']]}
return {'choices':[{'message':{'content':json.dumps(value)}}]}
return complete,calls
def run(result,plan,draft,**kw):
model,calls=scripted(draft,**kw)
return s.synthesize(result,plan,complete=model,diagnostics=True),calls
def test_natural_description_has_scoped_claim_review_citations_and_unchanged_notes():
result,plan=fixture();original=copy.deepcopy(result)
answer,calls=run(result,plan,draft_for(result))
assert answer['synthesis']['status']=='accepted' and len(calls)==3
assert 'Súčasťou riešenia' in answer['answer'] and 'Zaznamenané výsledky' in answer['answer']
assert 'dátum záznamu: 2020-05-14' in answer['answer']
assert 'Diplomová práca 2021' in answer['answer']
assert result['presentation_notes'][0] in answer['answer']
assert answer['source_urls']==result['source_urls']
assert answer['index_version']==result['index_version'] and answer['evidence']==result['evidence']
assert result==original
assert len(calls[1])==2 and s.VERIFY in calls[1][0]['content']
assert 'support_review' not in calls[2][-1]['content']
@pytest.mark.parametrize('text,reason',[
('Vytvoril systém s úspešnosťou 98 percent.','unsupported_number'),
('Prácu dokončil v roku 2099.','model_time_claim'),
('Dňa 14. 5. implementoval PageRank.','model_time_claim'),
('Je to jediný fungujúci systém.','exhaustive_claim'),
('Systém je opísaný [tu](/neexistujuci).','model_markup_or_link'),
('Systém funguje.\n[s]: /neexistujuci','model_markup_or_link'),
('Systém má <a href="/neexistujuci">repozitár</a>.','model_markup_or_link'),
])
def test_local_rejections_never_publish_draft(text,reason):
result,plan=fixture();answer,calls=run(result,plan,draft_for(result,text))
assert answer['answer']==result['answer'] and answer['synthesis']=={'status':'fallback','reason':reason}
assert len(calls)==(2 if reason=='model_time_claim' else 1)
def test_planned_activity_cannot_be_upgraded_to_achievement_before_review():
result,plan=fixture('planned','Implementujte vyhľadávanie pomocou PageRank.')
answer,_=run(result,plan,draft_for(result,'Implementoval vyhľadávanie pomocou PageRank.'))
assert answer['synthesis']['reason']=='phase_upgrade'
assert answer['answer']==result['answer']
def test_assignment_can_be_paraphrased_as_a_goal_without_extra_critical_call():
result,plan=fixture('assignment','Implementujte vyhľadávanie pomocou PageRank.')
answer,calls=run(result,plan,draft_for(result,'Cieľom je vytvoriť vyhľadávanie pomocou PageRank.'))
assert answer['synthesis']['status']=='accepted' and len(calls)==2
assert 'Zadanie a ciele' in answer['answer']
@pytest.mark.parametrize('review', ['unsupported','uncertain'])
def test_semantic_reviewer_rejects_fluent_unsupported_relationships(review):
result,plan=fixture(text='Python funguje. Model BERT bol vyhodnotený.')
answer,_=run(result,plan,draft_for(result,'Python bol použitý na trénovanie modelu BERT.'),support=review)
assert answer['synthesis']['reason']=='support_rejected' and answer['answer']==result['answer']
def test_second_critical_review_can_overrule_first_support_review():
result,plan=fixture()
answer,calls=run(result,plan,draft_for(result,'Diplomovú prácu úspešne dokončil.'),critic='unsupported')
assert len(calls)==6 and answer['synthesis']['reason']=='critical_support_rejected'
assert answer['answer']==result['answer']
def test_citing_all_goals_does_not_establish_their_semantic_coverage():
result,plan=fixture('assignment','Implementujte systém a navrhnite experimenty.')
answer,_=run(result,plan,draft_for(result,'Cieľom je implementácia systému.'),coverage='unsupported')
assert answer['synthesis']['reason']=='support_rejected'
def test_omitted_evidence_and_unknown_ids_are_local_failures():
result,plan=fixture();result['evidence'].append({**result['evidence'][0],'id':'E2','text':'Navrhnite experimenty.'})
draft=draft_for(result);draft['claims'][0]['evidence_ids'].pop()
assert run(result,plan,draft)[0]['synthesis']['reason']=='missing_evidence_coverage'
draft['claims'][0]['evidence_ids']=['invented']
assert run(result,plan,draft)[0]['synthesis']['reason']=='unknown_evidence'
def test_scope_cannot_silently_join_two_people_or_two_work_phases():
result,plan=fixture()
result['evidence'].append({**result['evidence'][0],'id':'E2','person':'Eva Nová','work_id':'W2'})
assert run(result,plan,draft_for(result))[0]['synthesis']['reason']=='mixed_scope_or_phase'
result['evidence'][1].update(person='Ján Holp',work_id=result['evidence'][0]['work_id'],phase='planned')
assert run(result,plan,draft_for(result))[0]['synthesis']['reason']=='mixed_scope_or_phase'
def test_wrong_known_person_cannot_receive_anothers_result():
result,plan=fixture();model,_=scripted(draft_for(result,'Eva Nová implementovala PageRank.'))
answer=s.synthesize(result,plan,complete=model,known_people=['Eva Nová'],diagnostics=True)
assert answer['synthesis']['reason']=='wrong_person'
@pytest.mark.parametrize('mutation', ['missing','duplicate','foreign'])
def test_review_requires_exactly_one_verdict_for_every_claim(mutation):
result,plan=fixture();draft=draft_for(result)
calls=[]
def complete(messages):
calls.append(1)
if len(calls)==1:value=draft
else:
verdict={'claim_id':0,'decision':'supported','reason':'entailed'}
value={'coverage':'supported','verdicts':[] if mutation=='missing' else [verdict,verdict] if mutation=='duplicate' else [{**verdict,'claim_id':1}]}
return {'choices':[{'message':{'content':json.dumps(value)}}]}
answer=s.synthesize(result,plan,complete=complete,diagnostics=True)
assert answer['synthesis']['status']=='fallback' and answer['answer']==result['answer']
@pytest.mark.parametrize('field,value', [('presentation','count'),('intent','metadata'),('intent','advisors')])
def test_exact_facts_lists_counts_and_academic_records_do_not_call_synthesis(field,value):
result,plan=fixture();setattr(plan,field,value)
def forbidden(_):pytest.fail('Unexpected synthesis call')
assert s.synthesize(result,plan,complete=forbidden) is result
@pytest.mark.parametrize('status', ['no_answer','ambiguous','error'])
def test_absence_and_failures_are_not_rephrased(status):
result,plan=fixture();result['status']=status
assert not s.eligible(result,plan)
def test_unavailable_model_and_too_large_input_keep_original_without_leaking_errors(monkeypatch):
result,plan=fixture()
def fail(_):raise RuntimeError('PRIVATE_KEY_VALUE')
answer=s.synthesize(result,plan,complete=fail,diagnostics=True)
assert answer['answer']==result['answer'] and 'PRIVATE' not in str(answer)
monkeypatch.setenv('RAG_LLM_MAX_TOKENS','256')
answer=s.synthesize(result,plan,complete=fail,diagnostics=True)
assert answer['synthesis']['reason']=='input_budget'
def test_comparison_is_formed_by_checked_per_work_cells_even_without_free_summary():
result,plan=fixture('assignment','Vypracujte prehľad metód vyhľadávania.')
other={**result['evidence'][0],'id':'E2','work_id':'W2','work_label':'Bakalárska práca 2020',
'text':'Vypracujte prehľad jazykových modelov.'}
result['evidence'].append(other);plan.presentation='comparison'
draft={'claims':[{'text':'Cieľom je prehľad metód vyhľadávania.','evidence_ids':[result['evidence'][0]['id']],'kind':'description'},
{'text':'Cieľom je prehľad jazykových modelov.','evidence_ids':['E2'],'kind':'description'}]}
answer,_=run(result,plan,draft)
assert answer['synthesis']['status']=='accepted' and '| Oblasť |' in answer['answer']
draft['claims'].insert(0,{'text':'Jedno zadanie sa týka vyhľadávania, druhé jazykových modelov.',
'evidence_ids':[result['evidence'][0]['id'],'E2'],'kind':'comparison'})
assert run(result,plan,draft)[0]['synthesis']['status']=='accepted'
plan.presentation='table'
answer,_=run(result,plan,draft)
assert answer['synthesis']['status']=='accepted' and '| Oblasť |' in answer['answer']
def test_diagnostic_free_fallback_is_byte_for_byte_the_original_contract():
result,plan=fixture()
assert s.synthesize(result,plan,complete=lambda _:None)==result
def test_content_lists_can_be_natural_but_exact_record_lists_stay_deterministic():
result,plan=fixture('assignment','Implementujte vyhľadávanie pomocou PageRank.')
plan.presentation='list'
answer,_=run(result,plan,draft_for(result,'Cieľom je vytvoriť vyhľadávanie pomocou PageRank.'))
assert answer['synthesis']['status']=='accepted' and '\n- ' in answer['answer']
plan.intent='metadata'
assert s.synthesize(result,plan,complete=lambda _:pytest.fail('exact list')) is result
def test_short_generation_ids_are_resolved_back_to_real_evidence_ids():
result,plan=fixture('assignment','Implementujte PageRank.')
value={'claims':[{'text':'Cieľom je implementácia PageRank.','kind':'description','evidence_ids':['E1']}]}
answer,_=run(result,plan,value)
assert answer['synthesis']['status']=='accepted'
assert answer['synthesis']['claims'][0]['evidence_ids']==[result['evidence'][0]['id']]
def test_explicit_all_grades_qualification_does_not_mean_exhaustive_search():
result,plan=fixture(text='Získané osnovy pre všetky ročníky základnej školy.')
answer,_=run(result,plan,draft_for(result,'Získal osnovy pre všetky ročníky základnej školy.'))
assert answer['synthesis']['status']=='accepted'
def test_partial_record_date_is_not_completed_from_work_year():
result,plan=fixture();result['evidence'][0].update(record_date=None,record_date_label='15.5.')
answer,_=run(result,plan,draft_for(result))
assert '15.5. (rok neuvedený)' in answer['answer']
def test_coverage_repair_has_one_attempt_and_all_guards_still_apply():
result,plan=fixture('assignment','Implementujte PageRank.')
result['evidence'].append({**result['evidence'][0],'id':'other','text':'Navrhnite experimenty.',
'meeting_id':'other-meeting','record_date':'2021-03-12'})
attempts=[]
def complete(messages):
payload=json.loads(messages[-1]['content']);attempts.append(payload)
if 'claims' in payload:
value={'coverage':'supported','verdicts':[{'claim_id':c['claim_id'],'decision':'supported','reason':'entailed'} for c in payload['claims']]}
else:
value={'claims':[{'kind':'description','text':'Cieľom je implementácia PageRank.','evidence_ids':['E1']}]}
if 'coverage_repair' in payload:
value['claims'].append({'kind':'description','text':'Má navrhnúť experimenty.','evidence_ids':['E2']})
return {'choices':[{'message':{'content':json.dumps(value)}}]}
answer=s.synthesize(result,plan,complete=complete,diagnostics=True)
assert answer['synthesis']['status']=='accepted'
assert len(attempts)==3 and attempts[1]['coverage_repair']['missing_evidence_ids']==['E2']
def test_common_delivery_uses_prose_but_never_treats_document_titles_as_people(monkeypatch):
from contextlib import nullcontext
from tools.answers import answer_delivery,question_plan
from tools.answers.question_plan import PlannedQuestion
from tools.knowledge import query
result,plan=fixture('assignment','Implementujte systém na získavanie informácií.')
monkeypatch.setattr(answer_delivery,'index_snapshot',lambda *a,**kw:nullcontext(result['index_version']))
monkeypatch.setattr(question_plan,'planning_catalog',lambda _: [
{'kind':'Person','name':'Ján Holp'}, {'kind':'Document','name':'Získavanie informácií'}])
monkeypatch.setattr(question_plan,'build_question_plan',lambda *a,**kw:PlannedQuestion(spec=plan,resolution={'status':'ok'}))
monkeypatch.setattr(query,'deliver_content',lambda *a,**kw:copy.deepcopy(result))
model,_=scripted(draft_for(result,'Cieľom je implementácia systému na získavanie informácií.'))
answer=answer_delivery.deliver_answer(plan.question,complete=model,include_diagnostics=True)
assert answer['synthesis']['status']=='accepted' and 'Cieľom je implementácia' in answer['answer']
def test_bare_claim_array_is_only_a_format_adapter_not_a_guard_bypass():
result,plan=fixture('assignment','Implementujte PageRank.')
value=draft_for(result,'Cieľom je implementácia PageRank.')['claims']
answer,_=run(result,plan,value)
assert answer['synthesis']['status']=='accepted'
value[0]['text']='Implementoval nový systém.'
answer,_=run(result,plan,value)
assert answer['synthesis']['reason']=='phase_upgrade'
def test_benchmark_name_cannot_leak_from_another_phase_or_record():
result,plan=fixture('planned','Vyhodnoťte Wiki 103 na Commonsense.')
answer,_=run(result,plan,draft_for(result,'Plánoval vyhodnotiť Wiki 103 na CommonsenseQA.'))
assert answer['synthesis']['reason']=='unsupported_technical_identifier'
def test_grammatical_model_name_and_real_task_acronym_remain_usable():
assert s._identifier_supported('PageRankom','Implementovaný PageRank.')
assert s._identifier_supported('NER','Rozpoznávanie pomenovaných entít.')
assert not s._identifier_supported('NER','Anotácia pomenovaných entít.')
assert s._identifier_supported('Wiki103','Vyhodnoťte Wiki 103 na Commonsense.')
assert s._identifier_supported('P-R-F1','Vyhodnoťte pomocou P-R-F1.')
def test_graph_shared_topic_is_rendered_from_path_with_both_witnesses():
result,plan=fixture('planned','Oboznámte sa so znalostným grafom.')
other={**result['evidence'][0],'id':'other','person':'Eva Nová','work_id':'W2',
'phase':'result','text':'Vytvorený znalostný graf.', 'source_path':'eva.md', 'source_url':'https://example.test/eva'}
result['evidence'].append(other);result['source_urls'].append(other['source_url']);plan.intent='similar'
result['graph_paths']=[{'anchor_id':result['evidence'][0]['id'],'peer_id':'other','concept':'knowledge graph'}]
value={'claims':[{'kind':'description','text':'Má sa oboznámiť so znalostným grafom.','evidence_ids':['E1']},
{'kind':'description','text':'Vytvorila znalostný graf.','evidence_ids':['E2']}]}
answer,_=run(result,plan,value)
assert answer['synthesis']['status']=='accepted'
assert 'Ján Holp a Eva Nová: doložená spoločná téma — znalostný graf. [1] [2]' in answer['answer']
result['graph_paths'][0]['peer_id']='invented'
answer,_=run(result,plan,value)
assert answer['synthesis']['reason']=='graph_witness_missing'
def test_a_false_cross_phase_comparison_does_not_borrow_achievements_for_plans():
result,plan=fixture('planned','Vyhodnoťte model na zvolenej množine.')
other={**result['evidence'][0],'id':'other','person':'Eva Nová','work_id':'W2',
'phase':'result','text':'Vyhodnotený model BERT na CommonsenseQA.'}
result['evidence'].append(other);plan.presentation='comparison'
value={'claims':[{'kind':'comparison','text':'Plánoval vyhodnotenie na CommonsenseQA, druhá práca už model vyhodnotila.',
'evidence_ids':['E1','E2']}]}
assert run(result,plan,value)[0]['synthesis']['reason']=='unsupported_comparison_summary'
def test_rejected_optional_summary_cannot_remove_facts_or_replace_scoped_cells():
result,plan=fixture('assignment','Implementujte PageRank.')
other={**result['evidence'][0],'id':'other','work_id':'W2','work_label':'Bakalárska práca 2020',
'text':'Vyhodnotený model BERT.','phase':'result'}
result['evidence'].append(other);plan.presentation='comparison'
value={'claims':[{'kind':'description','text':'Má implementovať PageRank.','evidence_ids':['E1']},
{'kind':'description','text':'Vyhodnotil model BERT.','evidence_ids':['E2']},
{'kind':'comparison','text':'Obe práce majú dokončené vyhodnotenie.','evidence_ids':['E1','E2']}]}
answer,_=run(result,plan,value)
assert answer['synthesis']['status']=='accepted' and answer['synthesis']['omitted_cross_phase_summaries']==1
assert 'Obe práce' not in answer['answer'] and 'Má implementovať' in answer['answer'] and 'Vyhodnotil model' in answer['answer']
value['claims'].pop(1)
answer,_=run(result,plan,value)
assert answer['synthesis']['status']=='fallback' and answer['synthesis']['reason']=='missing_evidence_coverage'
def test_optional_summary_cannot_replace_the_requested_comparison_cells():
result,plan=fixture('assignment','Implementujte PageRank.')
result['evidence'].append({**result['evidence'][0],'id':'other','work_id':'W2','text':'Natrénujte BERT.'})
plan.presentation='table'
value={'claims':[{'kind':'comparison','text':'Jedna práca má implementovať PageRank a druhá natrénovať BERT.',
'evidence_ids':['E1','E2']}]}
answer,_=run(result,plan,value)
assert answer['synthesis']['reason']=='missing_comparison_cells' and answer['answer']==result['answer']
def test_model_cannot_change_the_owners_surname_while_preserving_given_name():
result,plan=fixture('assignment','Implementujte PageRank.')
result['evidence'][0]['person']='Ondrej Megela'
answer,_=run(result,plan,draft_for(result,'Ondrej Megel má implementovať PageRank.'))
assert answer['synthesis']['reason']=='altered_person_name' and answer['answer']==result['answer']
def test_a_possible_workaround_cannot_become_a_required_or_successful_fix():
result,plan=fixture('progress','Možné riešenie je iný prepínač pri dotrénovaní.')
answer,_=run(result,plan,draft_for(result,'Problém si vyžiadal zmenu prepínača pri dotrénovaní.'))
assert answer['synthesis']['reason']=='lost_qualification'
result,plan=fixture('progress','Chyba optimizéra a jej riešenie.')
answer,_=run(result,plan,draft_for(result,'Vyriešil chybu optimizéra.'))
assert answer['synthesis']['reason']=='unsupported_resolution'
def test_qualified_workaround_is_preserved_and_reviewed():
result,plan=fixture('progress','Problém sa možno dá obísť skopírovaním modelu.')
answer,_=run(result,plan,draft_for(result,'Možným obídením problému je skopírovanie modelu.'))
assert answer['synthesis']['status']=='accepted'