GraphRAG from Scratch with Neo4j: Vectors Plus Graph Expansion
GraphRAG from scratch on Neo4j in Docker: ingest docs, retrieve by vector, expand graph neighborhoods, add community summaries, then compare to plain RAG.
This article is published in English.
Sources: Building GraphRAG from Scratch: Knowledge Graph-Powered RAG with Neo4j, Python, and LLMs
This walkthrough rebuilds the path from raw materials to a working system for: Building GraphRAG from Scratch: Knowledge Graph-Powered RAG with Neo4j, Python, and LLMs. The focus is operable steps, explicit checks, and code that you can drop into a repo without guessing intent. For Overview, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
Introduction
For Introduction, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs.
What is GraphRAG?
For What is GraphRAG?, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs.
Knowledge Graph vs GraphRAG
For Knowledge Graph vs GraphRAG, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs. For Knowledge Graph vs GraphRAG, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
Architecture
For Architecture, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms.
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
End-to-End Implementation
For End-to-End Implementation, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms.
1. Project Structure
For 1. Project Structure, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms. For 1. Project Structure, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
graphrag-neo4j-demo/
│
├── docker-compose.yml
├── requirements.txt
├── .env.example
├── app.py
└── documents/
└── brain_progression
2. Start Neo4j with Docker
For 2. Start Neo4j with Docker, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Inspect a sample neighborhood after the first batch. Cypher probes catch label drift and missing properties cheaper than end-to-end chat tests.
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. Install Python Dependencies
For 3. Install Python Dependencies, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Inspect a sample neighborhood after the first batch. Cypher probes catch label drift and missing properties cheaper than end-to-end chat tests.
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. Configure Environment Variables
For 4. Configure Environment Variables, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Inspect a sample neighborhood after the first batch. Cypher probes catch label drift and missing properties cheaper than end-to-end chat tests. For 4. Configure Environment Variables, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
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. Add Sample Documents
For 5. Add Sample Documents, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs.
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. Full GraphRAG Implementation
For 6. Full GraphRAG Implementation, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs.
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. Run the Ingestion Pipeline
For 7. Run the Ingestion Pipeline, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Model nodes and relationships for the questions you will ask, not for every noun in the document. Sparse, typed edges beat dense mystery graphs. For 7. Run the Ingestion Pipeline, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
python app.py ingest --folder documents --clear
8. Ask GraphRAG Questions
For 8. Ask GraphRAG Questions, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Use constraints and indexes before bulk ingest. Uniqueness on business keys turns later merges into predictable upserts instead of duplicate storms.
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. What Happens During Query Time?
For 9. What Happens During Query Time?, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state.
Step 1: Vector Retrieval
Question → Embedding → Neo4j Vector Index → Top-k Chunks
Step 2: Graph Expansion
Retrieved Chunk → Mentioned Entities → Related Entities → Relationship Paths
Step 3: Community Summary Retrieval
Entities → Communities → Community Summaries
10. Inspect the Graph in 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. Why This Is 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 vs 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. Final Thoughts
Vector search
+ Keyword search
+ Knowledge graph retrieval
+ Community summaries
+ Tool use
+ LLM reasoning
+ Evaluation