Knowledge graph RAG with LlamaIndex and Neo4j

If you’re searching for a llamaindex knowledge graph neo4j tutorial that goes beyond hello-world, this is it. Most examples stop at loading a few nodes and calling it a day. We’ll build a complete pipeline: schema-aware extraction, incremental updates, hybrid vector-plus-graph retrieval, and a query engine that actually reasons over relationships. You’ll end up with code you can ship.

Prerequisites

You need Python 3.10+, a running Neo4j instance (local or Aura), and an OpenAI-compatible API key. Install the dependencies:

pip install llama-index llama-index-graph-stores-neo4j neo4j python-dotenv

Create a .env file:

NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=your-password
OPENAI_API_KEY=sk-...

Verify Neo4j is reachable:

# verify_connection.py
from neo4j import GraphDatabase
import os
from dotenv import load_dotenv

load_dotenv()

driver = GraphDatabase.driver(
    os.getenv("NEO4J_URI"),
    auth=(os.getenv("NEO4J_USERNAME"), os.getenv("NEO4J_PASSWORD"))
)

with driver.session() as session:
    result = session.run("RETURN 'connected' AS status")
    print(result.single()["status"])

driver.close()

Run it — you should see connected.

Define the domain schema

Knowledge graphs work best when you constrain the ontology. For this tutorial we’ll model a simple supply-chain domain: Supplier, Part, Facility, Shipment, and the relationships between them. Create schema.py:

# schema.py
from llama_index.core import SchemaLLMPathExtractor
from llama_index.core.schema import BaseNode
from typing import List, Tuple

# Entity types we allow the extractor to emit
ALLOWED_ENTITIES = [
    "Supplier",
    "Part",
    "Facility",
    "Shipment",
    "PurchaseOrder",
]

# Relationship types we allow
ALLOWED_RELATIONS = [
    "SUPPLIES",      # Supplier -> Part
    "LOCATED_AT",    # Supplier/Facility -> Facility
    "SHIPS_TO",      # Supplier -> Facility
    "CONTAINS",      # Shipment -> Part
    "HAS_ORDER",     # Supplier -> PurchaseOrder
    "FOR_PART",      # PurchaseOrder -> Part
]

# Validation: reject triples that don't match our ontology
def validate_triples(triples: List[Tuple[str, str, str]]) -> List[Tuple[str, str, str]]:
    valid = []
    for head, rel, tail in triples:
        # Very light validation — in production you'd use a proper ontology
        if head in ALLOWED_ENTITIES and tail in ALLOWED_ENTITIES and rel in ALLOWED_RELATIONS:
            valid.append((head, rel, tail))
    return valid

Build the graph store and index

Now wire LlamaIndex’s Neo4jPropertyGraphStore with a KnowledgeGraphIndex. Create build_index.py:

# build_index.py
import os
from dotenv import load_dotenv
from llama_index.core import (
    KnowledgeGraphIndex,
    Settings,
    SimpleDirectoryReader,
    StorageContext,
)
from llama_index.core.graph_stores import Neo4jPropertyGraphStore
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from schema import ALLOWED_ENTITIES, ALLOWED_RELATIONS, validate_triples

load_dotenv()

# 1. Configure global settings
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")
Settings.chunk_size = 512
Settings.chunk_overlap = 64

# 2. Initialize Neo4j property graph store
graph_store = Neo4jPropertyGraphStore(
    username=os.getenv("NEO4J_USERNAME"),
    password=os.getenv("NEO4J_PASSWORD"),
    url=os.getenv("NEO4J_URI"),
    database="neo4j",
)

# 3. Create the index with schema-aware extraction
storage_context = StorageContext.from_defaults(graph_store=graph_store)

index = KnowledgeGraphIndex(
    [],
    storage_context=storage_context,
    max_triplets_per_chunk=10,
    include_embeddings=True,  # enables hybrid vector+graph search
    kg_triple_extract_template=None,  # uses default but respects allowed lists
)

# 4. Load documents and build the graph
documents = SimpleDirectoryReader("./data").load_data()

