GraphRAG з нуля за дапамою Neo4j: вектары плюс розгортанне графа
Створэнне GraphRAG з нуля на Neo4j у Docker: запрацоўкі дакументаў, пошук па вектарах, розшырэнне суседства графа, дадаць падсумкі спільнатэка, а потым параваненне з звычным RAG.
Крыніцы: Building GraphRAG from Scratch: Knowledge Graph-Powered RAG with Neo4j, Python, and LLMs
У гэтым карыце парадоксальнае выкананне шляху ад сыр'ёў да рабочай сістэмы для: Стварэння GraphRAG з нуля: RAG, які выкарыстоўвае графы знанняяў, з Neo4j, Python і LLM-амі. Акцэнт ставяцца на практычныя крокі, чысткія перакананні і код, які можна проста падставіць у рэпазітарый без неабяснення меты. Для агульнага відображэння, перш чым зменяць код, неабходна візначыць вхідныя даны, адпаведальнага за крок і крытэрыя завершэння. Аперацыёныя працавікі должны магчымае перадзваніць крок з вядомай точкі контролю без неабяснення схованага стану. Спрыятлівайце гэты ўрадок як даговор межа вхіднымі данымі і падтверджанымі выходнымі рэзультатамі. Дайце назвы артыфактам, візначыце перакананні на успех і адмовіцеся ад тыхоўага частковага завершэння.
Введэнне
Для пачатку, перад тым як зменіць код, неабяжна ўзначыць вхідныя даны, адпаведальнага за крок і крытэрыя завершэння. Аператары должны магчыма ўвайсці крок з вядомай точкі контролю, не спрабоўваючы здогадвацца пра схованы стан. Запісвайце час выканання і вартасьць токеноў або запытанняя пры функцыйнаых рэзультаатах. Візуабельнасьць вартасей з самага пачатку запобегае неспакойным рахункам, калі працэс пераходзіць з дэмавайнага рэжыма ў спяльныя сераўеры. Стварайце вузлі і зв’язкі для тых запытанняў, якія вы хочаце задаць, а не для кожнага іменніка ў дакумэнце. Разрозненыя, тыпаваныя рэбры лепшыя за ўжоўтучныя, загадковыя графы.
Што такое GraphRAG?
Для тага «Што такое GraphRAG?» неабяжна прадэфінаваць вхідныя даны, абяронца крока і крытэрыі завершэння пры змяне коду. Аперацыйныя працовнікі павінны магчыма ўвайсці крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра схованы стан. Конфігурацыю трэба залічыць праза код аплікацыі. Файлы сераўнавання, хранілішча секрэтных дадзеных і флагі функцый павінны знаходзіцца ў адном месцы, якое аперацыйныя працовнікі можуць пераглядаць, не чытаючы весь граф. Складзіце вузлы і зв’язкі модэля для запітанняў, якія вы будзеце ставіць, а не для кожнага іменніка ў дакуменце. Рэдкія, тыпаваныя рэбры лепшыя за ўжоўтучныя, загадковыя графы.
Граф знанняў проты GraphRAG
Для парадыгм Knowledge Graph і GraphRAG неабяжна ўзначыць вхідныя даны, адпаведальнага за кожны крок і крытэрыі завершэння пры перадзеяванні коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабяжна задокументаваць як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, пераказы людзям і апрацоўка некорэктных паведамленняў ёсць часткай продукту, а не етапамі далейшай наладкі. Ствараць вузлы і зв’язкі для тых запітанняў, якія вы будзеце ставіць, а не для кожнага іменніка ў дасведчэнні. Рыхлыя, тыпаваныя рэбры лепшыя за ўжоўтучныя, незрозумелыя графы. Для парадыгм Knowledge Graph і GraphRAG неабяжна ўзначыць вхідныя даны, адпаведальнага за кожны крок і крытэрыі завершэння пры перадзеяванні коду. Аператары должны магчымаць перзапуск кроку з вядомай точкі контролю, не падозрываючы прыхованы стан. Спрыяць гэтаму этапу як кантракту між вхіднымі данымі і перакананымі выходнымі рэзультатамі. Даць назвы элементам, узначыць крытэрыі успеху і не прабаваць прыймаць часткова завершаныя рэзультаты без паведамлення.
Архітэктура
Для архітэктуры неабяжна практыка задаць вхідныя даны, абавесніка кожнага крока і крэтырыя завершэння пры зміне коду. Аперацыйныя працавнікі должны магчымае запускаць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Запісваць трываласць выканання і кост токенаў або запытаў разам з функцыйнымі рэзультатамі. Відразлівасць костаў з самага пачатку запобегае неспакойным рахункам, калі працэс пераходзіць з дэмавай версіі ў спяльныя асяроддзя. Ўжываць обмежэнняі і індэксы пры масовай абрабоце даных. Унікальнасць бізнес-клучоў ператварае пазнейшыя з’еднання даных у прагнозавальныя операцыі адкрыціі/замены, а не ў сітуацыю дуплікацый.
Documents
↓
Chunking
↓
Embedding generation
↓
Vector index in Neo4j
↓
LLM-based entity and relationship extraction
↓
Knowledge graph construction
↓
Community detection
↓
Community summarization
↓
Hybrid retrieval:
- vector search
- graph expansion
- community summary retrieval
↓
LLM answer generation
Рэалізацыя «з канца ў канец»
Для абсалютнага выкарыстоўвання неабходна перад змянай коду адзначыць вхідныя даны, адпаведальнага за кожны крок і крытэрыя завершэння. Аператары должны магчымае перайсці на гэты крок з вядомага пункта контролю, не спрабоўваючы здогадвацца пра схованы стан. Конфігурацыю трэба залічыць паза кодам прыкладнення. Файлы сераўнавання, хранільнікі секрэтных дадзеных і флагі функцыйяў должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць, не чытаючы весь ланцуг задач. Перад масовым запісам трэба выкарыстоўваць обмежэння і індэксы. Унікальнасць ключоў дадзеных ператварае пазнейшыя з’еднання на прагнозаваныя апэратыўныя запісы, а не на сітуацыю з дуплікатамі.
1. Структура праекта
Для 1. Структуры праекта неабяжна прадзефінаваць вхідныя даны, адпаведальную особу за кожны крок і крэтыяры завершэння пры зміне коду. Аперацыйныя працавнікі должны магчымець перзапускаць крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра захаваны стан. Неабяжна задокументаваць як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, людзкія перакрыцця і апрацоўка некорэктных паведамленняў ёсць часткай продукту, а не чымсь, што дадаецца пазней. Ўжывайце обмежэнняі і індэксы пры масовай абрабоце. Унікальнасць бізнес-клучоў ператварае пазнейшыя з’еднання на прывідныя апэратыўныя дзеянні заместа стварэння дуплікатаў. Для 1. Структуры праекта неабяжна прадзефінаваць вхідныя даны, адпаведальную особу за кожны крок і крэтыяры завершэння пры зміне коду. Аперацыйныя працавнікі должны магчымець перзапускаць крок з вядомага пункта контролю, не спрабоўваючы здагадвацца пра захаваны стан. Спрыяйце цэму этапу як кантракту межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Даўце назвы артыфактам, прадзефінаваць перакрыцця успеху і не прабоўваць прыймаць часткова завершаныя рэзультаты без паведамлення.
graphrag-neo4j-demo/
│
├── docker-compose.yml
├── requirements.txt
├── .env.example
├── app.py
└── documents/
└── brain_progression
2. Запуск Neo4j за допаможчыю Docker
Для пункту 2. Запуск Neo4j за допаможчыю Docker неабяжна прадзеявіць параметры вводу, адпаведнага власніка крока і крытэрыя завершэння пры перадзмене коду. Аперацыйныя працавнікі должны магчымае перазапускаць крок з вядомай точкі контролю, не прабуючы спадарацца прыватны стан. Запісваюць час выканання і вартасць токена або запыту разам з функцыйнальнымі рэзултатамі. Відразы вартасці з самага пачатку запобегае неспакоўным рахункам, калі працэс пераходзіць з дэмовай асяроддзя ў спяльнаныя сераўеры. Пасля першага пакета дадзеных трэба адразу пераглянуць прыкладны район. Інструменты тыпу Cypher дапамагаюць выявіць змены ў атрыбутах і відсутнае значэнняя ўластыў дарожча, чым тэсты на весь цыкл.
version: "3.9"
services:
neo4j:
image: neo4j:5.22
container_name: graphrag-neo4j
ports:
- "7474:7474"
- "7687:7687"
environment:
NEO4J_AUTH: neo4j/password123
NEO4J_PLUGINS: '["apoc"]'
NEO4J_dbms_security_procedures_unrestricted: apoc.*
NEO4J_dbms_security_procedures_allowlist: apoc.*
volumes:
- neo4j_data:/data
- neo4j_logs:/logs
volumes:
neo4j_data:
neo4j_logs:
docker compose up -d
http://localhost:7474
Username: neo4j
Password: password123
3. Інсталяцыя залежнасцэй Python
Для пункта 3: перад змінайом код неабяжна установка залежнасцей Python, а таксама ваказваўце вхідных дадзеных, адпавядаючага адпаведальнага за крок і крэтарыяў выходу. Аператары должны магчымае перапрацаваць крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Конфігурацыю трэба залічыць праз аплявацыйны код. Файлы сераўіса, хранальнікі секрэтных дадзеных і флагі функцыяў должны знаходзіцца ў аднам месцы, якое аператары можаць пераглядаць, не чытаяўшы весь граф. Пасля першага пакета неабяжна адзірнуць прыкладны район. За дапамогою з’явоў Cypher можна легка выявіць змены атрыбутаў і відсутнае сваяйства, чым дэшэўа, чым тэсты на весь цяг.
neo4j==5.22.0
openai>=1.40.0
python-dotenv==1.0.1
numpy==1.26.4
networkx==3.3
tqdm==4.66.4
python -m venv .venv
source .venv/bin/activate
pip install -r requirements
python -m venv .venv
.venv\Scripts\activate
pip install -r requirements.txt
4. Канфігурацыя зменных сераўіса
Для пункту 4. Канфігурацыя зменных сэраўніка: перад зменай коду неабходна задаць вхідныя даны, адпаведальнага за крок і крэтыяры завершэння. Аператары должны магчымае перзапускаты крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Неабходна задокументаваць як шлях успеху, так і шлях вярнення да нормы. Перапрыбуткі, людзкія перакрыцця і апрацоўка некорэктных паведамленняў ёсць часткай продукту, а не чымсь, што дадаецца пазней. Пасля першай партыі неабходна адзірнуць прыкладны набор даных. Засобы тыпу Cypher працююць эфектываўшы ў выявленні змян у атрыбутах і ўтраты даных, калі порৱняваюцься з тэстамі чату ад пачатку да канца. Для пункту 4. Канфігурацыя зменных сэраўніка: перад зменай коду неабходна задаць вхідныя даны, адпаведальнага за крок і крэтыяры завершэння. Аператары должны магчымае перзапускаты крок з вядомай точкі контролю, не падозрываючы прыхованы стан. Спрыяйце таму, каб гэты этап быў розумеваны як контракт межа вхіднымі данымі і перакананымі выходнымі рэзультатамі. Назвайце всі элементы, задаць крэтыяры успеху і не прабывайце прыймаць часткова завершаныя рэзультаты без паведамлення.
OPENAI_API_KEY=your_api_key_here
OPENAI_CHAT_MODEL=gpt-4o-mini
OPENAI_EMBEDDING_MODEL=text-embedding-3-small
NEO4J_URI=bolt
cp .env.example .env
5. Дадзіце прыкладныя дакументы
Для пункта 5. Дадзіце прыкладныя дакументы неабходна ўсканаваць параметры вводу, адпаведальнага за выкананне крока і крэтарыяі завершэння працы, перш чым змяніць код. Аператары должны магчымае перадзеісцаваць крок з вядомай точкі контролю, не прабуючы спадароўваць сустоянне, якое залішылася непазначанае. Запісвайце час выканання і кост токенаў або запытаў праза рэзультатамі функцыйнасці. Відразлівае паказанне костаў запобегае неспадзяваным расчыткам, калі працэс пераходзіць з дэмавай версіі ў спяльныя асяроддзя. Ствараце вузлі і зв’язкі для тых пытанняў, якія вы хочаце задаць, а не для кожнага іменніка ў дакуменце. Рэдкія, тыпаваныя зв’язкі лепшыя за ўжоўтучаныя, незрозумелыя графы.
mkdir documents
Alzheimer's disease progression is commonly associated with amyloid beta accumulation, tau pathology, neurodegeneration, and cognitive decline.
Amyloid beta accumulation often appears early in the disease cascade and is associated with increased risk of progression from mild cognitive impairment to Alzheimer's dementia.
Tau pathology is frequently linked with neuronal injury and cognitive decline. Elevated tau levels are often measured using cerebrospinal fluid biomarkers or PET imaging.
Hippocampal atrophy is an MRI-based imaging marker associated with memory impairment and neurodegeneration. Patients with both amyloid positivity and hippocampal atrophy may show faster progression.
Mild Cognitive Impairment, also called MCI, is a clinical stage between normal aging and dementia. MCI patients with abnormal amyloid and tau biomarkers have higher risk of conversion to Alzheimer's dementia.
MMSE and ADAS-Cog are cognitive assessments often used to track cognitive decline over time. Declining MMSE scores may indicate worsening disease severity.
APOE4 is a genetic risk factor associated with increased probability of amyloid accumulation and Alzheimer's disease progression.
Neurodegeneration can be assessed through structural MRI, FDG-PET, and other imaging modalities. FDG-PET hypometabolism is associated with reduced brain glucose metabolism.
Disease progression modeling attempts to estimate temporal ordering of biomarkers, clinical symptoms, and imaging changes across disease stages.
6. Полная рэалізацыя GraphRAG
Для пункту 6. Чырвонай рэалізацыі GraphRAG неабходна прадзеяванне вхідных дадзеных, апрацоўкаваючага ўзела і крэтарыяў завершэння працы перад зменыма коду. Аперацыйныя працавнікі павінны магчымае перапрацоўваць даныя з вядомага пункту контролю, не прымусваныя здагадвацца пра схованы стан. Конфігурацыю трэба заставляць праза межамі коду прыкладнення. Файлы сераўіса, хранільнікі секрэтных дадзеных і флагі функцыяў павінны знаходзіцца ў адном месцы, якое працавнікі можуць пераглядаць, не чытаючы весь граф. Неабходна ствараць узелі і зв’язкі модэлю для тых запытанняў, якія вы будзеце ставіць, а не для кожнага іменніка ў дасягліве. Рэдкія, тыпаваныя рэбры лепшыя за ўжоўтучныя, незрозумелы графы.
import os
import re
import json
import argparse
from typing import List, Dict, Any
import networkx as nx
from tqdm import tqdm
from dotenv import load_dotenv
from neo4j import GraphDatabase
from openai import OpenAI
load_dotenv()
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_CHAT_MODEL = os.getenv("OPENAI_CHAT_MODEL", "gpt-4o-mini")
OPENAI_EMBEDDING_MODEL = os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-3-small")
NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USERNAME = os.getenv("NEO4J_USERNAME", "neo4j")
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "password123")
NEO4J_DATABASE = os.getenv("NEO4J_DATABASE", "neo4j")
EMBEDDING_DIMENSION = 1536
client = OpenAI(api_key=OPENAI_API_KEY)
def normalize_entity_name(name: str) -> str:
name = name.strip()
name = re.sub(r"\s+", " ", name)
return name.title()
class Neo4jGraphRAG:
def __init__(self):
self.driver = GraphDatabase.driver(
NEO4J_URI,
auth=(NEO4J_USERNAME, NEO4J_PASSWORD)
)
def close(self):
self.driver.close()
def run_query(self, query: str, params: Dict[str, Any] = None):
params = params or {}
with self.driver.session(database=NEO4J_DATABASE) as session:
return list(session.run(query, params))
def clear_database(self):
self.run_query("MATCH (n) DETACH DELETE n")
def create_constraints_and_indexes(self):
self.run_query("""
CREATE CONSTRAINT document_id IF NOT EXISTS
FOR (d:Document)
REQUIRE d.id IS UNIQUE
""")
self.run_query("""
CREATE CONSTRAINT chunk_id IF NOT EXISTS
FOR (c:Chunk)
REQUIRE c.id IS UNIQUE
""")
self.run_query("""
CREATE CONSTRAINT entity_name IF NOT EXISTS
FOR (e:Entity)
REQUIRE e.name IS UNIQUE
""")
self.run_query("""
CREATE CONSTRAINT community_id IF NOT EXISTS
FOR (c:Community)
REQUIRE c.id IS UNIQUE
""")
self.run_query(f"""
CREATE VECTOR INDEX chunk_embedding_index IF NOT EXISTS
FOR (c:Chunk)
ON (c.embedding)
OPTIONS {{
indexConfig: {{
`vector.dimensions`: {EMBEDDING_DIMENSION},
`vector.similarity_function`: 'cosine'
}}
}}
""")
def create_document(self, doc_id: str, path: str, text: str):
self.run_query("""
MERGE (d:Document {id: $doc_id})
SET d.path = $path,
d.text = $text
""", {
"doc_id": doc_id,
"path": path,
"text": text
})
def create_chunk(
self,
chunk_id: str,
doc_id: str,
text: str,
chunk_index: int,
embedding: List[float]
):
self.run_query("""
MATCH (d:Document {id: $doc_id})
MERGE (c:Chunk {id: $chunk_id})
SET c.text = $text,
c.chunk_index = $chunk_index,
c.embedding = $embedding
MERGE (d)-[:HAS_CHUNK]->(c)
""", {
"chunk_id": chunk_id,
"doc_id": doc_id,
"text": text,
"chunk_index": chunk_index,
"embedding": embedding
})
def upsert_entity(self, name: str, entity_type: str, description: str):
clean_name = normalize_entity_name(name)
if not clean_name:
return
self.run_query("""
MERGE (e:Entity {name: $name})
SET e.type = coalesce(e.type, $entity_type),
e.description = coalesce(e.description, $description)
""", {
"name": clean_name,
"entity_type": entity_type,
"description": description
})
def link_chunk_entity(self, chunk_id: str, entity_name: str):
clean_name = normalize_entity_name(entity_name)
if not clean_name:
return
self.run_query("""
MATCH (c:Chunk {id: $chunk_id})
MATCH (e:Entity {name: $entity_name})
MERGE (c)-[:MENTIONS]->(e)
""", {
"chunk_id": chunk_id,
"entity_name": clean_name
})
def upsert_relationship(
self,
source: str,
target: str,
relation: str,
evidence: str,
chunk_id: str
):
clean_source = normalize_entity_name(source)
clean_target = normalize_entity_name(target)
if not clean_source or not clean_target:
return
relation_label = relation.upper().replace(" ", "_")
self.run_query("""
MATCH (s:Entity {name: $source})
MATCH (t:Entity {name: $target})
MERGE (s)-[r:RELATED_TO {relation: $relation, chunk_id: $chunk_id}]->(t)
SET r.evidence = $evidence
""", {
"source": clean_source,
"target": clean_target,
"relation": relation_label,
"evidence": evidence,
"chunk_id": chunk_id
})
def vector_search_chunks(self, query_embedding: List[float], top_k: int = 5):
records = self.run_query("""
CALL db.index.vector.queryNodes(
'chunk_embedding_index',
$top_k,
$embedding
)
YIELD node, score
RETURN node.id AS id,
node.text AS text,
score
ORDER BY score DESC
""", {
"top_k": top_k,
"embedding": query_embedding
})
return [
{
"id": r["id"],
"text": r["text"],
"score": r["score"]
}
for r in records
]
def graph_expand_from_chunks(self, chunk_ids: List[str], depth: int = 2):
records = self.run_query("""
MATCH (c:Chunk)
WHERE c.id IN $chunk_ids
MATCH path = (c)-[:MENTIONS]->(:Entity)-[:RELATED_TO*0..2]-(:Entity)
WITH c, path
UNWIND nodes(path) AS n
WITH c, collect(DISTINCT n) AS ns, relationships(path) AS rels
RETURN
c.id AS chunk_id,
c.text AS chunk_text,
[n IN ns WHERE n:Entity | {
name: n.name,
type: n.type,
description: n.description
}] AS entities,
[r IN rels | {
source: startNode(r).name,
relation: r.relation,
target: endNode(r).name,
evidence: r.evidence
}] AS relationships
""", {
"chunk_ids": chunk_ids,
"depth": depth
})
expanded = []
for r in records:
expanded.append({
"chunk_id": r["chunk_id"],
"chunk_text": r["chunk_text"],
"entities": r["entities"],
"relationships": r["relationships"]
})
return expanded
def fetch_entity_graph(self):
records = self.run_query("""
MATCH (s:Entity)-[r:RELATED_TO]->(t:Entity)
RETURN s.name AS source,
t.name AS target,
r.relation AS relation,
r.evidence AS evidence
""")
return [
{
"source": r["source"],
"target": r["target"],
"relation": r["relation"],
"evidence": r["evidence"]
}
for r in records
]
def store_community(self, community_id: str, summary: str, entities: List[str]):
self.run_query("""
MERGE (c:Community {id: $community_id})
SET c.summary = $summary,
c.entity_count = $entity_count
""", {
"community_id": community_id,
"summary": summary,
"entity_count": len(entities)
})
for entity in entities:
self.run_query("""
MATCH (c:Community {id: $community_id})
MATCH (e:Entity {name: $entity})
MERGE (e)-[:IN_COMMUNITY]->(c)
""", {
"community_id": community_id,
"entity": entity
})
def get_communities_for_chunks(self, chunk_ids: List[str]):
records = self.run_query("""
MATCH (chunk:Chunk)-[:MENTIONS]->(e:Entity)-[:IN_COMMUNITY]->(comm:Community)
WHERE chunk.id IN $chunk_ids
RETURN DISTINCT comm.id AS id,
comm.summary AS summary,
comm.entity_count AS entity_count
ORDER BY comm.entity_count DESC
LIMIT 5
""", {
"chunk_ids": chunk_ids
})
return [
{
"id": r["id"],
"summary": r["summary"],
"entity_count": r["entity_count"]
}
for r in records
]
def read_documents(folder: str) -> List[Dict[str, str]]:
docs = []
for filename in os.listdir(folder):
if not filename.endswith((".txt", ".md")):
continue
path = os.path.join(folder, filename)
with open(path, "r", encoding="utf-8") as f:
text = f.read().strip()
docs.append({
"id": filename,
"path": path,
"text": text
})
return docs
def chunk_text(text: str, chunk_size: int = 900, overlap: int = 150) -> List[str]:
words = text.split()
chunks = []
start = 0
while start < len(words):
end = start + chunk_size
chunk = " ".join(words[start:end])
chunks.append(chunk)
if end >= len(words):
break
start = end - overlap
return chunks
def embed_text(text: str) -> List[float]:
response = client.embeddings.create(
model=OPENAI_EMBEDDING_MODEL,
input=text
)
return response.data[0].embedding
def extract_graph_from_chunk(text: str) -> Dict[str, Any]:
system_prompt = """
You are an expert knowledge graph extraction system.
Extract entities and relationships from the provided text.
Return only valid JSON with this structure:
{
"entities": [
{
"name": "entity name",
"type": "entity type",
"description": "short description"
}
],
"relationships": [
{
"source": "source entity name",
"target": "target entity name",
"relation": "relationship label",
"evidence": "short evidence phrase from the text"
}
]
}
Rules:
- Use concise entity names.
- Use consistent names across chunks.
- Relationship labels should be verb-like, for example ASSOCIATED_WITH, MEASURED_BY, or INCREASES_RISK_OF.
- Do not invent facts.
- Extract only facts supported by the text.
"""
response = client.chat.completions.create(
model=OPENAI_CHAT_MODEL,
temperature=0,
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": system_prompt
},
{
"role": "user",
"content": text
}
]
)
raw = response.choices[0].message.content
try:
data = json.loads(raw)
except json.JSONDecodeError:
data = {
"entities": [],
"relationships": []
}
data.setdefault("entities", [])
data.setdefault("relationships", [])
return data
def summarize_community(
entities: List[str],
relationships: List[Dict[str, Any]]
) -> str:
relationship_lines = []
for r in relationships:
relationship_lines.append(
f"{r['source']} --{r['relation']}--> {r['target']}. Evidence: {r.get('evidence', '')}"
)
prompt = f"""
You are summarizing a knowledge graph community.
Entities:
{json.dumps(entities, indent=2)}
Relationships:
{chr(10).join(relationship_lines)}
Write a concise technical summary of this community. Include:
1. Main theme
2. Important entities
3. Important relationships
4. Why this community matters for question answering
"""
response = client.chat.completions.create(
model=OPENAI_CHAT_MODEL,
temperature=0,
messages=[
{
"role": "system",
"content": "You summarize graph communities for a GraphRAG system."
},
{
"role": "user",
"content": prompt
}
]
)
return response.choices[0].message.content.strip()
def build_communities(graph_db: Neo4jGraphRAG):
edges = graph_db.fetch_entity_graph()
if not edges:
print("No entity graph found. Skipping community detection.")
return
graph = nx.Graph()
for edge in edges:
graph.add_edge(
edge["source"],
edge["target"],
relation=edge["relation"],
evidence=edge.get("evidence", "")
)
communities = list(nx.community.greedy_modularity_communities(graph))
print(f"Detected {len(communities)} graph communities.")
for i, community_nodes in enumerate(communities):
community_entities = sorted(list(community_nodes))
community_relationships = [
e for e in edges
if e["source"] in community_entities and e["target"] in community_entities
]
summary = summarize_community(
entities=community_entities,
relationships=community_relationships
)
community_id = f"community_{i}"
graph_db.store_community(
community_id=community_id,
summary=summary,
entities=community_entities
)
print(f"Stored {community_id}: {len(community_entities)} entities")
def ingest(folder: str, clear: bool = False):
graph_db = Neo4jGraphRAG()
try:
if clear:
print("Clearing database...")
graph_db.clear_database()
print("Creating constraints and indexes...")
graph_db.create_constraints_and_indexes()
docs = read_documents(folder)
print(f"Found {len(docs)} documents.")
for doc in docs:
print(f"Ingesting document: {doc['path']}")
graph_db.create_document(
doc_id=doc["id"],
path=doc["path"],
text=doc["text"]
)
chunks = chunk_text(doc["text"])
for idx, chunk in enumerate(tqdm(chunks)):
chunk_id = f"{doc['id']}::chunk_{idx}"
embedding = embed_text(chunk)
graph_db.create_chunk(
chunk_id=chunk_id,
doc_id=doc["id"],
text=chunk,
chunk_index=idx,
embedding=embedding
)
graph_data = extract_graph_from_chunk(chunk)
for entity in graph_data["entities"]:
graph_db.upsert_entity(
name=entity.get("name", ""),
entity_type=entity.get("type", "Unknown"),
description=entity.get("description", "")
)
graph_db.link_chunk_entity(
chunk_id=chunk_id,
entity_name=entity.get("name", "")
)
for rel in graph_data["relationships"]:
source = rel.get("source", "")
target = rel.get("target", "")
if not source or not target:
continue
graph_db.upsert_entity(
name=source,
entity_type="Unknown",
description=""
)
graph_db.upsert_entity(
name=target,
entity_type="Unknown",
description=""
)
graph_db.upsert_relationship(
source=source,
target=target,
relation=rel.get("relation", "RELATED_TO"),
evidence=rel.get("evidence", ""),
chunk_id=chunk_id
)
print("Building graph communities...")
build_communities(graph_db)
print("Ingestion complete.")
finally:
graph_db.close()
def build_context(
chunks: List[Dict[str, Any]],
expanded_graph: List[Dict[str, Any]],
communities: List[Dict[str, Any]]
) -> str:
context_parts = []
context_parts.append("## Retrieved Chunks")
for i, chunk in enumerate(chunks, start=1):
context_parts.append(
f"[Chunk {i} | id={chunk['id']} | score={chunk['score']:.4f}]\n{chunk['text']}"
)
context_parts.append("\n## Graph Neighborhood")
for item in expanded_graph:
context_parts.append(f"\nSource Chunk: {item['chunk_id']}")
if item["entities"]:
context_parts.append("Entities:")
for e in item["entities"]:
context_parts.append(
f"- {e.get('name')} ({e.get('type')}): {e.get('description')}"
)
if item["relationships"]:
context_parts.append("Relationships:")
for r in item["relationships"]:
context_parts.append(
f"- {r.get('source')} --{r.get('relation')}--> {r.get('target')}. Evidence: {r.get('evidence')}"
)
context_parts.append("\n## Community Summaries")
for community in communities:
context_parts.append(
f"[{community['id']}]\n{community['summary']}"
)
return "\n".join(context_parts)
def answer_question(question: str, top_k: int = 5):
graph_db = Neo4jGraphRAG()
try:
query_embedding = embed_text(question)
chunks = graph_db.vector_search_chunks(
query_embedding=query_embedding,
top_k=top_k
)
chunk_ids = [c["id"] for c in chunks]
expanded_graph = graph_db.graph_expand_from_chunks(
chunk_ids=chunk_ids,
depth=2
)
communities = graph_db.get_communities_for_chunks(chunk_ids)
context = build_context(
chunks=chunks,
expanded_graph=expanded_graph,
communities=communities
)
prompt = f"""
You are a GraphRAG assistant.
Answer the question using only the provided context.
Requirements:
- Use retrieved chunks, graph relationships, and community summaries.
- Explain relationship chains when useful.
- Do not invent facts.
- If the context is insufficient, say so clearly.
- Cite chunk ids when making factual claims.
Question:
{question}
Context:
{context}
"""
response = client.chat.completions.create(
model=OPENAI_CHAT_MODEL,
temperature=0,
messages=[
{
"role": "system",
"content": "You answer questions using graph-grounded retrieval context."
},
{
"role": "user",
"content": prompt
}
]
)
print("\nANSWER\n")
print(response.choices[0].message.content.strip())
print("\nRETRIEVED CHUNKS\n")
for c in chunks:
print(f"- {c['id']} | score={c['score']:.4f}")
finally:
graph_db.close()
def main():
parser = argparse.ArgumentParser(description="GraphRAG with Neo4j and Python")
subparsers = parser.add_subparsers(dest="command")
ingest_parser = subparsers.add_parser("ingest")
ingest_parser.add_argument("--folder", default="documents")
ingest_parser.add_argument("--clear", action="store_true")
query_parser = subparsers.add_parser("query")
query_parser.add_argument("--question", required=True)
query_parser.add_argument("--top-k", type=int, default=5)
args = parser.parse_args()
if args.command == "ingest":
ingest(folder=args.folder, clear=args.clear)
elif args.command == "query":
answer_question(question=args.question, top_k=args.top_k)
else:
parser.print_help()
if __name__ == "__main__":
main()
7. Запуск пайплайна інгестыі
Для пункту 7: перад змянайом код неабяжна запусціць канал практыкавання, задаць параметры вхідных дадзеных, адміністратара крока і крытэрыя завершэння. Аператары должны магчымаць перапуск крока з вядомай точкі контролю, не падозруючы пра схованы стан. Неабяжна задокументаваць як шлях успеху, так і шлях вярнення да нормальнага стану. Перапрыбуткі, людзкія перакрыцця і апрацоўка некорэктных паведамленняў ёсць часткай продукту, а не дадатковым удосконаленнем пазнейша. Ствараць мадэлі вузлаў і ўзаемных звязкаў для тых запытанняў, якія вы будзеце ставіць, а не для кожнага іменніка ў дакуменце. Рэдкія, тыпаваныя звязкі лепшыя за ўжоўтучныя, незрозумелыя графы. Для пункту 7: перад змянайом код неабяжна запусціць канал практыкавання, задаць параметры вхідных дадзеных, адміністратара крока і крытэрыя завершэння. Аператары должны магчымаць перапуск крока з вядомай точкі контролю, не падозруючы пра схованы стан. Спрыяць гэтаму этапу як кантракту між вхіднымі дадзенымі і перакананымі выходнымі рэзультатамі. Даць назвы элементам, задаць перакрыцці успеху і не прабаваць прыймаць часткова завершаныя рэзультаты без паведамлення.
python app.py ingest --folder documents --clear
8. Задаць запыткі GraphRAG
Для пункту 8. Задаць запыткі GraphRAG неабяжна пазначыць вхідныя даны, адпаведальнага за выкананне крока і крэтыяры завершэння пры перадзеі коду. Аперацыйныя працавнікі должны магчыма ўвайсці крок з вядомага пункта контролю без неабяжнасці вычысляць схованы стан. Запісваць час выканання і вартасць токеноў або запыткаў праза функцыйнаясць рэзультатаў. Відразлівасць вартасцей з самага пачатку запобегае неспакойным рахункам, калі працэс пераходзіць з дэмавайнага рэжыма ў спяльныя сераўры. Перад масовым запрашэнням даных трэба выкарыстоўваць обмежэнняі і індэксы. Унікальнасць на ключоўых даных дапамагае ператварыць пазнейшыя злучэння на прагнозаваныя дзеянні адменстрацыі, а не на сітуацыю з дуплікатамі.
python app.py query \
--question "How do amyloid beta, tau pathology, and hippocampal atrophy relate to Alzheimer's disease progression?"
python app.py query \
--question "Which biomarkers and imaging markers are associated with conversion from MCI to Alzheimer's dementia?"
python app.py query \
--question "Explain the relationship between APOE4, amyloid accumulation, and disease progression."
9. Што адбываецца пад час выканання запытка?
Для пункта 9 «Што адзначаецца за час выканання запіту?» неабяжна ўзначыць вхідныя даны, адпаведальнага за этап і крытэрыя завершэння пры перадзеіснавленні коду. Аператары должны магчымае перадзеіснавляць этап з вядомай точкі контролю, не падозрываючы прыхованы стан.
Этап 1: Адзыв вектараў
Question → Embedding → Neo4j Vector Index → Top-k Chunks
Этап 2: Пашырэнне графа
Retrieved Chunk → Mentioned Entities → Related Entities → Relationship Paths
Этап 3: Адзыв падсумку спільнатыпу
Entities → Communities → Community Summaries
10. Аналіз графа ў Neo4j
MATCH (e:Entity)-[r:RELATED_TO]->(t:Entity)
RETURN e, r, t
LIMIT 100
MATCH (c:Chunk)-[:MENTIONS]->(e:Entity)
RETURN c.id, e.name, e.type
LIMIT 50
MATCH (e:Entity)-[:IN_COMMUNITY]->(c:Community)
RETURN c.id, c.summary, collect(e.name) AS entities
11. Чаму гэта ёсць GraphRAG
Unstructured documents
↓
Chunking
↓
Entity extraction
↓
Relationship extraction
↓
Knowledge graph construction
↓
Vector index over chunks
↓
Graph traversal
↓
Community detection
↓
Community summaries
↓
LLM answer synthesis
Which chunks are relevant?
Which entities do those chunks mention?
Which entities are connected?
Which relationship paths explain the answer?
Which communities summarize the broader topic?
12. GraphRAG проты Vector RAG
APOE4 → increases risk of → Amyloid Accumulation
Amyloid Accumulation → associated with → Alzheimer’s Progression
Tau Pathology → linked with → Cognitive Decline
Hippocampal Atrophy → associated with → Memory Impairment
13. Заключныя меркі
Vector search
+ Keyword search
+ Knowledge graph retrieval
+ Community summaries
+ Tool use
+ LLM reasoning
+ Evaluation