RAG with Neo4j and LLMs - Supercharging AI with Knowledge Graphs

What is RAG and Why Combine it with Neo4j?

RAG (Retrieval-Augmented Generation) is a technique that enhances large language model (LLM) responses by providing them with relevant context retrieved from an external knowledge base. Instead of relying solely on knowledge contained in model parameters, RAG searches for specific information before generating a response.

Neo4j, as a graph database, is ideal for RAG because:

✅ Relational context: Captures complex relationships between entities
✅ Efficient navigation: Allows rapid traversal of relationships
✅ Structured knowledge: Organizes information semantically
✅ Contextual search: Finds relevant information considering connections

RAG Architecture with Neo4j

User → Question
    ↓
1. Question embedding
    ↓
2. Search in Neo4j (Vector Search)
    ↓
3. Retrieval of relational context
    ↓
4. Prompt construction with context
    ↓
5. LLM generates enriched response
    ↓
Contextualized response → User

Prerequisites

# Install required dependencies
pip install neo4j langchain openai tiktoken

You’ll need:

Neo4j 5.11+ (with Vector Index support)
OpenAI API key (or any compatible LLM)
Python 3.9+

Step 1: Configure Neo4j with Vectors

First, create a vector index in Neo4j to store embeddings:

// Create vector index for documents
CREATE VECTOR INDEX document_embeddings IF NOT EXISTS
FOR (d:Document)
ON d.embedding
OPTIONS {
  indexConfig: {
    `vector.dimensions`: 1536,
    `vector.similarity_function`: 'cosine'
  }
}

Step 2: Load Knowledge into the Graph

from neo4j import GraphDatabase
import openai

class KnowledgeGraphRAG:
    def __init__(self, uri, user, password, openai_key):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
        openai.api_key = openai_key

    def close(self):
        self.driver.close()

    def create_embedding(self, text):
        """Generate embedding using OpenAI"""
        response = openai.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

    def add_document(self, title, content, metadata=None):
        """Add document to graph with its embedding"""
        embedding = self.create_embedding(content)

        with self.driver.session() as session:
            session.run("""
                CREATE (d:Document {
                    title: $title,
                    content: $content,
                    embedding: $embedding,
                    metadata: $metadata,
                    created_at: datetime()
                })
            """, title=title, content=content,
                embedding=embedding, metadata=metadata)

    def add_relationship(self, doc1_title, doc2_title, rel_type):
        """Create relationship between documents"""
        with self.driver.session() as session:
            session.run("""
                MATCH (d1:Document {title: $doc1})
                MATCH (d2:Document {title: $doc2})
                CREATE (d1)-[r:%s]->(d2)
            """ % rel_type, doc1=doc1_title, doc2=doc2_title)

Step 3: Implement Vector Search

def retrieve_context(self, query, top_k=3):
    """Retrieve relevant context using vector search"""
    query_embedding = self.create_embedding(query)

    with self.driver.session() as session:
        result = session.run("""
            CALL db.index.vector.queryNodes(
                'document_embeddings',
                $top_k,
                $query_embedding
            ) YIELD node, score

            // Also get related documents
            OPTIONAL MATCH (node)-[r:RELATES_TO|REFERENCES]-(related:Document)

            RETURN
                node.title as title,
                node.content as content,
                score,
                collect(DISTINCT {
                    title: related.title,
                    type: type(r)
                }) as related_docs
            ORDER BY score DESC
        """, query_embedding=query_embedding, top_k=top_k)

        return [dict(record) for record in result]

Step 4: Generate Response with LLM

def generate_response(self, query, context_docs):
    """Generate response using retrieved context"""
    # Build context from Neo4j
    context_text = "\n\n".join([
        f"Document: {doc['title']}\n{doc['content']}\n"
        f"Related documents: {', '.join([r['title'] for r in doc['related_docs']])}"
        for doc in context_docs
    ])

    # Create augmented prompt
    prompt = f"""Based on the following context, answer the user's question.

Context:
{context_text}

Question: {query}

Detailed answer:"""

    # Call OpenAI
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "You are an expert assistant that answers based on the provided context."},
            {"role": "user", "content": prompt}
        ],
        temperature=0.7,
        max_tokens=500
    )

    return response.choices[0].message.content

def ask(self, query):
    """Main RAG method"""
    # 1. Retrieve context
    context_docs = self.retrieve_context(query, top_k=3)

    # 2. Generate response
    answer = self.generate_response(query, context_docs)

    # 3. Return with sources
    return {
        "answer": answer,
        "sources": [doc['title'] for doc in context_docs],
        "relevance_scores": [doc['score'] for doc in context_docs]
    }

Step 5: Complete System Usage

# Initialize
rag = KnowledgeGraphRAG(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password",
    openai_key="sk-..."
)

# Add knowledge to the graph
rag.add_document(
    title="Introduction to Neo4j",
    content="Neo4j is a native graph database that stores data in nodes and relationships...",
    metadata={"category": "database", "level": "beginner"}
)

rag.add_document(
    title="Cypher Query Language",
    content="Cypher is Neo4j's declarative query language, designed to work with graphs...",
    metadata={"category": "query-language", "level": "intermediate"}
)

# Create relationships between documents
rag.add_relationship(
    "Introduction to Neo4j",
    "Cypher Query Language",
    "PREREQUISITE"
)

# Ask questions
result = rag.ask("How can I query data in Neo4j?")

print("Answer:", result['answer'])
print("\nSources:", result['sources'])
print("Relevance:", result['relevance_scores'])

# Close connection
rag.close()

