orchestrata_ultimate.semantic_graph.models

Reference › orchestrata_ultimate › semantic_graph › models

Lightweight dataclasses used by the repository interfaces in interfaces. They deliberately depend on nothing in knowledge_graph, so the package can be reused with other backends. Entity is a node (id, labels, properties), Relationship is an edge (id, type, source and target ids, properties), Chunk is a text chunk with an optional embedding vector and metadata, and GraphRAGContext is the bundle that SemanticGraph.retrieve returns for a prompt: the query plus its entities, relationships and chunks. Each of the first three runs a __post_init__ check on construction. Both adapters and the ingestion code use these types for the shared memgraph semantic graph.

At a glance

Class diagram

Dependency diagram

API

Classes

Entity (dataclass)

Decorators: dataclass · Line 15

A node in the semantic graph.

Relationship (dataclass)

Decorators: dataclass · Line 28

An edge between two entities in the semantic graph.

Chunk (dataclass)

Decorators: dataclass · Line 47

A text chunk with an optional embedding vector.

GraphRAGContext (dataclass)

Decorators: dataclass · Line 61

Context bundle returned by SemanticGraph.retrieve for GraphRAG.

Source

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

"""Data models for semantic graph / RAG / GraphRAG.

Lightweight dataclasses used by the repository interfaces. They intentionally
avoid upward dependencies on orchestrata_ultimate.knowledge_graph so the
semantic_graph package can be reused with different backends.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any


@dataclass
class Entity:
    """A node in the semantic graph."""

    id: str
    labels: list[str] = field(default_factory=list)
    properties: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.id:
            raise ValueError("Entity.id must be non-empty")


@dataclass
class Relationship:
    """An edge between two entities in the semantic graph."""

    id: str
    type: str
    source_id: str
    target_id: str
    properties: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.id:
            raise ValueError("Relationship.id must be non-empty")
        if not self.type:
            raise ValueError("Relationship.type must be non-empty")
        if not self.source_id or not self.target_id:
            raise ValueError("Relationship source and target ids must be non-empty")


@dataclass
class Chunk:
    """A text chunk with an optional embedding vector."""

    id: str
    content: str
    embedding: list[float] | None = None
    metadata: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        if not self.id:
            raise ValueError("Chunk.id must be non-empty")


@dataclass
class GraphRAGContext:
    """Context bundle returned by SemanticGraph.retrieve for GraphRAG."""

    query: str = ""
    entities: list[Entity] = field(default_factory=list)
    relationships: list[Relationship] = field(default_factory=list)
    chunks: list[Chunk] = field(default_factory=list)

    def is_empty(self) -> bool:
        return not (self.entities or self.relationships or self.chunks)