How to Build a GraphRAG System with Neo4j

How to Build a GraphRAG System with Neo4j

Retrieval-Augmented Generation (RAG) has become the default pattern for grounding large language models in private or domain-specific data. Traditional RAG relies on vector similarity search over chunked text embeddings, which works well for factoid lookups but struggles with multi-hop reasoning, entity relationships, and global questions that span many documents. GraphRAG addresses these limitations by combining vector retrieval with a knowledge graph, letting the model traverse relationships between entities rather than matching isolated text fragments. In this tutorial, you'll build a complete GraphRAG pipeline using Neo4j as the graph database and OpenAI for embeddings and generation.

What Is GraphRAG?

GraphRAG is an extension of RAG that augments vector search with structured knowledge graph traversal. Instead of retrieving only the most semantically similar text chunks, GraphRAG also retrieves connected entities, their relationships, and the neighborhoods around them. This produces richer, more contextual prompts for the language model.

The core idea is that information in the real world is relational. A document about a drug trial mentions patients, treatments, side effects, and outcomes — all connected. A vector store flattens this into independent chunks, losing the connections. A graph preserves them explicitly as nodes and edges, enabling queries like "which patients experienced side effects from treatments developed by company X?" to be answered by traversing the graph rather than hoping the right chunk was embedded.

Why Neo4j?

Neo4j is a native graph database that stores data as nodes, relationships, and properties. It supports Cypher, a declarative query language purpose-built for graph traversal. Neo4j also ships with a vector index (since version 5.11), which means you can run hybrid queries that combine semantic similarity with graph traversal in a single Cypher statement. This makes it an ideal single-store backend for GraphRAG: you don't need a separate vector database and graph database glued together.

• Native graph storage — relationships are first-class citizens, not computed at query time.
• Built-in vector search — hybrid retrieval without a second system.
• Cypher — expressive queries for multi-hop traversal.
• LangChain and LlamaIndex integration — mature ecosystem support.
• GraphRAG package — Neo4j maintains a reference implementation.

Architecture Overview

A GraphRAG system has four main stages: ingestion, indexing, retrieval, and generation. During ingestion, source documents are split into chunks and entities are extracted (often with an LLM). During indexing, entities and relationships are written as graph nodes and edges, and chunk embeddings are stored alongside the graph. During retrieval, a user query is embedded and used to find relevant chunks and entities via vector search, then the graph is traversed to gather connected context. During generation, the assembled context is passed to an LLM to produce the final answer.

Prerequisites and Setup

You'll need Python 3.10+, a Neo4j instance (Aura Free cloud works fine, or a local Docker container), and an OpenAI API key. Install the required packages:

pip install neo4j langchain langchain-openai langchain-community \
            langchain-experimental pydantic python-dotenv

Start a local Neo4j instance with Docker if you don't want to use Aura:

docker run -d --name neo4j-graphrag \
  -p 7474:7474 -p 7687:7687 \
  -e NEO4J_AUTH=neo4j/yourpassword \
  neo4j:5.20

Create a .env file with your credentials:

OPENAI_API_KEY=sk-your-key-here
NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=yourpassword

Step 1: Connecting to Neo4j

First, establish a connection to Neo4j and initialize the OpenAI models you'll use throughout the pipeline.

import os
from dotenv import load_dotenv
from neo4j import GraphDatabase
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

load_dotenv()

driver = GraphDatabase.driver(
    os.environ["NEO4J_URI"],
    auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Verify the connection
with driver.session() as session:
    result = session.run("RETURN 1 AS test")
    print("Connected:", result.single()["test"])

Step 2: Creating the Vector Index

Neo4j needs a vector index to perform similarity search over chunk embeddings. Create it once before ingesting data. The index must specify the node label, the property holding the embedding, the dimensions, and the similarity function.

def create_vector_index(driver):
    with driver.session() as session:
        session.run("""
        CREATE VECTOR INDEX chunk_embeddings IF NOT EXISTS
        FOR (c:Chunk) ON (c.embedding)
        OPTIONS {
            indexConfig: {
                `vector.dimensions`: 1536,
                `vector.similarity_function`: 'cosine'
            }
        }
        """)
        # Also create constraints for entity uniqueness
        session.run(
            "CREATE CONSTRAINT entity_id IF NOT EXISTS "
            "FOR (e:Entity) REQUIRE e.id IS UNIQUE"
        )

create_vector_index(driver)

The dimension value 1536 matches OpenAI's text-embedding-3-small model. Adjust it if you use a different embedding model.

Step 3: Ingesting Documents and Extracting Entities

The ingestion pipeline splits documents into chunks, stores each chunk as a node, and uses an LLM to extract entities and relationships from each chunk. Those entities become graph nodes connected to their source chunk and to each other.

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
import uuid

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=100
)

