GraphRAG desde cero con Neo4j: vectores más expansión de grafos
GraphRAG desde cero en Neo4j dentro de Docker: ingresar documentos, recuperarlos mediante vectores, expandir los vecindarios del grafo, agregar resúmenes de comunidades y luego compararlo con RAG tradicional.
Fuentes: Building GraphRAG from Scratch: Knowledge Graph-Powered RAG with Neo4j, Python, and LLMs
Esta guía reconstruye el proceso desde las materias primas hasta un sistema funcional para: Construir GraphRAG desde cero: RAG impulsado por grafos de conocimiento con Neo4j, Python y LLMs. El enfoque está en pasos operativos claros, verificaciones explícitas y código que se puede incorporar directamente a un repositorio sin necesidad de adivinar la intención. Para tener una visión general, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Considere esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas.
Introducción
Como introducción, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde un punto de control conocido sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo en tokens o consultas junto con los resultados funcionales. La visibilidad temprana de los costos evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Modele nodos y relaciones para las preguntas que vaya a hacer, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que grafos densos y misteriosos.
¿Qué es GraphRAG?
Para “¿Qué es GraphRAG?”, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa desde un punto de control conocido sin tener que adivinar el estado oculto. Guarde la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de secretos y las banderas de funcionalidad deben encontrarse en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el grafo. Diseñe nodos y relaciones de modelo para las preguntas que vaya a hacer, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que los grafos densos y misteriosos.
Grafo de conocimiento vs GraphRAG
Para Knowledge Graph frente a GraphRAG, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Cree nodos y relaciones para las preguntas que vaya a hacer, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que los grafos densos e inexplicables. Para Knowledge Graph frente a GraphRAG, defina las entradas, el responsable de la etapa y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la etapa a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina verificaciones de éxito y rechace las completaciones parciales silenciosas.
Arquitectura
En el caso de la arquitectura, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo de tokens o consultas junto con los resultados funcionales. La visibilidad temprana del costo evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Utilice restricciones e índices antes de realizar la ingesta masiva. La unicidad en las claves empresariales convierte las fusiones posteriores en operaciones de inserción o actualización predecibles, en lugar de situaciones de duplicados excesivos.
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
Implementación de extremo a extremo
Para una implementación de extremo a extremo, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación. Los archivos de entorno, los almacenes de datos secretos y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el sistema. Utilice restricciones e índices antes de realizar la ingesta masiva. La unicidad en las claves empresariales convierte las fusiones posteriores en operaciones de inserción o actualización predecibles, en lugar de situaciones de duplicados excesivos.
1. Estructura del proyecto
En cuanto a la 1. Estructura del proyecto, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las intentonas, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Utilice restricciones e índices antes de la ingesta masiva. La unicidad en las claves empresariales convierte las fusiones posteriores en inserciones predecibles en lugar de situaciones de duplicados. En cuanto a la 1. Estructura del proyecto, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas.
graphrag-neo4j-demo/
│
├── docker-compose.yml
├── requirements.txt
├── .env.example
├── app.py
└── documents/
└── brain_progression
2. Iniciar Neo4j con Docker
Para “2. Iniciar Neo4j con Docker”, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando la tarea pasa de entornos de demostración a entornos compartidos. Inspeccione un vecindario de ejemplo después del primer lote; las pruebas con Cypher detectan cambios en las etiquetas y propiedades faltantes de manera más económica que las pruebas completas por chat.
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. Instalar dependencias de Python
Para el paso 3: instale las dependencias de Python, defina los parámetros de entrada, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación; los archivos de entorno, los almacenes de datos confidenciales y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el sistema.
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. Configurar variables de entorno
Para el punto 4: Configure las variables de entorno. Defina los datos de entrada, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Inspeccione un grupo de ejemplos después del primer lote; las pruebas Cypher detectan con mayor eficiencia los desvíos en las etiquetas y las propiedades faltantes que las pruebas de chat de extremo a extremo. Para el punto 4: Configure las variables de entorno. Defina los datos de entrada, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Considere esta etapa como un contrato entre los datos de entrada y los resultados validados. Asigne nombres a los artefactos, defina las verificaciones de éxito y rechace cualquier completación parcial silenciosa.
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. Agregar documentos de ejemplo
En la sección 5, Agregar documentos de ejemplo, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo de tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando el proceso pasa de entornos de demostración a entornos compartidos. Modele nodos y relaciones para las preguntas que vaya a hacer, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que grafos densos y misteriosos.
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. Implementación completa de GraphRAG
Para el punto 6: Para una implementación completa de GraphRAG, defina las entradas, el responsable de cada paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido, sin tener que adivinar el estado oculto. Mantenga la configuración fuera del código de la aplicación; los archivos de entorno, los almacenes de datos confidenciales y las banderas de funcionalidad deben estar en un lugar donde los operadores puedan auditarlos sin necesidad de leer todo el grafo. Diseñe nodos y relaciones de modelo para las preguntas que vaya a hacer, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que los grafos densos y complejos.
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. Ejecutar la tubería de ingestión
Para el punto 7: ejecute la tubería de ingestión, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Documente tanto la ruta óptima como la ruta de recuperación. Las reintentos, los controles humanos y el manejo de mensajes no entregados forman parte del producto, no son mejoras posteriores. Modele nodos y relaciones para las preguntas que planteará, no para cada sustantivo del documento. Los bordes escasos y tipados son mejores que grafos densos y misteriosos. Para el punto 7: ejecute la tubería de ingestión, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto. Trate esta etapa como un contrato entre las entradas y los resultados validados. Asigne nombres a los artefactos, defina comprobaciones de éxito y rechace las completaciones parciales silenciosas.
python app.py ingest --folder documents --clear
8. Hacer preguntas a GraphRAG
En la sección 8, Hacer preguntas a GraphRAG, defina las entradas, el responsable de la tarea y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar la tarea a partir de un punto de control conocido sin tener que adivinar el estado oculto. Registre los tiempos de ejecución y el costo en tokens o consultas junto con los resultados funcionales. Tener visibilidad del costo desde el principio evita facturas inesperadas cuando la tarea pasa de entornos de demostración a entornos compartidos. Utilice restricciones e índices antes de realizar la ingesta masiva. La unicidad en las claves empresariales convierte las fusiones posteriores en actualizaciones predecibles en lugar de situaciones de duplicados excesivos.
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. ¿Qué ocurre durante el tiempo de consulta?
Para el punto 9, “¿Qué sucede durante el tiempo de consulta?”, defina las entradas, el responsable del paso y los criterios de finalización antes de modificar el código. Los operadores deben poder volver a ejecutar el paso a partir de un punto de control conocido sin tener que adivinar el estado oculto.
Paso 1: Recuperación de vectores
Question → Embedding → Neo4j Vector Index → Top-k Chunks
Paso 2: Expansión del grafo
Retrieved Chunk → Mentioned Entities → Related Entities → Relationship Paths
Paso 3: Recuperación del resumen de la comunidad
Entities → Communities → Community Summaries
10. Inspeccionar el grafo en 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. Por qué esto es 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. Consideraciones finales
Vector search
+ Keyword search
+ Knowledge graph retrieval
+ Community summaries
+ Tool use
+ LLM reasoning
+ Evaluation
Lista de verificación operativa
Lecturas relacionadas
- Construir un pipeline simple de GraphRAG con Neo4j, Gemini y LangChain — Extracción de grafos relacionados con accidentes en la cocina restringida por esquemas, consultas de vecindad, rutas de múltiples saltos y conversión de texto a Cypher solo de lectura con Gemini.
- Memoria híbrida de agente: Fusión de BM25 y búsqueda vectorial con RRF en Python — Entienda por qué la búsqueda vectorial pura falla como memoria de agente, cómo la Fusión de Rango Recíproco combina resultados de BM25 y densos en Python, y cuándo los resúmenes de GraphRAG son útiles.