getzep--graphiti/spec/driver-operations-redesign.md

Driver Operations Redesign Spec

Status: Draft (in progress)

Goals

1. Operations interfaces become the core behavior — adding a new DB backend is as simple as implementing a driver with the operations interfaces filled out.
2. Operations interfaces are organized by object type (not one monolith).
3. DB-related functionality is closely linked to the Graphiti client via namespaces (graphiti.nodes.entity.save(node)), not scattered across data model classes.
4. No awkward override threading — no passing interfaces through multiple levels.
5. Data model classes (EntityNode, EntityEdge, etc.) become pure data (Pydantic models with no DB logic).
6. Phase 1 is non-breaking: existing methods on EntityNode/EntityEdge continue to work.

Architecture Overview

Three layers:

Graphiti Client (graphiti.py)
  └── Namespace Wrappers (thin orchestration: embeddings, tracing)
        └── Operations ABCs (pure DB I/O, implemented per driver)
              └── GraphDriver (connection + query execution)

User-Facing API

graphiti = Graphiti(uri, user, password)

# Node operations
await graphiti.nodes.entity.save(node)
await graphiti.nodes.entity.get_by_uuid("abc-123")
await graphiti.nodes.episode.retrieve_episodes(reference_time, last_n=5)

# Edge operations
await graphiti.edges.entity.save(edge)
await graphiti.edges.entity.get_between_nodes(source_uuid, target_uuid)

# Transactions
async with graphiti.driver.transaction() as tx:
    await graphiti.nodes.entity.save(node1, tx=tx)
    await graphiti.nodes.entity.save(node2, tx=tx)

# High-level search (orchestration stays on client)
results = await graphiti.search(query, ...)

Design Decisions

Decision	Choice	Rationale
Parameterized vs. bound instances	Parameterized (save(node))	Data classes stay pure, no hidden state, easier testing
Generic base vs. flat ops classes	Flat	Decoupled, easier to understand and debug
Embedding generation	Namespace layer	Driver stays pure DB I/O; namespace has access to both embedder and driver
driver param on ops methods	QueryExecutor passed explicitly each call	Ops depend on slim QueryExecutor ABC, not full GraphDriver — zero import cycles
build_fulltext_query	Lives on SearchOperations	Only consumed by search code
load_embeddings methods	Live on respective ops classes	They're per-object-type DB reads
Backwards compatibility	Keep existing data model methods in Phase 1	Non-breaking first, cleanup later
Transaction API	Context manager (async with driver.transaction() as tx)	Pythonic, clean, uniform across drivers
Transaction typing	Typed Transaction ABC	Type safety without coupling to specific drivers

QueryExecutor and Transaction: Breaking the Import Cycle

Operations ABCs need to call execute_query() and session() on the driver, but they must not import GraphDriver (which imports them). We solve this with a slim base class that GraphDriver extends. The Transaction ABC is also defined here since ops methods accept an optional transaction parameter.

# graphiti_core/driver/query_executor.py — standalone, no deps on ops or GraphDriver

class Transaction(ABC):
    """Minimal transaction interface. Yielded by GraphDriver.transaction()."""

    @abstractmethod
    async def run(self, query: str, **kwargs) -> Any: ...


class QueryExecutor(ABC):
    """Slim interface for executing queries. GraphDriver extends this."""

    @abstractmethod
    async def execute_query(self, query: str, **kwargs) -> Any: ...

    @abstractmethod
    def session(self, database: str | None = None) -> GraphDriverSession: ...

Dependency graph (strictly one-directional, no cycles):

QueryExecutor + Transaction    (standalone — no deps)
     ↑
Operations ABCs                (depend on QueryExecutor + Transaction only)
     ↑
GraphDriver                    (extends QueryExecutor, composes Operations ABCs)
     ↑
Namespaces                     (depend on GraphDriver)
     ↑
Graphiti                       (depends on Namespaces + GraphDriver)