EXTRACTION_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are an entity and relationship extractor.
Extract entities and their relationships from the text.
Return ONLY valid JSON with this schema:
{{
  "entities": [{{"name": string, "type": string, "description": string}}],
  "relationships": [{{"source": string, "target": string, "type": string, "description": string}}]
}}
Entity names must be lowercase. Be precise and concise."""),
    ("human", "{text}")
])

def extract_entities_and_relationships(text):
    chain = EXTRACTION_PROMPT | llm
    response = chain.invoke({"text": text})
    import json
    try:
        return json.loads(response.content)
    except json.JSONDecodeError:
        return {"entities": [], "relationships": []}

def ingest_document(driver, embeddings, document_text, source_name):
    chunks = text_splitter.split_text(document_text)
    chunk_embeddings = embeddings.embed_documents(chunks)

    with driver.session() as session:
        for i, (chunk_text, chunk_emb) in enumerate(zip(chunks, chunk_embeddings)):
            chunk_id = str(uuid.uuid4())

            # Store the chunk with its embedding
            session.run(
                """
                CREATE (c:Chunk {
                    id: $chunk_id,
                    text: $text,
                    embedding: $embedding,
                    source: $source,
                    position: $position
                })
                """,
                chunk_id=chunk_id,
                text=chunk_text,
                embedding=chunk_emb,
                source=source_name,
                position=i
            )

            # Extract entities and relationships from this chunk
            extracted = extract_entities_and_relationships(chunk_text)

            # Create entity nodes and link them to the chunk
            for entity in extracted["entities"]:
                session.run(
                    """
                    MERGE (e:Entity {name: $name})
                    ON CREATE SET e.id = randomUUID(),
                                  e.type = $type,
                                  e.description = $description
                    ON MATCH SET e.description = $description
                    WITH e
                    MATCH (c:Chunk {id: $chunk_id})
                    MERGE (c)-[:MENTIONS]->(e)
                    """,
                    name=entity["name"],
                    type=entity.get("type", "Unknown"),
                    description=entity.get("description", ""),
                    chunk_id=chunk_id
                )

            # Create relationships between entities
            for rel in extracted["relationships"]:
                session.run(
                    """
                    MATCH (src:Entity {name: $source})
                    MATCH (tgt:Entity {name: $target})
                    MERGE (src)-[r:RELATES {type: $type}]->(tgt)
                    ON CREATE SET r.description = $description
                    """,
                    source=rel["source"],
                    target=rel["target"],
                    type=rel.get("type", "RELATED_TO"),
                    description=rel.get("description", "")
                )

    print(f"Ingested {len(chunks)} chunks from {source_name}")

Now ingest a sample document to test the pipeline:

sample_text = """
Pfizer developed the COVID-19 vaccine Comirnaty in collaboration with BioNTech.
The vaccine uses mRNA technology and was approved by the FDA in August 2021.
Clinical trials showed 95% efficacy in preventing COVID-19 infection.
Common side effects include fatigue, headache, and fever.
Albert Bourla is the CEO of Pfizer and has led the company since 2019.
BioNTech was founded by Uğur Şahin and Özlem Türeci in Mainz, Germany.
"""

ingest_document(driver, embeddings, sample_text, "pfizer_covid.txt")

Step 4: Building the Retrieval Pipeline

Retrieval in GraphRAG combines vector similarity search with graph traversal. The query is embedded, the most similar chunks are found, and then the entities mentioned in those chunks are expanded to include their neighborhoods. This gives the LLM both the direct text evidence and the relational context.

def retrieve_context(driver, embeddings, query, top_k=5, hop_depth=2):
    query_embedding = embeddings.embed_query(query)

    with driver.session() as session:
        # Step 1: Find similar chunks via vector search
        chunk_result = session.run(
            """
            CALL db.index.vector.queryNodes('chunk_embeddings', $top_k, $embedding)
            YIELD node, score
            RETURN node.text AS text, node.source AS source, score
            """,
            top_k=top_k,
            embedding=query_embedding
        )
        chunks = [dict(record) for record in chunk_result]

        # Step 2: Find entities mentioned in those chunks and traverse
        entity_result = session.run(
            """
            CALL db.index.vector.queryNodes('chunk_embeddings', $top_k, $embedding)
            YIELD node, score
            MATCH (node)-[:MENTIONS]->(e:Entity)
            OPTIONAL MATCH path = (e)-[:RELATES*1..$hop_depth]-(connected:Entity)
            WITH e, connected, relationships(path) AS rels
            RETURN e.name AS entity,
                   e.type AS entity_type,
                   e.description AS entity_description,
                   collect(DISTINCT {
                       connected: connected.name,
                       connected_type: connected.type
                   }) AS neighbors
            """,
            top_k=top_k,
            embedding=query_embedding,
            hop_depth=hop_depth
        )
        entities = [dict(record) for record in entity_result]

    return chunks, entities

Step 5: Generating Answers

With the retrieved chunks and entity subgraph in hand, assemble a prompt that gives the LLM both the raw text evidence and the structured relationship context, then generate the answer.

from langchain_core.prompts import ChatPromptTemplate

ANSWER_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """You are a knowledgeable assistant answering questions
using the provided context. The context includes relevant text passages
and a knowledge graph of entities and their relationships.

