orchestrata_ultimate.semantic_graph.interfaces

Reference › orchestrata_ultimate › semantic_graph › interfaces

The repository contracts for the semantic graph, written as runtime-checkable Protocol classes so adapters can be swapped without inheritance. VectorStore stores chunks (add_chunks) and finds the top-k by embedding similarity (search_similar). GraphStore upserts entities and relationships, returns neighbours within a depth (get_neighbors) and can clear all data for tests. SemanticGraph inherits both and adds connect, close and retrieve, which assembles a GraphRAGContext for a query embedding. The two adapters, memgraph for the production memgraph database and memory for tests, satisfy them, and repository selects one. Data types are in models.

At a glance

Class diagram

Dependency diagram

API

Classes

VectorStore (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 19

Stores and searches text chunks by embedding similarity.

GraphStore (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 34

Stores and traverses entities and relationships.

SemanticGraph (protocol)

Bases: VectorStore, GraphStore, Protocol · Decorators: runtime_checkable · Line 57

Combined interface for RAG/GraphRAG backends.

Source

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

"""Repository interfaces for semantic graph / RAG / GraphRAG.

The interfaces are Protocols so adapters can be swapped without inheritance.
"""

from __future__ import annotations

from typing import Protocol, runtime_checkable

from orchestrata_ultimate.semantic_graph.models import (
    Chunk,
    Entity,
    GraphRAGContext,
    Relationship,
)


@runtime_checkable
class VectorStore(Protocol):
    """Stores and searches text chunks by embedding similarity."""

    async def add_chunks(self, chunks: list[Chunk]) -> None:
        """Store chunks in the vector store."""
        ...  # pragma: no cover - protocol method body is not executable

    async def search_similar(
        self, embedding: list[float], top_k: int = 5
    ) -> list[Chunk]:
        """Return the top-k chunks most similar to the query embedding."""
        ...  # pragma: no cover - protocol method body is not executable


@runtime_checkable
class GraphStore(Protocol):
    """Stores and traverses entities and relationships."""

    async def upsert_entities(self, entities: list[Entity]) -> None:
        """Create or update entities in the graph."""
        ...  # pragma: no cover - protocol method body is not executable

    async def upsert_relationships(self, relationships: list[Relationship]) -> None:
        """Create or update relationships in the graph."""
        ...  # pragma: no cover - protocol method body is not executable

    async def get_neighbors(
        self, entity_id: str, depth: int = 1
    ) -> tuple[list[Entity], list[Relationship]]:
        """Return entities and relationships reachable within ``depth`` hops."""
        ...  # pragma: no cover - protocol method body is not executable

    async def clear(self) -> None:
        """Remove all data from the graph store (useful for tests)."""
        ...  # pragma: no cover - protocol method body is not executable


@runtime_checkable
class SemanticGraph(VectorStore, GraphStore, Protocol):
    """Combined interface for RAG/GraphRAG backends."""

    async def connect(self) -> bool:
        """Verify connectivity to the backend.

        Returns True if healthy, False otherwise. Implementations should not
        raise for transient connection failures.
        """
        ...  # pragma: no cover - protocol method body is not executable

    async def close(self) -> None:
        """Close any underlying connections."""
        ...  # pragma: no cover - protocol method body is not executable

    async def retrieve(
        self,
        query_embedding: list[float],
        top_k: int = 5,
        graph_depth: int = 1,
    ) -> GraphRAGContext:
        """Assemble a GraphRAG context for the query embedding.

        Default strategy:
        1. Vector-search for top-k chunks.
        2. Traverse the graph around entities referenced by those chunks.
        3. Return the combined context.
        """
        ...  # pragma: no cover - protocol method body is not executable