All operations ABC methods take executor: QueryExecutor and optionally tx: Transaction | None. At runtime, the concrete driver (which is-a QueryExecutor) is passed through.

Transaction API

User-facing pattern

# Transactional — groups operations, auto-commits on exit, rolls back on exception
async with graphiti.driver.transaction() as tx:
    await graphiti.nodes.entity.save(node1, tx=tx)
    await graphiti.nodes.entity.save(node2, tx=tx)
    await graphiti.edges.entity.save(edge, tx=tx)

# Non-transactional — each operation executes independently (default)
await graphiti.nodes.entity.save(node)

Driver contract

# On GraphDriver
@abstractmethod
def transaction(self) -> AsyncContextManager[Transaction]: ...

Per-driver behavior

Driver	transaction() behavior
Neo4j	Opens a real transaction via session.begin_transaction(). Commits on clean exit, rolls back on exception.
FalkorDB	Returns a lightweight session wrapper. Queries execute immediately. No rollback on failure.
Kuzu	Same as FalkorDB — session wrapper, no rollback.
Neptune	Same as FalkorDB — session wrapper, no rollback.

Drivers that lack native transaction support are honest about it — the API is uniform but the guarantees differ. This matches the current behavior (where execute_write is faked on non-Neo4j drivers) but makes it explicit.

How tx flows through the layers

User code                          Namespace                           Ops ABC
─────────                          ─────────                           ───────
graphiti.nodes.entity.save(        EntityNodeNamespace.save(           EntityNodeOperations.save(
    node, tx=tx                        node, tx=tx                        executor, node, tx=tx
)                                  )                                   )
                                   │                                   │
                                   ├─ generate embeddings              ├─ if tx: tx.run(query)
                                   └─ delegate to ops                  └─ else: executor.execute_query(query)

Implementation sketch for Neo4j

class Neo4jTransaction(Transaction):
    def __init__(self, neo4j_tx):
        self._tx = neo4j_tx

    async def run(self, query: str, **kwargs) -> Any:
        result = await self._tx.run(query, **kwargs)
        return await result.data()


class Neo4jDriver(GraphDriver):
    @asynccontextmanager
    async def transaction(self):
        async with self._driver.session(database=self._database) as session:
            async with await session.begin_transaction() as tx:
                yield Neo4jTransaction(tx)
                await tx.commit()

Implementation sketch for non-transactional drivers (e.g., FalkorDB)

class FalkorTransaction(Transaction):
    """Thin wrapper — no real transaction, queries execute immediately."""

    def __init__(self, graph):
        self._graph = graph

    async def run(self, query: str, **kwargs) -> Any:
        return await self._graph.query(query, kwargs)


class FalkorDBDriver(GraphDriver):
    @asynccontextmanager
    async def transaction(self):
        graph = self.client.select_graph(self._database)
        yield FalkorTransaction(graph)
        # No commit/rollback — queries already executed

Layer 1: Operations ABCs

All operations ABCs are flat (no generic base class). Each object type defines its own complete set of methods independently.

EntityNodeOperations

