orchestrata_ultimate.llm.config

Reference › orchestrata_ultimate › llm › config

config decides which provider and model each volk uses. It holds no hardcoded assignments: the data comes from the YAML profile file spike-profiles.yaml, found by walking upward from the module (_find_profiles_path) and loaded into a module-level table by _load_profiles. The active profile is named by the SPIKE_PROFILE environment variable and defaults to default (_resolve_profile_name). client calls it on every request, and the code drivers import it too. The curated narrative is llm.

VolkConfig is a dataclass that loads the assignments of a profile; get_provider and get_model look an alias up and fall back to the profile's default entry, raising LLMConfigError only when neither exists. known_agents lists the aliases. A singleton is created lazily by _get_config, and the public wrappers are get_volk_provider, get_volk_model, get_active_profile, list_profiles and select_profile, which switches profile at runtime. PROVIDER_MODELS records the standard and reasoning model per provider. escalate_agent_model raises an agent's model from the standard to the reasoning tier for the rest of the session by updating the in-memory assignment; for OpenRouter it uses OPENROUTER_MODEL_ESCALATIONS to stay within the model's own family, and it returns the current model unchanged when it is already at the top tier.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

LLMConfigError

Bases: Exception · Line 89

Configuration error — missing profile or invalid YAML.

VolkConfig (dataclass)

Decorators: dataclass · Line 220

Per-volk LLM provider and model configuration.

Functions

Source

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

"""Per-volk LLM provider and model configuration.

The single source of truth is ``configs/spike-profiles.yaml``.
No hardcoded provider/model defaults. No env variable overrides.

Profile selection: ``SPIKE_PROFILE`` env var → ``"default"`` profile.
System raises ``LLMConfigError`` if no profile is found.

Available profiles are defined in ``configs/spike-profiles.yaml``:
    default    — all agents use deepseek
    claude     — all agents use anthropic/claude
    kimi       — all agents use kimi via openrouter
    budget     — all agents use ollama (local, free)
    mix        — best-of-breed per agent (anthropic + deepseek)

Per-agent reasoning tiers: Architect (U:Ar) and SeniorDev (U:SD) get
reasoning-tier models. All other agents get standard-tier models.

Usage:
    from orchestrata_ultimate.llm.config import get_volk_provider, get_volk_model

    provider = get_volk_provider("U:Cx")   # → "deepseek"
    model = get_volk_model("U:Ar")          # → "deepseek-v4-pro"
"""

from __future__ import annotations

import logging
import os
from collections.abc import Mapping
from dataclasses import dataclass, field

import yaml

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)

# Agent aliases that use reasoning-tier models
REASONING_AGENTS = frozenset({"U:Ar", "U:SD"})

# Provider → {standard_model, reasoning_model}
# Each entry in spike-profiles.yaml must reference a provider listed here.
PROVIDER_MODELS: dict[str, dict[str, str]] = {
    "deepseek": {
        "standard": "deepseek-v4-flash",
        "reasoning": "deepseek-v4-pro",
    },
    "claude": {
        "standard": "sonnet",
        "reasoning": "opus",
    },
    # anthropic: the native Anthropic API adapter (distinct from "claude",
    # the Claude CLI/OAuth provider above) -- used by the B8 design_authority
    # profile entry (configs/spike-profiles.yaml), always pinned to the
    # reasoning tier there, but both tiers are declared so escalation and
    # test_provider_models_match_profiles stay honest about what this
    # provider can serve.
    "anthropic": {
        "standard": "claude-sonnet-5",
        "reasoning": "claude-opus-4-8",
    },
    "openrouter": {
        "standard": "qwen/qwen3.7-plus",
        "reasoning": "qwen/qwen3.7-max",
    },
    "ollama": {
        "standard": "qwen3:1.7b",
        "reasoning": "qwen3:1.7b",
    },
}

