orchestrata_ultimate.semantic_graph.config

Reference › orchestrata_ultimate › semantic_graph › config

Connection and tuning configuration for semantic graph backends, resolved from environment variables. SemanticGraphConfig holds the backend name, Bolt URI and credentials, the vector index settings (name chunk_embedding_idx, cosine metric, capacity 100000, timeout 10 s), the chunk label (CodestralChunk by default), an optional HA controller URL, a list of candidate URIs, an optional database name, and the hybrid retrieval weights: 0.6 for vector similarity, 0.4 for keyword matches, caps on keyword terms and candidates, and small module and prominence boosts. from_env reads the keys, whose names default to SEMANTIC_GRAPH_BACKEND, MEMGRAPH_URI, MEMGRAPH_USERNAME and MEMGRAPH_PASSWORD and can be overridden by argument. Under RUL-003 the URI has no default: a missing or empty MEMGRAPH_URI raises instead of falling back to a localhost address, because nothing in this infrastructure listens there and the default would only disguise a configuration gap as an outage. The class describes the shared memgraph semantic graph. repository, the adapters and semantic_bridge use it.

At a glance

Class diagram

Dependency diagram

API

Classes

SemanticGraphConfig (dataclass)

Decorators: dataclass · Line 17

Connection config for a semantic graph backend.

Source

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

"""Configuration for semantic graph backends.

Values are resolved from environment variables. The connection URI has no
default (RUL-003): a missing/empty ``MEMGRAPH_URI`` raises rather than
silently falling back to ``bolt://localhost:7687`` — nothing in this
infrastructure listens on localhost, so that default could only ever turn a
config gap into a connection-refused that reads like an outage.
"""

from __future__ import annotations

import os
from dataclasses import dataclass, field


@dataclass
class SemanticGraphConfig:
    """Connection config for a semantic graph backend."""

    backend: str
    uri: str
    username: str
    password: str
    vector_index_name: str = "chunk_embedding_idx"
    vector_index_metric: str = "cos"
    vector_index_capacity: int = 100_000
    vector_index_timeout: float = 10.0
    chunk_label: str = "CodestralChunk"
    ha_controller_url: str | None = None
    uris: list[str] = field(default_factory=list)
    # Multi-tenancy: when set, every session the adapter opens targets this
    # database instead of the server default. Test infrastructure uses it to
    # give each parallel worker its OWN index/label universe on the shared
    # cluster (the same orchkg_w<N> pattern run_tests provisions for the KG),
    # which kills the cross-worker "deleted object" settle-window class
    # entirely (2026-08-11 analysis).
    database: str | None = None

    # Hybrid retrieval knobs (audit #2/#4): vector + keyword + module/entity
    # boosts, fused via weighted reciprocal-rank fusion (see ingestion.py).
    hybrid_vector_weight: float = 0.6
    hybrid_keyword_weight: float = 0.4
    hybrid_keyword_term_cap: int = 8
    hybrid_keyword_candidate_cap: int = 50
    hybrid_keyword_extra_cap: int = 5
    hybrid_module_boost: float = 0.02
    hybrid_prominence_boost: float = 0.01

    @classmethod
    def from_env(
        cls,
        backend_key: str = "SEMANTIC_GRAPH_BACKEND",
        uri_key: str = "MEMGRAPH_URI",
        username_key: str = "MEMGRAPH_USERNAME",
        password_key: str = "MEMGRAPH_PASSWORD",
        vector_index_name_key: str = "MEMGRAPH_VECTOR_INDEX_NAME",
        vector_index_metric_key: str = "MEMGRAPH_VECTOR_INDEX_METRIC",
        vector_index_capacity_key: str = "MEMGRAPH_VECTOR_INDEX_CAPACITY",
        vector_index_timeout_key: str = "MEMGRAPH_VECTOR_INDEX_TIMEOUT",
        chunk_label_key: str = "MEMGRAPH_CHUNK_LABEL",
        ha_controller_url_key: str = "MEMGRAPH_HA_CONTROLLER_URL",
        uris_key: str = "MEMGRAPH_URIS",
        default_backend: str = "memgraph",
    ) -> SemanticGraphConfig:
        """Build config from environment variables.

        Args:
            backend_key: env var name for the adapter backend to use.
            uri_key: env var name for the Bolt URI.
            username_key: env var name for the username.
            password_key: env var name for the password.
            vector_index_name_key: env var name for the vector index name.
            vector_index_metric_key: env var name for the vector similarity metric.
            vector_index_capacity_key: env var name for the vector index capacity.
            ha_controller_url_key: env var name for the HA controller HTTP URL.
            uris_key: env var name for comma-separated fallback Bolt URIs.
            default_backend: fallback backend when the env var is unset.

        Raises:
            RuntimeError: when *uri_key* is unset or empty — this resolver has
                no default address (RUL-003).
        """
        uri = os.environ.get(uri_key)
        if not uri:
            raise RuntimeError(
                f"No graph URI configured — set {uri_key} (see .env); this "
                "code has no default address. It used to fall back to "
                "bolt://localhost:7687, which nothing in this "
                "infrastructure listens on."
            )
        uris_raw = os.environ.get(uris_key, "")
        uris = [u.strip() for u in uris_raw.split(",") if u.strip()]
        return cls(
            backend=os.environ.get(backend_key, default_backend).lower(),
            uri=uri,
            username=os.environ.get(username_key, ""),
            password=os.environ.get(password_key, ""),
            vector_index_name=os.environ.get(
                vector_index_name_key, "chunk_embedding_idx"
            ),
            vector_index_metric=os.environ.get(vector_index_metric_key, "cos"),
            vector_index_capacity=int(
                os.environ.get(vector_index_capacity_key, "100000")
            ),
            vector_index_timeout=float(
                os.environ.get(vector_index_timeout_key, "10.0")
            ),
            chunk_label=os.environ.get(chunk_label_key, "CodestralChunk"),
            ha_controller_url=os.environ.get(ha_controller_url_key) or None,
            uris=uris,
            hybrid_vector_weight=float(
                os.environ.get("SEMANTIC_GRAPH_HYBRID_VECTOR_WEIGHT", "0.6")
            ),
            hybrid_keyword_weight=float(
                os.environ.get("SEMANTIC_GRAPH_HYBRID_KEYWORD_WEIGHT", "0.4")
            ),
            hybrid_keyword_term_cap=int(
                os.environ.get("SEMANTIC_GRAPH_HYBRID_KEYWORD_TERM_CAP", "8")
            ),
            hybrid_keyword_candidate_cap=int(
                os.environ.get("SEMANTIC_GRAPH_HYBRID_KEYWORD_CANDIDATE_CAP", "50")
            ),
            hybrid_keyword_extra_cap=int(
                os.environ.get("SEMANTIC_GRAPH_HYBRID_KEYWORD_EXTRA_CAP", "5")
            ),
            hybrid_module_boost=float(
                os.environ.get("SEMANTIC_GRAPH_HYBRID_MODULE_BOOST", "0.02")
            ),
            hybrid_prominence_boost=float(
                os.environ.get("SEMANTIC_GRAPH_HYBRID_PROMINENCE_BOOST", "0.01")
            ),
        )