orchestrata_ultimate.knowledge_graph.search

Reference › orchestrata_ultimate › knowledge_graph › search

SearchEngine provides full-text, type-filtered and relationship-based search over the operational knowledge graph (orchkg on orchgraph), through a graph_client GraphClient. It is traced to Spec 008 REQ-008-031 to -035. full_text_search returns an empty list when the client is not connected. It is engine-aware: resolve_kg_dialect is consulted, and because orchgraph has no db.index.fulltext.queryNodes procedure (measured live in July 2026), on the memgraph dialect the substring search _contains_search is the primary path. On the Neo4j dialect it calls the search_names full-text index and falls back to the substring search with a logged warning if that call fails. An explicit dialect argument overrides the resolution. search_by_type restricts matches to one node type, search_by_relationship lists the neighbours of a node along a relationship type in either direction, and search_count reports how many searches have run. The module is the search half of the read surface next to query_engine.

At a glance

Class diagram

Dependency diagram

API

Classes

SearchEngine

Decorators: traces(spec='008', story='US-001', reqs=['REQ-008-031', 'REQ-008-032', 'REQ-008-033', 'REQ-008-034', 'REQ-008-035']) · Line 33

Full-text and filtered search for the knowledge graph.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/knowledge_graph/search.py

# mypy: ignore-errors
"""Search engine for the knowledge graph.

Provides full-text search, type-filtered search, and
relationship-based search capabilities.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph_factory import resolve_kg_dialect

if TYPE_CHECKING:
    from orchestrata_ultimate.knowledge_graph.graph_client import GraphClient

logger = logging.getLogger(__name__)


@traces(
    spec="008",
    story="US-001",
    reqs=[
        "REQ-008-031",
        "REQ-008-032",
        "REQ-008-033",
        "REQ-008-034",
        "REQ-008-035",
    ],
)
class SearchEngine:
    """Full-text and filtered search for the knowledge graph."""

    def __init__(self, graph_client: GraphClient | None = None) -> None:
        self._client = graph_client
        self._search_count = 0

    async def full_text_search(
        self, query: str, limit: int = 100, dialect: str | None = None
    ) -> list[dict[str, Any]]:
        """Full-text search across all indexed nodes.

        Engine-aware: ``db.index.fulltext.queryNodes`` is Neo4j-only —
        orchgraph answers "There is no procedure named
        'db.index.fulltext.queryNodes'" (measured live 2026-07-26). The
        ``except`` path below would still reach the CONTAINS fallback, but
        only after a guaranteed-failing round-trip and a WARNING per call,
        so on Memgraph the substring search IS the primary path.

        Args:
            dialect: Optional override for ``resolve_kg_dialect`` -- "an
                explicit dialect argument always wins" there too. Exists so
                tests can force the Neo4j branch open against a live
                Memgraph and prove the except-fallback really fires,
                without env hacking or a mock.
        """
        if not self._client or not self._client.connected:
            return []

        self._search_count += 1
        if resolve_kg_dialect(dialect) == "memgraph":
            return await self._contains_search(query, limit)
        cypher = (
            "CALL db.index.fulltext.queryNodes('search_names', $query) "
            "YIELD node, score "
            "RETURN properties(node) AS props, score "
            "LIMIT $limit"
        )
        try:
            return await self._client.query(cypher, {"query": query, "limit": limit})
        except Exception as exc:
            logger.warning(
                "Full-text index search failed, falling back to CONTAINS search: %s",
                exc,
                exc_info=True,
            )
            return await self._contains_search(query, limit)

    async def search_by_type(
        self, query: str, node_type: str, limit: int = 100
    ) -> list[dict[str, Any]]:
        """Search nodes of a specific type."""
        if not self._client or not self._client.connected:
            return []

        from orchestrata_ultimate.knowledge_graph.schema import get_node_label

        label = get_node_label(node_type)
        self._search_count += 1

        cypher = (
            f"MATCH (n:{label}) "
            "WHERE n.name CONTAINS $query "
            "RETURN properties(n) AS props "
            "LIMIT $limit"
        )
        return await self._client.query(cypher, {"query": query, "limit": limit})

    async def search_by_relationship(
        self,
        node_id: str,
        rel_type: str,
        direction: str = "outgoing",
        limit: int = 100,
        node_type: str | None = None,
    ) -> list[dict[str, Any]]:
        """Find nodes connected by a specific relationship.

        node_type: optional type of the ANCHOR node (the one identified by
        ``node_id``) so its side of the MATCH plans a labeled index seek
        instead of an unlabeled AllNodesScan -- see
        graph_client.GraphClient.get_node for the identical trade. The
        OTHER side of the relationship is intentionally left unlabeled:
        its type is exactly what this search is trying to discover.
        """
        if not self._client or not self._client.connected:
            return []

        self._search_count += 1

        anchor_label = ""
        if node_type is not None:
            from orchestrata_ultimate.knowledge_graph.schema import get_node_label

            anchor_label = f":{get_node_label(node_type)}"

        if direction == "incoming":
            # Incoming: nodes that point TO the given node via rel_type.
            cypher = (
                f"MATCH (n)-[:{rel_type}]->(related{anchor_label} {{id: $node_id}}) "
                "RETURN properties(n) AS props LIMIT $limit"
            )
        else:
            # Outgoing: nodes that the given node points to via rel_type.
            cypher = (
                f"MATCH (n{anchor_label} {{id: $node_id}})-[:{rel_type}]->(related) "
                "RETURN properties(related) AS props LIMIT $limit"
            )
        return await self._client.query(cypher, {"node_id": node_id, "limit": limit})

    async def _contains_search(self, query: str, limit: int) -> list[dict[str, Any]]:
        """Fallback CONTAINS-based search over the fulltext index's universe.

        Scoped to the SAME labels the ``search_names`` fulltext index covers
        (Class|Function|Predicate) — the fallback must not return nodes the
        primary path could never return. Per-label UNION branches let each
        branch use its TEXT index (a label-less ``MATCH (n) WHERE n.name
        CONTAINS`` cannot use ANY index and full-scans the ~800k-node store).
        """
        # Guard mirrors the sibling public methods (full_text_search etc.):
        # the sole caller (full_text_search) already checks this, but the
        # guard can't be seen across methods, so it's repeated here.
        if not self._client or not self._client.connected:
            return []
        from orchestrata_ultimate.knowledge_graph.schema import (
            LABEL_CLASS,
            LABEL_FUNCTION,
            LABEL_PREDICATE,
        )

        cypher = (
            "CALL () { "
            f"MATCH (n:{LABEL_CLASS}) WHERE n.name CONTAINS $query RETURN n "
            f"UNION MATCH (n:{LABEL_FUNCTION}) WHERE n.name CONTAINS $query RETURN n "
            f"UNION MATCH (n:{LABEL_PREDICATE}) WHERE n.name CONTAINS $query RETURN n "
            "} WITH n "
            "RETURN properties(n) AS props LIMIT $limit"
        )
        return await self._client.query(cypher, {"query": query, "limit": limit})

    @property
    def search_count(self) -> int:
        return self._search_count