diff --git a/README.md b/README.md index 0f2408e..0d732c1 100644 --- a/README.md +++ b/README.md @@ -31,9 +31,9 @@ z rovnakého snapshotu a aktivujú spoločne. Pri chybe zostáva predchádzajúc pomocou embeddingov. Samostatné výsledky spája metódou RRF; osobu, typ a rok filtruje záväzne. Je vhodný na otázky na obsah, zadanie, metódy, priebeh a výsledky práce. -**GraphRAG** využíva znalostný graf osôb, dokumentov, prác, projektov, -tém, tagov a kategórií. Umožňuje vyhľadávať vzťahy, spoločné vlastnosti -a kombinácie podmienok, napríklad tag spolu s typom a rokom práce. +**GraphRAG** využíva metadátové dotazy a ohraničené prechody cez osoby, práce, +činnosti, témy a dôkazy v Neo4j. Prepojenia vysvetľuje zdrojovými výňatkami oboch prác. +Cypher šablóny sú pevné; jazykový model nevykonáva ľubovoľné databázové príkazy. Obsahový graf dopĺňa pojmy a vzťahy extrahované jazykovým modelom z profilov študentov. Každý poznatok uchováva doslovný výňatok, aktéra, konkrétnu prácu, diff --git a/test/test_local_graph.py b/test/test_local_graph.py new file mode 100644 index 0000000..d1d8600 --- /dev/null +++ b/test/test_local_graph.py @@ -0,0 +1,225 @@ +"""Witnessed local graph paths, fixed Cypher boundaries and production routing.""" +import copy +import os +import uuid + +import pytest + +from tools.answers.question_plan import build_question_plan +from tools.knowledge.sources import profile_units +from tools.knowledge.evidence import STRUCTURAL_KINDS, relation_has_local_support +from tools.knowledge import local_graph as local, store +from tools.knowledge.graph_answers import render_similarity, graph_details +from tools.knowledge.concepts import substantive_concept + + +@pytest.mark.parametrize('concept',['články','poznámky','prehľad literatúry','reading papers','zdrojové kódy','Python']) +def test_generic_study_activities_cannot_be_a_research_similarity(concept): + assert not substantive_concept(concept) + + +@pytest.mark.parametrize('concept',['znalostný graf','RAG','NER','poznámky o RAG','PEFT','anotácia pomenovaných entít']) +def test_substantive_topics_are_not_removed_with_generic_words(concept): + assert substantive_concept(concept) + + +def graph_index(): + rows=[] + entries=[('Alena Nová','alena_nova','Systém RAG používa PEFT.',['RAG','PEFT'],'RAG'), + ('Peter Starý','peter_stary','Systém chatbot používa PEFT.',['chatbot','PEFT'],'chatbot'), + ('Dana Mladá','dana_mlada','Systém RAG funguje.',['RAG'],None)] + for name,slug,text,concepts,subject in entries: + units=profile_units({'title':name,'path':f'pages/students/2022/{slug}/README.md', + 'content':'# Diplomová práca 2026\nStretnutie 5.5.2026\nStav:\n- '+text},include_records=True) + for row in units: + row.setdefault('concepts',[]);row.setdefault('relations',[]) + row.setdefault('annotation_status','structural') + if row['kind'] not in STRUCTURAL_KINDS: + row.update(annotation_status='knowledge',concepts=concepts,relations=[]) + if subject: + relation={'subject':subject,'predicate':'uses','object':'PEFT', + 'quote':subject+' používa PEFT.','verification':'supported'} + assert relation_has_local_support(row,relation) + row['relations']=[relation] + rows+=units + return {'version':'content-test','index_version':{'version':'source-test'}, + 'extraction_schema':2,'academic_schema':1,'units':rows,'profiles':3} + + +def bound(index,question='Kto pracoval na podobnej téme ako Alena Nová?'): + catalog=list({r['source_path']:{'kind':'Person','name':r['person'], + 'source_document':r['source_path']} for r in index['units']}.values()) + return build_question_plan(question,catalog,index_version=index['index_version']) + + +def witnessed_record(row): + return {'evidence':row,'person':{'name':row['person'],'source_path':row['source_path']}, + 'work':{'id':row['work_id'],'source_path':row['source_path'],'label':row['work_label'], + 'type':row['work_type'],'year':row['work_year']}, + 'activity':row['id'],'concepts':[local.concept_key(c) for c in row['concepts']]} + + +@pytest.mark.parametrize('field,value', [('person','Other Person'),('work_id','wrong-work'), + ('text','Invented result'),('phase','planned'),('source_path','other.md')]) +def test_graph_evidence_cannot_change_checkpoint_identity_or_fact(field,value): + index=graph_index();row=next(r for r in index['units'] if r.get('concepts')) + record=witnessed_record(copy.deepcopy(row));record['evidence'][field]=value + with pytest.raises(ValueError):local.validated_row(record,{row['id']:row}) + + +def test_graph_edges_must_bind_the_actual_work_and_concept(): + row=next(r for r in graph_index()['units'] if r.get('concepts')) + record=witnessed_record(row) + record['work']['id']='another-work' + with pytest.raises(ValueError):local.validated_row(record,{row['id']:row}) + record=witnessed_record(row);record['activity']=None + with pytest.raises(ValueError):local.validated_row(record,{row['id']:row}) + record=witnessed_record(row);record['concepts'].append('invented topic') + with pytest.raises(ValueError):local.validated_row(record,{row['id']:row}) + + +def test_large_witnesses_require_narrowing_instead_of_false_no_answer(): + index=graph_index();a,b=[next(r for r in index['units'] if r['person']==name and r.get('concepts')) + for name in ['Alena Nová','Peter Starý']] + a['text']='Doložený veľmi dlhý výňatok. '*1000 + result=local.LocalGraph(units=[a,b],seeds=[a],paths=[{'kind':'shared_topic','concept':'peft', + 'anchor_id':a['id'],'peer_id':b['id']}]) + response=render_similarity(bound(index).spec.question,index,bound(index),result) + assert response['status']=='ambiguous' and response['reason']=='graph_answer_budget' + + +def test_unverified_concept_cooccurrence_cannot_create_a_semantic_predicate(): + index=graph_index();row=next(r for r in index['units'] if r.get('relations')) + record={'relation':{'predicate':'uses','quote':'RAG používa PEFT.', + **{k:row.get(k) for k in ('actor','work_id','phase','source_path','source_url','line')}}, + 'evidence':row['id'],'subject':'rag','object':'peft'} + assert local.validated_relation(record,{row['id']:row}) + row['relations']=[] + with pytest.raises(ValueError):local.validated_relation(record,{row['id']:row}) + + +def test_projection_upgrade_cannot_mix_a_different_sqlite_snapshot(tmp_path,monkeypatch): + from tools.knowledge import graph_projection as projection + from contextlib import nullcontext + index=graph_index();index['_checkpoint_sha256']='checkpoint' + monkeypatch.setattr(projection,'acquire_reindex_lock',lambda *a:nullcontext()) + monkeypatch.setattr(projection,'publication_lock',lambda *a,**kw:nullcontext()) + monkeypatch.setattr(projection,'read_version',lambda *a:index['index_version']) + monkeypatch.setattr(store,'load_index',lambda *a:index) + monkeypatch.setattr(projection,'sqlite_version',lambda *a:{'version':'different'}) + monkeypatch.setattr(projection,'create_neo4j_driver',lambda *a:pytest.fail('No writes on snapshot mismatch')) + with pytest.raises(RuntimeError,match='different snapshots'): + projection.refresh_projection(tmp_path,tmp_path/'db.sqlite') + + +def test_similarity_shows_both_witnesses_and_only_two_verified_predicates_license_shared_use(): + index=graph_index();a,b=[next(r for r in index['units'] if r['person']==name and r.get('concepts')) + for name in ['Alena Nová','Peter Starý']] + path={'kind':'shared_topic','concept':'peft','anchor_id':a['id'],'peer_id':b['id'], + 'anchor_work_id':a['work_id'],'peer_work_id':b['work_id']} + result=local.LocalGraph(units=[a,b],seeds=[a],paths=[path]) + response=render_similarity(bound(index).spec.question,index,bound(index),result) + assert response['accepted_for_delivery'] and len(response['source_urls'])==2 + assert a['text'] in response['answer'] and b['text'] in response['answer'] + assert 'Spoločný overený vzťah' not in response['answer'] + result.relations=[{'evidence_id':r['id'],**r['relations'][0]} for r in [a,b]] + response=render_similarity(bound(index).spec.question,index,bound(index),result) + assert 'Spoločný overený vzťah: používa → PEFT' in response['answer'] + assert len(response['graph_paths'])==1 and len(response['graph_relations'])==2 + result.paths.append(dict(path)) + response=render_similarity(bound(index).spec.question,index,bound(index),result) + assert response['answer'].count(a['text'])==1 and response['answer'].count(b['text'])==1 + + +def test_planned_use_is_not_rendered_as_actual_shared_use(): + index=graph_index();rows=[r for r in index['units'] if r['person'] in {'Alena Nová','Peter Starý'} and r.get('concepts')] + a,b=rows;b['phase']='planned' + result=local.LocalGraph(units=rows,seeds=[a],paths=[{'kind':'shared_topic','concept':'peft', + 'anchor_id':a['id'],'peer_id':b['id']}],relations=[{'evidence_id':r['id'],**r['relations'][0]} for r in rows]) + response=render_similarity(bound(index).spec.question,index,bound(index),result) + assert 'Spoločný overený vzťah' not in response['answer'] and '(plán' in response['answer'] + + +def test_truncation_is_explicit_and_no_path_does_not_mean_no_similar_thesis_exists(): + index=graph_index();result=local.LocalGraph(truncation_reasons=['graph_concept_limit']) + response=render_similarity(bound(index).spec.question,index,bound(index),result) + assert response['status']=='no_answer' and response['truncated'] and response['complete'] is False + assert 'neexistuje' in response['answer'] and 'Zúž' in response['answer'] + + +def test_production_graph_delivery_uses_local_traversal_not_the_whole_graph(tmp_path,monkeypatch): + from tools.knowledge import query + from contextlib import nullcontext + index=graph_index();plan=bound(index) + monkeypatch.setattr(query,'index_snapshot',lambda *a,**kw:nullcontext()) + monkeypatch.setattr(query,'read_version',lambda *a:index['index_version']) + monkeypatch.setattr(query,'load_index',lambda *a:index) + monkeypatch.setattr(query,'read_graph',lambda *a:pytest.fail('Full corpus graph scan')) + called=[] + monkeypatch.setattr(local,'retrieve_local_graph',lambda *a,**kw:called.append(kw) or local.LocalGraph()) + response=query.deliver_content(plan.spec.question,engine='graphrag',limit=2,db_file=tmp_path/'db.sqlite',plan=plan) + assert called==[{'limit':2}] and response['reason']=='no_local_graph_connection' + + +@pytest.fixture +def live_graph(monkeypatch): + if os.getenv('RUN_NEO4J_TESTS')!='1':pytest.skip('Requires local Neo4j') + scope='local-graph-test-'+uuid.uuid4().hex + monkeypatch.setattr(store,'SCOPE',scope) + index=graph_index() + store.write_graph(index) + try:yield index + finally: + settings=store.get_neo4j_settings() + with store.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_live_bounded_paths_join_people_works_activities_concepts_and_checked_relations(live_graph): + index=live_graph + result=local.retrieve_local_graph(bound(index).spec.question,index,bound(index)) + assert result.paths and {r['person'] for r in result.units}=={'Alena Nová','Peter Starý','Dana Mladá'} + assert any(r['_graph_activity'] for r in result.units if r.get('concepts')) + assert {r['evidence_id'] for r in result.relations}=={r['id'] for r in index['units'] if r.get('relations')} + response=render_similarity(bound(index).spec.question,index,bound(index),result) + assert response['status']=='ok' and 'Spoločný overený vzťah' in response['answer'] + + +def test_live_wrong_year_has_no_neighborhood_fallback(live_graph): + plan=bound(live_graph,'Aké problémy riešila Alena Nová pri diplomovej práci v roku 1999?') + result=local.retrieve_local_graph(plan.spec.question,live_graph,plan) + assert result.units==[] and result.paths==[] + + +def test_live_shared_articles_do_not_introduce_an_unrelated_student(live_graph): + index=live_graph + for row in index['units']: + if row['person']=='Alena Nová' and row.get('concepts'): + row['text']+='\nPrečítané články.';row['end_line']+=1;row['concepts'].append('články') + extra=profile_units({'title':'Cyril Nový','path':'pages/students/2022/cyril_novy/README.md', + 'content':'# Diplomová práca 2026\nStav:\n- Prečítané články o nesúvisiacej téme.'},include_records=True) + for row in extra: + row.update(concepts=[] if row['kind'] in STRUCTURAL_KINDS else ['články'],relations=[], + annotation_status='structural' if row['kind'] in STRUCTURAL_KINDS else 'knowledge') + index['units']+=extra + store.write_graph(index) + result=local.retrieve_local_graph(bound(index).spec.question,index,bound(index)) + assert result.paths and not any(r['person']=='Cyril Nový' for r in result.units) + assert not any(p['concept']=='clanky' for p in result.paths) + + +def test_live_projection_version_and_tampered_evidence_fail_closed(live_graph): + with pytest.raises(RuntimeError):local.retrieve_local_graph(bound(live_graph).spec.question, + {**live_graph,'version':'different'},bound(live_graph)) + settings=store.get_neo4j_settings() + with store.create_neo4j_driver(settings) as driver,driver.session(database=settings.database) as session: + session.run("MATCH (e:KnowledgeEvidence {graph_scope:$scope,person:'Alena Nová'}) SET e.text='Invented result'",scope=store.SCOPE).consume() + with pytest.raises(ValueError):local.retrieve_local_graph(bound(live_graph).spec.question,live_graph,bound(live_graph)) + + +def test_live_evidence_budget_reports_actual_clipping(live_graph,monkeypatch): + monkeypatch.setattr(local,'EVIDENCE_LIMIT',1) + plan=bound(live_graph,'Aké metódy použila Alena Nová?') + result=local.retrieve_local_graph(plan.spec.question,live_graph,plan) + assert len(result.units)==1 and 'graph_evidence_limit' in result.truncation_reasons diff --git a/tools/knowledge/concepts.py b/tools/knowledge/concepts.py new file mode 100644 index 0000000..213698b --- /dev/null +++ b/tools/knowledge/concepts.py @@ -0,0 +1,35 @@ +"""Curated equivalents, never arbitrary related tasks or inferred relations.""" +import re +from tools.answers.context_policy import normalized + + +def concept_key(text: str) -> str: + value = normalized(text) + equivalents = { + 'rag': {'rag', 'retrieval augmented generation', 'generovanie s podporou vyhladavania'}, + 'ner': {'ner', 'named entity recognition', 'rozpoznavanie pomenovanych entit'}, + 'qa': {'qa', 'question answering', 'odpovedanie na otazky'}, + 'knowledge graph': {'knowledge graph', 'znalostny graf', 'znalostne grafy'}, + 'chatbot': {'chatbot', 'chatboty', 'chatbota'}, + } + return next((key for key, aliases in equivalents.items() if value in aliases), value) + + +GENERIC_CONCEPTS = {'python', 'nlp', 'system', 'model', 'student', 'praca', 'projekt', 'git', + 'docker', 'openwebui', 'rozhranie', 'webhook', 'generativny model', 'vyhladavat', 'data', 'text', + 'llm', 'language model', 'jazykovy model', 'neural network', 'neuronova siet', + 'machine learning', 'strojove ucenie', 'artificial intelligence', 'umela inteligencia', + 'api', 'web', 'database', 'application', 'software', 'studium'} + + +def substantive_concept(value): + """Reading/writing infrastructure is not a research-topic similarity.""" + words=normalized(value).split() + return bool(words) and normalized(value) not in GENERIC_CONCEPTS and not all(word in GENERIC_CONCEPTS or re.fullmatch( + r'clank\w*|poznam\w*|literatur\w*|resers\w*|report\w*|text\w*|kod\w*|skript\w*|' + r'zdroj\w*|prehlad\w*|teori\w*|konzult\w*|stretnut\w*|prezent\w*|' + r'implementac\w*|article\w*|paper\w*|note\w*|research|literature|code|script\w*|' + r'writing|reading|source\w*|meeting\w*|tutorial\w*|experiment\w*|metod\w*|method\w*|' + r'theor\w*|bibliograph\w*|citac\w*|citation\w*|vysled\w*|uloh\w*|zadani\w*|priprava|citanie|pisanie|' + r'o|na|a|s|z|pre|of|on|and|for|the|in|to',word) + for word in words) diff --git a/tools/knowledge/graph_answers.py b/tools/knowledge/graph_answers.py new file mode 100644 index 0000000..84ed204 --- /dev/null +++ b/tools/knowledge/graph_answers.py @@ -0,0 +1,128 @@ +"""Render witnessed graph paths; an ABOUT edge never means 'used' or 'finished'.""" +from collections import defaultdict + +from tools.answers.answer_errors import clarification_result +from tools.answers.context_policy import citation_check, text_urls +from tools.knowledge.concepts import concept_key + +PHASES={'result':'zaznamenaný výsledok','progress':'priebežný stav', + 'assignment':'zadanie','planned':'plán','unknown':'podklad'} +PREDICATES={'uses':'používa','evaluates':'vyhodnocuje','produces':'vytvára','about':'týka sa'} + + +def graph_details(answer, local): + """Keep witnesses for delivered evidence only; never append unseen claims.""" + ids={r['id'] for r in answer.get('evidence',[])} + paths=[p for p in local.paths if p['anchor_id'] in ids and p['peer_id'] in ids] + relations=[r for r in local.relations if r['evidence_id'] in ids] + reasons=list(dict.fromkeys([*answer.get('truncation_reasons',[]),*local.truncation_reasons])) + if local.truncation_reasons and answer.get('accepted_for_delivery'): + notice='Prechod grafom dosiahol limit výberu. Zúž otázku na prácu, tému alebo obdobie.' + if notice not in answer['answer']:answer={**answer,'answer':answer['answer']+'\n\n'+notice} + # Predicate claims are shown only with their own exact, phase-bound witness. + if relations and answer.get('status')=='ok': + by_id={r['id']:r for r in answer['evidence']} + lines=[] + for relation in relations[:3]: + row=by_id[relation['evidence_id']] + lines.append(f"- {row['person']} — {row['work_label']} ({PHASES.get(row['phase'],'podklad')}): " + f"{relation['subject']} → {PREDICATES[relation['predicate']]} → {relation['object']}. " + f"Dôkaz: „{relation['quote']}“") + text=answer['answer']; marker='\n\nZdroje:' + before,sep,after=text.rpartition(marker) + block='\n\nOverené významové vzťahy:\n'+'\n'.join(lines) + text=before+block+sep+after if sep else text+block + answer={**answer,'answer':text} + return {**answer,'complete':False,'truncated':bool(reasons), 'truncation_reasons':reasons, + 'graph_paths':paths,'graph_relations':relations, + 'graph_retrieval':{'strategy':'bounded_local_graph','seeds':len(local.seeds), + 'evidence':len(local.units),'paths':len(local.paths)}} + + +def render_similarity(question,index,bound,local,*,limit=5,complete=None): + spec=bound.spec + if spec.period or any(w.work_type or w.years or w.labels or w.numbers or w.excluded_labels or + w.excluded_numbers or w.unnumbered for w in spec.works): + return clarification_result('Má rok alebo typ práce obmedziť porovnávanú prácu, alebo aj hľadaných študentov? Uveď rozsah tejto podmienky.', + 'ambiguous_similarity_scope',engine='graphrag',index_version=index['index_version']) + by_id={u['id']:u for u in local.units} + valid=[p for p in local.paths if p['anchor_id'] in by_id and p['peer_id'] in by_id + and by_id[p['anchor_id']]['source_path']!=by_id[p['peer_id']]['source_path']] + candidates=list({p['peer_id']:by_id[p['peer_id']] for p in valid}.values()) + references=list({p['anchor_id']:by_id[p['anchor_id']] for p in valid}.values()) + if complete and candidates: + from tools.knowledge.query import select_topic_evidence + if len(candidates)>40:local.truncation_reasons.append('graph_selection_limit') + candidates=select_topic_evidence(question,candidates[:40],complete,references=references[:8]) + chosen={r['id'] for r in candidates} + groups=defaultdict(list) + for path in valid: + if path['peer_id'] in chosen:groups[by_id[path['peer_id']]['source_path']].append(path) + def priority(key): + rows=[by_id[p['peer_id']] for p in groups[key]] + return (-sum(r['phase'] in {'result','progress'} for r in rows),key) + paths=[];evidence={};lines=[];size=0 + if len(groups)>limit:local.truncation_reasons.append('graph_result_limit') + for source in sorted(groups,key=priority)[:limit]: + matches=groups[source] + peer=by_id[matches[0]['peer_id']] + selected=[];topics_seen=set() + for path in sorted(matches,key=lambda p:(by_id[p['peer_id']]['phase'] not in {'result','progress'},p['concept'])): + if path['concept'] in topics_seen:continue + selected.append(path);topics_seen.add(path['concept']) + if len(selected)==2:break + fits=[];used=set() + for path in selected: + ids={path['anchor_id'],path['peer_id']} - used + cost=sum(len(by_id[key]['text'])+len(by_id[key]['work_label'])+150 for key in ids) + if size+cost>18000: + local.truncation_reasons.append('graph_answer_budget');continue + fits.append(path);used|=ids;size+=cost + selected=fits + if not selected:continue + lines.append('**'+peer['person']+'**') + topics=sorted({p['concept'] for p in selected}) + lines.append('Prepája ich doložená téma: '+', '.join(topics)+'.') + shown=set() + for path in selected: + for endpoint in ('anchor_id','peer_id'): + row=by_id[path[endpoint]] + if row['id'] in shown:continue + shown.add(row['id']);evidence[row['id']]=row + stamp='; záznam '+row['record_date'] if row.get('record_date') else '' + lines.append(f"- {row['person']} — {row['work_label']} ({PHASES.get(row['phase'],'podklad')}{stamp}): {row['text']}") + paths.append(path) + # An explicit common predicate requires two verified relations, never + # just both students mentioning its endpoints. + left=[r for r in local.relations if r['evidence_id'] in shown and by_id[r['evidence_id']]['source_path']!=source] + right=[r for r in local.relations if r['evidence_id'] in shown and by_id[r['evidence_id']]['source_path']==source] + for a in left: + for b in right: + actual=lambda r:by_id[r['evidence_id']]['phase'] in {'result','progress'} + planned=lambda r:by_id[r['evidence_id']]['phase'] in {'assignment','planned'} + if a['predicate']==b['predicate'] and concept_key(a['object'])==concept_key(b['object']) and ( + actual(a) and actual(b) or planned(a) and planned(b)): + lines.append('Spoločný overený vzťah'+('' if actual(a) else ' v zadaní alebo pláne')+ + ': '+PREDICATES[a['predicate']]+' → '+a['object']+'.') + status='ok' if evidence else 'no_answer' + if not evidence and 'graph_answer_budget' in local.truncation_reasons: + return graph_details(clarification_result('Doložené prepojenia sa nezmestili do odpovede. Zúž otázku na konkrétnu prácu alebo tému.', + 'graph_answer_budget',engine='graphrag',index_version=index['index_version']),local) + if evidence: + lines.append('Prepojenie dokladá uvedenú tému alebo činnosť; nepreukazuje spoluprácu ani rovnaký stupeň dokončenia.') + else: + lines=['V ohraničenom lokálnom grafe sa nenašlo dostatočne doložené prepojenie k porovnávanej osobe. Nejde o potvrdenie, že podobná práca neexistuje.'] + urls=list(dict.fromkeys(r['source_url'] for r in evidence.values())) + if urls:lines.append('Zdroje: '+', '.join(urls)) + text='\n\n'.join([*bound.resolution.get('notices',[]),*lines]) + allowed=set(urls) + for row in evidence.values():allowed|=text_urls(row['text'])|text_urls(row['work_label']) + if not citation_check({'answer':text,'source_urls':urls},allowed)['citations_ok']: + from tools.knowledge.query import ContentAnswerRejected + raise ContentAnswerRejected('Graph path citations rejected') + return graph_details({'status':status,'reason':None if evidence else 'no_local_graph_connection', + 'engine':'graphrag','accepted_for_delivery':True,'claims':[],'answer':text,'source_urls':urls, + 'index_version':index['index_version'],'knowledge_version':index['version'], + 'evidence':list(evidence.values()),'complete':False,'truncated':False,'truncation_reasons':[], + 'answer_coverage':{'scope':'overview','status':'partial' if local.truncation_reasons else 'supported' if evidence else 'not_found', + 'warnings':[],'limitations':list(dict.fromkeys(local.truncation_reasons))}},local) diff --git a/tools/knowledge/graph_projection.py b/tools/knowledge/graph_projection.py new file mode 100644 index 0000000..3a5359b --- /dev/null +++ b/tools/knowledge/graph_projection.py @@ -0,0 +1,31 @@ +"""Re-project the current checkpoint atomically; no LLM calls or wiki changes.""" +from tools.rag.common import DATA_DIR, DB_FILE +from tools.sync.index_state import publication_lock, read_version, sqlite_version +from tools.rag.rebuild_index import acquire_reindex_lock +from tools.knowledge import store +from tools.graphrag.graph_db import get_neo4j_settings, create_neo4j_driver + + +def refresh_projection(data_dir=DATA_DIR, db_file=DB_FILE): + with acquire_reindex_lock(data_dir/'knowledge/.build.lock'), publication_lock(data_dir,exclusive=True): + index=store.load_index(data_dir,read_version(data_dir)) + if index is None or not index.get('_checkpoint_sha256'): + raise RuntimeError('A published, validated content checkpoint is required') + if sqlite_version(db_file)!=index['index_version']: + raise RuntimeError('SQLite and content checkpoint belong to different snapshots') + settings=get_neo4j_settings() + with create_neo4j_driver(settings) as driver, driver.session(database=settings.database) as session: + store.create_constraints(session) + with session.begin_transaction() as tx: + store.write_graph_transaction(tx,index,checkpoint_sha256=index['_checkpoint_sha256']) + tx.commit() + state=store.graph_state(session) + if state.get('version')!=index['version'] or state.get('projection_schema')!=store.CONTENT_GRAPH_SCHEMA: + raise RuntimeError('Graph projection verification failed') + return {'version':index['version'],'projection_schema':store.CONTENT_GRAPH_SCHEMA, + 'units':len(index['units']),'llm_requests':0} + + +if __name__=='__main__': + import json + print(json.dumps(refresh_projection())) diff --git a/tools/knowledge/local_graph.py b/tools/knowledge/local_graph.py new file mode 100644 index 0000000..ba39fc6 --- /dev/null +++ b/tools/knowledge/local_graph.py @@ -0,0 +1,239 @@ +"""Bounded, read-only graph neighborhoods with checkpoint-validated provenance.""" +from __future__ import annotations + +from dataclasses import dataclass, field +import json + +from neo4j import READ_ACCESS + +from tools.knowledge import store +from tools.knowledge.concepts import concept_key, substantive_concept +from tools.knowledge.evidence import claim_eligible, relation_has_local_support +from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings +from tools.answers.context_policy import normalized + +TIMEOUT = 8 +SEED_LIMIT = 32 +CONCEPT_LIMIT = 24 +EVIDENCE_LIMIT = 800 + +# Fixed templates only. User/model strings enter exclusively as parameters. +ROOT_QUERY = '''MATCH (p:KnowledgePerson {graph_scope:$scope})-[:HAS_EVIDENCE]->(e:KnowledgeEvidence {graph_scope:$scope}) +OPTIONAL MATCH (p)-[:HAS_WORK]->(w:KnowledgeWork {graph_scope:$scope})-[:EVIDENCED_BY]->(e) +WITH p,w,e WHERE (p.source_path IN $profiles OR + EXISTS { MATCH (w)-[:SUPERVISED_BY]->(v:KnowledgeAdvisor {graph_scope:$scope}) WHERE v.id IN $supervisors } OR + EXISTS { MATCH (w)-[:CONSULTED_BY]->(v:KnowledgeAdvisor {graph_scope:$scope}) WHERE v.id IN $consultants }) +AND ((w.id IN $works AND e.work_id=w.id) OR ($profile_records AND e.work_id IS NULL)) +AND ($date_field IS NULL OR ($start <= e[$date_field] AND e[$date_field] <= $end)) +OPTIONAL MATCH (w)-[:HAS_ACTIVITY]->(a:KnowledgeActivity {graph_scope:$scope})-[:SUPPORTED_BY]->(e) +OPTIONAL MATCH (e)-[:ABOUT]->(c:KnowledgeConcept {graph_scope:$scope}) +RETURN properties(p) AS person,properties(w) AS work,properties(e) AS evidence,a.id AS activity, + collect(DISTINCT c.key) AS concepts +ORDER BY CASE WHEN evidence.kind='work_record' THEN 0 ELSE 1 END,work.year DESC,evidence.line +LIMIT $budget''' + +SEED_QUERY = '''MATCH (p:KnowledgePerson {graph_scope:$scope})-[:HAS_WORK]->(w:KnowledgeWork {graph_scope:$scope}) + -[:EVIDENCED_BY]->(e:KnowledgeEvidence {graph_scope:$scope}) +WHERE e.id IN $ids AND e.work_id=w.id AND w.id IN $works + AND ($date_field IS NULL OR ($start <= e[$date_field] AND e[$date_field] <= $end)) +OPTIONAL MATCH (w)-[:HAS_ACTIVITY]->(a:KnowledgeActivity {graph_scope:$scope})-[:SUPPORTED_BY]->(e) +OPTIONAL MATCH (e)-[:ABOUT]->(c:KnowledgeConcept {graph_scope:$scope}) +RETURN properties(p) AS person,properties(w) AS work,properties(e) AS evidence,a.id AS activity, + collect(DISTINCT c.key) AS concepts LIMIT $budget''' +PERIOD_QUERY = SEED_QUERY.replace('e.id IN $ids AND ', '') + +NEIGHBOR_QUERY = '''MATCH (p:KnowledgePerson {graph_scope:$scope})-[:HAS_WORK]->(w:KnowledgeWork {graph_scope:$scope}) + -[:EVIDENCED_BY]->(e:KnowledgeEvidence {graph_scope:$scope})-[:ABOUT]->(c:KnowledgeConcept {graph_scope:$scope}) +WHERE c.key IN $keys AND NOT p.source_path IN $exclude AND w.id IN $works + AND e.work_id=w.id AND e.actor=p.name AND e.annotation_status='knowledge' + AND e.actor_status IN ['explicit_owner','profile_context'] + AND ($date_field IS NULL OR ($start <= e[$date_field] AND e[$date_field] <= $end)) +OPTIONAL MATCH (w)-[:HAS_ACTIVITY]->(a:KnowledgeActivity {graph_scope:$scope})-[:SUPPORTED_BY]->(e) +WITH p,w,e,a,collect(DISTINCT c.key) AS concepts +ORDER BY CASE WHEN e.phase IN ['result','progress'] THEN 0 ELSE 1 END,e.record_date DESC,e.line +WITH p,collect({work:properties(w),evidence:properties(e),activity:a.id,concepts:concepts})[..4] AS hits, + reduce(shared=[],group IN collect(concepts) | shared + [key IN group WHERE NOT key IN shared]) AS shared +ORDER BY size(shared) DESC,CASE WHEN any(hit IN hits WHERE hit.evidence.phase IN ['result','progress']) THEN 0 ELSE 1 END, + p.source_path LIMIT $peers +UNWIND hits AS hit +RETURN properties(p) AS person,hit.work AS work,hit.evidence AS evidence,hit.activity AS activity,hit.concepts AS concepts''' + +RELATION_QUERY = '''MATCH (r:KnowledgeRelation {graph_scope:$scope,origin:'llm_checked',verification:'supported'}) + -[:EVIDENCED_BY]->(e:KnowledgeEvidence {graph_scope:$scope}), + (r)-[:SUBJECT]->(s:KnowledgeConcept {graph_scope:$scope}), + (r)-[:OBJECT]->(o:KnowledgeConcept {graph_scope:$scope}) +WHERE e.id IN $ids AND r.evidence=e.id AND r.work_id=e.work_id AND r.source_path=e.source_path +RETURN properties(r) AS relation,e.id AS evidence,s.key AS subject,o.key AS object LIMIT $budget''' + +EXPAND_QUERY = '''MATCH (r:KnowledgeRelation {graph_scope:$scope,origin:'llm_checked',verification:'supported'}) + -[:EVIDENCED_BY]->(e:KnowledgeEvidence {graph_scope:$scope}), + (r)-[:SUBJECT]->(s:KnowledgeConcept {graph_scope:$scope}), + (r)-[:OBJECT]->(o:KnowledgeConcept {graph_scope:$scope}), + (p:KnowledgePerson {graph_scope:$scope})-[:HAS_WORK]->(w:KnowledgeWork {graph_scope:$scope})-[:EVIDENCED_BY]->(e) +WHERE (s.key IN $keys OR o.key IN $keys) AND w.id IN $works AND r.evidence=e.id + AND ($any_owner OR p.source_path IN $profiles) + AND e.work_id=w.id AND r.work_id=w.id AND r.source_path=p.source_path + AND ($date_field IS NULL OR ($start <= e[$date_field] AND e[$date_field] <= $end)) +OPTIONAL MATCH (w)-[:HAS_ACTIVITY]->(a:KnowledgeActivity {graph_scope:$scope})-[:SUPPORTED_BY]->(e) +RETURN properties(p) AS person,properties(w) AS work,properties(e) AS evidence,a.id AS activity, + [s.key,o.key] AS concepts,properties(r) AS relation,e.id AS evidence_id,s.key AS subject,o.key AS object +LIMIT $budget''' + + +@dataclass +class LocalGraph: + units: list[dict] = field(default_factory=list) + paths: list[dict] = field(default_factory=list) + relations: list[dict] = field(default_factory=list) + seeds: list[dict] = field(default_factory=list) + truncation_reasons: list[str] = field(default_factory=list) + + +def validated_row(record, by_id): + actual = dict(record['evidence']) + original = by_id.get(actual.get('id')) + if not original: + raise ValueError('Unknown graph evidence') + fields = ('person','actor','actor_status','source_path','source_url','text','line','end_line', + 'work_id','work_label','work_type','work_year','phase','section_scope','kind', + 'annotation_status','record_date','event_date','event_year','target_date','target_year') + if any(actual.get(k) != original.get(k) for k in fields): + raise ValueError('Graph assertion differs from its checkpoint') + person, work = dict(record['person']), dict(record['work'] or {}) + if person.get('source_path') != original['source_path'] or person.get('name') != original['person']: + raise ValueError('Cross-person graph edge') + if original.get('work_id') and any(work.get(k) != original.get(v) for k,v in + [('id','work_id'),('source_path','source_path'),('label','work_label'),('type','work_type'),('year','work_year')]): + raise ValueError('Cross-work graph edge') + if record.get('activity') and record['activity'] != original['id']: + raise ValueError('Cross-activity graph edge') + if (original.get('annotation_status')=='knowledge' and original.get('kind') in {'assignment','plan','progress','result'} + and claim_eligible(original) and record.get('activity')!=original['id']): + raise ValueError('Missing work-activity-evidence path') + keys = set(record['concepts']) + if not keys <= {concept_key(c) for c in original.get('concepts',[])}: + raise ValueError('Unverified concept edge') + return {**original, '_graph_keys': sorted(keys), '_graph_activity': record.get('activity')} + + +def seed_ids(question, index, spec): + """Classical retrieval proposes entry points; Neo4j must prove every path.""" + from tools.knowledge.query import _terms, activity_supported, concept_families + from tools.answers.content_selection import administrative_only + terms = _terms(question); families = concept_families(question) + candidates = [u for u in index['units'] if u.get('work_id') and claim_eligible(u) + and u.get('annotation_status')=='knowledge' and not administrative_only(u['text']) + and spec.matches_work(u['person'],u.get('work_label'),u.get('work_type'),u.get('work_year')) + and activity_supported(question,u)] + if spec.period and spec.period.field!='work_year': + candidates=[u for u in candidates if u.get(spec.period.field) and + spec.period.start<=u[spec.period.field]<=spec.period.end] + scores = {} + if index.get('vectors') and candidates: + import base64, zlib + import numpy as np + from tools.rag.embedding_utils import embed_query, model_profile, embedding_model_name + if model_profile(embedding_model_name())['fingerprint'] != index['embedding_fingerprint']: + raise RuntimeError('Graph seed embeddings belong to another model') + matrix=np.frombuffer(zlib.decompress(base64.b64decode(index['vectors'])),dtype='=.74: + ranked.append((hits+int(equivalent)+semantic,row)) + ranked.sort(key=lambda item:(-item[0],item[1]['source_path'],item[1]['line'])) + return [row['id'] for _,row in ranked[:SEED_LIMIT+1]] + + +def validated_relation(record, by_id): + row=by_id[record.get('evidence_id') or record['evidence']]; actual=dict(record['relation']) + match=next((r for r in row.get('relations',[]) if r.get('verification')=='supported' + and concept_key(r['subject'])==record['subject'] and concept_key(r['object'])==record['object'] + and r['predicate']==actual.get('predicate') and r['quote']==actual.get('quote') + and relation_has_local_support(row,r)),None) + if not match or any(actual.get(k)!=row.get(k) for k in ('actor','work_id','phase','source_path','source_url','line')): + raise ValueError('Unverified semantic graph relation') + return {'evidence_id':row['id'],**match} + + +def retrieve_local_graph(question, index, bound, *, limit=5): + spec=bound.spec; result=LocalGraph(); by_id={u['id']:u for u in index['units']} + people=[p for p in spec.people if p.role=='student'] + profiles=[p.source_document for p in people if p.source_document] + works=sorted({u['work_id'] for u in index['units'] if u.get('work_id') and + spec.matches_work(u['person'],u.get('work_label'),u.get('work_type'),u.get('work_year'))}) + ids=seed_ids(question,index,spec) if not profiles and not any(p.role in {'supervisor','consultant'} for p in spec.people) else [] + if len(ids)>SEED_LIMIT:result.truncation_reasons.append('graph_seed_limit') + ids=ids[:SEED_LIMIT] + period=spec.period if spec.period and spec.period.field!='work_year' else None + params={'scope':store.SCOPE,'profiles':profiles,'works':works,'ids':ids, + 'supervisors':[store.identifier(normalized(p.name)) for p in spec.people if p.role=='supervisor'], + 'consultants':[store.identifier(normalized(p.name)) for p in spec.people if p.role=='consultant'], + 'date_field':period.field if period else None,'start':period.start if period else None,'end':period.end if period else None, + 'profile_records':not any(w.work_type or w.years or w.labels or w.numbers or + w.excluded_labels or w.excluded_numbers or w.unnumbered for w in spec.works), + 'budget':EVIDENCE_LIMIT+1,'peers':min(61,max(11,limit*3+1))} + settings=get_neo4j_settings() + with create_neo4j_driver(settings,connection_timeout=TIMEOUT,max_transaction_retry_time=0) as driver,\ + driver.session(database=settings.database,default_access_mode=READ_ACCESS) as session, session.begin_transaction(timeout=TIMEOUT*4) as tx: + state=tx.run('MATCH (s:KnowledgeState {scope:$scope}) RETURN properties(s) AS state',scope=store.SCOPE).single() + state=dict(state['state']) if state else {} + if (state.get('version')!=index['version'] or state.get('projection_schema')!=store.CONTENT_GRAPH_SCHEMA + or state.get('index_version_json')!=json.dumps(index['index_version'],sort_keys=True) + or index.get('_checkpoint_sha256') and state.get('checkpoint_sha256')!=index['_checkpoint_sha256']): + raise RuntimeError('Content graph projection is stale or differs from checkpoint') + def fetch(query, **extra): + return [validated_row(dict(r),by_id) for r in tx.run(query,**{**params,**extra})] + root=fetch(ROOT_QUERY) if profiles or params['supervisors'] or params['consultants'] else fetch( + PERIOD_QUERY if spec.intent=='period' else SEED_QUERY) + if len(root)>EVIDENCE_LIMIT:result.truncation_reasons.append('graph_evidence_limit') + root=root[:EVIDENCE_LIMIT] + result.seeds=root + result.units=root + if spec.intent in {'similar','discovery','period'}: + ranked={} + for row in root: + if row.get('annotation_status')!='knowledge' or not claim_eligible(row):continue + for key in row['_graph_keys']: + if spec.intent!='similar' or substantive_concept(key): + ranked.setdefault(key,[]).append(row) + keys=sorted(ranked,key=lambda k:(-len(ranked[k]),k))[:CONCEPT_LIMIT] + if len(ranked)>CONCEPT_LIMIT:result.truncation_reasons.append('graph_concept_limit') + expansions=list(tx.run(EXPAND_QUERY,**{**params,'keys':keys, + 'any_owner':spec.intent!='similar','budget':SEED_LIMIT+1})) if keys else [] + if len(expansions)>SEED_LIMIT:result.truncation_reasons.append('graph_relation_limit') + for record in expansions[:SEED_LIMIT]: + row=validated_row(dict(record),by_id) + relation=validated_relation(dict(record),by_id) + result.relations.append(relation) + for key in row['_graph_keys']: + if (spec.intent!='similar' or substantive_concept(key)) and key not in keys: + if len(keys)>=CONCEPT_LIMIT+8: + result.truncation_reasons.append('graph_concept_limit');continue + keys.append(key);ranked.setdefault(key,[]).append(row) + root.append(row) + peers=fetch(NEIGHBOR_QUERY,keys=keys,exclude=profiles if spec.intent=='similar' else []) if keys else [] + if len({r['source_path'] for r in peers})>=params['peers']: + result.truncation_reasons.append('graph_peer_limit') + for row in peers: + for key in row['_graph_keys']: + if key not in keys:continue + anchors=[a for a in ranked[key] if spec.intent!='similar' or a['source_path']!=row['source_path']] + for anchor in sorted(anchors,key=lambda a:(a['phase'] not in {'result','progress'},a['line']))[:2]: + result.paths.append({'kind':'shared_topic','concept':key,'anchor_id':anchor['id'], + 'peer_id':row['id'],'anchor_work_id':anchor['work_id'],'peer_work_id':row['work_id'], + 'anchor_activity_id':anchor.get('_graph_activity'),'peer_activity_id':row.get('_graph_activity'), + 'concept_match':'approved_equivalent_key'}) + result.units=list({r['id']:r for r in [*root,*peers]}.values()) + records=tx.run(RELATION_QUERY,scope=store.SCOPE, + ids=[u['id'] for u in result.units],budget=EVIDENCE_LIMIT+1) + for record in records: + relation=validated_relation(dict(record),by_id) + if relation not in result.relations:result.relations.append(relation) + tx.commit() + return result diff --git a/tools/knowledge/query.py b/tools/knowledge/query.py index ca5a22a..93ba024 100644 --- a/tools/knowledge/query.py +++ b/tools/knowledge/query.py @@ -141,17 +141,7 @@ def _terms(text: str) -> set[str]: return {w[:7] if len(w) > 7 else w.rstrip("auiy") for w in words} -def concept_key(text: str) -> str: - """Only genuine abbreviations/equivalents; related tasks remain distinct.""" - value = normalized(text) - equivalents = { - "rag": {"rag", "retrieval augmented generation", "generovanie s podporou vyhladavania"}, - "ner": {"ner", "named entity recognition", "rozpoznavanie pomenovanych entit"}, - "qa": {"qa", "question answering", "odpovedanie na otazky"}, - "knowledge graph": {"knowledge graph", "znalostny graf", "znalostne grafy"}, - "chatbot": {"chatbot", "chatboty", "chatbota"}, - } - return next((key for key, aliases in equivalents.items() if value in aliases), value) +from tools.knowledge.concepts import concept_key def concept_families(text): @@ -253,6 +243,7 @@ def answer_from_index( complete=None, mode=None, plan=None, + local_graph=None, ) -> dict: mode = mode or intent(question) assert mode @@ -636,6 +627,7 @@ def answer_from_index( or equivalent or complete and semantic >= 0.74 + or complete and local_graph is not None ): scored.append((hits + semantic + int(equivalent), row)) filtered = [r for _, r in sorted(scored, key=lambda x: -x[0])] @@ -1006,7 +998,13 @@ def deliver_content(question: str, *, engine: str, limit: int, db_file, complete return public_failure({"reason": "content_index_unavailable", "engine": engine}) return None - units = read_graph(index) if engine == "graphrag" else index["units"] + local = None + if engine == 'graphrag' and plan: + from tools.knowledge.local_graph import retrieve_local_graph + local = retrieve_local_graph(question,index,plan,limit=limit) + units = local.units + else: + units = read_graph(index) if engine == "graphrag" else index["units"] from tools.knowledge.sources import source_lifecycle_notes from tools.rag.common import ZPWIKI_ROOT @@ -1023,9 +1021,16 @@ def deliver_content(question: str, *, engine: str, limit: int, db_file, complete if (note["source_path"], note.get("work_label"), note["text"]) not in existing ), ] + if local is not None and mode == 'similar': + from tools.knowledge.graph_answers import render_similarity + return render_similarity(question,index,plan,local,limit=limit,complete=complete) if mode == 'facets' and plan: from tools.knowledge.planned_answers import answer_facets - return answer_facets(plan,index,units,engine=engine,limit=limit) + result = answer_facets(plan,index,units,engine=engine,limit=limit) + if local is not None: + from tools.knowledge.graph_answers import graph_details + result = graph_details(result,local) + return result # Slow model calls must not hold the publication lock. Everything below # uses this captured source/graph generation, never a mixture of versions. if not mode: @@ -1049,7 +1054,7 @@ def deliver_content(question: str, *, engine: str, limit: int, db_file, complete units = with_work_context(units) relevance = None - if mode in {"discovery", "similar"} and index.get("vectors"): + if local is None and mode in {"discovery", "similar"} and index.get("vectors"): import base64 import zlib import numpy as np @@ -1085,7 +1090,7 @@ def deliver_content(question: str, *, engine: str, limit: int, db_file, complete else: scores = matrix @ embed_query(question) relevance = {u["id"]: float(s) for u, s in zip(index["units"], scores)} - return answer_from_index( + result = answer_from_index( question, index, engine=engine, @@ -1095,7 +1100,12 @@ def deliver_content(question: str, *, engine: str, limit: int, db_file, complete complete=complete, mode=mode, plan=plan, + local_graph=local, ) + if local is not None: + from tools.knowledge.graph_answers import graph_details + result = graph_details(result,local) + return result def needs_intent_model(question): diff --git a/tools/knowledge/store.py b/tools/knowledge/store.py index f0a8520..90db91b 100644 --- a/tools/knowledge/store.py +++ b/tools/knowledge/store.py @@ -7,8 +7,10 @@ from pathlib import Path from tools.knowledge.sources import identifier from tools.graphrag.graph_db import create_neo4j_driver, get_neo4j_settings from tools.answers.context_policy import normalized +from tools.knowledge.concepts import concept_key SCOPE = "zpwiki-content-v1" +CONTENT_GRAPH_SCHEMA = 2 def create_constraints(session) -> None: @@ -18,6 +20,7 @@ def create_constraints(session) -> None: ("person", "KnowledgePerson", "source_path"), ("work", "KnowledgeWork", "id"), ("advisor", "KnowledgeAdvisor", "id"), + ("activity", "KnowledgeActivity", "id"), ]: session.run( f"CREATE CONSTRAINT zp_content_{name} IF NOT EXISTS " @@ -27,11 +30,13 @@ def create_constraints(session) -> None: "CREATE CONSTRAINT zp_content_state IF NOT EXISTS " "FOR (n:KnowledgeState) REQUIRE n.scope IS UNIQUE" ).consume() + session.run('CREATE INDEX zp_content_concept_key IF NOT EXISTS ' + 'FOR (c:KnowledgeConcept) ON (c.graph_scope,c.key)').consume() def write_graph_transaction(tx, index: dict, *, checkpoint_sha256: str | None = None) -> None: """The caller owns the transaction and filesystem publication barrier.""" - from tools.knowledge.evidence import EXTRACTION_SCHEMA, relation_has_local_support + from tools.knowledge.evidence import EXTRACTION_SCHEMA, relation_has_local_support, claim_eligible rows = [] for unit in index["units"]: if index.get("extraction_schema") == EXTRACTION_SCHEMA and any( @@ -43,11 +48,14 @@ def write_graph_transaction(tx, index: dict, *, checkpoint_sha256: str | None = { **{k: v for k, v in unit.items() if k not in {"concepts", "relations"}}, "concepts": [ - {"id": identifier(normalized(c)), "name": c} for c in unit["concepts"] + {"id": identifier(normalized(c)), "name": c, 'key': concept_key(c)} for c in unit["concepts"] ], "relations_json": json.dumps(unit["relations"], ensure_ascii=False), } ) + tx.run( + "MATCH (n:KnowledgeActivity {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE + ).consume() tx.run( "MATCH (n:KnowledgeRelation {graph_scope:$scope}) DETACH DELETE n", scope=SCOPE ).consume() @@ -76,7 +84,7 @@ def write_graph_transaction(tx, index: dict, *, checkpoint_sha256: str | None = MERGE (p)-[:HAS_EVIDENCE]->(e) FOREACH (concept IN row.concepts | MERGE (c:KnowledgeConcept {id:concept.id,graph_scope:$scope}) - SET c.name=concept.name MERGE (e)-[:ABOUT]->(c))""", + SET c.name=concept.name,c.key=concept.key MERGE (e)-[:ABOUT]->(c))""", rows=evidence_rows, scope=SCOPE, ).consume() @@ -128,11 +136,29 @@ def write_graph_transaction(tx, index: dict, *, checkpoint_sha256: str | None = MATCH (p:KnowledgePerson {source_path:row.source_path,graph_scope:$scope}), (e:KnowledgeEvidence {id:row.evidence,graph_scope:$scope}) MERGE (w:KnowledgeWork {id:row.id,graph_scope:$scope}) - SET w.label=row.work_label,w.type=row.work_type,w.year=row.work_year + SET w.label=row.work_label,w.type=row.work_type,w.year=row.work_year,w.source_path=row.source_path MERGE (p)-[:HAS_WORK]->(w) MERGE (w)-[:EVIDENCED_BY]->(e)""", rows=works, scope=SCOPE, ).consume() + supported = [u for u in index['units'] if u.get('work_id') and claim_eligible(u)] + tx.run('''UNWIND $rows AS row + MATCH (w:KnowledgeWork {id:row.work_id,graph_scope:$scope}), + (e:KnowledgeEvidence {id:row.id,graph_scope:$scope}) + WHERE w.source_path=e.source_path AND w.label=e.work_label + MERGE (w)-[:EVIDENCED_BY]->(e)''', rows=supported, scope=SCOPE).consume() + activities = [u for u in supported if u.get('annotation_status') == 'knowledge' + and u.get('actor') == u.get('person') + and u.get('actor_status') in {'explicit_owner','profile_context'} + and u.get('kind') in {'assignment','plan','progress','result'}] + tx.run('''UNWIND $rows AS row + MATCH (w:KnowledgeWork {id:row.work_id,graph_scope:$scope}), + (e:KnowledgeEvidence {id:row.id,graph_scope:$scope}) + WHERE w.source_path=e.source_path AND w.label=e.work_label + CREATE (a:KnowledgeActivity {graph_scope:$scope,id:row.id, + work_id:row.work_id,kind:row.kind,phase:row.phase}) + CREATE (w)-[:HAS_ACTIVITY]->(a), (a)-[:SUPPORTED_BY]->(e)''', + rows=activities, scope=SCOPE).consume() for role, edge in [("supervisor", "SUPERVISED_BY"), ("consultant", "CONSULTED_BY")]: advisors = [ { @@ -181,11 +207,12 @@ def write_graph_transaction(tx, index: dict, *, checkpoint_sha256: str | None = MERGE (e)-[r:ATTRIBUTED_TO]->(p) SET r.basis=e.actor_status""", scope=SCOPE ).consume() tx.run( - "MERGE (s:KnowledgeState {scope:$scope}) SET s.version=$version, s.index_version_json=$index_version_json, s.checkpoint_sha256=$checkpoint_sha256", + "MERGE (s:KnowledgeState {scope:$scope}) SET s.version=$version, s.index_version_json=$index_version_json, s.checkpoint_sha256=$checkpoint_sha256,s.projection_schema=$projection_schema", scope=SCOPE, version=index["version"], index_version_json=json.dumps(index.get("index_version"), sort_keys=True), checkpoint_sha256=checkpoint_sha256, + projection_schema=CONTENT_GRAPH_SCHEMA, ).consume() diff --git a/tools/runtime/answer_health.py b/tools/runtime/answer_health.py index 57e89d0..7e95ba1 100644 --- a/tools/runtime/answer_health.py +++ b/tools/runtime/answer_health.py @@ -76,7 +76,10 @@ def _graph_readiness(state): actual = dict(row["state"]) if row else None if not state_matches_graph(state, actual): return readiness(False, "content_graph_version_mismatch") - return readiness(True, content_version=state["version"]) + from tools.knowledge.store import CONTENT_GRAPH_SCHEMA + if actual.get('projection_schema') != CONTENT_GRAPH_SCHEMA: + return readiness(False, 'content_graph_projection_required') + return readiness(True, content_version=state["version"], projection_schema=CONTENT_GRAPH_SCHEMA) except Exception: return readiness(False, "content_graph_unavailable")