class EntityNodeOperations(ABC):
    @abstractmethod
    async def save(self, executor: QueryExecutor, node: EntityNode,
                   tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def save_bulk(self, executor: QueryExecutor, nodes: list[EntityNode],
                        tx: Transaction | None = None,
                        batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete(self, executor: QueryExecutor, node: EntityNode,
                     tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete_by_group_id(self, executor: QueryExecutor,
                                  group_id: str, tx: Transaction | None = None,
                                  batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete_by_uuids(self, executor: QueryExecutor,
                               uuids: list[str], tx: Transaction | None = None,
                               batch_size: int = 100) -> None: ...

    @abstractmethod
    async def get_by_uuid(self, executor: QueryExecutor, uuid: str) -> EntityNode: ...

    @abstractmethod
    async def get_by_uuids(self, executor: QueryExecutor, uuids: list[str]) -> list[EntityNode]: ...

    @abstractmethod
    async def get_by_group_ids(self, executor: QueryExecutor, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[EntityNode]: ...

    @abstractmethod
    async def load_embeddings(self, executor: QueryExecutor, node: EntityNode) -> None: ...

    @abstractmethod
    async def load_embeddings_bulk(self, executor: QueryExecutor,
                                    nodes: list[EntityNode],
                                    batch_size: int = 100) -> None: ...

EpisodeNodeOperations

class EpisodeNodeOperations(ABC):
    @abstractmethod
    async def save(self, executor: QueryExecutor, node: EpisodicNode,
                   tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def save_bulk(self, executor: QueryExecutor, nodes: list[EpisodicNode],
                        tx: Transaction | None = None,
                        batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete(self, executor: QueryExecutor, node: EpisodicNode,
                     tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete_by_group_id(self, executor: QueryExecutor,
                                  group_id: str, tx: Transaction | None = None,
                                  batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete_by_uuids(self, executor: QueryExecutor,
                               uuids: list[str], tx: Transaction | None = None,
                               batch_size: int = 100) -> None: ...

    @abstractmethod
    async def get_by_uuid(self, executor: QueryExecutor, uuid: str) -> EpisodicNode: ...

    @abstractmethod
    async def get_by_uuids(self, executor: QueryExecutor,
                            uuids: list[str]) -> list[EpisodicNode]: ...

    @abstractmethod
    async def get_by_group_ids(self, executor: QueryExecutor, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[EpisodicNode]: ...

    @abstractmethod
    async def get_by_entity_node_uuid(self, executor: QueryExecutor,
                                       entity_node_uuid: str) -> list[EpisodicNode]: ...

    @abstractmethod
    async def retrieve_episodes(self, executor: QueryExecutor, reference_time: datetime,
                                 last_n: int = 3, group_ids: list[str] | None = None,
                                 source: str | None = None,
                                 saga: str | None = None) -> list[EpisodicNode]: ...

CommunityNodeOperations

class CommunityNodeOperations(ABC):
    @abstractmethod
    async def save(self, executor: QueryExecutor, node: CommunityNode,
                   tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def save_bulk(self, executor: QueryExecutor, nodes: list[CommunityNode],
                        tx: Transaction | None = None,
                        batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete(self, executor: QueryExecutor, node: CommunityNode,
                     tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete_by_group_id(self, executor: QueryExecutor,
                                  group_id: str, tx: Transaction | None = None,
                                  batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete_by_uuids(self, executor: QueryExecutor,
                               uuids: list[str], tx: Transaction | None = None,
                               batch_size: int = 100) -> None: ...

    @abstractmethod
    async def get_by_uuid(self, executor: QueryExecutor, uuid: str) -> CommunityNode: ...

    @abstractmethod
    async def get_by_uuids(self, executor: QueryExecutor,
                            uuids: list[str]) -> list[CommunityNode]: ...

    @abstractmethod
    async def get_by_group_ids(self, executor: QueryExecutor, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[CommunityNode]: ...

    @abstractmethod
    async def load_name_embedding(self, executor: QueryExecutor,
                                   node: CommunityNode) -> None: ...

SagaNodeOperations

class SagaNodeOperations(ABC):
    @abstractmethod
    async def save(self, executor: QueryExecutor, node: SagaNode,
                   tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def save_bulk(self, executor: QueryExecutor, nodes: list[SagaNode],
                        tx: Transaction | None = None,
                        batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete(self, executor: QueryExecutor, node: SagaNode,
                     tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete_by_group_id(self, executor: QueryExecutor,
                                  group_id: str, tx: Transaction | None = None,
                                  batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete_by_uuids(self, executor: QueryExecutor,
                               uuids: list[str], tx: Transaction | None = None,
                               batch_size: int = 100) -> None: ...

    @abstractmethod
    async def get_by_uuid(self, executor: QueryExecutor, uuid: str) -> SagaNode: ...

    @abstractmethod
    async def get_by_uuids(self, executor: QueryExecutor,
                            uuids: list[str]) -> list[SagaNode]: ...

    @abstractmethod
    async def get_by_group_ids(self, executor: QueryExecutor, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[SagaNode]: ...

EntityEdgeOperations

class EntityEdgeOperations(ABC):
    @abstractmethod
    async def save(self, executor: QueryExecutor, edge: EntityEdge,
                   tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def save_bulk(self, executor: QueryExecutor, edges: list[EntityEdge],
                        tx: Transaction | None = None,
                        batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete(self, executor: QueryExecutor, edge: EntityEdge,
                     tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete_by_uuids(self, executor: QueryExecutor,
                               uuids: list[str],
                               tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def get_by_uuid(self, executor: QueryExecutor, uuid: str) -> EntityEdge: ...

    @abstractmethod
    async def get_by_uuids(self, executor: QueryExecutor,
                            uuids: list[str]) -> list[EntityEdge]: ...

    @abstractmethod
    async def get_by_group_ids(self, executor: QueryExecutor, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[EntityEdge]: ...

    @abstractmethod
    async def get_between_nodes(self, executor: QueryExecutor,
                                 source_node_uuid: str,
                                 target_node_uuid: str) -> list[EntityEdge]: ...

    @abstractmethod
    async def get_by_node_uuid(self, executor: QueryExecutor,
                                node_uuid: str) -> list[EntityEdge]: ...

    @abstractmethod
    async def load_embeddings(self, executor: QueryExecutor, edge: EntityEdge) -> None: ...

    @abstractmethod
    async def load_embeddings_bulk(self, executor: QueryExecutor,
                                    edges: list[EntityEdge],
                                    batch_size: int = 100) -> None: ...

EpisodicEdgeOperations

class EpisodicEdgeOperations(ABC):
    @abstractmethod
    async def save(self, executor: QueryExecutor, edge: EpisodicEdge,
                   tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def save_bulk(self, executor: QueryExecutor, edges: list[EpisodicEdge],
                        tx: Transaction | None = None,
                        batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete(self, executor: QueryExecutor, edge: EpisodicEdge,
                     tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete_by_uuids(self, executor: QueryExecutor,
                               uuids: list[str],
                               tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def get_by_uuid(self, executor: QueryExecutor, uuid: str) -> EpisodicEdge: ...

    @abstractmethod
    async def get_by_uuids(self, executor: QueryExecutor,
                            uuids: list[str]) -> list[EpisodicEdge]: ...

    @abstractmethod
    async def get_by_group_ids(self, executor: QueryExecutor, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[EpisodicEdge]: ...

CommunityEdgeOperations

class CommunityEdgeOperations(ABC):
    @abstractmethod
    async def save(self, executor: QueryExecutor, edge: CommunityEdge,
                   tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete(self, executor: QueryExecutor, edge: CommunityEdge,
                     tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete_by_uuids(self, executor: QueryExecutor,
                               uuids: list[str],
                               tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def get_by_uuid(self, executor: QueryExecutor, uuid: str) -> CommunityEdge: ...

    @abstractmethod
    async def get_by_uuids(self, executor: QueryExecutor,
                            uuids: list[str]) -> list[CommunityEdge]: ...

    @abstractmethod
    async def get_by_group_ids(self, executor: QueryExecutor, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[CommunityEdge]: ...

HasEpisodeEdgeOperations

class HasEpisodeEdgeOperations(ABC):
    @abstractmethod
    async def save(self, executor: QueryExecutor, edge: HasEpisodeEdge,
                   tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def save_bulk(self, executor: QueryExecutor, edges: list[HasEpisodeEdge],
                        tx: Transaction | None = None,
                        batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete(self, executor: QueryExecutor, edge: HasEpisodeEdge,
                     tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete_by_uuids(self, executor: QueryExecutor,
                               uuids: list[str],
                               tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def get_by_uuid(self, executor: QueryExecutor, uuid: str) -> HasEpisodeEdge: ...

    @abstractmethod
    async def get_by_uuids(self, executor: QueryExecutor,
                            uuids: list[str]) -> list[HasEpisodeEdge]: ...

    @abstractmethod
    async def get_by_group_ids(self, executor: QueryExecutor, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[HasEpisodeEdge]: ...

NextEpisodeEdgeOperations

class NextEpisodeEdgeOperations(ABC):
    @abstractmethod
    async def save(self, executor: QueryExecutor, edge: NextEpisodeEdge,
                   tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def save_bulk(self, executor: QueryExecutor, edges: list[NextEpisodeEdge],
                        tx: Transaction | None = None,
                        batch_size: int = 100) -> None: ...

    @abstractmethod
    async def delete(self, executor: QueryExecutor, edge: NextEpisodeEdge,
                     tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def delete_by_uuids(self, executor: QueryExecutor,
                               uuids: list[str],
                               tx: Transaction | None = None) -> None: ...

    @abstractmethod
    async def get_by_uuid(self, executor: QueryExecutor, uuid: str) -> NextEpisodeEdge: ...

    @abstractmethod
    async def get_by_uuids(self, executor: QueryExecutor,
                            uuids: list[str]) -> list[NextEpisodeEdge]: ...

    @abstractmethod
    async def get_by_group_ids(self, executor: QueryExecutor, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[NextEpisodeEdge]: ...

SearchOperations

class SearchOperations(ABC):
    # Node search
    @abstractmethod
    async def node_fulltext_search(self, executor: QueryExecutor, query: str,
                                    search_filter: Any, group_ids: list[str] | None = None,
                                    limit: int = 10) -> list[EntityNode]: ...

    @abstractmethod
    async def node_similarity_search(self, executor: QueryExecutor, search_vector: list[float],
                                      search_filter: Any, group_ids: list[str] | None = None,
                                      limit: int = 10,
                                      min_score: float = 0.6) -> list[EntityNode]: ...

    @abstractmethod
    async def node_bfs_search(self, executor: QueryExecutor,
                               origin_uuids: list[str], search_filter: Any,
                               max_depth: int, group_ids: list[str] | None = None,
                               limit: int = 10) -> list[EntityNode]: ...

    # Edge search
    @abstractmethod
    async def edge_fulltext_search(self, executor: QueryExecutor, query: str,
                                    search_filter: Any, group_ids: list[str] | None = None,
                                    limit: int = 10) -> list[EntityEdge]: ...

    @abstractmethod
    async def edge_similarity_search(self, executor: QueryExecutor, search_vector: list[float],
                                      source_node_uuid: str | None,
                                      target_node_uuid: str | None,
                                      search_filter: Any,
                                      group_ids: list[str] | None = None,
                                      limit: int = 10,
                                      min_score: float = 0.6) -> list[EntityEdge]: ...

    @abstractmethod
    async def edge_bfs_search(self, executor: QueryExecutor,
                               origin_uuids: list[str], max_depth: int,
                               search_filter: Any, group_ids: list[str] | None = None,
                               limit: int = 10) -> list[EntityEdge]: ...

    # Episode search
    @abstractmethod
    async def episode_fulltext_search(self, executor: QueryExecutor, query: str,
                                       search_filter: Any,
                                       group_ids: list[str] | None = None,
                                       limit: int = 10) -> list[EpisodicNode]: ...

    # Community search
    @abstractmethod
    async def community_fulltext_search(self, executor: QueryExecutor, query: str,
                                         group_ids: list[str] | None = None,
                                         limit: int = 10) -> list[CommunityNode]: ...

    @abstractmethod
    async def community_similarity_search(self, executor: QueryExecutor,
                                           search_vector: list[float],
                                           group_ids: list[str] | None = None,
                                           limit: int = 10,
                                           min_score: float = 0.6) -> list[CommunityNode]: ...

    # Rerankers
    @abstractmethod
    async def node_distance_reranker(self, executor: QueryExecutor,
                                      node_uuids: list[str],
                                      center_node_uuid: str,
                                      min_score: float = 0) -> list[EntityNode]: ...

    @abstractmethod
    async def episode_mentions_reranker(self, executor: QueryExecutor,
                                         node_uuids: list[str],
                                         min_score: float = 0) -> list[EntityNode]: ...

    # Filter builders (sync)
    @abstractmethod
    def build_node_search_filters(self, search_filters: Any) -> Any: ...

    @abstractmethod
    def build_edge_search_filters(self, search_filters: Any) -> Any: ...

    # Fulltext query builder
    @abstractmethod
    def build_fulltext_query(self, query: str, group_ids: list[str] | None = None,
                              max_query_length: int = 8000) -> str: ...

GraphMaintenanceOperations

class GraphMaintenanceOperations(ABC):
    @abstractmethod
    async def clear_data(self, executor: QueryExecutor,
                          group_ids: list[str] | None = None) -> None: ...

    @abstractmethod
    async def build_indices_and_constraints(self, executor: QueryExecutor,
                                             delete_existing: bool = False) -> None: ...

    @abstractmethod
    async def delete_all_indexes(self, executor: QueryExecutor) -> None: ...

    @abstractmethod
    async def get_community_clusters(self, executor: QueryExecutor,
                                      group_ids: list[str] | None = None) -> list: ...

    @abstractmethod
    async def remove_communities(self, executor: QueryExecutor) -> None: ...

    @abstractmethod
    async def determine_entity_community(self, executor: QueryExecutor,
                                          entity: EntityNode) -> None: ...

    @abstractmethod
    async def get_mentioned_nodes(self, executor: QueryExecutor,
                                   episodes: list[EpisodicNode]) -> list[EntityNode]: ...

    @abstractmethod
    async def get_communities_by_nodes(self, executor: QueryExecutor,
                                        nodes: list[EntityNode]) -> list[CommunityNode]: ...

Layer 2: GraphDriver Composes Operations

class GraphDriver(QueryExecutor, ABC):
    # --- Core connection methods ---
    # execute_query() and session() inherited from QueryExecutor

    @abstractmethod
    async def close(self) -> None: ...

    @abstractmethod
    def transaction(self) -> AsyncContextManager[Transaction]: ...

    # --- Operations interfaces (all required, all abstract) ---
    @property
    @abstractmethod
    def entity_node_ops(self) -> EntityNodeOperations: ...

    @property
    @abstractmethod
    def episode_node_ops(self) -> EpisodeNodeOperations: ...

    @property
    @abstractmethod
    def community_node_ops(self) -> CommunityNodeOperations: ...

    @property
    @abstractmethod
    def saga_node_ops(self) -> SagaNodeOperations: ...

    @property
    @abstractmethod
    def entity_edge_ops(self) -> EntityEdgeOperations: ...

    @property
    @abstractmethod
    def episodic_edge_ops(self) -> EpisodicEdgeOperations: ...

    @property
    @abstractmethod
    def community_edge_ops(self) -> CommunityEdgeOperations: ...

    @property
    @abstractmethod
    def has_episode_edge_ops(self) -> HasEpisodeEdgeOperations: ...

    @property
    @abstractmethod
    def next_episode_edge_ops(self) -> NextEpisodeEdgeOperations: ...

    @property
    @abstractmethod
    def search_ops(self) -> SearchOperations: ...

    @property
    @abstractmethod
    def graph_ops(self) -> GraphMaintenanceOperations: ...

Example driver implementation:

class Neo4jDriver(GraphDriver):
    def __init__(self, uri, user, password):
        # ... connection setup ...
        self._entity_node_ops = Neo4jEntityNodeOps()
        self._episode_node_ops = Neo4jEpisodeNodeOps()
        self._community_node_ops = Neo4jCommunityNodeOps()
        self._saga_node_ops = Neo4jSagaNodeOps()
        self._entity_edge_ops = Neo4jEntityEdgeOps()
        self._episodic_edge_ops = Neo4jEpisodicEdgeOps()
        self._community_edge_ops = Neo4jCommunityEdgeOps()
        self._has_episode_edge_ops = Neo4jHasEpisodeEdgeOps()
        self._next_episode_edge_ops = Neo4jNextEpisodeEdgeOps()
        self._search_ops = Neo4jSearchOps()
        self._graph_ops = Neo4jGraphMaintenanceOps()

    @property
    def entity_node_ops(self) -> EntityNodeOperations:
        return self._entity_node_ops

    # ... etc for all ops properties ...

Layer 3: Namespace Wrappers

Thin wrappers on the Graphiti client that orchestrate non-DB concerns (embedding generation, tracing) before delegating to the driver's ops.

class EntityNodeNamespace:
    def __init__(self, driver: GraphDriver, embedder: EmbedderClient):
        self._driver = driver
        self._embedder = embedder
        self._ops = driver.entity_node_ops

    async def save(self, node: EntityNode,
                   tx: Transaction | None = None) -> EntityNode:
        await node.generate_name_embedding(self._embedder)
        await self._ops.save(self._driver, node, tx=tx)
        return node

    async def save_bulk(self, nodes: list[EntityNode],
                         tx: Transaction | None = None,
                         batch_size: int = 100) -> None:
        await self._ops.save_bulk(self._driver, nodes, tx=tx, batch_size=batch_size)

    async def delete(self, node: EntityNode,
                     tx: Transaction | None = None) -> None:
        await self._ops.delete(self._driver, node, tx=tx)

    async def delete_by_group_id(self, group_id: str,
                                  tx: Transaction | None = None,
                                  batch_size: int = 100) -> None:
        await self._ops.delete_by_group_id(self._driver, group_id, tx=tx, batch_size=batch_size)

    async def delete_by_uuids(self, uuids: list[str],
                               tx: Transaction | None = None,
                               batch_size: int = 100) -> None:
        await self._ops.delete_by_uuids(self._driver, uuids, tx=tx, batch_size=batch_size)

    async def get_by_uuid(self, uuid: str) -> EntityNode:
        return await self._ops.get_by_uuid(self._driver, uuid)

    async def get_by_uuids(self, uuids: list[str]) -> list[EntityNode]:
        return await self._ops.get_by_uuids(self._driver, uuids)

    async def get_by_group_ids(self, group_ids: list[str],
                                limit: int | None = None,
                                uuid_cursor: str | None = None) -> list[EntityNode]:
        return await self._ops.get_by_group_ids(self._driver, group_ids, limit, uuid_cursor)

    async def load_embeddings(self, node: EntityNode) -> None:
        await self._ops.load_embeddings(self._driver, node)

    async def load_embeddings_bulk(self, nodes: list[EntityNode],
                                    batch_size: int = 100) -> None:
        await self._ops.load_embeddings_bulk(self._driver, nodes, batch_size)


class NodeNamespace:
    """Accessed as graphiti.nodes"""
    def __init__(self, driver: GraphDriver, embedder: EmbedderClient):
        self.entity = EntityNodeNamespace(driver, embedder)
        self.episode = EpisodeNodeNamespace(driver)
        self.community = CommunityNodeNamespace(driver, embedder)
        self.saga = SagaNodeNamespace(driver)


class EdgeNamespace:
    """Accessed as graphiti.edges"""
    def __init__(self, driver: GraphDriver, embedder: EmbedderClient):
        self.entity = EntityEdgeNamespace(driver, embedder)
        self.episodic = EpisodicEdgeNamespace(driver)
        self.community = CommunityEdgeNamespace(driver)
        self.has_episode = HasEpisodeEdgeNamespace(driver)
        self.next_episode = NextEpisodeEdgeNamespace(driver)

Wired up in the Graphiti client:

class Graphiti:
    def __init__(self, ..., graph_driver: GraphDriver | None = None, ...):
        self.driver = graph_driver or Neo4jDriver(uri, user, password)
        self.embedder = embedder or OpenAIEmbedder()
        self.nodes = NodeNamespace(self.driver, self.embedder)
        self.edges = EdgeNamespace(self.driver, self.embedder)

        # High-level search orchestration stays as methods on Graphiti.
        # Low-level search queries delegate to self.driver.search_ops.

File Layout

graphiti_core/
  driver/
    query_executor.py                # QueryExecutor ABC (standalone, no deps)
    driver.py                        # GraphDriver(QueryExecutor) ABC, GraphDriverSession ABC
    operations/
      __init__.py                    # Re-exports all operations ABCs
      entity_node_ops.py             # EntityNodeOperations ABC
      episode_node_ops.py            # EpisodeNodeOperations ABC
      community_node_ops.py          # CommunityNodeOperations ABC
      saga_node_ops.py               # SagaNodeOperations ABC
      entity_edge_ops.py             # EntityEdgeOperations ABC
      episodic_edge_ops.py           # EpisodicEdgeOperations ABC
      community_edge_ops.py          # CommunityEdgeOperations ABC
      has_episode_edge_ops.py        # HasEpisodeEdgeOperations ABC
      next_episode_edge_ops.py       # NextEpisodeEdgeOperations ABC
      search_ops.py                  # SearchOperations ABC
      graph_ops.py                   # GraphMaintenanceOperations ABC
    neo4j/
      driver.py                      # Neo4jDriver(GraphDriver)
      operations/
        entity_node_ops.py           # Neo4jEntityNodeOps
        episode_node_ops.py          # Neo4jEpisodeNodeOps
        community_node_ops.py        # Neo4jCommunityNodeOps
        saga_node_ops.py             # Neo4jSagaNodeOps
        entity_edge_ops.py           # Neo4jEntityEdgeOps
        episodic_edge_ops.py         # Neo4jEpisodicEdgeOps
        community_edge_ops.py        # Neo4jCommunityEdgeOps
        has_episode_edge_ops.py      # Neo4jHasEpisodeEdgeOps
        next_episode_edge_ops.py     # Neo4jNextEpisodeEdgeOps
        search_ops.py                # Neo4jSearchOps
        graph_ops.py                 # Neo4jGraphMaintenanceOps
    falkordb/
      driver.py
      operations/
        ...                          # Same structure as neo4j/operations/
  namespaces/
    __init__.py
    nodes.py                         # NodeNamespace + EntityNodeNamespace, etc.
    edges.py                         # EdgeNamespace + EntityEdgeNamespace, etc.
  graphiti.py                        # Graphiti client with .nodes, .edges properties
  nodes.py                           # Data models (existing DB methods kept, deprecated)
  edges.py                           # Data models (existing DB methods kept, deprecated)
  search/
    search.py                        # High-level search orchestration (unchanged)
    search_utils.py                  # Will gradually migrate to use driver.search_ops

Migration Strategy

Phase 1: Non-Breaking (this round)

1. Define all operations ABCs in driver/operations/
2. Create Neo4j ops implementations (extract query logic from nodes.py, edges.py, search_utils.py)
3. Create namespace wrappers in namespaces/
4. Wire Graphiti with self.nodes, self.edges
5. Keep all existing methods on data model classes working as-is
6. Internal code can start using namespaces incrementally

Phase 2: Breaking Cleanup (later)

1. Remove DB methods from EntityNode, EntityEdge, etc.
2. Remove old SearchInterface and GraphOperationsInterface
3. Update all internal callers to use namespace API
4. Remove provider-branching from utility files
5. Remove search_interface and graph_operations_interface from driver

Resolved Questions

• Import cycles: Resolved via QueryExecutor ABC. Ops ABCs depend on QueryExecutor, not GraphDriver. No cycles, no __future__ workarounds.
• Embedding loading methods: Confirmed — live on the respective ops classes (per-object-type DB reads).
• build_fulltext_query: Confirmed — lives on SearchOperations.

Open Questions

None — all design questions resolved.