453 lines
24 KiB
Python
453 lines
24 KiB
Python
"""Adversarial, coverage and provenance cases from the extraction audit."""
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from tools.knowledge.sources import profile_units
|
|
from tools.knowledge.evidence import actor_binding, EXTRACTION_POLICY
|
|
from tools.knowledge.academic import completion_event, answer_academic
|
|
from tools.knowledge.extraction import extract_batch, MAX_BATCH_UNITS, VERIFY_PROMPT, extraction_key
|
|
from tools.knowledge.build import batches
|
|
|
|
|
|
def units(text):
|
|
return profile_units({'title':'Alena Nová','path':'pages/students/2020/alena/README.md',
|
|
'content':'# Diplomová práca 2026\n'+text},include_records=True)
|
|
|
|
|
|
def response(value):
|
|
return {'choices':[{'message':{'content':json.dumps(value,ensure_ascii=False)}}]}
|
|
|
|
|
|
def decision(key='E1', **kwargs):
|
|
return {'evidence_id':key,'decision':'knowledge','concepts':[],'relations':[],**kwargs}
|
|
|
|
|
|
@pytest.mark.parametrize('sentence',[
|
|
'Peter Nový úspešne obhájil diplomovú prácu v roku 2024.',
|
|
'Peter Nový neobhájil diplomovú prácu v roku 2024.',
|
|
'Diplomovú prácu úspešne obhájil Peter Nový v roku 2024.',
|
|
'Diplomová práca bola úspešne obhájená Petrom Novým v roku 2024.',
|
|
'Peter úspešne obhájil diplomovú prácu v roku 2024.',
|
|
'Jeho diplomová práca bola úspešne obhájená v roku 2024.',
|
|
])
|
|
def test_another_actor_is_not_replaced_by_profile_owner(sentence):
|
|
row=units('Stav:\n- '+sentence)[-1]
|
|
assert actor_binding(row)['actor_status'] not in {'profile_context','explicit_owner'}
|
|
assert completion_event(row) is None
|
|
|
|
|
|
def test_positive_own_defense_remains_supported_and_note_year_is_separate():
|
|
row=units('Stretnutie 5.5.2026\nStav:\n- Alena Nová úspešne obhájila diplomovú prácu dňa 3.4.2024.')[-1]
|
|
assert row['actor']=='Alena Nová' and row['actor_status']=='explicit_owner'
|
|
assert row['work_year']==2026 and row['record_date']=='2026-05-05'
|
|
assert row['event_date']=='2024-04-03'
|
|
assert completion_event(row)['event_date']=='2024-04-03'
|
|
|
|
|
|
def test_third_party_defense_is_not_joined_to_owners_supervisor(engine='rag'):
|
|
rows=units('Vedúci: Daniel Hládek\nStav:\n- Peter Nový úspešne obhájil diplomovú prácu v roku 2024.')
|
|
data={'version':'v','index_version':{'version':'v'},'academic_schema':1,'units':rows}
|
|
for engine in ('rag','graphrag'):
|
|
value=answer_academic('Ktorí študenti v roku 2024 úspešne ukončili práce pod vedením Daniela Hladeka?',data,engine=engine)
|
|
assert value['status']=='no_answer'
|
|
assert not any(r.get('completion_status')=='successful' for r in value['evidence'])
|
|
|
|
|
|
@pytest.mark.parametrize('label,expected',[('Zadanie','assignment'),('Výsledky','planned'),('Stav','planned')])
|
|
def test_has_to_implement_never_becomes_implemented(label,expected):
|
|
row=units(label+':\n- Má implementovať systém RAG do 3.4.2027.')[-1]
|
|
assert row['phase']==expected
|
|
assert row['event_date'] is None and row['target_date']=='2027-04-03'
|
|
|
|
|
|
def test_negation_wins_over_results_heading():
|
|
row=units('Výsledky:\n- Model ešte nie je implementovaný.')[-1]
|
|
assert row['phase']=='progress'
|
|
|
|
|
|
def test_global_backlog_does_not_inherit_last_meeting_date_or_identity():
|
|
rows=units('Stretnutie 5.5.2026\nStav:\n- Implementovaný systém RAG.\nÚlohy na ďalšie stretnutie:\n- Otestujte systém.\nZásobník úloh:\n- Vyhodnoťte RAG pomocou DeepEval.\n')
|
|
done=next(r for r in rows if 'Implementovaný' in r['text'])
|
|
follow=next(r for r in rows if 'Otestujte' in r['text'])
|
|
backlog=next(r for r in rows if 'DeepEval' in r['text'])
|
|
assert done['record_date']==follow['record_date']=='2026-05-05'
|
|
assert follow['section_scope']=='followup_task' and follow['event_date'] is None
|
|
assert backlog['section_scope']=='backlog' and backlog['record_date'] is None
|
|
assert backlog['date'] is None and backlog['meeting_id'] is None
|
|
|
|
|
|
def test_future_period_marker_ends_meeting_context():
|
|
row=units('Stretnutie 5.5.2026\nTýka sa ďalšieho obdobia:\n- Vytvorte systém RAG.')[-1]
|
|
assert row['section_scope']=='future_period' and row['record_date'] is None and row['meeting_id'] is None
|
|
|
|
|
|
@pytest.mark.parametrize('payload',[{'items':[]},{'items':[decision('E1')]}])
|
|
def test_missing_model_decisions_are_rejected_not_silently_empty(payload):
|
|
rows=[r for r in units('Stav:\n- Používa Python.\n- Vyhodnotil BERT.') if r['kind'] not in {'work_record','meeting'}]
|
|
calls=[]
|
|
def model(messages):
|
|
calls.append(1)
|
|
return response(payload)
|
|
with pytest.raises(ValueError,match='two attempts'):
|
|
extract_batch(rows,model)
|
|
assert len(calls)==2
|
|
|
|
|
|
def test_every_item_including_batch_tail_has_explicit_decision():
|
|
rows=units('Stav:\n'+'\n'.join('- Implementovaný model BERT '+str(i)+'.' for i in range(17)))
|
|
rows=[r for r in rows if r['kind']!='work_record']
|
|
seen=[]
|
|
def model(messages):
|
|
payload=json.loads(messages[1]['content']);seen.extend(r['text'] for r in payload)
|
|
return response({'items':[decision(r['evidence_id'],concepts=['BERT']) for r in payload]})
|
|
result=[r for batch in batches(rows) for r in extract_batch(batch,model)]
|
|
assert len(result)==17 and len(seen)==17
|
|
assert max(map(len,list(batches(rows))))<=MAX_BATCH_UNITS
|
|
assert result[-1]['annotation_status']=='knowledge' and result[-1]['concepts']==['BERT']
|
|
|
|
|
|
@pytest.mark.parametrize('text,subject,predicate,obj,quote',[
|
|
('Používa Python. Vyhodnotil BERT.','Python','produces','BERT','Používa Python. Vyhodnotil BERT.'),
|
|
('Systém RAG používa Python.','Python','uses','RAG','Systém RAG používa Python.'),
|
|
('Systém RAG nepoužíva Python.','RAG','uses','Python','Systém RAG nepoužíva Python.'),
|
|
('Python sa používa pri testovaní BERT.','Python','uses','BERT','Python sa používa pri testovaní BERT.'),
|
|
])
|
|
def test_endpoint_cooccurrence_wrong_direction_and_negation_do_not_create_edges(text,subject,predicate,obj,quote):
|
|
row=units('Stav:\n- '+text)[-1]
|
|
payload={'items':[decision(concepts=[subject,obj],relations=[{'subject':subject,'predicate':predicate,'object':obj,'quote':quote}])]}
|
|
result=extract_batch([row],lambda _:response(payload))[0]
|
|
assert result['relations']==[] and result['discarded_relations']==1
|
|
assert result['text']==row['text']
|
|
|
|
|
|
@pytest.mark.parametrize('verdict,expected',[('supported',1),('unsupported',0),('uncertain',0)])
|
|
def test_direct_relation_also_requires_independent_semantic_verification(verdict,expected):
|
|
row=units('Stav:\n- Systém RAG používa Python.')[-1]
|
|
prompts=[]
|
|
def model(messages):
|
|
prompts.append(messages[0]['content'])
|
|
if messages[0]['content']==VERIFY_PROMPT:
|
|
return response({'items':[{'relation_id':'R1','decision':verdict}]})
|
|
return response({'items':[decision(concepts=['RAG','Python'],relations=[{'subject':'RAG','predicate':'uses','object':'Python','quote':'Systém RAG používa Python.'}])]})
|
|
result=extract_batch([row],model)[0]
|
|
assert len(prompts)==2 and len(result['relations'])==expected
|
|
assert result['annotation_policy']==EXTRACTION_POLICY
|
|
if expected:assert result['relations'][0]['verification']=='supported'
|
|
|
|
|
|
def test_actor_and_record_scope_are_in_cache_identity():
|
|
row=units('Stav:\n- Systém RAG používa Python.')[-1]
|
|
for key,value in [('actor','Peter Nový'),('record_date','2025-01-01'),('section_scope','backlog'),('work_label','Iná práca')]:
|
|
assert extraction_key([row],{'model':'x'})!=extraction_key([{**row,key:value}],{'model':'x'})
|
|
|
|
|
|
def test_overlong_evidence_is_explicitly_uncertain_and_never_shortened():
|
|
row=units('Stav:\n- '+('Systém RAG. '*900))[-1]
|
|
def forbidden(messages):pytest.fail('An incomplete annotation view must not be classified as full evidence')
|
|
value=extract_batch([row],forbidden)[0]
|
|
assert value['annotation_status']=='uncertain' and value['annotation_input_limited']
|
|
assert value['text']==row['text'] and value['relations']==[]
|
|
|
|
|
|
def test_missing_relation_verification_never_counts_as_supported():
|
|
row=units('Stav:\n- Systém RAG používa Python.')[-1]
|
|
attempts=[]
|
|
def model(messages):
|
|
if messages[0]['content']==VERIFY_PROMPT:
|
|
attempts.append(1)
|
|
return response({'items':[]})
|
|
return response({'items':[decision(concepts=['RAG','Python'],relations=[{'subject':'RAG','predicate':'uses','object':'Python','quote':'Systém RAG používa Python.'}])]})
|
|
with pytest.raises(ValueError,match='verification rejected'):
|
|
extract_batch([row],model)
|
|
assert len(attempts)==2
|
|
|
|
|
|
def test_global_backlog_is_not_rendered_as_last_meeting_notes():
|
|
rows=units('Stretnutie 5.5.2026\nStav:\n- Implementovaný systém RAG.\nZásobník úloh:\n- Vyhodnoťte systém pomocou DeepEval.')
|
|
data={'version':'v','index_version':{'version':'v'},'academic_schema':1,'units':rows}
|
|
value=answer_academic('Kedy mala Alena Nová posledné stretnutie a o čom bolo?',data,engine='rag')
|
|
assert 'Implementovaný systém RAG' in value['answer'] and 'DeepEval' not in value['answer']
|
|
|
|
|
|
def test_passive_success_statement_remains_a_documented_work_outcome():
|
|
row=units('Stav:\n- Práca úspešne obhájená.')[-1]
|
|
assert completion_event(row)['status']=='successful'
|
|
assert completion_event(row)['event_year'] is None
|
|
|
|
|
|
def test_example_topic_is_not_promoted_to_the_students_research_assertion():
|
|
row=units('Zadanie:\n- Implementujte systém RAG.\nNapr.:\n- Téma rozpoznávanie entít.')[-1]
|
|
assert row['section_scope']=='example'
|
|
value=extract_batch([row],lambda _:response({'items':[decision(concepts=['rozpoznávanie entít'])]}))[0]
|
|
assert value['annotation_status']=='irrelevant' and not value['concepts'] and not value['relations']
|
|
|
|
|
|
def test_known_person_is_not_lost_when_all_their_annotations_are_uncertain():
|
|
from tools.knowledge.query import answer_from_index
|
|
row=units('Stav:\n- Vytvorený systém RAG.')[-1]
|
|
row.update(annotation_status='uncertain',annotation_reason='Uncertain scope',concepts=[],relations=[])
|
|
value=answer_from_index('Ako pokročila Alena Nová?',{'version':'v','index_version':{'version':'v'},'units':[row]},engine='rag')
|
|
assert value['name_resolution']['status']=='ok' and value['status']=='no_answer'
|
|
|
|
|
|
def builder_fixture(tmp_path):
|
|
import hashlib
|
|
from tools.sync.sync_indexes import source_files
|
|
from tools.sync.index_state import durable_json
|
|
wiki=tmp_path/'wiki';data=tmp_path/'data';data.mkdir()
|
|
path=wiki/'pages/students/2020/alena/README.md';path.parent.mkdir(parents=True)
|
|
path.write_text('---\ntitle: Alena Nová\n---\n# Diplomová práca 2026\nStav:\n'+'\n'.join('- Implementovaný systém RAG '+str(i)+'.' for i in range(8)))
|
|
version={'version':'v','source_sha256':hashlib.sha256(json.dumps(source_files(wiki),sort_keys=True).encode()).hexdigest()}
|
|
durable_json(data/'index_version.json',version)
|
|
return wiki,data
|
|
|
|
|
|
def test_builder_subdivides_incomplete_batch_without_losing_or_faking_decisions(tmp_path):
|
|
from tools.knowledge.build import build
|
|
wiki,data=builder_fixture(tmp_path)
|
|
seen=[]
|
|
def model(messages):
|
|
payload=json.loads(messages[1]['content'])
|
|
seen.append(len(payload))
|
|
if len(payload)>2:
|
|
return response({'items':[]})
|
|
return response({'items':[decision(r['evidence_id'],concepts=['RAG']) for r in payload]})
|
|
report=build(wiki,data,model,{'model':'fixture'},graph=False,vectors=False,workers=1)
|
|
value=json.loads((data/'knowledge/index.json').read_text())
|
|
assert report['model_requests']>report['llm_batches']
|
|
assert value['coverage']['decisions']['knowledge']==8
|
|
assert len({r['id'] for r in value['units']})==len(value['units'])
|
|
assert 8 in seen and 2 in seen
|
|
|
|
|
|
def test_unresolved_single_unit_cannot_replace_active_checkpoint(tmp_path):
|
|
from tools.knowledge.build import build
|
|
wiki,data=builder_fixture(tmp_path)
|
|
def valid(messages):
|
|
payload=json.loads(messages[1]['content'])
|
|
return response({'items':[decision(r['evidence_id'],concepts=['RAG']) for r in payload]})
|
|
build(wiki,data,valid,{'model':'good'},graph=False,vectors=False,workers=1)
|
|
before=(data/'knowledge/index.json').read_bytes()
|
|
with pytest.raises(ValueError,match='two attempts'):
|
|
build(wiki,data,lambda _:response({'items':[]}),{'model':'broken'},graph=False,vectors=False,workers=1)
|
|
assert before==(data/'knowledge/index.json').read_bytes()
|
|
|
|
|
|
def test_graph_import_refuses_forged_supported_flag_before_any_write():
|
|
from tools.knowledge.store import write_graph_transaction
|
|
row=units('Stav:\n- Používa Python. Vyhodnotil BERT.')[-1]
|
|
row.update(concepts=['Python','BERT'],relations=[{'subject':'Python','predicate':'produces','object':'BERT','quote':'Používa Python. Vyhodnotil BERT.','verification':'supported'}])
|
|
class Forbidden:
|
|
def run(self,*args,**kwargs):pytest.fail('Invalid relationship reached graph writer')
|
|
with pytest.raises(ValueError,match='Unverified'):
|
|
write_graph_transaction(Forbidden(),{'extraction_schema':2,'units':[row]})
|
|
|
|
|
|
def test_old_annotation_policy_is_explicitly_unavailable(tmp_path):
|
|
from tools.knowledge.publication import checkpoint_state
|
|
from tools.knowledge.store import load_index
|
|
old={'version':'old','index_version':{'version':'v'},'units':[],'extraction_policy':'legacy'}
|
|
checkpoint_state(tmp_path,old)
|
|
with pytest.raises(RuntimeError,match='policy changed'):
|
|
load_index(tmp_path,{'version':'v'})
|
|
|
|
|
|
@pytest.mark.parametrize('text',[
|
|
'Vyhodnotil model BERT publikovaný dňa 3.4.2024.',
|
|
'Úspešne obhájila diplomovú prácu z roku 2024.',
|
|
'Úspešne obhájila diplomovú prácu do 3.4.2024.',
|
|
])
|
|
def test_publication_work_year_and_deadline_are_not_exact_event_timing(text):
|
|
row=units('Stav:\n- '+text)[-1]
|
|
assert row['event_date'] is None and row['event_year'] is None
|
|
|
|
|
|
def test_paper_publication_date_is_not_a_planned_deadline():
|
|
row=units('Zadanie:\n- Preštudujte článok z 3.4.2024.')[-1]
|
|
assert row['target_date'] is None and row['event_date'] is None
|
|
|
|
|
|
def test_as_of_status_date_does_not_become_execution_date():
|
|
row=units('Stav:\n- Implementovaný systém RAG ku dňu 3.4.2024.')[-1]
|
|
assert row['event_date'] is None
|
|
|
|
|
|
@pytest.mark.skipif(__import__('os').getenv('RUN_NEO4J_TESTS')!='1',reason='Requires local Neo4j')
|
|
def test_verified_graph_relation_carries_exact_actor_work_phase_and_source(monkeypatch):
|
|
import uuid
|
|
from tools.knowledge import store
|
|
from tools.graphrag.graph_db import create_neo4j_driver,get_neo4j_settings
|
|
scope='assertion-test-'+uuid.uuid4().hex
|
|
monkeypatch.setattr(store,'SCOPE',scope)
|
|
row=units('Stretnutie 5.5.2026\nStav:\n- Systém RAG používa Python.')[-1]
|
|
row.update(annotation_status='knowledge',concepts=['RAG','Python'],relations=[{'subject':'RAG','predicate':'uses','object':'Python','quote':'Systém RAG používa Python.','verification':'supported'}])
|
|
value={'version':'test','index_version':{'version':'source'},'extraction_schema':2,'units':[row]}
|
|
settings=get_neo4j_settings()
|
|
try:
|
|
store.write_graph(value)
|
|
with create_neo4j_driver(settings) as driver,driver.session(database=settings.database) as session:
|
|
found=session.run('''MATCH (r:KnowledgeRelation {graph_scope:$scope})-[:SUBJECT]->(a),
|
|
(r)-[:OBJECT]->(b),(r)-[:EVIDENCED_BY]->(e)-[:ATTRIBUTED_TO]->(p)
|
|
RETURN properties(r) AS relation,a.name AS subject,b.name AS object,p.name AS actor''',scope=scope).single()
|
|
assert found['subject']=='RAG' and found['object']=='Python' and found['actor']=='Alena Nová'
|
|
assert found['relation']['work_id']==row['work_id'] and found['relation']['phase']==row['phase']
|
|
assert found['relation']['record_date']=='2026-05-05'
|
|
assert found['relation'].get('event_date') is None
|
|
assert found['relation']['quote']=='Systém RAG používa Python.'
|
|
finally:
|
|
with create_neo4j_driver(settings) as driver,driver.session(database=settings.database) as session:
|
|
session.run('MATCH (n {graph_scope:$scope}) DETACH DELETE n',scope=scope).consume()
|
|
session.run('MATCH (s:KnowledgeState {scope:$scope}) DELETE s',scope=scope).consume()
|
|
|
|
|
|
def test_output_budget_exhaustion_uses_smaller_batches_without_fake_empty_rows(tmp_path):
|
|
from tools.knowledge.build import build
|
|
wiki,data=builder_fixture(tmp_path)
|
|
def model(messages):
|
|
payload=json.loads(messages[1]['content'])
|
|
if len(payload)>2:
|
|
raise RuntimeError('Incomplete answer model output')
|
|
return response({'items':[decision(r['evidence_id'],concepts=['RAG']) for r in payload]})
|
|
result=build(wiki,data,model,{'model':'limited'},graph=False,vectors=False,workers=1)
|
|
assert result['decisions']['knowledge']==8
|
|
|
|
|
|
def test_provider_failure_is_not_misreported_as_uncertain_content(tmp_path):
|
|
from tools.knowledge.build import build
|
|
wiki,data=builder_fixture(tmp_path)
|
|
def unavailable(messages):raise RuntimeError('Answer model is not configured')
|
|
with pytest.raises(RuntimeError,match='not configured'):
|
|
build(wiki,data,unavailable,{'model':'missing'},graph=False,vectors=False,workers=1)
|
|
assert not (data/'knowledge/index.json').exists()
|
|
|
|
|
|
def test_extra_relation_candidates_are_all_validated_without_losing_unit_decision():
|
|
row=units('Stav:\n- Používa Python. Vyhodnotil BERT.')[-1]
|
|
candidates=[{'subject':'Python','predicate':'produces','object':'BERT','quote':'Používa Python. Vyhodnotil BERT.'} for _ in range(20)]
|
|
value=extract_batch([row],lambda _:response({'items':[decision(concepts=['Python','BERT'],relations=candidates)]}))[0]
|
|
assert value['annotation_status']=='knowledge'
|
|
assert value['discarded_relations']==20 and value['relations']==[]
|
|
assert len(json.loads(value['rejected_relations_json']))==20
|
|
|
|
|
|
def test_invalid_candidate_is_rejected_without_discarding_a_valid_unit_decision():
|
|
row=units('Stav:\n- Systém RAG používa Python.')[-1]
|
|
value=extract_batch([row],lambda _:response({'items':[decision(concepts=['RAG','Python'],relations=[{'subject':'RAG','predicate':'anything','object':'Python'}])]}))[0]
|
|
assert value['annotation_status']=='knowledge' and value['discarded_relations']==1 and not value['relations']
|
|
|
|
|
|
def test_uncertain_decision_never_activates_even_model_supplied_concepts_or_edges():
|
|
row=units('Stav:\n- Systém RAG používa Python.')[-1]
|
|
value=extract_batch([row],lambda _:response({'items':[decision(decision='uncertain',concepts=['RAG','Python'],relations=[{'subject':'RAG','predicate':'uses','object':'Python','quote':'Systém RAG používa Python.'}])]}))[0]
|
|
assert value['annotation_status']=='uncertain' and not value['concepts'] and not value['relations']
|
|
|
|
|
|
@pytest.mark.parametrize('text',[
|
|
'Zaujala ma metóda prekladu pomocou špeciálnych znakov.',
|
|
'Model má problém s vyhľadávaním na internete.',
|
|
'Vyhodnoťte model a navrhnite jeho zlepšenia.',
|
|
'Používa Python. Vyhodnotil BERT.',
|
|
])
|
|
def test_non_person_subject_and_model_possessive_do_not_invent_other_actors(text):
|
|
row=units('Stav:\n- '+text)[-1]
|
|
assert row['actor']=='Alena Nová' and row['actor_status']=='profile_context'
|
|
|
|
|
|
def test_other_persons_planned_activity_is_not_assigned_to_profile_owner():
|
|
row=units('Stav:\n- Peter má implementovať vyhľadávací systém.')[-1]
|
|
assert row['actor']!='Alena Nová' and row['actor_status']=='uncertain'
|
|
|
|
|
|
def test_examples_of_questions_are_not_activated_as_students_results():
|
|
row=units('Stretnutie 5.5.2026\nPríklady otázok:\n- Kedy dosiahol aký výsledok?')[-1]
|
|
assert row['section_scope']=='example'
|
|
value=extract_batch([row],lambda _:response({'items':[decision()]}))[0]
|
|
assert value['annotation_status']=='irrelevant'
|
|
|
|
|
|
def test_bounded_workers_start_next_batch_without_waiting_for_slow_first_batch(tmp_path):
|
|
import threading
|
|
from tools.knowledge.build import build
|
|
wiki,data=builder_fixture(tmp_path)
|
|
path=wiki/'pages/students/2020/alena/README.md'
|
|
path.write_text(path.read_text()+'\n'+'\n'.join('- Implementovaný systém RAG '+str(i)+'.' for i in range(8,24)))
|
|
import hashlib
|
|
from tools.sync.sync_indexes import source_files
|
|
from tools.sync.index_state import durable_json
|
|
durable_json(data/'index_version.json',{'version':'v','source_sha256':hashlib.sha256(json.dumps(source_files(wiki),sort_keys=True).encode()).hexdigest()})
|
|
third_started=threading.Event()
|
|
lock=threading.Lock()
|
|
active=0
|
|
peak=0
|
|
def model(messages):
|
|
nonlocal active,peak
|
|
payload=json.loads(messages[1]['content'])
|
|
number=int(payload[0]['text'].split()[-1].rstrip('.'))
|
|
with lock:
|
|
active+=1;peak=max(peak,active)
|
|
try:
|
|
if number==0:
|
|
assert third_started.wait(5),'A slow first batch blocked an otherwise free worker'
|
|
elif number>=16:
|
|
third_started.set()
|
|
return response({'items':[decision(r['evidence_id'],concepts=['RAG']) for r in payload]})
|
|
finally:
|
|
with lock:active-=1
|
|
report=build(wiki,data,model,{'model':'bounded'},graph=False,vectors=False,workers=2)
|
|
assert report['decisions']['knowledge']==24 and peak==2
|
|
assert third_started.is_set()
|
|
|
|
|
|
@pytest.mark.parametrize('text',[
|
|
'Dôsledne používajte jednotnú terminológiu.',
|
|
'Vypracujte používateľskú a systémovú príručku.',
|
|
])
|
|
def test_imperative_and_user_manual_are_not_other_persons_finite_actions(text):
|
|
row=units('Zadanie:\n- '+text)[-1]
|
|
assert row['actor']=='Alena Nová' and row['actor_status']=='profile_context'
|
|
assert row['phase']=='assignment'
|
|
|
|
|
|
def test_supervisor_in_an_illustrative_example_is_not_a_real_supervision_fact():
|
|
from tools.knowledge.sources import advisor_name
|
|
from tools.knowledge.evidence import claim_eligible
|
|
row=units('Príklady:\nVedúci: Daniel Hládek')[-1]
|
|
assert row['kind']=='supervisor' and row['section_scope']=='example'
|
|
assert advisor_name(row,'supervisor') is None
|
|
assert not claim_eligible(row)
|
|
|
|
|
|
def test_uncertain_topic_cannot_be_reintroduced_through_work_context():
|
|
from tools.knowledge.sources import with_work_context
|
|
rows=units('Téma: Rozpoznávanie pomenovaných entít\nStav:\n- Implementovaný systém RAG.')
|
|
topic=next(r for r in rows if r['kind']=='topic')
|
|
topic['annotation_status']='uncertain'
|
|
result=next(r for r in with_work_context(rows) if r['kind']=='result')
|
|
assert result['context_quotes']==[]
|
|
|
|
|
|
def test_uncertain_assertion_cannot_influence_semantic_similarity_anchor(tmp_path,monkeypatch):
|
|
import base64,zlib
|
|
import numpy as np
|
|
from contextlib import nullcontext
|
|
from tools.knowledge import query
|
|
from tools.rag import embedding_utils
|
|
supported=units('Stav:\n- Implementovaný systém RAG.')[-1]
|
|
unsupported={**supported,'id':'uncertain','annotation_status':'uncertain','text':'Iná téma.'}
|
|
other={**supported,'id':'other','person':'Boris Nový','source_path':'pages/students/2020/boris/README.md'}
|
|
matrix=np.asarray([[1,0],[0,1],[1,0]],dtype='<f4')
|
|
value={'version':'v','index_version':{'version':'v'},'academic_schema':1,
|
|
'units':[supported,unsupported,other],'embedding_fingerprint':'test',
|
|
'vector_shape':list(matrix.shape),'vectors':base64.b64encode(zlib.compress(matrix.tobytes())).decode()}
|
|
monkeypatch.setattr(query,'index_snapshot',lambda *a,**k:nullcontext())
|
|
monkeypatch.setattr(query,'load_index',lambda *a:value)
|
|
monkeypatch.setattr(query,'read_version',lambda *a:{'version':'v'})
|
|
monkeypatch.setattr(query,'answer_from_index',lambda *a,**k:k['relevance'])
|
|
monkeypatch.setattr(embedding_utils,'embedding_model_name',lambda:'test')
|
|
monkeypatch.setattr(embedding_utils,'model_profile',lambda *a:{'fingerprint':'test'})
|
|
result=query.deliver_content('Kto pracoval na podobnej téme ako Alena Nová?',engine='rag',limit=5,db_file=tmp_path/'db.sqlite')
|
|
assert result['other']==pytest.approx(1.0)
|