How to Configure Semantica with Neo4j Graph Store: A Complete Setup Guide

Configure Semantica to use Neo4j by setting backend: neo4j in your YAML config with Bolt URI and credentials, then instantiate Neo4jStore with .connect() and .close() lifecycle methods.

Semantica's graph persistence layer uses a pluggable backend architecture where Neo4jStore serves as the primary production implementation. This guide walks through installation, YAML configuration, programmatic usage, and key source files for integrating Neo4j with the semantica-agi/semantica repository.

---

Installation and Dependencies

The Neo4j Python driver is required but optional in the base package. Install it before configuring the graph store:

# Standalone driver installation

pip install neo4j

# Or install Semantica with Neo4j extras

pip install "semantica[neo4j]"

If the driver is absent, Neo4jStore logs a warning at import time and raises ProcessingError on connection attempts. The driver implements Bolt protocol communication with Neo4j 4.x and 5.x servers.

---

YAML Configuration for Neo4j Graph Store

Semantica discovers backend settings through a graph_db configuration block. The loader expects this structure in config.yaml or passes equivalent values via CLI arguments.

Required Configuration Fields

graph_db:
  backend: neo4j              # Activates Neo4jStore instantiation

  uri: bolt://localhost:7687  # Neo4j Bolt endpoint

  user: neo4j                 # Authentication principal

  password: ${NEO4J_PASSWORD} # Credential (environment variable recommended)

  database: neo4j             # Target database name (optional, defaults to "neo4j")

  encrypted: false            # TLS encryption toggle (optional)

Configuration Precedence

1. CLI flags override file values
2. Environment variables referenced with ${VAR} syntax expand at runtime
3. Default values apply for omitted optional fields

CLI Usage Patterns

# Inline backend selection

semantica --store neo4j kg build -s source_data.txt

# Config file with explicit path

semantica --config ./production.yaml kg build -s source_data.txt

# Environment-only credentials (12-factor style)

export NEO4J_URI=bolt://neo4j.prod.internal:7687
export NEO4J_USER=semantica_app
export NEO4J_PASSWORD=$(cat /run/secrets/neo4j_password)
semantica --store neo4j worker start

---

Programmatic Graph Store Usage

The Neo4jStore class in semantica/graph_store/neo4j_store.py wraps the official neo4j.GraphDatabase.driver with connection pooling, transaction management, and Semantica-native CRUD operations.

Complete Working Example

from semantica.graph_store.neo4j_store import Neo4jStore, ProcessingError

# 1. Initialize with configuration values

store = Neo4jStore(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="secure_credential",
    database="semantica_kg",
)

# 2. Establish connection (validates credentials and network reachability)

store.connect()

try:
    # 3. Create nodes with labels and properties

    alice = store.create_node(
        labels=["Person", "Researcher"],
        properties={
            "name": "Alice Chen",
            "department": "Machine Learning",
            "h_index": 42,
        },
    )
    bob = store.create_node(
        labels=["Person", "Engineer"],
        properties={"name": "Bob Martinez", "department": "Infrastructure"},
    )

    # 4. Create typed relationships between nodes

    collaboration = store.create_relationship(
        start_node_id=alice["id"],
        end_node_id=bob["id"],
        rel_type="COLLABORATED_WITH",
        properties={"project": "knowledge-graph-pipeline", "year": 2024},
    )

    # 5. Execute arbitrary Cypher with parameterized queries

    results = store.execute_query(
        """
        MATCH (researcher:Researcher)-[:COLLABORATED_WITH]-(peer)
        WHERE researcher.department = $dept
        RETURN researcher.name AS source, peer.name AS collaborator, peer.department AS peer_dept
        """,
        {"dept": "Machine Learning"},
    )

    for record in results["records"]:
        print(f"{record['source']} collaborated with {record['collaborator']}")

    # 6. Graph traversal helpers

    alice_neighbors = store.get_neighbors(
        node_id=alice["id"], 
        direction="outgoing",
        rel_types=["COLLABORATED_WITH"]
    )

finally:
    # 7. Explicit cleanup releases driver connections

    store.close()

Core API Methods Reference

