99923c0352
* feat: add driver operations architecture with abstract interfaces and concrete implementations Introduces a clean operations-based architecture for graph driver operations, replacing inline query logic with abstract interfaces (ABCs) and concrete implementations for both Neo4j and FalkorDB backends. Key changes: - Add QueryExecutor and Transaction ABCs for database-agnostic query execution - Add 11 operations ABCs covering all node, edge, search, and graph maintenance operations - Implement all 11 operations for Neo4j with real transaction commit/rollback - Implement all 11 operations for FalkorDB with RedisSearch fulltext and vecf32 embeddings - Add NodeNamespace and EdgeNamespace convenience wrappers on Graphiti class - Wire operations into Neo4jDriver and FalkorDriver with property accessors - Fix circular import by moving STOPWORDS to graphiti_core.driver.falkordb package - Include design spec documenting architecture decisions and migration plan Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: address PR review feedback - Fix ruff UP037: remove quoted type annotations in driver.py (redundant with `from __future__ import annotations`) - Extract duplicate record parsers into shared record_parsers.py module, eliminating identical _entity_node_from_record, _entity_edge_from_record, _episodic_node_from_record, and _community_node_from_record across 10 files in both Neo4j and FalkorDB operations - Fix MAX_QUERY_LENGTH inconsistency in FalkorDB search_ops build_fulltext_query (was 8000, now uses module constant 128) - Make namespace attributes unconditional with NotImplementedError for drivers that don't implement required operations Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: make namespace init graceful for drivers missing operations KuzuDriver doesn't implement the new operations interfaces, so the NotImplementedError on init broke Kuzu tests. Now attributes are only set when the driver provides them, and __getattr__ gives a clear error on access. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
176 lines
8.8 KiB
Python
176 lines
8.8 KiB
Python
"""
|
|
Database query utilities for different graph database backends.
|
|
|
|
This module provides database-agnostic query generation for Neo4j and FalkorDB,
|
|
supporting index creation, fulltext search, and bulk operations.
|
|
"""
|
|
|
|
from typing_extensions import LiteralString
|
|
|
|
from graphiti_core.driver.driver import GraphProvider
|
|
|
|
# Mapping from Neo4j fulltext index names to FalkorDB node labels
|
|
NEO4J_TO_FALKORDB_MAPPING = {
|
|
'node_name_and_summary': 'Entity',
|
|
'community_name': 'Community',
|
|
'episode_content': 'Episodic',
|
|
'edge_name_and_fact': 'RELATES_TO',
|
|
}
|
|
# Mapping from fulltext index names to Kuzu node labels
|
|
INDEX_TO_LABEL_KUZU_MAPPING = {
|
|
'node_name_and_summary': 'Entity',
|
|
'community_name': 'Community',
|
|
'episode_content': 'Episodic',
|
|
'edge_name_and_fact': 'RelatesToNode_',
|
|
}
|
|
|
|
|
|
def get_range_indices(provider: GraphProvider) -> list[LiteralString]:
|
|
if provider == GraphProvider.FALKORDB:
|
|
return [
|
|
# Entity node
|
|
'CREATE INDEX FOR (n:Entity) ON (n.uuid, n.group_id, n.name, n.created_at)',
|
|
# Episodic node
|
|
'CREATE INDEX FOR (n:Episodic) ON (n.uuid, n.group_id, n.created_at, n.valid_at)',
|
|
# Community node
|
|
'CREATE INDEX FOR (n:Community) ON (n.uuid)',
|
|
# Saga node
|
|
'CREATE INDEX FOR (n:Saga) ON (n.uuid, n.group_id, n.name)',
|
|
# RELATES_TO edge
|
|
'CREATE INDEX FOR ()-[e:RELATES_TO]-() ON (e.uuid, e.group_id, e.name, e.created_at, e.expired_at, e.valid_at, e.invalid_at)',
|
|
# MENTIONS edge
|
|
'CREATE INDEX FOR ()-[e:MENTIONS]-() ON (e.uuid, e.group_id)',
|
|
# HAS_MEMBER edge
|
|
'CREATE INDEX FOR ()-[e:HAS_MEMBER]-() ON (e.uuid)',
|
|
# HAS_EPISODE edge
|
|
'CREATE INDEX FOR ()-[e:HAS_EPISODE]-() ON (e.uuid, e.group_id)',
|
|
# NEXT_EPISODE edge
|
|
'CREATE INDEX FOR ()-[e:NEXT_EPISODE]-() ON (e.uuid, e.group_id)',
|
|
]
|
|
|
|
if provider == GraphProvider.KUZU:
|
|
return []
|
|
|
|
return [
|
|
'CREATE INDEX entity_uuid IF NOT EXISTS FOR (n:Entity) ON (n.uuid)',
|
|
'CREATE INDEX episode_uuid IF NOT EXISTS FOR (n:Episodic) ON (n.uuid)',
|
|
'CREATE INDEX community_uuid IF NOT EXISTS FOR (n:Community) ON (n.uuid)',
|
|
'CREATE INDEX saga_uuid IF NOT EXISTS FOR (n:Saga) ON (n.uuid)',
|
|
'CREATE INDEX relation_uuid IF NOT EXISTS FOR ()-[e:RELATES_TO]-() ON (e.uuid)',
|
|
'CREATE INDEX mention_uuid IF NOT EXISTS FOR ()-[e:MENTIONS]-() ON (e.uuid)',
|
|
'CREATE INDEX has_member_uuid IF NOT EXISTS FOR ()-[e:HAS_MEMBER]-() ON (e.uuid)',
|
|
'CREATE INDEX has_episode_uuid IF NOT EXISTS FOR ()-[e:HAS_EPISODE]-() ON (e.uuid)',
|
|
'CREATE INDEX next_episode_uuid IF NOT EXISTS FOR ()-[e:NEXT_EPISODE]-() ON (e.uuid)',
|
|
'CREATE INDEX entity_group_id IF NOT EXISTS FOR (n:Entity) ON (n.group_id)',
|
|
'CREATE INDEX episode_group_id IF NOT EXISTS FOR (n:Episodic) ON (n.group_id)',
|
|
'CREATE INDEX community_group_id IF NOT EXISTS FOR (n:Community) ON (n.group_id)',
|
|
'CREATE INDEX saga_group_id IF NOT EXISTS FOR (n:Saga) ON (n.group_id)',
|
|
'CREATE INDEX relation_group_id IF NOT EXISTS FOR ()-[e:RELATES_TO]-() ON (e.group_id)',
|
|
'CREATE INDEX mention_group_id IF NOT EXISTS FOR ()-[e:MENTIONS]-() ON (e.group_id)',
|
|
'CREATE INDEX has_episode_group_id IF NOT EXISTS FOR ()-[e:HAS_EPISODE]-() ON (e.group_id)',
|
|
'CREATE INDEX next_episode_group_id IF NOT EXISTS FOR ()-[e:NEXT_EPISODE]-() ON (e.group_id)',
|
|
'CREATE INDEX name_entity_index IF NOT EXISTS FOR (n:Entity) ON (n.name)',
|
|
'CREATE INDEX saga_name IF NOT EXISTS FOR (n:Saga) ON (n.name)',
|
|
'CREATE INDEX created_at_entity_index IF NOT EXISTS FOR (n:Entity) ON (n.created_at)',
|
|
'CREATE INDEX created_at_episodic_index IF NOT EXISTS FOR (n:Episodic) ON (n.created_at)',
|
|
'CREATE INDEX valid_at_episodic_index IF NOT EXISTS FOR (n:Episodic) ON (n.valid_at)',
|
|
'CREATE INDEX name_edge_index IF NOT EXISTS FOR ()-[e:RELATES_TO]-() ON (e.name)',
|
|
'CREATE INDEX created_at_edge_index IF NOT EXISTS FOR ()-[e:RELATES_TO]-() ON (e.created_at)',
|
|
'CREATE INDEX expired_at_edge_index IF NOT EXISTS FOR ()-[e:RELATES_TO]-() ON (e.expired_at)',
|
|
'CREATE INDEX valid_at_edge_index IF NOT EXISTS FOR ()-[e:RELATES_TO]-() ON (e.valid_at)',
|
|
'CREATE INDEX invalid_at_edge_index IF NOT EXISTS FOR ()-[e:RELATES_TO]-() ON (e.invalid_at)',
|
|
]
|
|
|
|
|
|
def get_fulltext_indices(provider: GraphProvider) -> list[LiteralString]:
|
|
if provider == GraphProvider.FALKORDB:
|
|
from typing import cast
|
|
|
|
from graphiti_core.driver.falkordb import STOPWORDS
|
|
|
|
# Convert to string representation for embedding in queries
|
|
stopwords_str = str(STOPWORDS)
|
|
|
|
# Use type: ignore to satisfy LiteralString requirement while maintaining single source of truth
|
|
return cast(
|
|
list[LiteralString],
|
|
[
|
|
f"""CALL db.idx.fulltext.createNodeIndex(
|
|
{{
|
|
label: 'Episodic',
|
|
stopwords: {stopwords_str}
|
|
}},
|
|
'content', 'source', 'source_description', 'group_id'
|
|
)""",
|
|
f"""CALL db.idx.fulltext.createNodeIndex(
|
|
{{
|
|
label: 'Entity',
|
|
stopwords: {stopwords_str}
|
|
}},
|
|
'name', 'summary', 'group_id'
|
|
)""",
|
|
f"""CALL db.idx.fulltext.createNodeIndex(
|
|
{{
|
|
label: 'Community',
|
|
stopwords: {stopwords_str}
|
|
}},
|
|
'name', 'group_id'
|
|
)""",
|
|
"""CREATE FULLTEXT INDEX FOR ()-[e:RELATES_TO]-() ON (e.name, e.fact, e.group_id)""",
|
|
],
|
|
)
|
|
|
|
if provider == GraphProvider.KUZU:
|
|
return [
|
|
"CALL CREATE_FTS_INDEX('Episodic', 'episode_content', ['content', 'source', 'source_description']);",
|
|
"CALL CREATE_FTS_INDEX('Entity', 'node_name_and_summary', ['name', 'summary']);",
|
|
"CALL CREATE_FTS_INDEX('Community', 'community_name', ['name']);",
|
|
"CALL CREATE_FTS_INDEX('RelatesToNode_', 'edge_name_and_fact', ['name', 'fact']);",
|
|
]
|
|
|
|
return [
|
|
"""CREATE FULLTEXT INDEX episode_content IF NOT EXISTS
|
|
FOR (e:Episodic) ON EACH [e.content, e.source, e.source_description, e.group_id]""",
|
|
"""CREATE FULLTEXT INDEX node_name_and_summary IF NOT EXISTS
|
|
FOR (n:Entity) ON EACH [n.name, n.summary, n.group_id]""",
|
|
"""CREATE FULLTEXT INDEX community_name IF NOT EXISTS
|
|
FOR (n:Community) ON EACH [n.name, n.group_id]""",
|
|
"""CREATE FULLTEXT INDEX edge_name_and_fact IF NOT EXISTS
|
|
FOR ()-[e:RELATES_TO]-() ON EACH [e.name, e.fact, e.group_id]""",
|
|
]
|
|
|
|
|
|
def get_nodes_query(name: str, query: str, limit: int, provider: GraphProvider) -> str:
|
|
if provider == GraphProvider.FALKORDB:
|
|
label = NEO4J_TO_FALKORDB_MAPPING[name]
|
|
return f"CALL db.idx.fulltext.queryNodes('{label}', {query})"
|
|
|
|
if provider == GraphProvider.KUZU:
|
|
label = INDEX_TO_LABEL_KUZU_MAPPING[name]
|
|
return f"CALL QUERY_FTS_INDEX('{label}', '{name}', {query}, TOP := $limit)"
|
|
|
|
return f'CALL db.index.fulltext.queryNodes("{name}", {query}, {{limit: $limit}})'
|
|
|
|
|
|
def get_vector_cosine_func_query(vec1, vec2, provider: GraphProvider) -> str:
|
|
if provider == GraphProvider.FALKORDB:
|
|
# FalkorDB uses a different syntax for regular cosine similarity and Neo4j uses normalized cosine similarity
|
|
return f'(2 - vec.cosineDistance({vec1}, vecf32({vec2})))/2'
|
|
|
|
if provider == GraphProvider.KUZU:
|
|
return f'array_cosine_similarity({vec1}, {vec2})'
|
|
|
|
return f'vector.similarity.cosine({vec1}, {vec2})'
|
|
|
|
|
|
def get_relationships_query(name: str, limit: int, provider: GraphProvider) -> str:
|
|
if provider == GraphProvider.FALKORDB:
|
|
label = NEO4J_TO_FALKORDB_MAPPING[name]
|
|
return f"CALL db.idx.fulltext.queryRelationships('{label}', $query)"
|
|
|
|
if provider == GraphProvider.KUZU:
|
|
label = INDEX_TO_LABEL_KUZU_MAPPING[name]
|
|
return f"CALL QUERY_FTS_INDEX('{label}', '{name}', cast($query AS STRING), TOP := $limit)"
|
|
|
|
return f'CALL db.index.fulltext.queryRelationships("{name}", $query, {{limit: $limit}})'
|