# OpenRouter hosts many model FAMILIES behind one provider, so the
# provider-level standard→reasoning pair above is wrong for it: escalating
# a deepseek-flash volk must yield deepseek-pro, not a jump to another
# family. Escalation on openrouter is therefore family-aware: models listed
# here escalate within their family; models absent from this map are
# already at the top of theirs (noop).
OPENROUTER_MODEL_ESCALATIONS: dict[str, str] = {
    "deepseek/deepseek-v4-flash": "deepseek/deepseek-v4-pro",
    "qwen/qwen3.7-plus": "qwen/qwen3.7-max",
    "qwen/qwen3-coder-next": "qwen/qwen3.7-max",
}


# ── Profile loading ─────────────────────────────────────────────────


class LLMConfigError(Exception):
    """Configuration error — missing profile or invalid YAML."""


_INSTALLED_PROFILES_PATH = "/opt/orchestrata/configs/spike-profiles.yaml"


def _find_profiles_path(fallback: str = _INSTALLED_PROFILES_PATH) -> str:
    """Locate spike-profiles.yaml by walking upward from this file.

    *fallback* is the well-known install path tried last. It is a parameter so
    a caller (a test, in practice) can point it at a path known to be absent:
    on a host where /opt/orchestrata IS bootstrapped the default exists, and a
    "not found" check that depends on the host's deployment state is not a
    check (measured 2026-10-05).
    """
    from pathlib import Path

    here = Path(__file__).resolve()
    for parent in here.parents:
        candidate = parent / "configs" / "spike-profiles.yaml"
        if candidate.exists():
            return str(candidate)
    # Fallback to well-known install path
    if Path(fallback).exists():
        return fallback
    raise LLMConfigError(
        "spike-profiles.yaml not found. Create configs/spike-profiles.yaml "
        "at the project root or at /opt/orchestrata/configs/"
    )


_PROFILES_PATH = _find_profiles_path()


def _load_profiles() -> dict[str, dict[str, dict[str, str]]]:
    """Load all profiles from the YAML file.

    Returns:
        Mapping of profile_name → {agent_alias: {provider, model}}.

    Raises:
        LLMConfigError: If the file is missing or malformed.
    """
    try:
        with open(_PROFILES_PATH, encoding="utf-8") as fh:
            data = yaml.safe_load(fh)
    except FileNotFoundError as e:
        raise LLMConfigError(f"Profile file not found: {_PROFILES_PATH}") from e
    except yaml.YAMLError as e:
        raise LLMConfigError(f"Invalid YAML in {_PROFILES_PATH}: {e}") from e

    if not isinstance(data, dict):
        raise LLMConfigError(
            f"Expected a YAML mapping in {_PROFILES_PATH}, got {type(data).__name__}"
        )

    # Validate every entry has required fields
    for profile_name, agents in data.items():
        if not isinstance(agents, dict):
            raise LLMConfigError(
                f"Profile '{profile_name}' must be a mapping of agent → {{provider, model}}"
            )
        for agent_alias, cfg in agents.items():
            if not isinstance(cfg, dict):
                raise LLMConfigError(
                    f"Profile '{profile_name}' / agent '{agent_alias}': "
                    f"expected {{provider, model}}, got {type(cfg).__name__}"
                )
            if "provider" not in cfg:
                raise LLMConfigError(
                    f"Profile '{profile_name}' / agent '{agent_alias}': "
                    f"missing required field 'provider'"
                )
            if "model" not in cfg:
                raise LLMConfigError(
                    f"Profile '{profile_name}' / agent '{agent_alias}': "
                    f"missing required field 'model'"
                )

    return data


# Load profiles at import time
_PROFILES: dict[str, dict[str, dict[str, str]]] = _load_profiles()


def _resolve_profile_name(env: Mapping[str, str] | None = None) -> str:
    """Determine which profile to use.

    Checks ``SPIKE_PROFILE`` env var, falls back to ``"default"``.

    Args:
        env: Optional mapping to read ``SPIKE_PROFILE`` from. Defaults to
            ``os.environ`` when not provided, preserving normal runtime behaviour
            while allowing tests to pass an explicit dictionary.
    """
    if env is None:
        env = os.environ
    return env.get("SPIKE_PROFILE", "default")


