Graph Engine and Adapters

A minimal guide to how Cognee talks to a graph database without your code ever depending on which one is actually configured. You’ll create two nodes and an edge, read them back, and clean up — using only interface methods, on an isolated local graph.

Before You Start

This is a low-level infrastructure exercise, not a replacement for cognee.remember() or Cognee pipelines — direct graph writes like this skip vector indexing, relational records, dataset provenance, and normal access-control behavior.
This guide defines a custom Person(DataPoint) model — see DataPoints for the concept, or Custom Data Models and Custom Graph Model to go deeper.

Code in Action

import os
import asyncio

import cognee
from cognee.low_level import DataPoint
from cognee.infrastructure.databases.graph import get_graph_engine


class Person(DataPoint):
    name: str
    metadata: dict = {"index_fields": ["name"], "identity_fields": ["name"]}


async def main():
    graph_db_path = os.path.join(os.path.dirname(__file__), ".artifacts", "graph_engine_guide_db")

    cognee.config.set_graph_db_config(
        {
            "graph_database_provider": "ladybug",
            "graph_file_path": graph_db_path,
            "graph_database_subprocess_enabled": False,
        }
    )

    graph_engine = await get_graph_engine()

    alice = Person(name="Alice")
    bob = Person(name="Bob")

    try:
        await graph_engine.add_nodes([alice, bob])
        await graph_engine.add_edge(
            str(alice.id),
            str(bob.id),
            "knows",
            edge_properties={"since": "2020"},
        )

        stored_alice = await graph_engine.get_node(str(alice.id))
        print(f"Alice node: {stored_alice}")

        bob_neighbors = await graph_engine.get_neighbors(str(bob.id))
        print(f"Bob's neighbors: {bob_neighbors}")

        alice_knows_bob = await graph_engine.has_edge(str(alice.id), str(bob.id), "knows")
        print(f"Alice knows Bob: {alice_knows_bob}")
    finally:
        await graph_engine.delete_nodes([str(alice.id), str(bob.id)])


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

What Just Happened

Step 1: Configure an Isolated Graph and Get the Engine

graph_db_path = os.path.join(os.path.dirname(__file__), ".artifacts", "graph_engine_guide_db")

cognee.config.set_graph_db_config(
    {
        "graph_database_provider": "ladybug",
        "graph_file_path": graph_db_path,
        "graph_database_subprocess_enabled": False,
    }
)

graph_engine = await get_graph_engine()

graph_file_path points Ladybug at a throwaway directory so this exercise never touches your default Cognee graph. graph_database_subprocess_enabled=False keeps Ladybug running inside your Python process instead of as a separate background process — simpler for a short script like this. get_graph_engine() is async and returns whatever adapter the configuration selected — here, a Ladybug adapter — typed as GraphDBInterface. Everything from this point on is written against that interface, not against Ladybug specifically.

Step 2: Define a Minimal Node Model

class Person(DataPoint):
    name: str
    metadata: dict = {"index_fields": ["name"], "identity_fields": ["name"]}

Person inherits from DataPoint, Cognee’s base model for graph nodes. Declaring "identity_fields": ["name"] gives each Person a deterministic id based on its name, so Person(name="Alice") always resolves to the same node id instead of a random one.

Step 3: Create Nodes and an Edge

await graph_engine.add_nodes([alice, bob])
await graph_engine.add_edge(
    str(alice.id),
    str(bob.id),
    "knows",
    edge_properties={"since": "2020"},
)

add_nodes(nodes) takes a list of DataPoint instances and writes them as one bulk operation. The Ladybug, Postgres, and Turso adapters split a large list into several idempotent bulk statements rather than sending one statement per call, so a single write cannot outgrow the backend’s per-statement limits; the chunk size is an internal, per-adapter detail and is not configurable. add_edge(source_id, target_id, relationship_name, edge_properties) creates one directed edge — here, Alice -[knows]-> Bob with a small property dictionary. Both take plain string ids, so DataPoint.id (a UUID) is converted with str(...) before being passed in.

Step 4: Read Data Back

stored_alice = await graph_engine.get_node(str(alice.id))
bob_neighbors = await graph_engine.get_neighbors(str(bob.id))
alice_knows_bob = await graph_engine.has_edge(str(alice.id), str(bob.id), "knows")

get_node(node_id) returns a single node’s properties as a dictionary, or None if it does not exist. get_neighbors(node_id) returns the properties of every node connected to the given node — here, Bob’s only neighbor is Alice. has_edge(source_id, target_id, relationship_name) checks whether a specific directed, labeled edge exists, returning a plain bool.

Step 5: Clean Up

finally:
    await graph_engine.delete_nodes([str(alice.id), str(bob.id)])

delete_nodes(node_ids) removes the listed nodes and any edges attached to them. Running it in a finally block ensures the example nodes are removed even if an earlier step raises, leaving the isolated graph empty again.

One Interface, Many Databases

