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.

Never fabricate edges to avoid an empty edge list: get_graph_data() must return only edges that actually exist in your store. A graph that holds nodes but no relationships returns (nodes, []) — do not pad the result with placeholder self-loops or any other synthetic edge. Cognee decides whether a projection is empty from the node list alone, so an empty edge list is a valid answer and never something to work around: a projection with nodes and no edges is passed through with edges == [], and only a projection with no nodes raises EntityNotFoundError. The built-in Ladybug adapter used to do exactly this — one (node_id, node_id, "SELF") edge per node — to satisfy an older edge-based emptiness check, and those fake triplets reached retrieval and visualization as real relationships; the fabrication and the check are both gone.

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 with no nodes (a filtered result with nodes but no edges is used as-is, not treated as a miss).

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

Optional update_node: GraphDBInterface.update_node(node_id: str, values: Dict[str, Any]) -> bool merges values into an existing node’s properties. The base implementation raises NotImplementedError, so it is an optional extension — omitting it is legal, and only the built-in Ladybug adapter implements it today. The contract is merge, not replace: patch only the keys named in values and carry every unnamed field (created_at, updated_at, version, …) through untouched — a read-modify-write of the stored properties is the straightforward way to get this right. Return True when the node existed and was patched, and False when it does not exist; treat an empty values as nothing-to-patch and return False, as the Ladybug implementation does. If you omit the method, callers such as the user-preference store catch the NotImplementedError and fall back to rewriting the whole node through add_nodes, so make sure your add_nodes is a true upsert.

async def update_node(self, node_id: str, values: Dict[str, Any]) -> bool:
    """Merge values into an existing node, leaving unnamed fields untouched."""
    node = await self.get_node(node_id)
    if node is None or not values:
        return False
    # ... write back node's current properties with values layered on top ...
    return True

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))

Optional get_top_degree_node_ids: This method is declared on GraphDBInterface but is not abstract, so overriding it is optional — your adapter loads either way. It picks the seed nodes for a bare visualize_graph() call, and the inherited default reads the whole graph via get_graph_data() and counts degree in Python, which is O(graph) in memory and logs a warning once per adapter type per process. Override it if you expect large graphs. The contract: return the ids (list[str]) of up to top_k well-connected nodes — ids only, never node objects or degree counts; raise ValueError when top_k < 1, before reading the graph; treat isolated nodes as valid seeds, so a graph with nodes but no edges still returns a nonempty list. The result is approximate by contract, so sample rather than aggregate exactly — the built-in adapters rank both endpoints of at most 200,000 sampled edges inside the store, then top the list up with an id-only node query when the sample is sparse.

Optional iter_bounded_neighborhood: Also declared on GraphDBInterface without being abstract, so overriding it is optional. It is the bounded, chunked neighborhood read behind visualize_graph(): an async iterator taking node_ids, depth, max_nodes, chunk_size=2000, and property_keys=None, yielding (nodes, edges) chunks with at most max_nodes members (seeds first, then by hop distance) and every edge between two members exactly once, in the chunk holding its later endpoint. Raise ValueError when depth, max_nodes, or chunk_size is below 1, before reading anything. The inherited default calls get_neighborhood() and applies the cap in Python, which is O(neighborhood) in memory and logs a warning once per adapter type per process; an override should stop the traversal once it holds max_nodes ids and read node content one chunk at a time. postgres_demo is the worked example.

Optional get_entity_type_names: Also declared on GraphDBInterface without being abstract, so overriding it is optional. It is the one-hop is_a lookup that lets a bounded visualize_graph() read label entities whose EntityType node falls outside the window. The contract: take a list[str] of entity ids and return a dict[str, str] mapping each entity id to the name of the EntityType it points at through is_a; leave out entities with no is_a edge or an unnamed type rather than mapping them to an empty value, and return {} for an empty list without touching the store. The inherited default goes through get_neighborhood(node_ids=entity_ids, depth=1, edge_types=["is_a"]), which hydrates the entities with all their properties and returns every edge among them; an override should read only the is_a edges and the target names, and never traverse outward from a type node. See Graph engine adapters for which built-in adapters implement it natively.

Optional get_graph_counts: Also declared on GraphDBInterface without being abstract, so overriding it is optional. async def get_graph_counts(self) -> tuple[int, int] takes no arguments and returns (num_nodes, num_edges) — the two counts and nothing else, and it should return (0, 0) for an empty graph. It backs the dataset graph-summary path, which needs only those numbers. The inherited default calls your get_graph_metrics(include_optional=False) and reads the counts out of its result, which also pays for the connected-component walk that method does, and logs a warning once per adapter type per process. Implement it as two aggregation queries rather than collecting rows and taking len(). See Graph engine adapters for which built-in adapters implement it natively.

Optional delete_edge_triples: Also declared on GraphDBInterface without being abstract, but unlike the methods above its inherited default does not fall back to anything: it raises UnsupportedProvenanceCapability. async def delete_edge_triples(self, edges: list[EdgeIdentity]) -> None deletes each edge whose source id, target id and relationship name all match, and should return without touching the store for an empty list. Delete only the relationship (in Cypher, DELETE r), never a detach delete — the endpoint nodes stay in the graph even when the deleted edge was their last one. Ladybug, Neo4j, Neptune (shared with neptune_analytics), Postgres, and Turso implement it. On an adapter that skips it, rewriting stored conflicts in improve(review_conflicts=True) and the stale-edge and node-set sweeps on code-graph re-ingestion raise UnsupportedProvenanceCapability; pruning prefers edges from a user’s preference node catches the error and rewrites those edges as neutral instead of deleting them; provenance-driven deletes never reach it, because they run only on a graph marked as storing its own provenance.

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