Amazon Neptune Integration

Use Amazon Neptune Database or Neptune Analytics as a cloud-native graph database backend for Graphiti

Amazon Neptune provides fully managed graph database services with two options: Neptune Database (property graph) and Neptune Analytics (in-memory analytics).

Installation

Install Graphiti with Neptune support:

pip

pip install graphiti-core[neptune]

Prerequisites

• Neptune Database Cluster or Neptune Analytics Graph
• Amazon OpenSearch Serverless Collection (for fulltext search)
• AWS Credentials configured with appropriate IAM permissions

Required IAM Permissions

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "neptune-db:*",
        "neptune-graph:*",
        "aoss:*"
      ],
      "Resource": "*"
    }
  ]
}

Configuration

Environment Variables

.env

# AWS Credentials (or use IAM roles)
AWS_ACCESS_KEY_ID=your-access-key
AWS_SECRET_ACCESS_KEY=your-secret-key
AWS_DEFAULT_REGION=us-east-1

# Neptune endpoints
NEPTUNE_ENDPOINT=your-cluster.cluster-xxx.us-east-1.neptune.amazonaws.com
AOSS_ENDPOINT=your-collection.us-east-1.aoss.amazonaws.com

Neptune Database Setup

Connect to a Neptune Database cluster:

from graphiti_core import Graphiti
from graphiti_core.driver.neptune_driver import NeptuneDriver

# Create Neptune driver for Database cluster
driver = NeptuneDriver(
    host="neptune-db://your-cluster.cluster-xxx.us-east-1.neptune.amazonaws.com",
    aoss_host="your-collection.us-east-1.aoss.amazonaws.com",
    port=8182,       # Neptune Database port
    aoss_port=443    # OpenSearch Serverless port
)

# Initialize Graphiti
graphiti = Graphiti(graph_driver=driver)

Neptune Analytics Setup

Connect to a Neptune Analytics graph:

from graphiti_core.driver.neptune_driver import NeptuneDriver

# Create Neptune driver for Analytics graph
driver = NeptuneDriver(
    host="neptune-graph://your-graph-id",
    aoss_host="your-collection.us-east-1.aoss.amazonaws.com",
    aoss_port=443
)

# Port parameter is ignored for Neptune Analytics
graphiti = Graphiti(graph_driver=driver)

Driver Implementation

The NeptuneDriver (graphiti_core/driver/neptune_driver.py:139) provides:

• Dual Backend Support: Works with both Neptune Database and Neptune Analytics
• OpenSearch Integration: Uses Amazon OpenSearch Serverless for fulltext search
• AWS Authentication: Automatic IAM-based authentication via boto3
• Managed Service: Fully managed, scalable graph infrastructure

Connection Parameters

Parameter	Type	Default	Description
host	str	Required	Neptune endpoint with protocol prefix
aoss_host	str	Required	OpenSearch Serverless collection endpoint
port	int	8182	Neptune Database port (ignored for Analytics)
aoss_port	int	443	OpenSearch Serverless port

Host Format

• Neptune Database: neptune-db://<cluster-endpoint>
• Neptune Analytics: neptune-graph://<graph-id>

OpenSearch Integration

Neptune uses Amazon OpenSearch Serverless for fulltext indexing. The driver automatically manages four indices:

Index Definitions

1. node_name_and_summary: Entity node fulltext search
2. community_name: Community node search
3. episode_content: Episode content search
4. edge_name_and_fact: Relationship fact search

Indices are automatically created during initialization:

# Rebuild OpenSearch indices
await driver.build_indices_and_constraints(delete_existing=True)

# Note: Index creation takes ~60 seconds

Complete Example

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

async def main():
    # Initialize Neptune driver
    driver = NeptuneDriver(
        host=f"neptune-db://{os.environ['NEPTUNE_ENDPOINT']}",
        aoss_host=os.environ['AOSS_ENDPOINT'],
        port=8182,
        aoss_port=443
    )
    
    # Initialize Graphiti
    graphiti = Graphiti(graph_driver=driver)
    
    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)
        )
        print("Added episode to Neptune")
        
        # 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()
        print("Neptune connection closed")

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

When to Use Neptune

Choose Neptune if you:

• Need a fully managed, production-ready graph database
• Want AWS-native integration and IAM security
• Require high availability and automated backups
• Need elastic scalability for varying workloads
• Prefer serverless deployment options (Neptune Analytics)

Choose Neptune Database if you:

• Need persistent storage with automatic backups
• Want multi-AZ replication
• Require read replicas for high read throughput

Choose Neptune Analytics if you:

• Need fast in-memory analytics
• Want to pay only for query execution time
• Require rapid scaling for analytical workloads

Performance Considerations

• Cold Starts: OpenSearch index creation takes ~60 seconds
• Batch Operations: Use bulk operations for large data imports
• Connection Pooling: Driver manages connection pooling automatically
• Query Optimization: Monitor CloudWatch metrics for query performance

Monitoring and Debugging

CloudWatch Metrics

Monitor Neptune through CloudWatch:

• Query execution time
• Connection count
• Storage usage (Database only)
• Request count

Enable Query Logging

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("graphiti_core.driver.neptune_driver")
logger.setLevel(logging.DEBUG)

Cost Optimization

• Neptune Database: Use instance sizing based on workload
• Neptune Analytics: Pay per query; optimize query patterns
• OpenSearch Serverless: Configure appropriate OCU (capacity units)
• Data Transfer: Minimize cross-region data transfer

Security Best Practices

• IAM Roles: Use IAM roles instead of access keys when possible
• VPC Configuration: Deploy Neptune in private subnets
• Encryption: Enable encryption at rest and in transit
• Network Security: Use security groups to restrict access

Related Resources

• Amazon Neptune Documentation
• Neptune Database Guide
• Neptune Analytics Guide
• OpenSearch Serverless