def _get_active_profile(env: Mapping[str, str] | None = None) -> str:
    """Return the active profile name, falling back to 'default'.

    Args:
        env: Optional mapping to read ``SPIKE_PROFILE`` from. Defaults to
            ``os.environ`` when not provided.
    """
    name = _resolve_profile_name(env)
    if name not in _PROFILES:
        available = ", ".join(sorted(_PROFILES.keys()))
        logger.warning(
            "Profile '%s' not found in %s. Available: %s. Falling back to 'default'.",
            name,
            _PROFILES_PATH,
            available,
        )
        if "default" not in _PROFILES:
            raise LLMConfigError(
                f"Profile '{name}' not found and no 'default' profile exists. "
                f"Available profiles: {available}"
            )
        return "default"
    return name


# ── VolkConfig ───────────────────────────────────────────────────────


@dataclass
class VolkConfig:
    """Per-volk LLM provider and model configuration.

    Loaded entirely from the active YAML profile. No hardcoded defaults.
    No env variable overrides. The profile is the single source of truth.
    """

    profile_name: str = field(default_factory=_get_active_profile)

    def __post_init__(self) -> None:
        """Load assignments from the active profile."""
        profile_data = _PROFILES.get(self.profile_name, {})
        self._assignments: dict[str, dict[str, str]] = {
            k: dict(v) for k, v in profile_data.items()
        }
        # "default" is the SPEC-028 REQ-028-021/028 sentinel fallback entry,
        # not a real volk alias — exclude it from the roster so TUI/CLI agent
        # listings (gateway.handlers.ultimatetui.handle_agents,
        # integrations.cheetahclaws._handle_agents) don't show a phantom
        # "default" agent.
        self._known_agents = set(self._assignments.keys()) - {"default"}

    @traces(spec="028", reqs=["REQ-028-021", "REQ-028-028"])
    def get_provider(self, agent_alias: str | None) -> str:
        """Get the LLM provider for an agent.

        Args:
            agent_alias: Agent alias like "U:Ar" or "J:Junior", or ``None``
                for an unresolved/unstamped volk.

        Returns:
            Provider name (e.g. "deepseek", "anthropic") — from the agent's
            own entry, or the profile's ``default`` sentinel entry
            (REQ-028-021/028) when the alias is unrecognised or ``None``.

        Raises:
            LLMConfigError: If the agent is not configured AND no ``default``
                fallback entry exists in the active profile.
        """
        entry = self._assignments.get(agent_alias) if agent_alias is not None else None
        if entry is None:
            entry = self._assignments.get("default")
        if entry is None:
            raise LLMConfigError(
                f"Agent '{agent_alias}' not configured in profile "
                f"'{self.profile_name}', and no 'default' fallback entry "
                f"exists. Add an explicit entry for it to {_PROFILES_PATH}."
            )
        return entry["provider"]

    @traces(spec="028", reqs=["REQ-028-021", "REQ-028-028"])
    def get_model(self, agent_alias: str | None) -> str:
        """Get the LLM model for an agent.

        Args:
            agent_alias: Agent alias like "U:Ar" or "J:Junior", or ``None``
                for an unresolved/unstamped volk.

        Returns:
            Model name (e.g. "deepseek-v4-pro", "claude-sonnet-5") — from
            the agent's own entry, or the profile's ``default`` sentinel
            entry (REQ-028-021/028) when the alias is unrecognised or
            ``None``.

        Raises:
            LLMConfigError: If the agent is not configured AND no ``default``
                fallback entry exists in the active profile.
        """
        entry = self._assignments.get(agent_alias) if agent_alias is not None else None
        if entry is None:
            entry = self._assignments.get("default")
        if entry is None:
            raise LLMConfigError(
                f"Agent '{agent_alias}' not configured in profile "
                f"'{self.profile_name}', and no 'default' fallback entry "
                f"exists. Add an explicit entry for it to {_PROFILES_PATH}."
            )
        return entry["model"]

    @property
    def known_agents(self) -> set[str]:
        """Return the set of agent aliases that have assignments."""
        return self._known_agents


