orchestrata_ultimate.knowledge_graph.semantic_bridge

Reference › orchestrata_ultimate › knowledge_graph › semantic_bridge

The bridge between the two graphs (Spec 107, REQ-107-015). It reads text chunks from the shared semantic/RAG graph, the memgraph database described in the semantic_graph package, and maps their file paths back to File root entities in the operational graph, the orchkg database on orchgraph. The neo4j driver import is Bolt-client naming only: Memgraph speaks Bolt. Everything is synchronous because the context assembler and kernel pipeline are.

search_semantic_roots builds search terms from the query text and the basenames of the target files (_file_name_terms, capped at max_terms), opens a sync driver with _memgraph_driver from SemanticGraphConfig and verifies connectivity. _query_memgraph_chunks returns chunks whose content contains a term, limited per term. _query_neo4j_files looks the chunk paths up as File nodes through a SyncGraphRepository (created lazily when none is passed), optionally scoped to a project. The result is a list of SemanticRoot objects (path, line count, up to a few chunks, and a snippet helper). If either graph is unreachable the function logs at debug level and returns an empty list instead of raising. render_semantic_roots formats roots as the legacy markdown block. The context provider semantic_context and kernel code import it.

At a glance

Class diagram

Dependency diagram

API

Classes

SemanticRoot (dataclass)

Decorators: dataclass · Line 126

One Neo4j File root matched by semantic chunk search.

Functions

Source

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

