Startseite / Artikel

GraphRAG von Grund auf mit Neo4j: Vektoren plus Graphenerweiterung

GraphRAG von Grund auf auf Neo4j in Docker: Dokumente einlesen, über Vektoren abrufen, Graph-Nachbarn erweitern, Zusammenfassungen der Communities hinzufügen und anschließend mit herkömmlichem RAG vergleichen.

Verlag: REACTAPP.TOP17 Min. Lesezeit3815 Wörter

Quellen: Building GraphRAG from Scratch: Knowledge Graph-Powered RAG with Neo4j, Python, and LLMs

Dieser Leitfaden zeigt Schritt für Schritt den Weg von Rohstoffen bis zu einem funktionsfähigen System für: „Building GraphRAG from Scratch: Knowledge Graph-Powered RAG mit Neo4j, Python und LLMs“. Der Schwerpunkt liegt auf ausführbaren Schritten, expliziten Überprüfungen sowie Code, den man ohne Rätseln über die Absicht direkt in ein Repository einfügen kann. Zur Übersicht sollten Sie vor der Änderung des Codes die Eingaben, den Verantwortlichen für den Schritt sowie die Abschlusskriterien definieren. Die Betreiber sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Erzeugnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab.

Einführung

Zur Einführung sollten Sie die Eingabedaten, den Verantwortlichen für den jeweiligen Schritt sowie die Abbruchkriterien definieren, bevor Sie Code ändern. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erfassen Sie neben den funktionalen Ergebnissen auch die Laufzeiten sowie die Kosten für Tokens oder Abfragen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo-Umgebung in gemeinsam genutzte Umgebungen wechselt. Modellieren Sie nur die Knoten und Beziehungen zu den Fragen, die Sie stellen möchten – nicht jedes Substantiv im Dokument. Spärliche, typisierte Verbindungen sind besser als dichte, unklare Graphen.

Was ist GraphRAG?

Für „Was ist GraphRAG?“ sollten Sie die Eingaben, den Verantwortlichen für den jeweiligen Schritt sowie die Abbruchkriterien definieren, bevor Sie Code ändern. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort gesammelt sein, den die Operator überprüfen können, ohne den gesamten Graphen durchlesen zu müssen. Definieren Sie Modellknoten und -beziehungen für die Fragen, die Sie stellen werden – nicht für jedes Substantiv im Dokument. Spärliche, typisierte Kanten sind besser als dichte, unklare Graphen.

Wissensgraph vs. GraphRAG

Für Knowledge Graph gegenüber GraphRAG sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallplan. Wiederholungsversuche, menschliche Überprüfungen und die Handhabung fehlerhafter Nachrichten gehören zum Produkt selbst, nicht zu späteren Optimierungen. Definieren Sie Modellknoten und -beziehungen für die Fragen, die Sie stellen werden, und nicht für jedes Substantiv im Dokument. Spärliche, typisierte Kanten sind besser als dichte, unklare Graphen. Für Knowledge Graph gegenüber GraphRAG sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Ergebnisdokumente, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Ergebnisse ab.

Architektur

Für die Architektur sollten Eingabedaten, der Verantwortliche für den jeweiligen Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Neben den funktionalen Ergebnissen sollten Zeiten sowie Kosten für Token oder Abfragen aufgezeichnet werden. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von Demonstrationsumgebungen in gemeinsam genutzte Umgebungen übergeht. Verwenden Sie Einschränkungen und Indizes vor der Masseneingabe. Die Eindeutigkeit von Geschäfts Schlüsseln sorgt dafür, dass spätere Zusammenführungen zu vorhersehbaren Upsert-Operationen statt zu Doppelgängerproblemen führen.

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-Implementierung