Use both the text passages and the entity relationships to formulate
your answer. If the context is insufficient, say so clearly.
Always cite which entities and relationships informed your answer.

TEXT PASSAGES:
{text_passages}

ENTITY GRAPH:
{entity_graph}"""),
    ("human", "{question}")
])

def format_passages(chunks):
    return "\n\n".join(
        f"[Passage {i+1}] (score: {c['score']:.3f}, source: {c['source']})\n{c['text']}"
        for i, c in enumerate(chunks)
    )

def format_entity_graph(entities):
    lines = []
    for e in entities:
        neighbor_str = ", ".join(
            f"{n['connected']}({n['connected_type']})"
            for n in e["neighbors"] if n["connected"]
        )
        lines.append(
            f"- {e['entity']} ({e['entity_type']}): {e['entity_description']}\n"
            f"  Connected to: {neighbor_str or 'none'}"
        )
    return "\n".join(lines)

def answer_question(driver, embeddings, question):
    chunks, entities = retrieve_context(driver, embeddings, question)

    chain = ANSWER_PROMPT | llm
    response = chain.invoke({
        "question": question,
        "text_passages": format_passages(chunks),
        "entity_graph": format_entity_graph(entities)
    })
    return response.content

# Test it
answer = answer_question(driver, embeddings,
    "Who developed the COVID-19 vaccine and what are its side effects?")
print(answer)

Step 6: Adding a Conversational Interface

To make the system usable, wrap it in a simple REPL loop that maintains conversation history so follow-up questions can reference prior context.

def chat_loop(driver, embeddings):
    print("GraphRAG Chat (type 'quit' to exit)\n")
    while True:
        question = input("You: ").strip()
        if question.lower() in ("quit", "exit", "q"):
            break
        if not question:
            continue
        print("\nAssistant:", answer_question(driver, embeddings, question))
        print()

chat_loop(driver, embeddings)

Best Practices

• Normalize entity names. LLM extraction is noisy. Use a canonicalization step — merge "Pfizer Inc." and "pfizer" into one node. A second LLM call or a fuzzy matching pass over existing entity names helps.
• Use community detection for global questions. For questions like "what are the main themes in this corpus?", run Louvain or Leiden algorithms on the graph to detect communities, then summarize each community. Neo4j's Graph Data Science library supports this natively.
• Tune hop depth carefully. A hop depth of 1 or 2 is usually enough. Going deeper pulls in too much noise and slows queries. Benchmark retrieval quality at each depth on your data.
• Cache embeddings. Re-embedding the same chunks wastes API calls. Store embeddings in the graph and only recompute when the source text changes.
• Handle extraction failures gracefully. LLM JSON output can be malformed. Wrap parsing in try/except and log failures so you can monitor extraction quality over time.
• Use structured output when available. OpenAI's function calling and response_format produce more reliable structured output than free-text JSON parsing. Prefer with_structured_output in LangChain.
• Monitor graph size. Entity extraction can produce very large graphs from large corpora. Set limits on nodes per chunk and prune low-confidence relationships periodically.
• Evaluate with a test set. Build a small set of question-answer pairs and measure retrieval recall and answer faithfulness. Tools like RAGAS or TruLens can automate this.

Advanced: Using Neo4j's GraphRAG Python Package

Neo4j maintains an official neo4j-graphrag package that abstracts much of this pipeline. If you'd rather not build from scratch, here's the equivalent retrieval using the package:

pip install neo4j-graphrag

from neo4j_graphrag.retrievers import VectorRetriever
from neo4j_graphrag.generation import GraphRAG
from neo4j_graphrag.llm import OpenAILLM
from neo4j_graphrag.embeddings import OpenAIEmbeddings

neo4j_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
retriever = VectorRetriever(
    driver=driver,
    index_name="chunk_embeddings",
    embedder=neo4j_embeddings,
    return_properties=["text", "source"]
)

llm_pkg = OpenAILLM(model_name="gpt-4o-mini")
rag = GraphRAG(retriever=retriever, llm=llm_pkg)

result = rag.search(
    "Who developed the COVID-19 vaccine?",
    retriever_config={"top_k": 5}
)
print(result.answer)

This package handles the plumbing but gives you less control over graph traversal. For production systems with custom retrieval logic, the manual approach shown earlier is often preferable.

Conclusion

GraphRAG with Neo4j gives you a retrieval architecture that captures both the semantic content of your documents and the relationships between the entities they describe. By storing chunks, embeddings, entities, and relationships in a single database, you can run hybrid queries that combine vector similarity with multi-hop graph traversal — something no standalone vector store can do. The pipeline you built here covers the full lifecycle: document ingestion, entity extraction, vector indexing, hybrid retrieval, and answer generation. As you move toward production, focus on entity normalization, extraction quality, retrieval evaluation, and community detection for global questions. With those pieces in place, GraphRAG becomes a powerful foundation for building LLM applications that reason over complex, interconnected knowledge.