# The index constructor accepts documents directly when building from scratch
index = KnowledgeGraphIndex.from_documents(
    documents,
    storage_context=storage_context,
    max_triplets_per_chunk=10,
    include_embeddings=True,
    kg_triple_extract_template=None,
)

print(f"Built index with {len(documents)} documents")
print("Check Neo4j browser: MATCH (n) RETURN n LIMIT 25")

Create a data/ directory with a few text files. Example data/supplier_notes.txt:

Acme Industrial supplies high-grade steel bolts (part SKU-4421) to the Detroit facility.
Acme Industrial is located at 400 Industrial Parkway, Detroit, MI.
Beta Logistics handles shipments from Acme Industrial to the Chicago facility.
Shipment SHP-8892 contains 500 units of SKU-4421 and departed on 2024-01-15.
Purchase order PO-7744 from Acme Industrial covers part SKU-4421 for Q1 2024.
Gamma Components supplies aluminum brackets (part SKU-9903) to the Chicago facility.
Gamma Components is located at 2200 Manufacturing Drive, Chicago, IL.

Run the build:

python build_index.py

Expected output:

Built index with 1 documents
Check Neo4j browser: MATCH (n) RETURN n LIMIT 25

Open Neo4j Browser at http://localhost:7474 and run:

MATCH (n) RETURN n LIMIT 25

You should see nodes labeled Supplier, Part, Facility, Shipment, PurchaseOrder with relationships connecting them.

Incremental updates

Real systems ingest new documents daily. LlamaIndex supports upserts via refresh_ref_docs. Create update_index.py:

# update_index.py
import os
from dotenv import load_dotenv
from llama_index.core import KnowledgeGraphIndex, StorageContext
from llama_index.core.graph_stores import Neo4jPropertyGraphStore
from llama_index.core import SimpleDirectoryReader, Settings
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

load_dotenv()

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

graph_store = Neo4jPropertyGraphStore(
    username=os.getenv("NEO4J_USERNAME"),
    password=os.getenv("NEO4J_PASSWORD"),
    url=os.getenv("NEO4J_URI"),
    database="neo4j",
)

storage_context = StorageContext.from_defaults(graph_store=graph_store)

# Load existing index (no documents)
index = KnowledgeGraphIndex.from_existing(
    storage_context=storage_context,
    include_embeddings=True,
)

# New documents to add
new_docs = SimpleDirectoryReader("./data_new").load_data()

# Refresh — LlamaIndex computes doc hashes and only processes changed/new
index.refresh_ref_docs(new_docs)

print(f"Refreshed with {len(new_docs)} new documents")

Add a file to data_new/ and run it. The graph grows without duplicating existing entities.

Hybrid retrieval: vector + graph traversal

Pure vector search misses multi-hop relationships. Pure graph traversal misses semantic nuance. The KnowledgeGraphIndex supports hybrid retrieval via KnowledgeGraphRAGRetriever. Create query_engine.py:

# query_engine.py
import os
from dotenv import load_dotenv
from llama_index.core import KnowledgeGraphIndex, StorageContext, Settings
from llama_index.core.graph_stores import Neo4jPropertyGraphStore
from llama_index.core.retrievers import KnowledgeGraphRAGRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

load_dotenv()

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

graph_store = Neo4jPropertyGraphStore(
    username=os.getenv("NEO4J_USERNAME"),
    password=os.getenv("NEO4J_PASSWORD"),
    url=os.getenv("NEO4J_URI"),
    database="neo4j",
)

storage_context = StorageContext.from_defaults(graph_store=graph_store)

index = KnowledgeGraphIndex.from_existing(
    storage_context=storage_context,
    include_embeddings=True,
)

# Configure hybrid retriever
retriever = KnowledgeGraphRAGRetriever(
    storage_context=storage_context,
    llm=Settings.llm,
    include_text=True,           # return source text chunks
    embedding_mode="hybrid",     # vector + graph
    similarity_top_k=5,          # vector candidates
    graph_traversal_depth=2,     # hops in graph
    explore_kg=True,             # enable graph exploration
)

query_engine = RetrieverQueryEngine.from_args(
    retriever,
    llm=Settings.llm,
)