Example Output

Answer: To query data in Neo4j, you use Cypher Query Language,
the declarative language designed specifically for working with graphs.
Cypher allows you to MATCH patterns, WHERE to filter,
and RETURN to specify what data you want to retrieve. For example:

MATCH (n:Person)-[:KNOWS]->(friend)
WHERE n.name = 'Alice'
RETURN friend.name

This query finds all of Alice's friends in the graph.

Sources: ['Cypher Query Language', 'Introduction to Neo4j']
Relevance: [0.89, 0.76]

Advantages of RAG with Neo4j vs Simple Vector Search

Feature	Simple Vectors	Neo4j RAG
Relational context	❌ No	✅ Yes
Knowledge navigation	❌ Limited	✅ Complete
Explainability	⚠️ Low	✅ High
Updates	⚠️ Complex	✅ Simple
Multi-hop reasoning	❌ No	✅ Yes

Advanced Use Cases

1. Multi-hop RAG (Multi-step Reasoning)

def multi_hop_retrieve(self, query, depth=2):
    """Retrieve context by navigating the graph in depth"""
    query_embedding = self.create_embedding(query)

    with self.driver.session() as session:
        result = session.run("""
            CALL db.index.vector.queryNodes(
                'document_embeddings',
                3,
                $query_embedding
            ) YIELD node, score

            // Navigate up to 2 levels deep
            CALL apoc.path.subgraphAll(node, {
                maxLevel: $depth,
                relationshipFilter: 'RELATES_TO|REFERENCES|PREREQUISITE'
            })
            YIELD nodes, relationships

            RETURN node.title as main_doc,
                   [n IN nodes | n.title] as related_docs,
                   [r IN relationships | type(r)] as relationship_types,
                   score
        """, query_embedding=query_embedding, depth=depth)

        return [dict(record) for record in result]

2. RAG with Temporal Filtering

def retrieve_recent_context(self, query, days=30):
    """Retrieve only recent context"""
    query_embedding = self.create_embedding(query)

    with self.driver.session() as session:
        result = session.run("""
            CALL db.index.vector.queryNodes(
                'document_embeddings',
                5,
                $query_embedding
            ) YIELD node, score

            WHERE node.created_at > datetime() - duration({days: $days})

            RETURN node.title, node.content, score
            ORDER BY score DESC
        """, query_embedding=query_embedding, days=days)

        return [dict(record) for record in result]

3. RAG with User Personalization

def personalized_retrieve(self, query, user_id):
    """Retrieve personalized context based on user preferences"""
    query_embedding = self.create_embedding(query)

    with self.driver.session() as session:
        result = session.run("""
            // Find relevant documents
            CALL db.index.vector.queryNodes(
                'document_embeddings',
                10,
                $query_embedding
            ) YIELD node, score

            // Boost documents related to user
            OPTIONAL MATCH (u:User {id: $user_id})-[:INTERESTED_IN]->(t:Topic)
            OPTIONAL MATCH (node)-[:ABOUT]->(t)

            WITH node, score,
                 CASE WHEN t IS NOT NULL THEN 1.5 ELSE 1.0 END as user_boost

            RETURN node.title, node.content,
                   (score * user_boost) as adjusted_score
            ORDER BY adjusted_score DESC
            LIMIT 5
        """, query_embedding=query_embedding, user_id=user_id)

        return [dict(record) for record in result]

Optimization and Best Practices

1. Intelligent Chunking

def smart_chunk_document(self, long_text, chunk_size=500, overlap=50):
    """Split long documents into chunks with overlap"""
    words = long_text.split()
    chunks = []

    for i in range(0, len(words), chunk_size - overlap):
        chunk = ' '.join(words[i:i + chunk_size])
        chunks.append(chunk)

    return chunks

def add_long_document(self, title, content):
    """Add long document in related chunks"""
    chunks = self.smart_chunk_document(content)

    for i, chunk in enumerate(chunks):
        chunk_title = f"{title} - Part {i+1}"
        self.add_document(chunk_title, chunk)

        # Relate consecutive chunks
        if i > 0:
            prev_chunk = f"{title} - Part {i}"
            self.add_relationship(prev_chunk, chunk_title, "NEXT_CHUNK")

2. Embedding Cache

from functools import lru_cache

@lru_cache(maxsize=1000)
def create_embedding_cached(self, text):
    """Cached version of create_embedding"""
    return self.create_embedding(text)

3. Relevance Monitoring

def ask_with_confidence(self, query, min_score=0.7):
    """RAG with confidence threshold"""
    context_docs = self.retrieve_context(query, top_k=3)

    # Check minimum relevance
    if not context_docs or context_docs[0]['score'] < min_score:
        return {
            "answer": "I didn't find sufficiently relevant information to answer with confidence.",
            "confidence": "low",
            "sources": []
        }

    answer = self.generate_response(query, context_docs)

    return {
        "answer": answer,
        "confidence": "high",
        "sources": [doc['title'] for doc in context_docs],
        "avg_score": sum(d['score'] for d in context_docs) / len(context_docs)
    }

Conclusion

RAG with Neo4j combines the best of both worlds: the generation capability of LLMs and structured knowledge in graphs. This architecture enables:

🎯 More accurate answers based on specific knowledge
🔍 Complete traceability of sources and reasoning
🔄 Dynamic updates to knowledge without retraining models
🌐 Relational context that simple vectors cannot capture
📊 Explainability of why certain responses were generated

The future of conversational AI lies in combining language models with structured knowledge bases, and Neo4j is the perfect platform for building these systems.