Method	Signature	Purpose
connect()	connect() -> None	Validates URI scheme, authenticates, and initializes driver
create_node()	create_node(labels: list, properties: dict) -> dict	Returns node with auto-generated id field
create_relationship()	create_relationship(start_node_id, end_node_id, rel_type, properties) -> dict	Creates edge with mandatory type string
execute_query()	execute_query(query: str, parameters: dict = None) -> dict	Runs Cypher with automatic parameter serialization
get_neighbors()	get_neighbors(node_id, direction, rel_types) -> list	Traversal helper without raw Cypher
close()	close() -> None	Idempotent driver shutdown

---

Neo4jStore Architecture and Implementation

Understanding the internal structure helps debug connection issues and extend functionality.

File: semantica/graph_store/neo4j_store.py

This module contains three nested adapter classes that map Neo4j driver objects to Semantica's uniform interface:

• Neo4jDriver — wraps GraphDatabase.driver with retry logic and logging
• Neo4jSession — manages transaction scope and automatic rollback on exceptions
• Neo4jTransaction — implements run() with parameter binding and result parsing

The Neo4jStore class composes these adapters and implements:

class Neo4jStore:
    def __init__(self, uri: str, user: str, password: str, database: str = "neo4j"):
        self.uri = uri
        self.auth = (user, password)
        self.database = database
        self._driver = None
        self._logger = get_logger(__name__)

    def connect(self) -> None:
        """Initialize Bolt driver and verify connectivity."""
        # Raises ProcessingError on authentication or network failure

    def execute_query(self, query: str, parameters: Optional[dict] = None) -> dict:
        """Execute Cypher with automatic result marshaling to Python primitives."""

Exception Handling

All graph operations raise domain-specific exceptions defined in semantica/utils/exceptions.py:

Exception	Trigger Condition	Recovery Strategy
ProcessingError	Connection failure, timeout, Neo4j transient error	Exponential backoff retry
ValidationError	Malformed node/relationship parameters, invalid Cypher	Fix input and resubmit
NotFoundError	Node or relationship lookup miss	Create missing entity or skip

---

Testing and Validation

The test suite in tests/test_graph_store.py provides patterns for verifying your Neo4j configuration:

# Mock-based unit test pattern from test_graph_store.py

from unittest.mock import MagicMock, patch

def test_neo4j_store_connect_success():
    with patch("semantica.graph_store.neo4j_store.GraphDatabase") as mock_db:
        mock_driver = MagicMock()
        mock_db.driver.return_value = mock_driver
        
        store = Neo4jStore(uri="bolt://test", user="neo4j", password="test")
        store.connect()
        
        mock_db.driver.assert_called_once_with(
            "bolt://test", 
            auth=("neo4j", "test")
        )
        assert store._driver is mock_driver

For integration testing against a real Neo4j instance, use Docker:

docker run -d \
  --name semantica-neo4j-test \
  -p 7687:7687 \
  -e NEO4J_AUTH=neo4j/testpassword \
  neo4j:5.13-community

# Run with test config

NEO4J_PASSWORD=testpassword semantica --config test.yaml kg validate

---

Summary

• Install the neo4j Python driver as an optional dependency before configuration
• Configure via YAML graph_db block with backend: neo4j, Bolt URI, and credentials
• Instantiate Neo4jStore programmatically with matching constructor parameters
• Manage lifecycle explicitly through .connect() and .close() calls
• Leverage CRUD helpers and execute_query() for both simple operations and complex Cypher
• Reference semantica/graph_store/neo4j_store.py for implementation details and extension points

---

Frequently Asked Questions

What authentication methods does Neo4jStore support?

Basic username/password authentication via the auth tuple is fully supported. SSO, LDAP, and Kerberos require manual driver configuration—pass a pre-configured driver instance to Neo4jStore instead of URI credentials. Certificate-based authentication is supported through encrypted: true with appropriate trust store configuration.

How do I handle connection pooling and concurrent access?

The underlying Neo4j driver manages connection pooling automatically. Neo4jStore instances are thread-safe for read operations but require external synchronization for write-heavy workloads. Use separate Neo4jStore instances per process in multiprocessing deployments to avoid driver contention.

Can I switch from Neo4j to an in-memory graph store for testing?

Semantica's backend abstraction supports pluggable implementations, though only Neo4jStore is production-ready in the current codebase. For testing, mock the store interface or use NEO4J_TEST_CONTAINER environment variable to spin up ephemeral Neo4j Docker instances via test fixtures in tests/test_graph_store.py.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}