GraphRAG de zéro avec Neo4j : vecteurs et expansion graphique
Créer GraphRAG à partir de zéro sur Neo4j dans Docker : importer des documents, récupérer des informations via des vecteurs, étendre les voisins du graphe, ajouter des résumés de communautés, puis comparer avec un RAG classique.
Sources: Building GraphRAG from Scratch: Knowledge Graph-Powered RAG with Neo4j, Python, and LLMs
Ce guide reconstitue le parcours allant des matières premières à un système fonctionnel pour : Construire GraphRAG de zéro : RAG alimenté par des graphes de connaissances avec Neo4j, Python et les LLM. L’accent est mis sur des étapes opérationnelles, des vérifications explicites et du code que vous pouvez intégrer directement dans un dépôt sans devoir deviner l’intention. Pour une vue d’ensemble, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans avoir à deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Nommez les artefacts, définissez des vérifications de succès et refusez toute complétion partielle silencieuse.
Introduction
Pour l’introduction, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Modélisez les nœuds et les relations en fonction des questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et bien définies valent mieux que des graphes complexes et mystérieux.
Qu’est-ce que GraphRAG ?
Pour « Qu’est-ce que GraphRAG ? », il convient de définir les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les flags fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du graphe. Modélisez les nœuds et les relations pour les questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes complexes et mystérieux.
Graphe de connaissances vs GraphRAG
Pour Knowledge Graph contre GraphRAG, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez ensemble le parcours normal et les procédures de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations ultérieures. Modélisez les nœuds et les relations pour les questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes denses et mystérieux. Pour Knowledge Graph contre GraphRAG, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définites des vérifications de succès et refusez les complétions partielles silencieuses.
Architecture
Pour l’architecture, définissez les entrées, le responsable de chaque étape ainsi que les critères de fin avant de modifier du code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans avoir à deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce des coûts évite les factures inattendues lorsque le processus passe d’un environnement de démonstration à des environnements partagés. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse.
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
Mise en œuvre bout en bout
Pour une mise en œuvre bout en bout, définissez les entrées, le responsable de chaque étape ainsi que les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu, sans avoir à deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse.
1. Structure du projet
Pour 1. Structure du projet, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez conjointement le parcours normal et le parcours de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non livrés font partie intégrante du produit, et non d’une mise en forme ultérieure. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité des clés métier permet de transformer les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de créer des doublons en masse. Pour 1. Structure du projet, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définissez des vérifications de succès et refusez les terminations partielles silencieuses.
graphrag-neo4j-demo/
│
├── docker-compose.yml
├── requirements.txt
├── .env.example
├── app.py
└── documents/
└── brain_progression
2. Démarrer Neo4j avec Docker
Pour « 2. Démarrer Neo4j avec Docker », définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des tokens ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce du coût permet d’éviter des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés. Vérifiez un échantillon de voisins après le premier lot ; les sondes Cypher permettent de détecter plus facilement les écarts dans les étiquettes et les propriétés manquantes que des tests de chat de bout en bout.
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. Installer les dépendances Python
Pour 3. Installez les dépendances Python, définites les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les stocks de secrets et les flags fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans devoir lire l’ensemble du système. Vérifiez un échantillon du voisinage après le premier lot. Les sondes Cypher permettent de détecter les écarts de labels et les propriétés manquantes à moindre coût que des tests de conversation du début à la fin.
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. Configurer les variables d’environnement
Pour le point 4 : Configurez les variables d’environnement, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez conjointement le parcours normal et les scénarios de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages non traités font partie intégrante du produit, et non d’améliorations ultérieures. Inspectez un échantillon de données après le premier lot. Les sondes Cypher permettent de détecter plus rapidement les écarts dans les étiquettes et les propriétés manquantes que les tests de conversation du début à la fin. Pour le point 4 : Configurez les variables d’environnement, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définissez des vérifications de succès et refusez les terminations partielles silencieuses.
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. Ajouter des documents d’exemple
Pour la section 5. Ajouter des documents d’exemple, définissez les entrées, le responsable de l’étape et les critères de fin avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût des jetons ou des requêtes à côté des résultats fonctionnels. Une visibilité précoce du coût permet d’éviter des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés.
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. Implémentation complète de GraphRAG
Pour 6. Implémentation complète de GraphRAG, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Conservez la configuration en dehors du code de l’application. Les fichiers d’environnement, les bases de données secrètes et les indicateurs fonctionnels doivent être regroupés en un seul endroit que les opérateurs peuvent auditer sans avoir à lire l’ensemble du graphe. Modélisez les nœuds et les relations pour les questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes denses et mystérieux.
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. Exécuter le pipeline d’ingestion
Pour le point 7 : exécutez le pipeline d’ingestion, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Documentez conjointement le parcours normal et les procédures de récupération. Les tentatives répétées, les contrôles humains et la gestion des messages échoués font partie intégrante du produit, et non d’améliorations ultérieures. Modélisez les nœuds et les relations en fonction des questions que vous poserez, et non pour chaque nom dans le document. Des arêtes sparses et typées valent mieux que des graphes complexes et mystérieux. Pour le point 7 : exécutez le pipeline d’ingestion, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Considérez cette étape comme un contrat entre les entrées et les sorties validées. Donnez des noms aux artefacts, définissez des vérifications de succès et refusez les terminations partielles silencieuses.
python app.py ingest --folder documents --clear
8. Poser des questions à GraphRAG
Pour la section 8. Poser des questions à GraphRAG, définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché. Enregistrez les temps d’exécution ainsi que le coût en tokens ou en requêtes à côté des résultats fonctionnels. Une visibilité précoce du coût permet d’éviter des factures inattendues lorsque le processus passe de l’environnement de démonstration à des environnements partagés. Utilisez des contraintes et des index avant l’ingestion en masse. L’unicité basée sur les clés métier transforme les fusionnements ultérieurs en opérations d’insertion/mise à jour prévisibles, plutôt que de provoquer des douzaines de doublons.
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. Que se passe-t-il pendant le temps de traitement de la requête ?
Pour la section 9 « Que se passe-t-il pendant le temps de requête ? », définissez les entrées, le responsable de l’étape et les critères d’arrêt avant de modifier le code. Les opérateurs doivent pouvoir relancer l’étape à partir d’un point de contrôle connu sans deviner l’état caché.
Étape 1 : Récupération des vecteurs
Question → Embedding → Neo4j Vector Index → Top-k Chunks
Étape 2 : Expansion du graphe
Retrieved Chunk → Mentioned Entities → Related Entities → Relationship Paths
Étape 3 : Récupération du résumé de la communauté
Entities → Communities → Community Summaries
10. Examiner le graphe dans 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. Pourquoi c’est 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. Réflexions finales
Vector search
+ Keyword search
+ Knowledge graph retrieval
+ Community summaries
+ Tool use
+ LLM reasoning
+ Evaluation