dp-zp-agent/tools/graphrag/tests/test_graph_search_entity_v2.py

173 lines
8.3 KiB
Python

from __future__ import annotations
import pytest
from tools.graphrag import graph_search as gs
from .test_graph_search import FakeSession
def catalog():
return [
{'kind': 'Person', 'element_id': 'alice', 'id': 'person:alice', 'name': 'Alena Nová',
'source_document': 'alice.md'},
{'kind': 'Person', 'element_id': 'boris', 'id': 'person:boris', 'name': 'Boris Malý',
'source_document': 'boris.md'},
{'kind': 'Person', 'element_id': 'cross', 'id': 'person:cross', 'name': 'Boris Novák',
'source_document': 'cross.md'},
{'kind': 'Keyword', 'element_id': 'rag', 'id': 'rag', 'name': 'RAG'},
{'kind': 'Keyword', 'element_id': 'zp', 'id': 'zp', 'name': 'zp'},
]
def thesis():
return {'person_element_id': 'alice', 'person': 'Alena Nová', 'thesis_element_id': 'thesis-a',
'thesis_type': 'bachelor_thesis', 'year': 2024, 'title': 'Ukážkový výskum',
'document_path': 'alice.md', 'source_url': 'https://example.test/alice'}
def test_complete_names_cannot_be_assembled_from_different_people():
data = catalog() + [{'kind': 'Person', 'element_id': 'mixed', 'id': 'person:mixed', 'name': 'Boris Nová'}]
result = gs.resolve_entities('Porovnaj osoby Alena Nová a Boris Malý.', data)
assert {row['element_id'] for row in result if row['kind'] == 'Person'} == {'alice', 'boris'}
@pytest.mark.parametrize('query', [
'Akú prácu robila Alena Nová?',
'Aký je názov práce Aleny Novej?',
'Vyhľadaj prácu Borisa Malého.',
])
def test_controlled_name_inflections_still_resolve(query):
resolved = gs.resolve_entities(query, catalog())
assert any(row['kind'] == 'Person' for row in resolved)
def test_exact_full_name_wins_over_inflection_of_another_name():
data = [{'kind': 'Person', 'element_id': 'jan', 'id': 'jan', 'name': 'Ján Novák'},
{'kind': 'Person', 'element_id': 'jana', 'id': 'jana', 'name': 'Jana Novák'}]
assert [row['element_id'] for row in gs.resolve_entities('Čomu sa venuje Jana Novák?', data)] == ['jana']
@pytest.mark.parametrize('query', [
'Aký názov diplomovej práce má Zuzana Neexistujúca?',
'Aký názov diplomovej práce má zuzana neexistujuca?',
'Akú prácu robil Neznámy Študent?',
'Čomu sa venuje neznama osoba?',
'Aký je názov práce osoby neznama osoba?',
'Z kandidátov Alena Nová, neznama osoba, Boris Malý vyber osoby s prácou v roku 2024. Koľko ich je?',
])
def test_explicit_unknown_person_never_becomes_global_search(query):
session = FakeSession({gs.CATALOG_QUERY: catalog(), gs.THESIS_FACTS_QUERY: [thesis()]})
result = gs.search_graph_with_session(session, query)
assert result['status'] == 'no_answer'
assert result['reason'] == 'unknown_person'
assert result['facts'] == [] and result['paths'] == []
assert result['aggregation'] is None
assert gs.THESIS_FACTS_QUERY not in {statement for statement, _ in session.calls}
def test_legitimate_global_listing_keeps_unrestricted_query():
session = FakeSession({gs.CATALOG_QUERY: catalog(), gs.THESIS_FACTS_QUERY: [thesis()]})
result = gs.search_graph_with_session(session, 'Aké bakalárske práce sú uvedené v roku 2024?')
assert result['status'] == 'ok'
assert result['facts'][0]['title'] == 'Ukážkový výskum'
assert next(params for statement, params in session.calls if statement == gs.THESIS_FACTS_QUERY)['person_ids'] == []
def test_duplicate_full_name_is_explicitly_ambiguous():
data = catalog() + [{'kind': 'Person', 'element_id': 'duplicate', 'id': 'person:duplicate',
'name': 'Alena Nová', 'source_document': 'duplicate.md'}]
session = FakeSession({gs.CATALOG_QUERY: data, gs.THESIS_FACTS_QUERY: [thesis()]})
result = gs.search_graph_with_session(session, 'Aký názov práce má Alena Nová?')
assert result['status'] == 'ambiguous'
assert result['reason'] == 'ambiguous_person'
assert result['facts'] == []
def test_explicit_document_author_role_is_not_replaced_by_same_named_student():
data = catalog() + [{'kind': 'DocumentAuthor', 'element_id': 'author', 'id': 'author:alice', 'name': 'Alena Nová'}]
session = FakeSession({gs.CATALOG_QUERY: data})
result = gs.search_graph_with_session(session, 'Ktoré dokumenty napísal autor Alena Nová?')
params = next(params for statement, params in session.calls if statement == gs.PERSON_DOCUMENT_AUTHOR_FACTS_QUERY)
assert params['person_ids'] == []
assert params['author_ids'] == ['author']
assert not any(row['kind'] == 'Person' for row in result['resolved_entities'])
def test_unknown_document_author_does_not_list_every_document():
session = FakeSession({gs.CATALOG_QUERY: catalog()})
result = gs.search_graph_with_session(session, 'Ktoré dokumenty napísal autor Neznámy Autor?')
assert result['status'] == 'no_answer'
assert gs.PERSON_DOCUMENT_AUTHOR_FACTS_QUERY not in {statement for statement, _ in session.calls}
def test_unknown_multiword_tag_is_not_truncated_to_known_prefix():
session = FakeSession({gs.CATALOG_QUERY: catalog(), gs.THESIS_FACTS_QUERY: [thesis()]})
result = gs.search_graph_with_session(session,
'Aký názov práce má Alena Nová, ak má kľúčové slovo RAG neexistujuci?')
assert result['status'] == 'no_answer'
assert result['reason'] == 'unknown_keyword'
assert result['facts'] == []
def test_plural_tag_listing_does_not_filter_by_zp_from_zp_wiki():
session = FakeSession({gs.CATALOG_QUERY: catalog()})
gs.search_graph_with_session(session, 'Aké kľúčové slová má Alena Nová v ZP Wiki?')
params = next(params for statement, params in session.calls if statement == gs.PERSON_KEYWORD_FACTS_QUERY)
assert params['keyword_ids'] == []
def test_unsupported_count_domain_does_not_claim_zero():
session = FakeSession({gs.CATALOG_QUERY: catalog()})
result = gs.search_graph_with_session(session, 'Koľko kategórií má Alena Nová?')
assert result['status'] == 'no_answer'
assert result['reason'] == 'unsupported_count_domain'
assert result['aggregation'] is None
def test_zero_matching_candidates_is_answerable_count_zero():
session = FakeSession({gs.CATALOG_QUERY: catalog()})
result = gs.search_graph_with_session(session,
'Z kandidátov Alena Nová a Boris Malý vyber osoby s prácou v roku 2099. Koľko ich je?')
assert result['status'] == 'ok'
assert result['aggregation']['count'] == 0
assert result['aggregation']['complete']
@pytest.mark.parametrize('query', [
'Aký názov diplomovej práce má Zuzana?',
'Aký názov práce má osoba Zuzana?',
'Čomu sa venuje Zuzana?',
])
def test_single_token_unknown_subject_never_becomes_global_lookup(query):
session = FakeSession({gs.CATALOG_QUERY: catalog(), gs.THESIS_FACTS_QUERY: [thesis()]})
result = gs.search_graph_with_session(session, query)
assert result['status'] == 'no_answer'
assert 'unknown_person' in result['reason']
assert result['facts'] == []
assert all(cypher != gs.THESIS_FACTS_QUERY for cypher, _ in session.calls)
def test_trailing_instruction_is_not_an_unknown_candidate_name():
query = 'Over prepojenie osôb Alena Nová a Boris Malý cez kľúčové slovo RAG a pomenuj spoločné kľúčové slovo.'
assert gs.requested_name_phrases(query) == ['alena nova', 'boris maly']
def test_document_subject_is_not_misread_as_the_named_document_author():
query = 'Kto je autor dokumentu osoby Alena Nová, ktorý patrí do kategórie bp2024?'
assert gs.named_author_phrases(query) == []
result = gs.search_graph_with_session(FakeSession({gs.CATALOG_QUERY: catalog()}), query)
assert not result['constraints']['unresolved_document_author']
def test_tag_before_named_people_is_not_extended_with_person_clause():
query = 'Over spoločné kľúčové slovo RAG pri osobách Alena Nová a Boris Malý.'
assert gs.explicit_keyword_phrase(query) == 'rag'
result = gs.search_graph_with_session(FakeSession({gs.CATALOG_QUERY: catalog()}), query)
assert not result['constraints']['unresolved_keyword']
def test_explicit_bachelor_project_is_not_a_thesis_query():
constraints = gs.extract_query_constraints('Aký bakalársky projekt mal Peter Vzor v roku 2024?')
assert constraints['work_types'] == ['bachelor_project']
assert not constraints['thesis_types']