"""Semantic-search bridge: Memgraph chunks → Neo4j file roots.

Reusable component that queries the Memgraph semantic-graph chunk store,
maps chunk file paths back to operational Neo4j ``File`` root entities, and
returns structured roots for context injection. The provider layer
(``context/providers/semantic_context.py``) renders these roots as markdown.

All backend access is synchronous: the unified assembler and kernel pipeline
are synchronous, and the Memgraph adapter normally exposes an async interface.
The bridge therefore uses the sync Neo4j Bolt driver directly (Memgraph speaks
Bolt) and the operational graph's ``SyncGraphRepository`` seam for Neo4j lookups.

If either graph is unreachable, the bridge degrades to an empty list
rather than raising.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any

from neo4j import basic_auth

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph.repository import create_graph_repository
from orchestrata_ultimate.core.graph_factory import NO_AUTH, create_graph
from orchestrata_ultimate.semantic_graph.config import SemanticGraphConfig
from orchestrata_ultimate.semantic_graph.ingestion import _extract_query_terms

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph.interfaces import SyncGraphRepository

_logger = logging.getLogger(__name__)


def _memgraph_driver(config: SemanticGraphConfig) -> Any:
    """Create a sync Bolt driver for Memgraph.

    Memgraph nodes are normally unauthenticated; ``NO_AUTH`` is passed
    explicitly so the factory does not fall back to NEO4J_USERNAME/
    NEO4J_PASSWORD credentials.
    """
    auth: Any = NO_AUTH
    if config.username or config.password:
        auth = basic_auth(config.username, config.password)
    return create_graph(uri=config.uri, auth=auth)


def _file_name_terms(paths: list[str]) -> list[str]:
    """Extract search terms from target-file basenames/stems."""
    terms: list[str] = []
    for path in paths:
        if not path:
            continue
        base = path.split("/")[-1]
        stem = base.split(".")[0]
        if stem:
            terms.append(stem)
    return terms


def _query_memgraph_chunks(
    driver: Any,
    chunk_label: str,
    terms: list[str],
    limit_per_term: int,
) -> list[dict[str, Any]]:
    """Return chunk records whose content contains any query term."""
    results: list[dict[str, Any]] = []
    seen_ids: set[str] = set()
    cypher = (
        f"MATCH (c:{chunk_label}) "
        "WHERE toLower(c.content) CONTAINS $term "
        "RETURN c.id AS id, c.content AS content, c.file_path AS file_path "
        "LIMIT $limit"
    )
    with driver.session() as session:
        for term in terms:
            try:
                records = session.run(cypher, {"term": term, "limit": limit_per_term})
                for record in records:
                    cid = record["id"]
                    if cid in seen_ids:
                        continue
                    seen_ids.add(cid)
                    results.append(dict(record))
            except Exception as exc:  # pragma: no cover - degrade per-term
                _logger.debug(
                    "Semantic keyword query failed for term %r: %s", term, exc
                )
    return results


def _query_neo4j_files(
    repo: SyncGraphRepository,
    paths: list[str],
    project_id: str | None,
) -> list[dict[str, Any]]:
    """Return ``File`` root-entity records for the given paths."""
    if not paths:
        return []
    or_clauses: list[str] = ["f.path IN $paths"]
    params: dict[str, Any] = {"paths": paths}
    if project_id is not None:
        or_clauses.append("f.id IN $ids")
        params["ids"] = [f"{project_id}:{p}" for p in paths]
    where_clause = "(" + " OR ".join(or_clauses) + ")"
    if project_id is not None:
        where_clause += " AND f.project_id = $project_id"
        params["project_id"] = project_id
    cypher = (
        "MATCH (f:File) "
        f"WHERE {where_clause} "
        "RETURN f.path AS path, f.line_count AS lines "
        "LIMIT 20"
    )
    try:
        return repo.execute_read(cypher, params)
    except Exception as exc:
        _logger.debug("graph file lookup failed: %s", exc)
        return []


@dataclass
class SemanticRoot:
    """One Neo4j ``File`` root matched by semantic chunk search."""

    path: str
    line_count: int | None = None
    chunks: list[dict[str, Any]] = field(default_factory=list)

    def snippet(self) -> str:
        """Return the first line of the first chunk, truncated."""
        if not self.chunks:
            return ""
        content = self.chunks[0].get("content", "")
        first_line = content.splitlines()[0] if content else ""
        if len(first_line) > 120:
            first_line = first_line[:117] + "..."
        return first_line


@traces(spec="107", reqs=["REQ-107-015"])
def search_semantic_roots(
    query: str,
    target_files: list[str],
    *,
    project_id: str | None = None,
    config: SemanticGraphConfig | None = None,
    graph_repo: SyncGraphRepository | None = None,
    max_terms: int = 6,
    limit_per_term: int = 5,
    max_chunks_per_file: int = 3,
) -> list[SemanticRoot]:
    """Search Memgraph chunks and map them back to the graph ``File`` roots.

    Args:
        query: Raw user/intent text used to extract keyword terms.
        target_files: Files the volk is being asked to act on; their basenames
            are also used as search terms.
        project_id: Optional project scope for Neo4j ``File`` lookups.
        config: Semantic-graph config (Memgraph URI, chunk label). Defaults to
            ``SemanticGraphConfig.from_env()``.
        neo4j_repo: Optional operational-graph repository for mapping chunk
            file paths back to ``File`` root entities. When omitted, a default
            repository is created lazily.
        max_terms: Maximum keyword terms extracted from the combined query.
        limit_per_term: Maximum chunks returned per term.
        max_chunks_per_file: Maximum chunks kept per matched root file.

    Returns:
        A list of :class:`SemanticRoot` records, one per Neo4j ``File`` that
        was matched. Empty when no relevant chunks are found or any backend is
        unreachable.
    """
    query_text = query + " " + " ".join(_file_name_terms(target_files))
    terms = _extract_query_terms(query_text, max_terms=max_terms)
    if not terms:
        return []

    config = config or SemanticGraphConfig.from_env()
    # `Any` (not `Any | None`): pyright treats an explicit `| None` member as
    # always-live for reportOptionalMemberAccess even after an `Any`-typed
    # reassignment (assigning `Any` doesn't narrow away a declared `None`
    # union member). `_memgraph_driver()` -> `Any` already covers "unset".
    memgraph_driver: Any = None
    try:
        memgraph_driver = _memgraph_driver(config)
        memgraph_driver.verify_connectivity()
    except Exception as exc:
        _logger.debug("Memgraph unreachable; semantic bridge degraded: %s", exc)
        if memgraph_driver is not None:
            try:
                memgraph_driver.close()
            except Exception as close_exc:  # pragma: no cover - defensive cleanup
                _logger.debug("Failed to close Memgraph driver: %s", close_exc)
        return []

    try:
        chunks = _query_memgraph_chunks(
            memgraph_driver,
            config.chunk_label,
            terms,
            limit_per_term=limit_per_term,
        )
        if not chunks:
            return []

        chunk_paths = [c["file_path"] for c in chunks if c.get("file_path")]
        repo = graph_repo or create_graph_repository()
        file_records = _query_neo4j_files(repo, chunk_paths, project_id)

        by_path: dict[str, SemanticRoot] = {}
        record_by_path = {r["path"]: r for r in file_records if r.get("path")}
        for chunk in chunks:
            path = chunk.get("file_path")
            if not path:
                continue
            if path not in by_path:
                lines = record_by_path.get(path, {}).get("lines")
                by_path[path] = SemanticRoot(path=path, line_count=lines)
            if len(by_path[path].chunks) < max_chunks_per_file:
                by_path[path].chunks.append(chunk)

        return list(by_path.values())
    except Exception as exc:
        _logger.warning("Semantic bridge assembly failed: %s", exc)
        return []
    finally:
        if memgraph_driver is not None:
            try:
                memgraph_driver.close()
            except Exception as close_exc:  # pragma: no cover - defensive cleanup
                _logger.debug("Failed to close Memgraph driver: %s", close_exc)


@traces(spec="107", reqs=["REQ-107-015"])
def render_semantic_roots(
    roots: list[SemanticRoot],
    terms: list[str],
    *,
    max_lines: int = 30,
) -> str:
    """Render semantic roots as the legacy markdown block.

    Preserves the output shape of the original ``render_semantic_context``
    provider so callers can swap to the bridge without changing their
    assertions.
    """
    if not roots:
        return ""

    lines: list[str] = ["# SEMANTIC CONTEXT"]
    lines.append(
        f"Search terms: {', '.join(f'`{t}`' for t in terms)} | "
        f"chunks: {sum(len(r.chunks) for r in roots)} | matched files: {len(roots)}"
    )

    for root in roots[:5]:
        in_kg = "✓" if root.line_count is not None else "?"
        first_line = root.snippet()
        lines.append(f"- [{in_kg}] `{root.path}`: {first_line}")

    if any(r.line_count is not None for r in roots):
        lines.append("## Matched root files")
        for root in roots[:5]:
            if root.line_count is None:
                continue
            lines.append(f"- `{root.path}` ({root.line_count} lines)")

    while len(lines) > max_lines:
        lines.pop()

    if len(lines) <= 2:
        return ""

    return "\n".join(lines)