diff --git a/graphiti_core/driver/arcadedb/operations/search_ops.py b/graphiti_core/driver/arcadedb/operations/search_ops.py index b8ee20967a..7e0e6d4756 100644 --- a/graphiti_core/driver/arcadedb/operations/search_ops.py +++ b/graphiti_core/driver/arcadedb/operations/search_ops.py @@ -116,8 +116,16 @@ async def node_fulltext_search( if filter_queries: filter_query = ' AND ' + (' AND '.join(filter_queries)) - # Use toLower/CONTAINS for fulltext search (compatible with ArcadeDB Cypher) - search_terms = lucene_sanitize(query).split() + # Use toLower/CONTAINS for fulltext search (compatible with ArcadeDB Cypher). + # NOTE: use the raw query terms here, NOT lucene_sanitize(query) -- that + # escapes special/reserved characters with a literal backslash (e.g. + # 'Alice' -> '\Alice'), which is correct for building a real Lucene + # query string but breaks a literal CONTAINS substring match against + # unescaped stored text (verified: 'Alice' never matches + # toLower(n.name) CONTAINS '\alice'). lucene_sanitize is still used + # above via _build_arcadedb_fulltext_query purely as an empty/oversized + # query gate, not for the actual match text. + search_terms = query.split() if not search_terms: return [] @@ -171,16 +179,14 @@ async def node_similarity_search( filter_queries.append('n.group_id IN $group_ids') filter_params['group_ids'] = group_ids - filter_query = '' - if filter_queries: - filter_query = ' WHERE ' + (' AND '.join(filter_queries)) + filter_queries.append('n.name_embedding IS NOT NULL') + filter_query = ' WHERE ' + (' AND '.join(filter_queries)) # Fetch candidate nodes with embeddings cypher = ( 'MATCH (n:Entity)' + filter_query + """ - WHERE n.name_embedding IS NOT NULL RETURN """ + get_entity_node_return_query(GraphProvider.ARCADEDB) @@ -286,8 +292,11 @@ async def edge_fulltext_search( if filter_queries: filter_query = ' AND ' + (' AND '.join(filter_queries)) - # Use CONTAINS for fulltext search - search_terms = lucene_sanitize(query).split() + # Use CONTAINS for fulltext search. NOTE: raw query terms, NOT + # lucene_sanitize(query) -- see node_fulltext_search for why (it + # Lucene-escapes with a literal backslash, which never matches + # unescaped stored text via CONTAINS). + search_terms = query.split() if not search_terms: return [] @@ -342,24 +351,22 @@ async def edge_similarity_search( filter_queries.append('e.group_id IN $group_ids') filter_params['group_ids'] = group_ids - if source_node_uuid is not None: - filter_params['source_uuid'] = source_node_uuid - filter_queries.append('n.uuid = $source_uuid') + if source_node_uuid is not None: + filter_params['source_uuid'] = source_node_uuid + filter_queries.append('n.uuid = $source_uuid') - if target_node_uuid is not None: - filter_params['target_uuid'] = target_node_uuid - filter_queries.append('m.uuid = $target_uuid') + if target_node_uuid is not None: + filter_params['target_uuid'] = target_node_uuid + filter_queries.append('m.uuid = $target_uuid') - filter_query = '' - if filter_queries: - filter_query = ' WHERE ' + (' AND '.join(filter_queries)) + filter_queries.append('e.fact_embedding IS NOT NULL') + filter_query = ' WHERE ' + (' AND '.join(filter_queries)) # Fetch candidate edges with embeddings cypher = ( 'MATCH (n:Entity)-[e:RELATES_TO]->(m:Entity)' + filter_query + """ - WHERE e.fact_embedding IS NOT NULL RETURN DISTINCT """ + get_entity_edge_return_query(GraphProvider.ARCADEDB) @@ -460,8 +467,9 @@ async def episode_fulltext_search( group_filter_query += '\nAND e.group_id IN $group_ids' filter_params['group_ids'] = group_ids - # Use CONTAINS for fulltext search - search_terms = lucene_sanitize(query).split() + # Use CONTAINS for fulltext search. NOTE: raw query terms, NOT + # lucene_sanitize(query) -- see node_fulltext_search for why. + search_terms = query.split() if not search_terms: return [] @@ -511,8 +519,9 @@ async def community_fulltext_search( group_filter_query = 'AND c.group_id IN $group_ids' filter_params['group_ids'] = group_ids - # Use CONTAINS for fulltext search - search_terms = lucene_sanitize(query).split() + # Use CONTAINS for fulltext search. NOTE: raw query terms, NOT + # lucene_sanitize(query) -- see node_fulltext_search for why. + search_terms = query.split() if not search_terms: return [] @@ -553,17 +562,17 @@ async def community_similarity_search( ) -> list[CommunityNode]: query_params: dict[str, Any] = {} - group_filter_query = '' + filter_queries = ['c.name_embedding IS NOT NULL'] if group_ids is not None: - group_filter_query += ' WHERE c.group_id IN $group_ids' + filter_queries.append('c.group_id IN $group_ids') query_params['group_ids'] = group_ids # Fetch candidate communities with embeddings cypher = ( 'MATCH (c:Community)' - + group_filter_query + + ' WHERE ' + + (' AND '.join(filter_queries)) + """ - WHERE c.name_embedding IS NOT NULL RETURN """ + COMMUNITY_NODE_RETURN diff --git a/graphiti_core/driver/arcadedb/search_interface.py b/graphiti_core/driver/arcadedb/search_interface.py new file mode 100644 index 0000000000..0a8e42519c --- /dev/null +++ b/graphiti_core/driver/arcadedb/search_interface.py @@ -0,0 +1,134 @@ +""" +Copyright 2024, Zep Software, Inc. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +""" + +from typing import Any + +from pydantic import PrivateAttr + +from graphiti_core.driver.arcadedb.operations.search_ops import ArcadeDBSearchOperations +from graphiti_core.driver.search_interface.search_interface import SearchInterface + + +class ArcadeDBSearchInterface(SearchInterface): + """Routes the fulltext/similarity search dispatchers in + graphiti_core.search.search_utils to ArcadeDBSearchOperations instead of + the generic Neo4j-fulltext-procedure path. + + ArcadeDB does not support the `db.index.fulltext.query{Nodes,Relationships}` + Bolt procedures that search_utils.py's generic (non-Neptune/Kuzu) branch + assumes, so edge/node/episode/community fulltext search crash for ArcadeDB + today. ArcadeDBSearchOperations already implements the CONTAINS-based + equivalent correctly -- it was just never wired up. `GraphDriver` extends + `QueryExecutor`, so the driver instance can be passed directly as the + `executor` argument these methods expect. + + 6 of the 12 methods SearchInterface declares are implemented here: + edge_fulltext_search, edge_similarity_search, node_fulltext_search, + node_similarity_search and episode_fulltext_search are implemented because + search_utils.py calls them WITHOUT a `try/except NotImplementedError` guard + around the `search_interface` dispatch -- leaving them unimplemented would + turn "search_interface not set" (falls through to the generic path) into + "search_interface set but incomplete" (crashes with an uncaught + NotImplementedError) for functionality that already works today via the + generic path. community_fulltext_search IS guarded, but is additionally + implemented anyway because its generic fallback has the identical + fulltext-procedure bug. The remaining 6 methods (edge_bfs_search, + node_bfs_search, community_similarity_search, node_distance_reranker, + episode_mentions_reranker, get_embeddings_for_communities) are + intentionally left unimplemented: their generic paths use provider-agnostic + Cypher (no fulltext procedure) and already work correctly against + ArcadeDB, and all of their call sites in search_utils.py DO guard with + `except NotImplementedError: pass`, so they safely keep using that + already-working path. + """ + + _ops: ArcadeDBSearchOperations = PrivateAttr(default_factory=ArcadeDBSearchOperations) + + async def edge_fulltext_search( + self, + driver: Any, + query: str, + search_filter: Any, + group_ids: list[str] | None = None, + limit: int = 100, + ) -> list[Any]: + return await self._ops.edge_fulltext_search(driver, query, search_filter, group_ids, limit) + + async def edge_similarity_search( + self, + driver: Any, + 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 = 100, + min_score: float = 0.7, + ) -> list[Any]: + return await self._ops.edge_similarity_search( + driver, + search_vector, + source_node_uuid, + target_node_uuid, + search_filter, + group_ids, + limit, + min_score, + ) + + async def node_fulltext_search( + self, + driver: Any, + query: str, + search_filter: Any, + group_ids: list[str] | None = None, + limit: int = 100, + ) -> list[Any]: + return await self._ops.node_fulltext_search(driver, query, search_filter, group_ids, limit) + + async def node_similarity_search( + self, + driver: Any, + search_vector: list[float], + search_filter: Any, + group_ids: list[str] | None = None, + limit: int = 100, + min_score: float = 0.7, + ) -> list[Any]: + return await self._ops.node_similarity_search( + driver, search_vector, search_filter, group_ids, limit, min_score + ) + + async def episode_fulltext_search( + self, + driver: Any, + query: str, + search_filter: Any, + group_ids: list[str] | None = None, + limit: int = 100, + ) -> list[Any]: + return await self._ops.episode_fulltext_search( + driver, query, search_filter, group_ids, limit + ) + + async def community_fulltext_search( + self, + driver: Any, + query: str, + group_ids: list[str] | None = None, + limit: int = 100, + ) -> list[Any]: + return await self._ops.community_fulltext_search(driver, query, group_ids, limit) diff --git a/graphiti_core/driver/arcadedb_driver.py b/graphiti_core/driver/arcadedb_driver.py index 13f370ece8..970581a19c 100644 --- a/graphiti_core/driver/arcadedb_driver.py +++ b/graphiti_core/driver/arcadedb_driver.py @@ -43,6 +43,7 @@ ) from graphiti_core.driver.arcadedb.operations.saga_node_ops import ArcadeDBSagaNodeOperations from graphiti_core.driver.arcadedb.operations.search_ops import ArcadeDBSearchOperations +from graphiti_core.driver.arcadedb.search_interface import ArcadeDBSearchInterface from graphiti_core.driver.driver import GraphDriver, GraphDriverSession, GraphProvider from graphiti_core.driver.operations.community_edge_ops import CommunityEdgeOperations from graphiti_core.driver.operations.community_node_ops import CommunityNodeOperations @@ -57,6 +58,7 @@ from graphiti_core.driver.operations.search_ops import SearchOperations from graphiti_core.driver.query_executor import Transaction from graphiti_core.graph_queries import get_fulltext_indices, get_range_indices +from graphiti_core.utils.datetime_utils import convert_datetimes_to_strings logger = logging.getLogger(__name__) @@ -99,6 +101,7 @@ def __init__( self._next_episode_edge_ops = ArcadeDBNextEpisodeEdgeOperations() self._search_ops = ArcadeDBSearchOperations() self._graph_ops = ArcadeDBGraphMaintenanceOperations() + self.search_interface = ArcadeDBSearchInterface() self.aoss_client = None @@ -164,10 +167,26 @@ async def execute_query(self, cypher_query_: LiteralString, **kwargs: Any) -> Ea params = kwargs.pop('params', None) if params is None: params = {} - kwargs.setdefault('database_', self._database) + database_ = kwargs.pop('database_', self._database) + + # ArcadeDB's Bolt plugin stores a native datetime write as a + # zone-less ChronoLocalDateTime, but an incoming native datetime + # query parameter is decoded as a zone-aware OffsetDateTime -- the + # two are not directly comparable, and Cypher's implicit WHERE/ORDER + # BY coercion between them raises a ClassCastException at the engine + # level (confirmed against a real instance: reproducible whenever a + # range index on the property was created before any data existed, + # which is exactly what build_indices_and_constraints() always does). + # Serializing to an ISO string here, combined with the ARCADEDB-only + # datetime(...) wrap in retrieve_episodes(), sidesteps it. + params = convert_datetimes_to_strings(params) + query_params = convert_datetimes_to_strings(kwargs) + query_params['database_'] = database_ try: - result = await self.client.execute_query(cypher_query_, parameters_=params, **kwargs) + result = await self.client.execute_query( + cypher_query_, parameters_=params, **query_params + ) except Exception as e: logger.error(f'Error executing ArcadeDB query: {e}\n{cypher_query_}\n{params}') raise