Für eine End-zu-End-Einführung sollten die Eingaben, der Verantwortliche für den jeweiligen Schritt sowie die Abbruchkriterien vor dem Codeändern definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Die Konfiguration sollte außerhalb des Anwendungscode gespeichert werden. Umgebungsdateien, Geheimdatenspeicher und Feature-Flags sollten an einem Ort zusammengefasst sein, den die Operator überprüfen können, ohne den gesamten Ablauf durchzulesen. Verwenden Sie Einschränkungen und Indizes vor dem Masseneinlesen. Die Eindeutigkeit der Geschäftschlüssel sorgt dafür, dass spätere Zusammenführungen zu vorhersehbaren Upserts statt zu Doppelungsproblemen führen.

1. Projektstruktur

Zur 1. Projektstruktur sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholte Versuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen. Verwenden Sie Einschränkungen und Indizes vor dem Masseneinlesen. Die Eindeutigkeit der Geschäftschlüssel sorgt dafür, dass spätere Zusammenführungen zu vorhersehbaren Aktualisierungen statt zu Doppelungen führen. Zur 1. Projektstruktur sollten die Eingaben, der Verantwortliche für den Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definiert werden. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, teilweise abgeschlossene Vorgänge ab.

graphrag-neo4j-demo/
│
├── docker-compose.yml
├── requirements.txt
├── .env.example
├── app.py
└── documents/
    └── brain_progression

2. Neo4j mit Docker starten

Zur Schritt 2 „Neo4j mit Docker starten“ sollten Sie die Eingabedaten, den Verantwortlichen für diesen Schritt sowie die Abbruchkriterien vor dem Ändern des Codes definieren. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erfassen Sie neben den funktionalen Ergebnissen auch die Ausführungsdauer sowie die Kosten für Token oder Abfragen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Weg von Demonstrationsumgebungen in gemeinsam genutzte Umgebungen wechselt. Überprüfen Sie nach der ersten Batch-Verarbeitung ein Beispiel aus dem Neighborhood. Cypher-Proben erkennen Label-Änderungen und fehlende Eigenschaften kostengünstiger als umfassende 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. Python-Abhängigkeiten installieren

Zur Nummer 3: Installieren Sie vor dem Ändern des Codes die Python-Abhängigkeiten, definieren Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckten Zustände schließen zu müssen. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher sowie Feature-Flags sollten an einem Ort gesammelt sein, den die Operator überprüfen können, ohne den gesamten Ablauf durchzulesen. Überprüfen Sie nach der ersten Batch-Verarbeitung ein Beispiel aus dem Umfeld – Cypher-Proben erkennen Label-Abweichungen sowie fehlende Eigenschaften kostengünstiger als 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. Konfigurieren Sie Umgebungsvariablen

Für Schritt 4: Konfigurieren Sie die Umgebungsvariablen, definieren Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien, bevor Sie Code ändern. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallplan. Wiederholte Versuche, menschliche Überprüfungen und die Handhabung fehlerhafter Nachrichten gehören zum Produkt selbst und nicht zu späteren Optimierungen. Überprüfen Sie nach der ersten Charge ein Beispiel aus dem Umfeld. Cypher-Proben erkennen Leistungsabfälle und fehlende Eigenschaften kostengünstiger als End-to-End-Chat-Tests. Für Schritt 4: Konfigurieren Sie die Umgebungsvariablen, definieren Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien, bevor Sie Code ändern. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf verborgene Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Artefakte, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab.

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. Beispieldokumente hinzufügen

Zur Aufgabe „5. Beispieldokumente hinzufügen“ sollten Sie vor dem Ändern des Codes die Eingabedaten, den Verantwortlichen für diesen Schritt sowie die Abbruchkriterien definieren. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erfassen Sie außerdem die Ausführungsdauer sowie die Kosten für Token oder Abfragen zusammen mit den funktionalen Ergebnissen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Ablauf von einer Demo-Umgebung in eine gemeinsam genutzte Umgebung wechselt. Modellieren Sie nur die Knoten und Beziehungen zu den Fragen, die Sie stellen möchten – nicht jedes Substantiv im Dokument. Spärliche, typisierte Verbindungen sind besser als dichte, unklare Graphen.

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. Vollständige Implementierung von GraphRAG

