fix(arcadedb): wire up fulltext/similarity search (BM25 currently crashes)

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

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)

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