Neo4j + AI: Building Intelligent Applications with Knowledge Graphs | Graphs | graphwiz.ai
Neo4j + AI: Building Intelligent Applications with Knowledge Graphs
neo4jknowledge-graphsgraphragllmcyphergraph-ai
Neo4j and AI are a natural pairing. Graph databases excel at storing and traversing relationships — exactly what LLMs struggle with. By using Neo4j as the structured knowledge layer, you can ground AI responses in verifiable facts, enable multi-hop reasoning across entities, and build agents that explore and enrich the graph autonomously.
This article walks through four production-grade integration patterns, from simplest to most sophisticated, with concrete code examples for each.
Why Neo4j for AI
Neo4j is the leading graph database for relationship-rich data. When you combine Neo4j with AI, you get a stack where the database handles structured knowledge and the LLM handles natural language. The critical advantage is that graph queries are deterministic — every result traces back to explicit nodes and edges, unlike vector similarity search where relevance is probabilistic.
Three patterns dominate the Neo4j + AI landscape: graph-backed RAG, natural language to Cypher, and autonomous graph agents. Each solves a different problem, and production systems typically combine all three.
Use Case 1: Graph-Backed RAG
Replace flat vector search with Cypher-powered retrieval. Instead of retrieving the top-k most similar text chunks, you query the graph for precisely the entities and relationships you need:
The Full Pipeline
Here is a complete Python implementation that queries a Neo4j knowledge graph, retrieves a structured subgraph, and feeds it into an LLM prompt:
from neo4j import GraphDatabase
from openai import OpenAI
import json
# --- Step 1: Query the knowledge graph ---
driver = GraphDatabase.driver(
"neo4j+s://localhost:7687",
auth=("neo4j", "password")
)
def retrieve_subgraph(tx, topic: str) -> dict:
"""Return a structured subgraph around a topic."""
result = tx.run("""
MATCH (t:Topic {name: $topic})
OPTIONAL MATCH (t)<-[:ABOUT]-(p:Paper)-[:AUTHORED_BY]->(a:Author)
OPTIONAL MATCH (p)-[:CITES]->(cited:Paper)
RETURN t.name AS topic,
collect(DISTINCT {
title: p.title,
abstract: p.abstract,
authors: collect(DISTINCT a.name),
citations: collect(DISTINCT cited.title)
}) AS papers
""", topic=topic)
return result.single().data()
# --- Step 2: Format as LLM context ---
with driver.session() as session:
subgraph = session.execute_read(retrieve_subgraph, "GraphRAG")
context = f"""
Topic: {subgraph['topic']}
Related papers and their connections:
{json.dumps(subgraph['papers'], indent=2)}
"""
# --- Step 3: Generate grounded response ---
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "Answer based only on the provided knowledge graph context."},
{"role": "user", "content": f"{context}\n\nWhat are the main research trends in this area?"}
]
)
print(response.choices[0].message.content)
driver.close()
Feed the structured results into your LLM prompt for grounded, relationship-aware answers. Unlike vector RAG, every fact in the result is verifiable against the graph — the LLM cannot hallucinate an author who did not write the paper.
Hybrid Search Pattern
For production systems, combine graph traversal with vector similarity. Neo4j supports both full-text indexes and vector indexes, enabling weighted reciprocal rank fusion (WRRF) to merge results:
// Hybrid search: vector similarity + graph traversal
CALL db.index.vector.queryNodes('paper-embeddings', 10, $queryEmbedding)
YIELD node AS paper, score
MATCH (paper)-[:ABOUT]->(t:Topic)
MATCH (paper)<-[:AUTHORED]-(a:Author)
RETURN paper.title, a.name, t.name AS topic, score
ORDER BY score DESC
LIMIT 5
This pattern retrieves semantically similar papers, then enriches them with graph structure. The vector search finds candidates; the graph traversal adds context that vector similarity cannot infer.
Here is a concrete Python example using the neo4j-graphrag library:
from neo4j import GraphDatabase
from neo4j_graphrag.llm import OpenAILLM
from neo4j_graphrag.retrievers import VectorCypherRetriever
from neo4j_graphrag.generation import GraphRAG
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
retriever = VectorCypherRetriever(
driver=driver,
index_name="paper_embeddings",
retrieval_query="""
MATCH (node)-[:ABOUT]->(t:Topic)
OPTIONAL MATCH (node)<-[:AUTHORED]-(a:Author)
RETURN node.abstract AS text,
score,
{title: node.title, authors: collect(a.name)} AS metadata
""",
)
llm = OpenAILLM(model_name="gpt-4o")
rag = GraphRAG(retriever=retriever, llm=llm)
response = rag.search("What are the latest advances in GraphRAG?")
print(response.answer)
The retrieval_query enriches vector search results with graph context — the LLM sees not just the text but also the authors and topic relationships.
Use Case 2: Natural Language to Cypher
Let users ask questions in plain English and have an LLM translate them into Cypher:
You are a Neo4j Cypher expert. Given a user question, generate a
Cypher query against this schema:
{schema}
Rules:
- Always use parameterised queries ($param), never string interpolation
- Return only the Cypher query, no explanation
- If the question cannot be answered, return // UNANSWERABLE
User question: {question}
Cypher query:
Use GraphCypherQAChain from LangChain or write your own prompt template.
Validation and Error Handling
Generated Cypher can contain syntax errors, hallucinated property names, or unsafe write operations. Always validate before execution:
import re
from neo4j import GraphDatabase
def validate_and_execute(cypher: str, driver, params: dict = None) -> list:
"""Validate and safely execute generated Cypher."""
# Block write operations in read-only mode
if re.search(r'\b(CREATE|DELETE|SET|MERGE|REMOVE)\b', cypher, re.IGNORECASE):
raise ValueError("Write operations are not permitted in read-only mode")
# Run an EXPLAIN first to catch syntax errors
with driver.session() as session:
try:
explain = session.run(f"EXPLAIN {cypher}", params or {})
explain.consume() # succeeds = valid syntax
except Exception as e:
# Retry with schema-aware hints
raise RuntimeError(f"Invalid Cypher: {e}") from e
# Execute if validation passes
with driver.session() as session:
result = session.run(cypher, params or {})
return [record.data() for record in result]
Pair this with a prompt that constrains the LLM to use only known labels and property names extracted from the database schema:
def get_schema(driver) -> str:
with driver.session() as session:
result = session.run("""
CALL apoc.meta.schema()
YIELD label, properties
RETURN label, properties
""")
return "\n".join(
f"{r['label']}: {', '.join(r['properties'].keys())}"
for r in result
)
Inject the schema string into your LLM prompt template so the generated Cypher stays within the bounds of your actual graph model.
Use Case 3: Autonomous Graph Agents
Build agents that explore and enrich the graph autonomously. A multi-agent architecture separates concerns:
Query Agent
Given a user question, the agent traverses the graph to find relevant sub-graphs. Unlike fixed-pipeline RAG, it can follow unexpected relationship paths:
async def query_agent(question: str, driver, llm):
"""Explore the graph to answer a question."""
# Step 1: Identify starting entities
entities = await llm.extract_entities(question)
# Step 2: Query the graph for each entity
subgraphs = []
for entity in entities:
with driver.session() as session:
result = await session.run(
"MATCH (e:Entity {name: $name})-[r*1..3]-(neighbour) "
"RETURN e, r, neighbour LIMIT 50",
name=entity
)
subgraphs.append(await result.data())
# Step 3: Let the LLM synthesise across subgraphs
answer = await llm.synthesise(question, subgraphs)
return answer
Extraction Agent
Processes new documents, extracts entities and relationships, and inserts them into the graph:
async def extraction_agent(document: str, driver, llm):
""""Extract entities and relationships from a document into the graph."""
# Extract structured data using the LLM
triples = await llm.extract_triples(document)
# Insert into Neo4j in a single transaction
with driver.session() as session:
await session.run("""
UNWIND $triples AS t
MERGE (s:Entity {name: t.subject})
MERGE (o:Entity {name: t.object})
CALL apoc.create.relationship(s, t.relation, {}, o) YIELD rel
RETURN count(rel)
""", triples=triples)
Maintenance Agent
Detects stale or contradictory relationships and flags them for review. A scheduled job runs daily:
def maintenance_agent(driver):
"""Identify outdated or contradictory relationships."""
with driver.session() as session:
# Find relationships older than 90 days with no recent activity
stale = session.run("""
MATCH (s)-[r]->(o)
WHERE r.last_updated < datetime() - duration({days: 90})
RETURN s.name, type(r), o.name, r.last_updated
ORDER BY r.last_updated
LIMIT 20
""")
return [record.data() for record in stale]
Integration Pattern Decision Table
Pattern
Best For
Query Method
Latency
Graph-backed RAG
Factual QA, multi-hop reasoning
Cypher
10-50ms
Hybrid search
Open-ended discovery
Vector + Cypher
50-200ms
Text-to-Cypher
Ad-hoc user queries
LLM → Cypher
1-5s
Autonomous agents
Complex workflows
Multi-turn
5-30s
Use Case Comparison
Use Case
Input
Output
Key Technology
Best For
Graph-Backed RAG
User query
Grounded answer + graph paths
VectorCypherRetriever, GraphRAG
Question answering over connected documents
NL to Cypher
Natural language
Executable Cypher query
GraphCypherQAChain, prompt templates
Analytics dashboards, self-serve querying
Autonomous Agents
Task description
Graph mutations, reports
Agent loops, LLM function calling
Automated knowledge enrichment, monitoring
Hybrid Search with Neo4j
Combine vector similarity with graph traversal for richer retrieval:
from neo4j_graphrag.retrievers import HybridRetriever
hybrid_retriever = HybridRetriever(
driver=driver,
vector_index="paper_embeddings",
fulltext_index="paper_fulltext",
retrieval_query="""
MATCH (node)-[:CITES]->(cited:Paper)
RETURN node.abstract AS text,
score,
{cited_titles: collect(cited.title)} AS metadata
""",
)
results = hybrid_retriever.search(
query_text="knowledge graph reasoning",
top_k=5,
vector_weight=0.7, # tune the blend
)
The vector_weight parameter lets you control the balance: push towards 1.0 for semantic similarity dominance, or lower towards 0.0 for keyword matches via the fulltext index. The Cypher enrichment layer then adds citation relationships that neither vector nor keyword search can capture alone.
Schema Extraction and Graph Loading
Before you can query, you need to model and load your data. Here is a pipeline that extracts schema from JSON documents and ingests it into Neo4j:
import json
from typing import Dict, Any
from neo4j import GraphDatabase
class GraphLoader:
def __init__(self, uri: str, user: str, password: str):
self.driver = GraphDatabase.driver(uri, auth=(user, password))
def infer_schema(self, doc: Dict[str, Any], parent_label: str = None) -> list:
"""Infer node labels and relationships from a document."""
statements = []
main_label = doc.get("type", "Document").capitalize()
props = {k: v for k, v in doc.items() if k != "type" and not isinstance(v, (dict, list))}
statements.append(
(f"MERGE (n:{main_label} {{id: $id}}) SET n += $props", {"id": doc["id"], "props": props})
)
for key, value in doc.items():
if isinstance(value, dict):
child_label = key.capitalize()
statements.append(
(f"MERGE (c:{child_label} {{id: $child_id}}) "
f"MERGE (n:{main_label} {{id: $parent_id}})-[:HAS_{key.upper()}]->(c)",
{"child_id": value["id"], "parent_id": doc["id"]})
)
return statements
def load_document(self, doc: Dict[str, Any]) -> None:
with self.driver.session() as session:
for query, params in self.infer_schema(doc):
session.run(query, params)
def close(self) -> None:
self.driver.close()
# Usage
loader = GraphLoader("bolt://localhost:7687", "neo4j", "password")
with open("documents.json") as f:
for doc in json.load(f):
loader.load_document(doc)
loader.close()
This pattern automatically creates nodes and relationships from nested JSON structure, giving you a populated graph without manual Cypher scripting for every document type.
Production Considerations
When moving from prototype to production, address these concerns:
Connection Pooling
The Neo4j Python driver manages a pool of Bolt connections. Configure it for your workload:
Use a single driver instance per application — creating a new driver for every request exhausts connections and adds latency.
Session Management
Always use sessions as context managers. Avoid long-lived sessions; open them per-transaction and close promptly:
with driver.session(database="knowledge", fetch_size=500) as session:
tx = session.begin_transaction()
try:
tx.run("MERGE (a:Author {name: $name})", name="Ada Lovelace")
tx.commit()
except Exception:
tx.rollback()
raise
Monitoring and Observability
Enable query logging and use apoc.monitor.query() to track slow queries:
CALL apoc.monitor.query(true) YIELD query, elapsed, params
WHERE elapsed > 1000
RETURN query, elapsed, params
ORDER BY elapsed DESC
For the LLM side, log prompt templates, generated Cypher, and response times. A centralised logging setup helps debug hallucinated queries and unexpected graph traversal patterns.
Getting Started
The fastest way to experiment with these patterns is to run Neo4j locally:
from neo4j import GraphDatabase
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
# Seed sample data
with driver.session() as session:
session.run("""
MERGE (a:Author {name: "Ada Lovelace"})
MERGE (p:Paper {title: "Notes on the Analytical Engine", abstract: "First computer algorithm"})
MERGE (t:Topic {name: "Computing"})
MERGE (a)-[:AUTHORED]->(p)
MERGE (p)-[:ABOUT]->(t)
""")
result = session.run(
"MATCH (a:Author)-[:AUTHORED]->(p:Paper)-[:ABOUT]->(t:Topic) "
"RETURN a.name AS author, p.title AS paper, t.name AS topic"
)
for record in result:
print(f"{record['author']} wrote '{record['paper']}' about {record['topic']}")
driver.close()
The Neo4j knowledge graph infrastructure documented in AGENTS.md connects directly to these patterns. The graph is already running — you just need to query it. For production deployments, use the neo4j+s:// URI scheme for TLS encryption and environment variables for credentials:
import os
driver = GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USER"], os.environ["NEO4J_PASSWORD"])
)