Für die vollständige GraphRAG-Implementierung sollten Sie vor dem Ändern des Codes die Eingaben, den Verantwortlichen für den jeweiligen Schritt sowie die Abbruchkriterien definieren. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Bewahren Sie die Konfiguration außerhalb des Anwendungscode auf. Umgebungsdateien, Geheimdatenspeicher sowie Feature-Flags sollten an einem Ort gesammelt sein, den die Operator überprüfen können, ohne den gesamten Graphen durchlesen zu müssen. Definieren Sie Modellknoten und -beziehungen nur für die Fragen, die Sie stellen möchten – nicht für jedes Substantiv im Dokument. Spärliche, typisierte Kanten sind besser als dichte, unklare Graphen.

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. Führen Sie den Ingestion-Pipeline aus

Für Schritt 7: Führen Sie vor dem Ändern des Codes den Ingestion-Pipeline aus, definieren Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Dokumentieren Sie gemeinsam den erfolgreichen Ablauf sowie den Notfallweg. Wiederholungsversuche, menschliche Überprüfungen und die Handhabung von Fehlern gehören zum Produkt selbst, nicht zu späteren Optimierungen. Modellieren Sie Knoten und Beziehungen für die Fragen, die Sie stellen werden – nicht für jedes Substantiv im Dokument. Spärliche, typisierte Verbindungen sind besser als dichte, unklare Graphen. Für Schritt 7: Führen Sie vor dem Ändern des Codes den Ingestion-Pipeline aus, definieren Sie die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Betrachten Sie diese Phase als Vertrag zwischen den Eingaben und den validierten Ausgaben. Benennen Sie die Erzeugnisse, definieren Sie Erfolgskontrollen und lehnen Sie stille, unvollständige Abschlüsse ab.

python app.py ingest --folder documents --clear

8. GraphRAG-Fragen stellen

Zur Aufgabe 8 „GraphRAG-Fragen stellen“ sollten Sie vor dem Ändern des Codes die Eingaben, den Verantwortlichen für den Schritt sowie die Abbruchkriterien definieren. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen. Erfassen Sie neben den funktionalen Ergebnissen auch die Laufzeiten sowie die Kosten für Tokens oder Abfragen. Eine frühzeitige Sichtbarkeit der Kosten verhindert überraschende Rechnungen, wenn der Prozess von Demonstrationsumgebungen in gemeinsam genutzte Umgebungen übergeht. Wenden Sie Einschränkungen und Indizes vor dem Masseneinlesen an. Die Eindeutigkeit der Geschäftschlüssel sorgt dafür, dass spätere Zusammenführungen zu vorhersehbaren Upserts statt zu Doppelgängerproblemen führen.

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. Was geschieht während der Abfristzeit?

Zu Punkt 9 „Was geschieht während der Abfristunde?“ sollten die Eingaben, der Verantwortliche für den Schritt sowie die Beendigungskriterien definiert werden, bevor der Code geändert wird. Die Operator sollten in der Lage sein, den Schritt von einem bekannten Checkpoint aus erneut auszuführen, ohne auf versteckte Zustände schließen zu müssen.

Schritt 1: Vektorabruf

Question → Embedding → Neo4j Vector Index → Top-k Chunks

Schritt 2: Grapherweiterung

Retrieved Chunk → Mentioned Entities → Related Entities → Relationship Paths

Schritt 3: Zusammenfassung der Community abrufen

Entities → Communities → Community Summaries

10. Den Graph in Neo4j untersuchen

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. Warum dies GraphRAG ist

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 gegenüber 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. Abschließende Gedanken

Vector search
+ Keyword search
+ Knowledge graph retrieval
+ Community summaries
+ Tool use
+ LLM reasoning
+ Evaluation

Operative Checkliste