# ── Module-level singleton ──────────────────────────────────────────

_CONFIG: VolkConfig | None = None


@traces(spec="028", reqs=["REQ-028-021"])
def _get_config() -> VolkConfig:
    """Get or create the singleton VolkConfig."""
    global _CONFIG
    if _CONFIG is None:
        _CONFIG = VolkConfig()
    return _CONFIG


# ── Public API ───────────────────────────────────────────────────────


@traces(spec="028", reqs=["REQ-028-021", "REQ-028-028"])
def get_volk_provider(agent_alias: str) -> str:
    """Get the LLM provider for an agent.

    Args:
        agent_alias: Agent alias like "U:Ar" or "J:Junior".

    Returns:
        Provider name from the active profile, or from the profile's
        ``default`` sentinel entry (SPEC-028 REQ-028-021/028) when the
        alias is not configured.

    Raises:
        LLMConfigError: If the agent is not configured AND no ``default``
            fallback entry exists in the active profile.
    """
    return _get_config().get_provider(agent_alias)


@traces(spec="028", reqs=["REQ-028-021", "REQ-028-028"])
def get_volk_model(agent_alias: str | None) -> str:
    """Get the LLM model for an agent.

    Args:
        agent_alias: Agent alias like "U:Ar" or "J:Junior". ``None`` (an
            unresolved/unstamped volk) and any alias absent from the active
            profile fall back to the profile's ``default`` sentinel entry
            (SPEC-028 REQ-028-021/028) when one exists.

    Returns:
        Model name from the active profile (or its ``default`` entry).

    Raises:
        LLMConfigError: If the agent is not configured AND no ``default``
            fallback entry exists in the active profile.
    """
    return _get_config().get_model(agent_alias)


def get_active_profile() -> str:
    """Return the name of the currently active profile."""
    return _get_config().profile_name


def list_profiles() -> list[str]:
    """Return the names of all available profiles."""
    return sorted(_PROFILES.keys())


def select_profile(name: str) -> None:
    """Switch to a different profile at runtime.

    Args:
        name: Profile name (must exist in spike-profiles.yaml).

    Raises:
        LLMConfigError: If the profile does not exist.
    """
    global _CONFIG
    if name not in _PROFILES:
        available = ", ".join(sorted(_PROFILES.keys()))
        raise LLMConfigError(f"Profile '{name}' not found. Available: {available}")
    _CONFIG = VolkConfig(profile_name=name)
    logger.info("Switched to profile: %s", name)


def escalate_agent_model(agent_alias: str) -> str:
    """Escalate an agent's model from standard to reasoning tier.

    When a standard model produces subpar results, the agent can call
    this to retry with the provider's reasoning-tier model. Only affects
    the in-memory singleton config for the current session.

    Args:
        agent_alias: Agent alias like "U:Cd" or "J:Junior".

    Returns:
        New model name, or current model if already at reasoning tier.

    Raises:
        LLMConfigError: If the agent is not configured.
    """
    config = _get_config()
    provider = config.get_provider(agent_alias)
    current_model = config.get_model(agent_alias)

    if provider == "openrouter":
        # Family-aware: escalate within the model's own family; a model not
        # in the map is already at the top of its family (noop).
        reasoning_model = OPENROUTER_MODEL_ESCALATIONS.get(current_model, current_model)
    else:
        # Get the reasoning-tier model for this provider
        tier_models = PROVIDER_MODELS.get(provider, {})
        reasoning_model = tier_models.get("reasoning", current_model)

    if current_model == reasoning_model:
        return current_model  # Already at top tier

    # Update in-memory assignment
    config._assignments[agent_alias]["model"] = reasoning_model
    logger.info(
        "Escalated %s: %s → %s (provider: %s)",
        agent_alias,
        current_model,
        reasoning_model,
        provider,
    )
    return reasoning_model