229 lines
12 KiB
Python
229 lines
12 KiB
Python
"""Regressions for actual Claude questions and adversarial same-person joins."""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tools.answer_evaluation.checked_generation import generate_checked_answer
|
|
from tools.answer_evaluation.context_policy import answer_context_issues
|
|
from tools.answer_evaluation.graph_context import graph_context
|
|
from tools.answer_evaluation.render_answer import render_graph_answer
|
|
from tools.answers.answer_delivery import deliver_answer
|
|
from tools.graphrag import graph_search as gs
|
|
from tools.graphrag.tests.test_graph_search import FakeSession
|
|
from tools.retrieval.question_constraints import unsupported_constraint
|
|
|
|
|
|
PEOPLE=['Ján Pták','Ján Malinovský','Valerii Kutsenko','Jakub Schwarc']
|
|
PROMPT=Path('tools/answer_evaluation/graph_answer_prompt.txt').read_text()
|
|
|
|
|
|
def person(index,name):
|
|
return {'kind':'Person','name':name,'element_id':f'p{index}',
|
|
'source_document':f'pages/students/2022/p{index}/README.md'}
|
|
|
|
|
|
def source(index):
|
|
return {'document_path':f'pages/students/2022/p{index}/README.md',
|
|
'source_url':f'https://example.test/p{index}'}
|
|
|
|
|
|
@pytest.fixture
|
|
def tagged_projects():
|
|
catalog=[person(i,name) for i,name in enumerate(PEOPLE,1)]
|
|
catalog.append({'kind':'Keyword','name':'RAG','id':'rag','element_id':'krag'})
|
|
projects=[{'person':name,'person_element_id':f'p{i}','year':2026,
|
|
'work_type':'diploma_project','work_id':f'work:p{i}',
|
|
'heading':'Diplomový projekt 2026','title':f'Téma projektu {name}',**source(i)}
|
|
for i,name in enumerate(PEOPLE,1)]
|
|
tags=[{'person':name,'person_element_id':f'p{i}','keyword':'RAG',
|
|
'keyword_id':'rag','keyword_element_id':'krag',**source(i)}
|
|
for i,name in enumerate(PEOPLE[:3],1)]
|
|
# A fifth person has the tag, but no work in the requested year/type.
|
|
tags.append({'person':'Iná Osoba','person_element_id':'p5','keyword':'RAG',
|
|
'keyword_id':'rag','keyword_element_id':'krag',**source(5)})
|
|
return FakeSession({gs.CATALOG_QUERY:catalog,gs.WORK_FACTS_QUERY:projects,
|
|
gs.PERSON_KEYWORD_FACTS_QUERY:tags})
|
|
|
|
|
|
@pytest.mark.parametrize('question',[
|
|
'Ktorí študenti majú tag RAG a zároveň diplomový projekt z roku 2026?',
|
|
'Ktorí študenti s tagom RAG majú diplomový projekt v roku 2026?',
|
|
'Ktorí študenti majú tag RAG a diplomový projekt z roku 2026?',
|
|
'Ktorí študenti majú tag „RAG“ a zároveň diplomový projekt z roku 2026?',
|
|
'Ktorí študenti s tagom RAG a zároveň diplomovým projektom z roku 2026?',
|
|
'Ktorí študenti s tagom RAG a diplomovým projektom z roku 2026?',
|
|
])
|
|
def test_tag_and_project_are_witnessed_by_the_same_person(tagged_projects,question):
|
|
r=gs.search_graph_with_session(tagged_projects,question,limit=50)
|
|
assert r['status']=='ok' and r['constraints']['keyword_phrase']=='rag'
|
|
assert {f['person'] for f in r['facts']}==set(PEOPLE[:3])
|
|
work=[params for q,params in tagged_projects.calls if q==gs.WORK_FACTS_QUERY][0]
|
|
tags=[params for q,params in tagged_projects.calls if q==gs.PERSON_KEYWORD_FACTS_QUERY][0]
|
|
assert work['years']==[2026] and work['work_types']==['diploma_project']
|
|
assert tags['keyword_ids']==['krag']
|
|
context=graph_context(r)
|
|
assert {c['person'] for c in context['answer_candidates']}==set(PEOPLE[:3])
|
|
assert all(c['work_type']=='diploma_project' for c in context['answer_candidates'])
|
|
|
|
|
|
def test_diploma_thesis_cannot_be_replaced_by_a_project(tagged_projects):
|
|
r=gs.search_graph_with_session(tagged_projects,
|
|
'Ktorí študenti majú tag RAG a zároveň diplomovú prácu z roku 2026?',limit=50)
|
|
assert r['status']=='no_answer' and r['reason']=='no_matching_facts'
|
|
assert not r['facts'] and not r['constraints']['unresolved_keyword']
|
|
|
|
|
|
@pytest.mark.parametrize('limit',[1,2,3,4,5,6])
|
|
def test_tag_predicate_stays_bound_to_visible_records_after_truncation(tagged_projects,limit):
|
|
q='Ktorí študenti majú tag RAG a zároveň diplomový projekt z roku 2026?'
|
|
raw=gs.search_graph_with_session(tagged_projects,q,limit=limit)
|
|
assert len(raw['facts'])<=limit and raw['truncated']==(limit<3)
|
|
context=graph_context(raw)
|
|
visible={r['person'] for r in raw['facts'] if r['fact_type']=='person_work'}
|
|
assert {c['person'] for c in context['answer_candidates']}==visible
|
|
assert all(r['person'] in visible for r in raw['supporting_facts'])
|
|
assert all(r['fact_type']=='person_keyword' for r in raw['supporting_facts'])
|
|
claims=[{k:c[k] for k in ['person','year','keyword','work_type']} for c in context['answer_candidates']]
|
|
answer={'status':'ok','claims':claims,'source_urls':[s['source_url'] for s in context['sources']]}
|
|
rendered=render_graph_answer(answer,context)
|
|
assert not answer_context_issues(rendered,context)
|
|
assert 'Jakub Schwarc' not in rendered['answer'] and 'Iná Osoba' not in rendered['answer']
|
|
assert ('Výpis je neúplný.' in rendered['answer'])==(limit<3)
|
|
|
|
|
|
def test_count_model_input_keeps_actual_total_and_only_claim_fields(tagged_projects):
|
|
import json
|
|
from tools.answer_evaluation.answer_format import Claim
|
|
q='Koľko študentov má tag RAG a zároveň diplomový projekt z roku 2026?'
|
|
context=graph_context(gs.search_graph_with_session(tagged_projects,q,limit=1))
|
|
before=copy.deepcopy(context)
|
|
def select(messages):
|
|
sent=json.loads(messages[1]['content'])['graph_context']
|
|
assert sent['aggregation']['count']==3 and 'candidate_count' not in sent['aggregation']
|
|
assert 'count=3' in messages[0]['content']
|
|
assert len(sent['answer_candidates'])==1
|
|
assert set(sent['answer_candidates'][0])<=set(Claim.model_fields)
|
|
a={'status':'ok','answer':'','count':sent['aggregation']['count'],
|
|
'claims':sent['answer_candidates'],'source_urls':context['answer_candidates'][0]['source_urls']}
|
|
return {'choices':[{'message':{'content':json.dumps(a)}}]}
|
|
r=generate_checked_answer(q,context,PROMPT,select)
|
|
assert r['accepted_for_delivery'] and r['answer']['count']==3 and context==before
|
|
|
|
|
|
def test_unknown_tag_cannot_trigger_a_global_project_listing(tagged_projects):
|
|
r=gs.search_graph_with_session(tagged_projects,
|
|
'Ktorí študenti majú tag XNEZNAMY a zároveň diplomový projekt z roku 2026?')
|
|
assert r['status']=='no_answer' and r['reason']=='unknown_keyword' and not r['facts']
|
|
|
|
|
|
@pytest.mark.parametrize('question,expected',[
|
|
('Ľudia s tagom strojové učenie a zároveň diplomovou prácou.','strojove ucenie'),
|
|
('Ľudia s tagom „výskum a aplikácie“ a zároveň diplomovým projektom.','vyskum a aplikacie'),
|
|
('Ľudia s tagom RAG z roku 2026.','rag'),
|
|
])
|
|
def test_multiword_tag_and_condition_boundary(question,expected):
|
|
assert gs.explicit_keyword_phrase(question)==expected
|
|
|
|
|
|
@pytest.mark.parametrize('question,code',[
|
|
('Ktorí študenti majú aspoň tri práce?','quantity_condition'),
|
|
('Ktorí študenti majú aspoň 3 práce?','quantity_condition'),
|
|
('Kto má najmenej tri práce?','quantity_condition'),
|
|
('Študenti s najviac dvoma projektmi?','quantity_condition'),
|
|
('Študenti s minimálne piatimi prácami?','quantity_condition'),
|
|
('ktorý študenti týmto tempom ukončia svoje diplomové práce v roku 2027?','completion_prediction'),
|
|
('Ktorí študenti ukončia diplomové práce v roku 2027?','completion_prediction'),
|
|
('Odhadni termín dokončenia diplomovej práce Jána Holpa.','completion_prediction'),
|
|
('Kedy Ján Holp ukončil diplomovú prácu?','work_event_date'),
|
|
('Aké metódy predikcie počasia používa diplomová práca?',None),
|
|
('Aký je názov práce „Kto dokončí prácu v roku 2027“?',None),
|
|
])
|
|
def test_unsupported_condition_wording(question,code):
|
|
assert unsupported_constraint(question)==code
|
|
|
|
|
|
@pytest.mark.parametrize('engine',['rag','graphrag'])
|
|
@pytest.mark.parametrize('question,word',[
|
|
('Ktorí študenti majú aspoň tri práce?','počet'),
|
|
('Ktorí študenti týmto tempom ukončia diplomové práce v roku 2027?','predpovedať'),
|
|
])
|
|
def test_unsupported_condition_never_calls_a_model(engine,question,word):
|
|
r=deliver_answer(question,engine=engine,complete=lambda _:pytest.fail('No model for unsupported condition'))
|
|
assert r['status']=='ambiguous' and r['accepted_for_delivery'] and word in r['answer']
|
|
assert not r['claims'] and not r['source_urls']
|
|
|
|
|
|
def selected(context):
|
|
fields=['person','record_id','title','year','work_type','work_label','keyword','category','document_author']
|
|
rows=context.get('required_academic_records') or context['facts']
|
|
claims=[{k:r[k] for k in fields if r.get(k) is not None} for r in rows]
|
|
return {'status':'ok','answer':'','claims':claims,
|
|
'source_urls':[s['source_url'] for s in context['sources']]}
|
|
|
|
|
|
def generated(answer,context):
|
|
return generate_checked_answer(context.get('question',''),context,PROMPT,
|
|
lambda _: {'choices':[{'message':{'content':json.dumps(answer)}}]})
|
|
|
|
|
|
def test_titleless_records_survive_and_are_delivered_without_invented_titles():
|
|
catalog=[person(1,'Ján Holp')]
|
|
records=[{'person':'Ján Holp','person_element_id':'p1','work_id':f'project-{i}',
|
|
'title':None,'heading':label,'work_type':'diploma_project','year':2020,**source(1)}
|
|
for i,label in enumerate(['Diplomový projekt 2020','Diplomový projekt 2 2020'])]
|
|
s=FakeSession({gs.CATALOG_QUERY:catalog,gs.WORK_FACTS_QUERY:records})
|
|
raw=gs.search_graph_with_session(s,'Aký názov má diplomový projekt Jána Holpa z roku 2020?')
|
|
assert raw['status']=='ok' and len(raw['facts'])==2
|
|
context=graph_context(raw);a=selected(context)
|
|
result=generated(a,context)
|
|
assert result['accepted_for_delivery']
|
|
assert result['answer']['answer'].count('názov v zdroji nie je uvedený')==2
|
|
assert len(result['answer']['claims'])==2 and not any(c.get('title') for c in result['answer']['claims'])
|
|
bad=copy.deepcopy(a);bad['claims'][0]['title']='Vymyslený názov'
|
|
assert 'unsupported_academic_binding' in answer_context_issues(bad,context,check_text=False)
|
|
bad=copy.deepcopy(a);bad['claims'].pop()
|
|
assert 'missing_academic_record_details' in answer_context_issues(bad,context,check_text=False)
|
|
|
|
|
|
def shared_context(three=False):
|
|
people=PEOPLE[:3] if three else PEOPLE[:2]
|
|
raw={'status':'ok','reason':None,'complete':True,'truncated':False,'aggregation':None,
|
|
'query':'Aké spoločné kľúčové slová majú '+' a '.join(people)+'?',
|
|
'constraints':gs.extract_query_constraints('spoločné kľúčové slová'),
|
|
'resolved_entities':[person(i,p) for i,p in enumerate(people,1)],
|
|
'sources':[source(i) for i in range(1,len(people)+1)],'facts':[]}
|
|
for keyword,left,right in [('NLP',1,2),('RAG',1,2)]+([('NLP',1,3)] if three else []):
|
|
raw['facts'].append({'fact_type':'shared_keyword','keyword':keyword,
|
|
'left_person':people[left-1],'right_person':people[right-1],
|
|
'left_document_path':source(left)['document_path'],'right_document_path':source(right)['document_path'],
|
|
'left_source_url':source(left)['source_url'],'right_source_url':source(right)['source_url']})
|
|
return graph_context(raw)
|
|
|
|
|
|
def test_shared_tags_are_summarized_once_and_require_all_peoples_evidence():
|
|
context=shared_context();a=selected(context)
|
|
result=generated(a,context)
|
|
assert result['accepted_for_delivery']
|
|
text=result['answer']['answer']
|
|
assert 'Spoločné kľúčové slová: **NLP**, **RAG**.' in text
|
|
assert text.count('NLP')==1 and text.count('RAG')==1 and 'Pri osobe' not in text
|
|
bad=copy.deepcopy(a);bad['claims'].pop()
|
|
assert 'missing_shared_property_evidence' in answer_context_issues(bad,context,check_text=False)
|
|
|
|
|
|
def test_pairwise_shared_tag_does_not_mean_shared_by_everyone():
|
|
context=shared_context(three=True)
|
|
assert {r['value'] for r in context['shared_properties']}=={'NLP'}
|
|
text=render_graph_answer(selected(context),context)['answer']
|
|
assert 'Spoločné kľúčové slová: **NLP**.' in text and '**RAG**' not in text
|
|
|
|
|
|
def test_incomplete_shared_tags_cannot_claim_exhaustive_results():
|
|
context=shared_context();context.update(complete=False,truncated=True,truncation_reasons=['fact_limit'])
|
|
result=generated(selected(context),context)
|
|
assert result['accepted_for_delivery']
|
|
assert 'Doložené spoločné kľúčové slová' in result['answer']['answer']
|
|
assert 'Výpis je neúplný.' in result['answer']['answer']
|