Graph Database Integration

Add a new graph database adapter to Cognee.

Adding a New Graph Database to cognee

This guide describes how to integrate a new graph database engine into cognee.

Repository Options

Cognee used both the core and the community repositories to host graph-database adapters.

🚨 From now on every graph-database adapter – except Kuzu – will live in the cognee-community repository.

NetworkX has already been migrated and the remaining adapters will follow shortly.

Therefore all new adapter contributions must target the community repository.
The core repository will keep only the built-in Kuzu integration.

For Community Repository

To add a new adapter to cognee-community:

1. Fork and clone the cognee-community repository, and branch from main — unlike the core Cognee repo, cognee-community has no dev branch.
2. Create your adapter in packages/<engine_name>/cognee_community_graph_adapter_<engine_name>/
3. Inside that directory add __init__.py, <engine_name>_adapter.py, and register.py (see the Redis example).
4. At the package root packages/<engine_name>/ add __init__.py, pyproject.toml, and README.md.
5. Run the shared graph conformance suite in packages/shared/contract_suite/graph_contract.py against your adapter.
6. Submit a pull request to the community repository

Below are the recommended steps in more detail.

---

Why cognee-community?

cognee-community is the extension hub for Cognee.
Anything that is not part of the core lives here—adapters for third-party databases, pipelines, community contributed additional tasks, etc.
Placing your adapter in this repository means:

• Your code is released under the community license and can evolve independently of the core.
• It can be installed with pip install cognee-community-graph-adapter-(engine_mane) without pulling in heavyweight drivers for users who don’t need them. For example, for NetworkX it is pip install cognee-community-graph-adapter-networkx
• These packages can be called with cognee core package using the registration step described below.

