GraphRAG с нуля с использованием Neo4j: векторы плюс расширение графа
Разработка GraphRAG с нуля на Neo4j в Docker: загрузка документов, поиск по векторам, расширение соседей графа, добавление кратких обзоров сообществ, затем сравнение с обычным RAG.
Источники: Building GraphRAG from Scratch: Knowledge Graph-Powered RAG with Neo4j, Python, and LLMs
В этом руководстве пошагово описывается процесс создания рабочей системы, начиная с сырьевых материалов: «Создание GraphRAG с нуля: RAG на основе графовых знаний с использованием Neo4j, Python и больших языков моделирования». Основное внимание уделяется практическим шагам, четкой проверке результатов и коду, который можно просто скопировать в репозиторий без необходимости догадываться о намерениях автора. Для получения общего представления необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед внесением изменений в код. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии системы. Рассматривайте этот этап как контракт между входными данными и проверенными результатами. Дайте названия создаваемым элементам, определите критерии успеха и не допускайте безответственного частичного выполнения задачи.
Введение
Во введении необходимо определить входные данные, ответственного за выполнение шага и критерии завершения перед изменением кода. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Рядом с функциональными результатами следует записывать время выполнения и стоимость токенов или запросов. Отображение затрат с самого начала предотвращает неожиданные счета при переходе от демо-среды к общедоступным средам. Моделируйте узлы и связи только для тех вопросов, которые вы собираетесь задать, а не для каждого существительного в документе. Редкие, типизированные связи лучше, чем плотные графы с непонятными структурами.
Что такое 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, определить входные данные, ответственного за выполнение шага и критерии завершения. Операторы должны иметь возможность перезапустить шаг с известной точки контроля, не догадываясь о скрытом состоянии. Конфигурацию следует хранить отдельно от кода приложения. Файлы среды, хранилища секретов и флаги функций должны находиться в одном месте, чтобы операторы могли их проверять, не читая весь граф выполнения.
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