Cognee can store its graph in several different backends — Ladybug (the default local, file-based engine), Postgres, Neo4j, and others (see Graph Stores for how to configure each one). Application code that queries or writes to the graph should not need a different code path for each one. Cognee solves this with three pieces:

configuration -> get_graph_engine() -> GraphDBInterface -> concrete adapter

Configuration picks which provider is active (e.g. "ladybug" or "postgres_demo").
get_graph_engine() is a factory function. It reads the configuration and returns an object implementing the interface — you never construct an adapter yourself.
GraphDBInterface is an abstract base class that declares the methods every adapter must provide (add_node, get_node, has_edge, and so on).
Concrete adapters (LadybugAdapter, PostgresDemoAdapter, …) implement that interface against a specific database.

Because every adapter satisfies the same interface, code written against get_graph_engine() works unchanged no matter which provider is configured.

Seed selection: `get_top_degree_node_ids`

GraphDBInterface.get_top_degree_node_ids(top_k) — the method that picks seed nodes for the default graph visualization — is declared on the interface but, unlike most of its methods, is not abstract: it ships with a working inherited implementation, so an adapter that predates it keeps loading. Its contract:

It returns node ids only (list[str]), never node objects or degree counts.
It is approximate by contract: an adapter may sample instead of counting exactly, and the order of near-equal nodes may vary between calls. Do not treat the result as a ranking — only as “up to top_k nodes worth starting from”.
Isolated nodes are valid seeds, so a graph with nodes but no edges still yields a non-empty result.
top_k must be ≥ 1; a non-positive value raises ValueError before the graph is read.

Ladybug, Neo4j, Neptune, Postgres, and Turso override it and do the ranking inside the store, over a bounded sample of at most 200,000 edges (Postgres and Turso materialize that sample once and reuse it for both edge directions). That bounds the aggregation and the returned data; it is not a latency or query-planning guarantee. An adapter that does not override get_top_degree_node_ids — a third-party adapter from cognee-community, for instance — still works: it inherits the default, which reads the whole graph and counts degree in Python. That is O(graph) in memory and can exhaust a worker on a multi-million-node graph, so the fallback logs a warning once per adapter type per process:

<AdapterName> has no native get_top_degree_node_ids; falling back to a full graph read to rank 10 seeds. This is O(graph) in memory.

If you maintain an adapter and expect large graphs, override the method with a bounded, in-store implementation rather than moving the exact count into the database — an exact aggregate is expensive too. Adding a new graph database covers the signature alongside the other methods a new adapter implements.

Do not import or instantiate LadybugAdapter, PostgresDemoAdapter, or any other adapter directly. Always go through get_graph_engine() — adapter modules and class names are internal and do move (the Postgres adapter was renamed to PostgresDemoAdapter and its old module removed), while code written against get_graph_engine() keeps working unchanged.

Comparing Adapters

LadybugAdapter and PostgresDemoAdapter both implement add_nodes, add_edge, get_node, get_neighbors, and has_edge from GraphDBInterface — but the two implementations look nothing alike internally. Ladybug builds parameterized Cypher-style statements against an embedded Kuzu database; Postgres issues SQL against relational tables that model nodes and edges. Neither difference is visible to code written against the interface, which is the point: swapping graph_database_provider from "ladybug" to "postgres_demo" (with matching connection settings) does not require changing any of the code above.

Every adapter also has a raw query() method for running commands written directly in that database’s own language — Cypher for Ladybug/Neo4j, SQL for Postgres. Using it ties your code to one specific database, which is exactly what this guide is trying to avoid. That’s why it’s left out here — everything above uses only the shared GraphDBInterface methods, which work the same way no matter which database is configured.

DataPoints

Learn more about DataPoint, the base model Person builds on

BaseRetriever Guide

See the same abstract-contract pattern applied to retrievers

Getting Started

Core Concepts

Setup Configuration

Guides

Examples

CLI

OSS

Before You Start

Code in Action

What Just Happened

Step 1: Configure an Isolated Graph and Get the Engine

Step 2: Define a Minimal Node Model

Step 3: Create Nodes and an Edge

Step 4: Read Data Back

Step 5: Clean Up

One Interface, Many Databases

Seed selection: `get_top_degree_node_ids`

Comparing Adapters

DataPoints

BaseRetriever Guide

​Before You Start

​Code in Action

​What Just Happened

​Step 1: Configure an Isolated Graph and Get the Engine

​Step 2: Define a Minimal Node Model

​Step 3: Create Nodes and an Edge

​Step 4: Read Data Back

​Step 5: Clean Up

​One Interface, Many Databases

​Seed selection: get_top_degree_node_ids

​Comparing Adapters

DataPoints

BaseRetriever Guide

Before You Start

Code in Action

What Just Happened

Step 1: Configure an Isolated Graph and Get the Engine

Step 2: Define a Minimal Node Model

Step 3: Create Nodes and an Edge

Step 4: Read Data Back

Step 5: Clean Up

One Interface, Many Databases

Seed selection: `get_top_degree_node_ids`

Comparing Adapters