Skip to main content
Neo4j is the recommended graph database backend for Graphiti, providing robust, production-ready storage for temporal knowledge graphs.

​
Installation

Install Graphiti with Neo4j support: 
pip install graphiti-core

​
Prerequisites

You need a running Neo4j instance. The simplest way to get started is with Neo4j Desktop. Alternatively, run Neo4j with Docker: 
docker run -p 7687:7687 -p 7474:7474 \
  -e NEO4J_AUTH=neo4j/password \
  neo4j:5.26

​
Configuration

​
Environment Variables

Set up your Neo4j connection credentials:
.env
NEO4J_URI=bolt://localhost:7687
NEO4J_USER=neo4j
NEO4J_PASSWORD=your-password

​
Basic Setup

Initialize Graphiti with Neo4j using the default driver: 
import os
from graphiti_core import Graphiti

# Using environment variables
graphiti = Graphiti(
    os.environ["NEO4J_URI"],
    os.environ["NEO4J_USER"],
    os.environ["NEO4J_PASSWORD"]
)

​
Custom Database Name

By default, Graphiti uses the neo4j database. To use a custom database: 
from graphiti_core import Graphiti
from graphiti_core.driver.neo4j_driver import Neo4jDriver

# Create driver with custom database name
driver = Neo4jDriver(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password",
    database="my_custom_database"
)

# Pass driver to Graphiti
graphiti = Graphiti(graph_driver=driver)

​
Driver Implementation

The Neo4jDriver (graphiti_core/driver/neo4j_driver.py:60) implements the GraphDriver interface with the following features:
  • Connection Management: Async connection pooling via neo4j.AsyncGraphDatabase
  • Transaction Support: Full ACID transactions with commit/rollback semantics
  • Index Management: Automatic creation of range and fulltext indices
  • Query Execution: Direct Cypher query execution with parameter binding

​
Connection Parameters

ParameterTypeDefaultDescription
uristrRequiredBolt connection URI (e.g., bolt://localhost:7687)
userstrRequiredNeo4j username
passwordstrRequiredNeo4j password
databasestr"neo4j"Database name to connect to

​
Index Management

Graphiti automatically creates the following indices on initialization:
  • Range Indices: For efficient UUID, group_id, and timestamp lookups
  • Fulltext Indices: For BM25 text search on entity/edge facts and episode content
To rebuild indices: 
await graphiti.build_indices_and_constraints(delete_existing=True)

​
Health Check

Verify Neo4j connectivity: 
try:
    await graphiti.driver.health_check()
    print("Neo4j connection successful")
except Exception as e:
    print(f"Neo4j health check failed: {e}")

​
Complete Example

import asyncio
import os
from datetime import datetime, timezone
from graphiti_core import Graphiti
from graphiti_core.nodes import EpisodeType

async def main():
    # Initialize with Neo4j
    graphiti = Graphiti(
        os.getenv("NEO4J_URI", "bolt://localhost:7687"),
        os.getenv("NEO4J_USER", "neo4j"),
        os.getenv("NEO4J_PASSWORD", "password")
    )
    
    try:
        # Add an episode
        await graphiti.add_episode(
            name="California Politics 1",
            episode_body="Kamala Harris is the Attorney General of California.",
            source=EpisodeType.text,
            reference_time=datetime.now(timezone.utc)
        )
        
        # Search the graph
        results = await graphiti.search("Who was the California Attorney General?")
        for result in results:
            print(f"Fact: {result.fact}")
    
    finally:
        await graphiti.close()

if __name__ == "__main__":
    asyncio.run(main())

​
Production Considerations

  • Connection Pooling: Neo4j driver automatically manages connection pools
  • Database Selection: Use separate databases for different environments (dev/staging/prod)
  • Backup: Configure regular Neo4j backups for data durability
  • Performance: Monitor query performance and adjust indices as needed
  • Security: Use strong passwords and restrict network access in production

Build docs developers (and LLMs) love

Get started for freeTalk to us