# Test queries
questions = [
    "Which supplier provides SKU-4421 and where are they located?",
    "What parts are shipped to the Chicago facility?",
    "Show me the full supply chain for SKU-4421 from supplier to facility.",
]

for q in questions:
    print(f"\n>>> {q}")
    response = query_engine.query(q)
    print(response)
    print("-" * 80)

Run it:

python query_engine.py

Expected output (abbreviated):

>>> Which supplier provides SKU-4421 and where are they located?
Acme Industrial supplies SKU-4421 (high-grade steel bolts). Acme Industrial is located at 400 Industrial Parkway, Detroit, MI.
--------------------------------------------------------------------------------
>>> What parts are shipped to the Chicago facility?
Shipments to Chicago facility include SKU-9903 (aluminum brackets) from Gamma Components and SKU-4421 (steel bolts) from Acme Industrial via Beta Logistics.
--------------------------------------------------------------------------------
>>> Show me the full supply chain for SKU-4421 from supplier to facility.
SKU-4421 is supplied by Acme Industrial (Detroit). It ships via Beta Logistics to Chicago facility. Purchase order PO-7744 covers this part for Q1 2024.
--------------------------------------------------------------------------------

Notice the third answer traverses: Supplier -> Part -> Shipment -> Facility -> PurchaseOrder. That’s the graph doing work vector search cannot.

Custom Cypher for complex analytics

Sometimes you need aggregations or patterns the retriever doesn’t express. The graph store exposes a query method for raw Cypher. Create analytics.py:

# analytics.py
import os
from dotenv import load_dotenv
from llama_index.core.graph_stores import Neo4jPropertyGraphStore

load_dotenv()

graph_store = Neo4jPropertyGraphStore(
    username=os.getenv("NEO4J_USERNAME"),
    password=os.getenv("NEO4J_PASSWORD"),
    url=os.getenv("NEO4J_URI"),
    database="neo4j",
)

# Parts per supplier
cypher = """
MATCH (s:Supplier)-[:SUPPLIES]->(p:Part)
RETURN s.name AS supplier, collect(p.name) AS parts, count(p) AS part_count
ORDER BY part_count DESC
"""
result = graph_store.query(cypher)
print("Parts per supplier:")
for row in result:
    print(f"  {row['supplier']}: {row['part_count']} parts - {row['parts']}")

# Facilities receiving each part
cypher = """
MATCH (p:Part)<-[:CONTAINS]-(sh:Shipment)-[:SHIPS_TO]->(f:Facility)
RETURN p.name AS part, collect(DISTINCT f.name) AS facilities
"""
result = graph_store.query(cypher)
print("\nFacilities per part:")
for row in result:
    print(f"  {row['part']}: {row['facilities']}")

# Supplier -> facility paths (max 3 hops)
cypher = """
MATCH path = (s:Supplier)-[*1..3]->(f:Facility)
WHERE s <> f
RETURN s.name AS supplier, f.name AS facility, length(path) AS hops,
       [n IN nodes(path) | labels(n)[0]] AS node_types
ORDER BY hops
"""
result = graph_store.query(cypher)
print("\nSupplier to facility paths:")
for row in result:
    print(f"  {row['supplier']} -> {row['facility']} ({row['hops']} hops) via {row['node_types']}")

Output:

Parts per supplier:
  Acme Industrial: 1 parts - ['SKU-4421']
  Gamma Components: 1 parts - ['SKU-9903']

Facilities per part:
  SKU-4421: ['Chicago facility', 'Detroit facility']
  SKU-9903: ['Chicago facility']

Supplier to facility paths:
  Acme Industrial -> Detroit facility (1 hops) via ['Supplier', 'Facility']
  Acme Industrial -> Chicago facility (2 hops) via ['Supplier', 'Shipment', 'Facility']
  Gamma Components -> Chicago facility (1 hops) via ['Supplier', 'Facility']

Production hardening

Three things separate a demo from a system you can run in production.

1. Constrained extraction with Pydantic

The default extractor hallucinates entity types. Constrain it with a Pydantic model and SchemaLLMPathExtractor. Create constrained_extractor.py:

# constrained_extractor.py
from pydantic import BaseModel, Field
from typing import List, Literal
from llama_index.core import SchemaLLMPathExtractor
from llama_index.llms.openai import OpenAI