If you are unfamiliar with the layout, have a look at the existing folders under packages/* in the community repo—each sub-folder represents a separate provider implemented in exactly the way you are about to do.

---

1. Implement the Adapter

File: packages/graph/<engine_name>/cognee_community_graph_adapter_<engine_name>/<engine_name>_adapter.py

Your adapter must subclass GraphDBInterface, implementing all required CRUD and utility methods (e.g., add_node, add_edge, extract_node, etc.). Here is a sample skeleton with placeholders:

"""Adapter for <engine_name> graph database."""

import json
import asyncio
from typing import Dict, Any, List, Optional, Tuple

from cognee.shared.logging_utils import get_logger
from cognee.infrastructure.databases.graph.graph_db_interface import GraphDBInterface

logger = get_logger()

class <EngineName>Adapter(GraphDBInterface):
    """Adapter for <engine_name> graph database operations."""

    def __init__(
        self,
        graph_database_url: str = "",
        graph_database_username: Optional[str] = None,
        graph_database_password: Optional[str] = None,
    ):
        self.graph_database_url = graph_database_url
        self.graph_database_username = graph_database_username
        self.graph_database_password = graph_database_password
        self.connection = None

    async def query(self, query: str, params: Optional[Dict[str, Any]] = None) -> List[Tuple]:
        """Execute an async query.
        
        If your graph database library provides an async SDK, call it directly here.
        If it only provides a synchronous client, you can run the call via
        `loop.run_in_executor()` or a similar technique to avoid blocking the event loop.
        """

        loop = asyncio.get_running_loop()
        params = params or {}

        def blocking_query():
            try:
                # Example usage with your driver
                # cursor = self.connection.execute(query, params)
                # results = cursor.fetchall()
                return []
            except Exception as e:
                logger.error(f"<engine_name> query execution failed: {e}")
                raise

        return await loop.run_in_executor(self.executor, blocking_query)

    # -- Example: Add a node
    async def add_node(self, node_data: Any) -> None:
        """Add a single node to <engine_name>."""
        # Implement logic:
        # 1. Extract relevant fields (id, text, type, properties, etc.).
        # 2. Construct a CREATE query.
        # 3. Call self.query(query_str, params).
        pass

    # -- Example: Retrieve all nodes/edges
    async def get_graph_data(self) -> Tuple[List, List]:
        """Retrieve all nodes and edges from <engine_name>."""
        # Return (nodes, edges) where each node is `(node_id, properties_dict)`
        # and each edge is `(source_id, target_id, relationship_label, properties_dict)`.
        return ([], [])

    # -- Additional methods (delete_node, add_edge, etc.) ...

Keep the method signatures consistent with GraphDBInterface. Reference the KuzuAdapter or the Neo4jAdapter for a more comprehensive example.

Adapter instance reuse: Cognee’s graph engine factory caches adapter instances keyed by their configuration parameters. Multiple calls with identical settings return the same adapter object. Design your adapter to be safe for reuse — avoid per-instance mutable state that cannot be safely shared, and prefer lazy or thread-safe initialization where state is required. When an entry is evicted (e.g. cache eviction or cache_clear), the factory calls your adapter’s close(). For capacity eviction the close is deferred until every leased reference to that instance is released, but the dataset-queue teardown and the idle reaper force-close immediately even while idle references are still held (holders transparently re-resolve to a fresh instance on next use) — so implement close() idempotently and safe to run while stale references remain; if it raises, the error is logged and swallowed rather than propagated to the caller.

Optional get_id_filtered_graph_data: This method is not part of GraphDBInterface, so it is optional. If you implement it, graph-completion searches project only the vector-search neighborhood instead of loading the full graph via get_graph_data(); if you omit it, Cognee falls back to get_graph_data(). The contract is edge-driven and matches the built-in Ladybug, Neo4j, and Postgres adapters: given target_ids, return (nodes, edges) where edges are every edge with either endpoint in target_ids, and nodes are all endpoint nodes of those edges (same (node_id, properties) / (source_id, target_id, relationship_label, properties) shapes as get_graph_data()). Return ([], []) when target_ids is empty; Cognee also falls back to the full graph if the filtered result comes back empty.

async def get_id_filtered_graph_data(self, target_ids: List[str]) -> Tuple[List, List]:
    """Return the subgraph touching target_ids: edges with either endpoint in
    the set, plus all endpoint nodes of those edges."""
    if not target_ids:
        return [], []
    # ... query edges where source/target is in target_ids, then their endpoint nodes ...
    return nodes, edges

Declaring Cypher support: GraphDBInterface declares supports_cypher_queries: bool = True, so adapters are assumed to speak Cypher through query(). Override it to False on your adapter class when query() executes something else — the built-in Postgres and Turso adapters do this because their query() runs SQL against the graph tables. SearchType.CYPHER and SearchType.NATURAL_LANGUAGE then raise SearchTypeNotSupported (naming your adapter class) instead of handing your backend a Cypher string it cannot parse. Keep the flag on the class rather than setting it in __init__: tests and tooling read the capability directly off the adapter class without instantiating it (no database connection needed), as Cognee’s own adapter tests do.

class <EngineName>Adapter(GraphDBInterface):
    # ``query()`` executes <engine_name>'s own query language, not Cypher.
    supports_cypher_queries = False

Optional get_triplets_batch — order before you paginate: GraphDBInterface.get_triplets_batch(offset, limit) is an optional extension; the base implementation raises NotImplementedError, and only adapters that override it can back Memify’s triplet-embedding pipeline (the built-in Ladybug, Neo4j, Postgres, and Turso adapters do). If you implement it, your query must apply a total ordering before the offset/limit, because get_triplet_datapoints reads the whole graph with a single offset loop, advancing the offset by each batch’s size until a batch comes back short or empty, and that loop is only exhaustive if every call slices the same stable sequence. Paginating an unordered result set silently skips and duplicates rows. Sort on (source node id, target node id, relationship name) to match the built-in adapters. In Cypher this means putting ORDER BY in a WITH clause ahead of SKIP/LIMIT rather than after RETURN, so the skip applies to an already-ordered stream.

MATCH (start_node:Node)-[relationship:EDGE]->(end_node:Node)
WITH start_node, relationship, end_node
ORDER BY start_node.id, end_node.id, relationship.relationship_name
SKIP $offset LIMIT $limit
RETURN start_node, relationship, end_node

Chunk bulk writes: Do not send an entire add_nodes / add_edges payload as one statement — a large ingest (tens of thousands of nodes or edges) can then exceed a per-statement or per-call deadline and never finish. Define a module-level _WRITE_CHUNK_SIZE and loop over the rows in slices of that size, issuing one statement per chunk; the built-in adapters use 2000 (Ladybug), 1000 (Postgres), and 500 (Turso), so pick a bound that suits your backend. Write each chunk as an idempotent MERGE/upsert: chunks are separate statements, so a run can fail partway through, and idempotent writes plus Cognee’s pipeline rollback ledger make the partial progress safe to re-apply. Chunking is internal to the adapter and does not change how many data points the pipeline batches per call.

_WRITE_CHUNK_SIZE = 2000

total = len(rows)
for start in range(0, total, _WRITE_CHUNK_SIZE):
    chunk = rows[start : start + _WRITE_CHUNK_SIZE]
    await self.query(merge_query, {"nodes": chunk})

Match edge endpoints through an index: In a bulk edge write, bind each endpoint with a property-map match on the indexed id (MATCH (from:Node {id: edge.from_id})) rather than a cartesian MATCH (from:Node), (to:Node) WHERE from.id = ... AND to.id = .... The cartesian form plans as a scan over the node table for every edge, which degrades badly as the graph grows; the property-map form is a primary-key index seek.

Whichever form you use, and in existence checks as much as in writes, match on the id property Cognee stores — the string UUID the pipeline hands you — and never on a backend-internal node identifier such as Neo4j’s id(n) or an autoincrementing row number. Those identifiers are a different value (and usually a different type) from the id Cognee passes, so the comparison does not error; it silently matches nothing. The neo4j adapter’s has_edges compared id(a) against the string UUID for exactly this reason and reported every edge as absent, which made the cognify dedup step below write a fresh copy of every edge on each re-cognify.

Let existence checks fail loudly: In has_edges — and in any read or existence-check method whose empty result is meaningful — do not catch a store error and return an empty list. An empty return must mean “the backend answered, and nothing matched”; it must never mean “the query failed”. has_edges is the batch check the cognify dedup uses to decide which edges are new, so a swallowed failure tells it that none of the edges exist, it writes all of them, those writes fail against the same broken store, and the run finishes reporting success with nothing persisted. Log the error and re-raise, as the neo4j adapter does, and reserve [] for the genuine empty-input short-circuit.

A successful check has a shape contract too: return the subset of the input triples that exist, as (source_id, target_id, relationship_name) strings. Not booleans, and not a list aligned one-to-one with the input — the caller, find_existing_edge_identities, unpacks each returned item into three values and treats the result as the set of edges to skip, so a per-input boolean list is not a compatible substitute.

async def has_edges(self, edges: List[Tuple[str, str, str]]) -> List[Tuple[str, str, str]]:
    if not edges:  # empty input: an answer, not a failure
        return []
    try:
        results = await self.query(query, {"edges": edge_params})
        return [(str(row[0]), str(row[1]), str(row[2])) for row in results]
    except Exception as e:
        logger.error(f"Failed to check edges in batch: {e}")
        raise  # a failed existence check is not an empty existence check

Carry edge properties out of get_connections: get_connections(node_id) returns (source_node, edge, target_node) triples for every edge into or out of the node, and the middle element must carry the edge’s stored properties alongside its relationship_name. Consumers read those properties by key, and a missing key does not raise — it produces a quietly wrong answer. Cognee’s document-deletion path derives the EdgeType vector-row id of a chunk’s contains edges from edge["edge_text"]; when that key is absent it falls back to the relationship name, computes a different id, and leaves the real vector rows behind. Watch for driver helpers that flatten a relationship and discard its properties on the way out: Neo4j’s result.data() reduces a relationship to (start_props, type, end_props), so the adapter has to request properties(relation) explicitly and merge it in.

edge = {"relationship_name": relationship_type}
edge.update(relationship_properties or {})  # edge_text, weights, timestamps, ...
connections.append((source_properties, edge, target_properties))

2. Test with a Dedicated Script

Your contribution should have an example showcasing how this integration should be configured and used.

File: packages/graph/engine_name/examples/example.py

Create a script that loads cognee and the integration package, registers it to use your new <engine_name> provider, and runs basic usage checks (for example, remembering data, recalling it, and pruning isolated test state). For example:

import sys
import asyncio
import pathlib
from os import path

# NOTE: Importing the register module we let cognee know it can use the Networkx adapter
import packages.graph.networkx.register

async def main():
    from cognee import config, prune, remember, recall, SearchType

    system_path = pathlib.Path(__file__).parent
    config.system_root_directory(path.join(system_path, ".cognee-system"))
    config.data_root_directory(path.join(system_path, ".cognee-data"))

    config.set_graph_db_config({
        "graph_database_provider": "engine-name",
    })

    await prune.prune_data()
    await prune.prune_system()

    text = """
    Natural language processing (NLP) is an interdisciplinary
    subfield of computer science and information retrieval.
    """
    dataset_name = "test_dataset"

    await remember(text, dataset_name=dataset_name)

    query_text = "Tell me about NLP"

    search_results = await recall(
        query_text,
        SearchType.GRAPH_COMPLETION,
        datasets=[dataset_name],
    )

    for result_text in search_results:
        print(result_text)

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

3. Create a Test Workflow

File: .github/workflows/engine_name/test_engine_name.yml

Create a GitHub Actions workflow to run your integration tests. This ensures any pull requests that modify your new engine (or the shared graph code) will be tested automatically. See an example here.

name: test | <engine_name>

on:
  workflow_dispatch:
  pull_request:
    types: [labeled, synchronize]

concurrency:
  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
  cancel-in-progress: true

env:
  RUNTIME__LOG_LEVEL: ERROR

jobs:
  run_<engine_name>_integration_test:
    name: test
    runs-on: ubuntu-22.04

    defaults:
      run:
        shell: bash

    steps:
      - name: Check out
        uses: actions/checkout@master

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11.x'

      - name: Install Poetry
        uses: snok/install-poetry@v1.4.1
        with:
          virtualenvs-create: true
          virtualenvs-in-project: true
          installer-parallel: true

      - name: Install dependencies
        # If your pyproject.toml has an extra named '<engine_name>', use:
        run: poetry install -E <engine_name> --no-interaction

      - name: Run <engine_name> tests
        env:
          ENV: 'dev'
          LLM_MODEL: ${{ secrets.LLM_MODEL }}
          LLM_ENDPOINT: ${{ secrets.LLM_ENDPOINT }}
          LLM_API_KEY: ${{ secrets.LLM_API_KEY }}
          LLM_API_VERSION: ${{ secrets.LLM_API_VERSION }}
          EMBEDDING_MODEL: ${{ secrets.EMBEDDING_MODEL }}
          EMBEDDING_ENDPOINT: ${{ secrets.EMBEDDING_ENDPOINT }}
          EMBEDDING_API_KEY: ${{ secrets.EMBEDDING_API_KEY }}
          EMBEDDING_API_VERSION: ${{ secrets.EMBEDDING_API_VERSION }}
          GRAPH_DATABASE_PROVIDER: ''
          GRAPH_DATABASE_URL: ''
          GRAPH_DATABASE_PASSWORD: ''
        run: poetry run python ./cognee_community_graph_adapter_engine_name/examples/example.py
        working-directory: ./packages/graph/engine_name

Tips:

• Rename <engine_name> appropriately.
• Ensure your pyproject.toml has an extras entry for any new dependencies.

5. Poetry Extras

If your new graph engine requires a special Python client or system libraries, update:

pyproject.toml:

[tool.poetry.dependencies]
python = "^3.11"
cognee
your-graph-db-client

[tool.poetry.extras]
...

---

6. Final Checklist

1. Implement your <EngineName>Adapter in packages/<engine_name>/cognee_community_graph_adapter_<engine_name>/<engine_name>_adapter.py.

2. Add a register helper (register.py) and call it before configuring Cognee:

   from cognee.infrastructure.databases.graph import use_graph_adapter
   
   from .<engine_name>_adapter import <EngineName>Adapter
   
   use_graph_adapter("engine_name", <EngineName>Adapter)
   

3. Run the shared conformance suite in packages/shared/contract_suite/graph_contract.py against your adapter — it is the common contract every community graph adapter is expected to satisfy.

4. Register a dataset-database handler from the same register.py if your backend can isolate storage per user + dataset:

   from cognee.infrastructure.databases.dataset_database_handler import (
       use_dataset_database_handler,
   )
   
   use_dataset_database_handler("engine_name", <EngineName>DatasetDatabaseHandler, "engine_name")
   

   Without a handler, your adapter can only be used with ENABLE_BACKEND_ACCESS_CONTROL=false.

5. Create a test or example script example.py.

6. Create a test workflow: .github/workflows/engine_name/test_<engine_name>.yml.

7. Add required dependencies to pyproject.toml extras.

8. Open a PR to verify that your new integration passes CI.

That’s all! This approach keeps cognee’s architecture flexible, allowing you to swap in any graph DB provider easily. Review the previous implementations in the core and the community repos.

Join the Conversation!

Have questions about creating custom tasks? Join our community to discuss implementation strategies and best practices!

Join the community