orchestrata_ultimate.verification.models.base_model

Reference › orchestrata_ultimate › verification › models › base_model

The contract and plumbing of the LLM layer (Spec 035 US-005). LLMModel is a runtime-checkable Protocol: a name property and three asynchronous methods, verify, analyze and generate_proof, each taking and returning dictionaries. The supervisor agent and verification agent depend on this protocol, not on a concrete model.

ModelRegistry maps names to models with register, get (returns None for an unknown name), list_models and count. ModelFallbackChain wraps an ordered list of models and satisfies LLMModel structurally; its name is the member names joined with a vertical bar. For each operation it tries the models in order and returns the first success, logging a warning for verify and generate_proof failures, and raises RuntimeError("All models failed...") chained to the last error when every model fails. The concrete wrappers are ClaudeOpusModel, DeepseekR1Model, GrokMiniMaxModel and QwenMaxModel.

At a glance

Class diagram

Dependency diagram

API

Classes

LLMModel (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='035', story='US-005', reqs=['REQ-035-US-005-007', 'REQ-035-US-005-010', 'REQ-035-US-005-019']) · Line 33

Protocol for LLM model wrappers.

ModelRegistry

Decorators: traces(spec='035', story='US-005', reqs=['REQ-035-US-005-005', 'REQ-035-US-005-008', 'REQ-035-US-005-015', 'REQ-035-US-… · Line 75

Registry of available LLM models.

ModelFallbackChain

Decorators: traces(spec='035', story='US-005', reqs=['REQ-035-US-005-006', 'REQ-035-US-005-009', 'REQ-035-US-005-016', 'REQ-035-US-… · Line 121

Chain of models with fallback on failure.

Source

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

"""LLM model base protocol and registry.

Wiring status (2026-07-06, W6.2): ``LLMModel``/``ModelRegistry``/
``ModelFallbackChain`` were fully built and unit-tested but had zero
callers. They are now wired into ``VerificationAgent`` (REQ-035-US-005-005:
select an appropriate model, either directly or via ``ModelRegistry``
lookup) and ``SupervisorAgent`` (REQ-035-US-005-001/011: LLM-backed proof
generation with a deterministic fallback). While wiring this in,
``ModelFallbackChain`` was found incomplete relative to the ``LLMModel``
protocol it implements -- it chained ``verify``/``analyze`` but not
``generate_proof``, and had no ``name`` property (REQ-035-US-005-009: support
model fallback chain). Both are added below so the chain is fully
protocol-conformant and can be handed anywhere a single ``LLMModel`` is
expected.
"""

# Spec: 035 - Verification Agent

import logging
from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


@runtime_checkable
@traces(
    spec="035",
    story="US-005",
    reqs=["REQ-035-US-005-007", "REQ-035-US-005-010", "REQ-035-US-005-019"],
)
class LLMModel(Protocol):
    """Protocol for LLM model wrappers."""

    @property
    @traces(spec="035", story="US-005", reqs=["REQ-035-US-005-010"])
    def name(self) -> str: ...

    @traces(
        spec="035",
        story="US-005",
        reqs=[
            "REQ-035-US-005-006",
            "REQ-035-US-005-016",
            "REQ-035-US-005-020",
            "REQ-035-US-005-023",
        ],
    )
    async def verify(
        self, constraint: dict[str, Any], context: dict[str, Any]
    ) -> dict[str, Any]: ...

    @traces(spec="035", story="US-005", reqs=["REQ-035-US-005-006"])
    async def analyze(self, results: list[dict[str, Any]]) -> dict[str, Any]: ...

    @traces(
        spec="035", story="US-005", reqs=["REQ-035-US-005-009", "REQ-035-US-005-016"]
    )
    async def generate_proof(
        self, constraint: dict[str, Any], context: dict[str, Any]
    ) -> dict[str, Any]: ...


@traces(
    spec="035",
    story="US-005",
    reqs=[
        "REQ-035-US-005-005",
        "REQ-035-US-005-008",
        "REQ-035-US-005-015",
        "REQ-035-US-005-017",
    ],
)
class ModelRegistry:
    """Registry of available LLM models."""

    def __init__(self) -> None:
        self._models: dict[str, LLMModel] = {}

    @traces(spec="035", story="US-5", reqs=["REQ-035-US-005-005"])
    def register(self, name: str, model: LLMModel) -> None:
        self._models[name] = model

    @traces(
        spec="035", story="US-005", reqs=["REQ-035-US-005-005", "REQ-035-US-005-015"]
    )
    def get(self, name: str) -> LLMModel | None:
        return self._models.get(name)

    @traces(
        spec="035",
        story="US-005",
        reqs=[
            "REQ-035-US-005-005",
            "REQ-035-US-005-009",
            "REQ-035-US-005-019",
            "REQ-035-US-005-023",
        ],
    )
    def list_models(self) -> list[str]:
        return list(self._models.keys())

    @property
    def count(self) -> int:
        return len(self._models)


@traces(
    spec="035",
    story="US-005",
    reqs=[
        "REQ-035-US-005-006",
        "REQ-035-US-005-009",
        "REQ-035-US-005-016",
        "REQ-035-US-005-018",
        "REQ-035-US-005-020",
        "REQ-035-US-005-023",
    ],
)
class ModelFallbackChain:
    """Chain of models with fallback on failure."""

    def __init__(self, models: list[LLMModel]) -> None:
        self._models = models

    @property
    @traces(spec="035", story="US-005", reqs=["REQ-035-US-005-009"])
    def name(self) -> str:
        """Composite name -- the chain satisfies ``LLMModel`` structurally."""
        return "|".join(getattr(m, "name", "unknown") for m in self._models)

    @traces(
        spec="035",
        story="US-005",
        reqs=[
            "REQ-035-US-005-006",
            "REQ-035-US-005-016",
            "REQ-035-US-005-020",
            "REQ-035-US-005-023",
        ],
    )
    async def verify(
        self, constraint: dict[str, Any], context: dict[str, Any]
    ) -> dict[str, Any]:
        last_error = None
        for model in self._models:
            try:
                return await model.verify(constraint, context)
            except Exception as exc:
                logger.warning(
                    "Model %s failed: %s", getattr(model, "name", "unknown"), exc
                )
                last_error = exc
        raise RuntimeError(
            f"All models failed. Last error: {last_error}"
        ) from last_error

    @traces(spec="035", story="US-005", reqs=["REQ-035-US-005-006"])
    async def analyze(self, results: list[dict[str, Any]]) -> dict[str, Any]:
        last_error = None
        for model in self._models:
            try:
                return await model.analyze(results)
            except Exception as exc:
                last_error = exc
        raise RuntimeError(
            f"All models failed. Last error: {last_error}"
        ) from last_error

    @traces(
        spec="035", story="US-005", reqs=["REQ-035-US-005-009", "REQ-035-US-005-016"]
    )
    async def generate_proof(
        self, constraint: dict[str, Any], context: dict[str, Any]
    ) -> dict[str, Any]:
        last_error = None
        for model in self._models:
            try:
                return await model.generate_proof(constraint, context)
            except Exception as exc:
                logger.warning(
                    "Model %s failed: %s", getattr(model, "name", "unknown"), exc
                )
                last_error = exc
        raise RuntimeError(
            f"All models failed. Last error: {last_error}"
        ) from last_error

    @property
    @traces(spec="035", story="US-005", reqs=["REQ-035-US-005-009"])
    def model_count(self) -> int:
        return len(self._models)