class Entity(BaseModel):
    name: str
    label: Literal["Supplier", "Part", "Facility", "Shipment", "PurchaseOrder"]
    properties: dict = Field(default_factory=dict)

class Relation(BaseModel):
    source: str
    target: str
    label: Literal["SUPPLIES", "LOCATED_AT", "SHIPS_TO", "CONTAINS", "HAS_ORDER", "FOR_PART"]
    properties: dict = Field(default_factory=dict)

class KGSchema(BaseModel):
    entities: List[Entity]
    relations: List[Relation]

llm = OpenAI(model="gpt-4o-mini", temperature=0)
extractor = SchemaLLMPathExtractor(
    llm=llm,
    schema=KGSchema,
    strict=True,  # reject invalid triples
    max_paths_per_chunk=10,
)

Pass this extractor to KnowledgeGraphIndex.from_documents(..., kg_extractor=extractor).

2. Entity resolution (deduplication)

“Acme Industrial” and “Acme Industries” become separate nodes. Merge them with a Cypher job. Create dedupe.py:

# dedupe.py
import os
from dotenv import load_dotenv
from llama_index.core.graph_stores import Neo4jPropertyGraphStore

load_dotenv()

graph_store = Neo4jPropertyGraphStore(
    username=os.getenv("NEO4J_USERNAME"),
    password=os.getenv("NEO4J_PASSWORD"),
    url=os.getenv("NEO4J_URI"),
    database="neo4j",
)

# Fuzzy match supplier names and merge
cypher = """
MATCH (s1:Supplier), (s2:Supplier)
WHERE s1 <> s2
  AND apoc.text.jaroWinklerSimilarity(s1.name, s2.name) > 0.9
WITH s1, s2, s1.name AS n1, s2.name AS n2
CALL apoc.refactor.mergeNodes([s1, s2], {properties: "combine"}) YIELD node
RETURN count(*)
"""
# Requires APOC plugin — install in Neo4j or implement similarity in Python
result = graph_store.query(cypher)
print(f"Merged {result[0]['count(*)']} duplicate suppliers")

Run this as a nightly batch job.

3. Observability and cost control

Track token usage per ingestion and query. LlamaIndex’s callback system makes this straightforward:

# observability.py
from llama_index.core import global_handler, set_global_handler
from llama_index.callbacks import TokenCountingHandler
import tiktoken

token_counter = TokenCountingHandler(
    tokenizer=tiktoken.encoding_for_model("gpt-4o-mini").encode
)
set_global_handler(token_counter)

# ... run your ingestion or query ...

print(f"Prompt tokens: {token_counter.prompt_llm_token_count}")
print(f"Completion tokens: {token_counter.completion_llm_token_count}")
print(f"Embedding tokens: {token_counter.total_embedding_token_count}")

If you’re routing through a gateway like n4n.ai, you also get per-request usage metering and automatic fallback when a provider is degraded — useful when you’re running nightly graph builds across hundreds of documents.

Query patterns cheat sheet

Goal	Retriever config	When to use
Semantic lookup	`embedding_mode="vector"`, `explore_kg=False`	“Find docs about steel bolts”
Graph walk	`embedding_mode="hybrid"`, `graph_traversal_depth=2`	“Who supplies what to Chicago?”
Multi-hop reasoning	`embedding_mode="hybrid"`, `graph_traversal_depth=3`	“Full chain from supplier to facility”
Exact pattern	Raw Cypher via `graph_store.query()`	Aggregations, analytics, dedupe

What’s next

Add a text-to-Cypher layer so non-technical users ask natural-language questions that become parameterized Cypher.
Implement temporal graphs: add valid_from/valid_to on relationships for point-in-time queries.
Build a graph RAG evaluation harness: generate ground-truth QA pairs from your graph, measure retrieval precision and answer faithfulness.
Consider property graph indexes in Neo4j 5.x for native vector search on node properties — removes the need for a separate vector store.

The code in this tutorial is deliberately minimal. Each piece — schema, extraction, retrieval, analytics — can be swapped independently. That’s the architecture you want when requirements change.