orchestrata_ultimate.core.protocols.providers

Reference › orchestrata_ultimate › core › protocols › providers

The provider-manager types of SPEC-011, the canonical and production-wired version that interfaces.providers re-exports. It describes how a request for a language model is shaped, routed and costed. ModelCapability lists what a model can do (reasoning, coding, vision, tool use, long context, embedding, summarisation) and RoutingStrategy how a model is chosen (cost, latency or quality optimised, round robin and others). Dataclasses cover requests and responses (Message, ProviderRequest, ProviderResponse, RawProviderResponse), accounting (TokenUsage, CostRecord, CostEstimate, BudgetStatus, UsageReport) and configuration (ModelPricing, ModelInfo, ProviderConfig).

ProviderProtocol is the contract for one provider: complete, get_models, supports_capability and health reporting. ProviderManagerProtocol is the contract for the manager: execute a request, estimate_cost, get_usage, register_provider and deregister_provider, get_remaining_budget, cancel and get_provider_health. The implementation is in the providers package.

At a glance

Class diagram

Dependency diagram

API

Classes

ModelCapability (enum)

Bases: StrEnum · Line 22

LLM model capabilities for capability-based routing.

RoutingStrategy (enum)

Bases: StrEnum · Decorators: traces(spec='011', story='US-004', reqs=['REQ-011-011', 'REQ-011-012', 'REQ-011-013', 'REQ-011-014']) · Line 39

Strategy for provider selection.

TokenUsage (dataclass)

Decorators: dataclass · Line 49

Token usage for a provider request.

CostRecord (dataclass)

Decorators: dataclass · Line 58

Cost record for a provider request.

CostEstimate (dataclass)

Decorators: dataclass, traces(spec='011', story='US-003', reqs=['REQ-011-018']) · Line 69

Estimated cost for a provider request.

Message (dataclass)

Decorators: dataclass, traces(spec='103', story='US-001', reqs=['REQ-103-001']) · Line 81

A message in a conversation.

ProviderRequest (dataclass)

Decorators: dataclass, traces(spec='011', story='US-001', reqs=['REQ-011-004']), traces(spec='103', story='US-001', reqs=['REQ-103-002']) · Line 96

Request to the provider manager.

ProviderResponse (dataclass)

Decorators: dataclass, traces(spec='011', story='US-001', reqs=['REQ-011-005']), traces(spec='103', story='US-001', reqs=['REQ-103-003']) · Line 118

Response from the provider manager.

BudgetStatus (dataclass)

Decorators: dataclass · Line 135

Budget status for a scope.

UsageReport (dataclass)

Decorators: dataclass · Line 150

Aggregated usage report.

ModelPricing (dataclass)

Decorators: dataclass, traces(spec='011', story='US-005', reqs=['REQ-011-028']) · Line 162

Pricing configuration for a model.

ModelInfo (dataclass)

Decorators: dataclass, traces(spec='011', story='US-005', reqs=['REQ-011-028']) · Line 174

Information about an available model.

ProviderConfig (dataclass)

Decorators: dataclass · Line 185

Configuration for a provider adapter.

RawProviderResponse (dataclass)

Decorators: dataclass, traces(spec='103', story='US-001', reqs=['REQ-103-003']) · Line 200

Raw response from a provider adapter before normalization.

ProviderProtocol (protocol)

Bases: Protocol · Decorators: traces(spec='011', story='US-006', reqs=['REQ-011-029']) · Line 217

Protocol for provider adapter implementations.

ProviderManagerProtocol (protocol)

Bases: ComponentProtocol, Protocol · Decorators: traces(spec='011', story='US-001', reqs=['REQ-011-001', 'REQ-011-003']) · Line 241

Unified LLM provider abstraction.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/protocols/providers.py

"""
Provider manager protocol types for Orchestrata Ultimate (SPEC-011).

Depends on :mod:`orchestrata_ultimate.core.protocols.base` for the
``ComponentProtocol`` base.
"""

import uuid
from dataclasses import dataclass, field
from datetime import datetime
from enum import StrEnum
from typing import Any, Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols.base import ComponentProtocol

# ============================================================================
# Provider Manager Types (SPEC-011)
# ============================================================================


class ModelCapability(StrEnum):
    """LLM model capabilities for capability-based routing."""

    REASONING = "reasoning"
    CODING = "coding"
    VISION = "vision"
    TOOL_USE = "tool_use"
    LONG_CONTEXT = "long_context"
    EMBEDDING = "embedding"
    SUMMARIZATION = "summarization"


@traces(
    spec="011",
    story="US-004",
    reqs=["REQ-011-011", "REQ-011-012", "REQ-011-013", "REQ-011-014"],
)
class RoutingStrategy(StrEnum):
    """Strategy for provider selection."""

    COST_OPTIMIZED = "cost_optimized"
    LATENCY_OPTIMIZED = "latency_optimized"
    QUALITY_OPTIMIZED = "quality_optimized"
    ROUND_ROBIN = "round_robin"


@dataclass
class TokenUsage:
    """Token usage for a provider request."""

    input_tokens: int = 0
    output_tokens: int = 0
    total_tokens: int = 0


@dataclass
class CostRecord:
    """Cost record for a provider request."""

    input_cost: float = 0.0
    output_cost: float = 0.0
    total_cost: float = 0.0
    currency: str = "USD"


@dataclass
@traces(spec="011", story="US-003", reqs=["REQ-011-018"])
class CostEstimate:
    """Estimated cost for a provider request."""

    estimated_input_tokens: int = 0
    estimated_output_tokens: int = 0
    estimated_cost: CostRecord = field(default_factory=CostRecord)
    provider: str = ""
    model: str = ""


@dataclass
@traces(spec="103", story="US-001", reqs=["REQ-103-001"])
class Message:
    """A message in a conversation."""

    role: str = "user"
    content: str = ""
    # OpenAI-compatible tool-calling fields (optional, backward-compatible).
    tool_calls: list[dict[str, Any]] | None = None  # assistant: requested calls
    tool_call_id: str | None = None  # tool result: id of the call it answers
    name: str | None = None  # tool result: the tool's name
    reasoning_content: str | None = None  # thinking models must echo this back


@dataclass
@traces(spec="011", story="US-001", reqs=["REQ-011-004"])
@traces(spec="103", story="US-001", reqs=["REQ-103-002"])
class ProviderRequest:
    """Request to the provider manager."""

    request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    messages: list[Message] = field(default_factory=list)
    model: str | None = None
    capabilities: list[ModelCapability] | None = None
    parameters: dict[str, Any] | None = None
    routing_strategy: RoutingStrategy = RoutingStrategy.COST_OPTIMIZED
    max_tokens: int | None = None
    volk_id: str | None = None
    session_id: str | None = None
    project_id: str | None = None
    # Function/tool-calling (OpenAI-compatible). When set, forwarded to the
    # provider; the response may then carry tool_calls instead of/with content.
    tools: list[dict[str, Any]] | None = None
    tool_choice: str | None = None


@dataclass
@traces(spec="011", story="US-001", reqs=["REQ-011-005"])
@traces(spec="103", story="US-001", reqs=["REQ-103-003"])
class ProviderResponse:
    """Response from the provider manager."""

    content: str = ""
    model_used: str = ""
    provider_used: str = ""
    usage: TokenUsage = field(default_factory=TokenUsage)
    cost: CostRecord = field(default_factory=CostRecord)
    latency_ms: float = 0.0
    metadata: dict[str, Any] = field(default_factory=dict)
    tool_calls: list[dict[str, Any]] = field(default_factory=list)
    # Thinking models (deepseek-v4-pro) return reasoning_content that must be
    # echoed back in subsequent turns within the same conversation.
    reasoning_content: str | None = None


@dataclass
class BudgetStatus:
    """Budget status for a scope."""

    scope: str = ""
    scope_id: str = ""
    budget_limit_usd: float = 0.0
    spent_usd: float = 0.0
    remaining_usd: float = 0.0
    usage_percent: float = 0.0
    warning_threshold: float = 0.8
    is_warning: bool = False
    is_exceeded: bool = False


@dataclass
class UsageReport:
    """Aggregated usage report."""

    total_requests: int = 0
    total_tokens: int = 0
    total_cost: float = 0.0
    by_provider: dict[str, dict[str, Any]] = field(default_factory=dict)
    by_model: dict[str, dict[str, Any]] = field(default_factory=dict)


@dataclass
@traces(spec="011", story="US-005", reqs=["REQ-011-028"])
class ModelPricing:
    """Pricing configuration for a model."""

    input_per_1m: float = 0.0
    output_per_1m: float = 0.0
    capabilities: list[ModelCapability] = field(default_factory=list)
    context_window: int = 8192
    quality_rank: int = 50


@dataclass
@traces(spec="011", story="US-005", reqs=["REQ-011-028"])
class ModelInfo:
    """Information about an available model."""

    name: str = ""
    provider: str = ""
    capabilities: list[ModelCapability] = field(default_factory=list)
    context_window: int = 8192
    pricing: ModelPricing = field(default_factory=ModelPricing)


@dataclass
class ProviderConfig:
    """Configuration for a provider adapter."""

    api_key: str = ""
    base_url: str | None = None
    models: list[str] = field(default_factory=list)
    rate_limit_rpm: int = 60
    rate_limit_tpm: int = 100_000
    timeout_seconds: float = 30.0
    max_retries: int = 3
    pricing: dict[str, ModelPricing] = field(default_factory=dict)


@dataclass
@traces(spec="103", story="US-001", reqs=["REQ-103-003"])
class RawProviderResponse:
    """Raw response from a provider adapter before normalization."""

    content: str = ""
    input_tokens: int = 0
    output_tokens: int = 0
    model: str = ""
    metadata: dict[str, Any] = field(default_factory=dict)
    tool_calls: list[dict[str, Any]] = field(default_factory=list)


# ============================================================================
# Provider Manager Protocols (SPEC-011)
# ============================================================================


@traces(spec="011", story="US-006", reqs=["REQ-011-029"])
class ProviderProtocol(Protocol):
    """Protocol for provider adapter implementations."""

    async def complete(
        self,
        messages: list[Message],
        model: str,
        parameters: dict[str, Any],
    ) -> RawProviderResponse:
        """Send completion request to provider."""
        ...

    @traces(spec="011", story="US-001", reqs=["REQ-011-028"])
    def get_models(self) -> list[ModelInfo]:
        """Return available models with metadata."""
        ...

    @traces(spec="011", story="US-001", reqs=["REQ-011-007"])
    def supports_capability(self, capability: ModelCapability) -> bool:
        """Check if provider supports a capability."""
        ...


@traces(spec="011", story="US-001", reqs=["REQ-011-001", "REQ-011-003"])
class ProviderManagerProtocol(ComponentProtocol, Protocol):
    """Unified LLM provider abstraction.

    Manages multiple LLM backends, providing request routing,
    load balancing, cost tracking, fallback chains, and rate limit management.
    """

    @traces(spec="011", story="US-001", reqs=["REQ-011-006"])
    async def execute(self, request: ProviderRequest) -> ProviderResponse:
        """Execute an LLM request. Handles routing, fallback, cost tracking."""
        ...

    @traces(spec="011", story="US-001", reqs=["REQ-011-018"])
    async def estimate_cost(self, request: ProviderRequest) -> CostEstimate:
        """Estimate cost for a request without executing it."""
        ...

    @traces(spec="011", story="US-001", reqs=["REQ-011-023"])
    async def get_usage(
        self,
        *,
        volk_id: str | None = None,
        session_id: str | None = None,
        project_id: str | None = None,
        date_from: datetime | None = None,
        date_to: datetime | None = None,
    ) -> UsageReport:
        """Query cumulative usage and cost."""
        ...

    @traces(spec="011", story="US-001", reqs=["REQ-011-024", "REQ-011-025"])
    def register_provider(
        self,
        name: str,
        provider: ProviderProtocol,
        config: ProviderConfig,
    ) -> None:
        """Register a provider adapter with configuration."""
        ...

    @traces(spec="011", story="US-001", reqs=["REQ-011-024"])
    def deregister_provider(self, name: str) -> bool:
        """Remove a provider. Returns True if found and removed."""
        ...

    @traces(spec="011", story="US-001", reqs=["REQ-011-045"])
    async def get_remaining_budget(
        self,
        *,
        volk_id: str | None = None,
        project_id: str | None = None,
        portfolio_id: str | None = None,
    ) -> BudgetStatus:
        """Query remaining budget for a scope."""
        ...

    async def cancel(self, request_id: str) -> bool:
        """Cancel an in-flight request. Returns True if successfully cancelled."""
        ...

    @traces(spec="015", story="US-001", reqs=["REQ-015-055"])
    def get_provider_health(self) -> dict[str, Any]:
        """Return health status of all registered providers."""
        ...


__all__ = [
    "ModelCapability",
    "RoutingStrategy",
    "TokenUsage",
    "CostRecord",
    "CostEstimate",
    "Message",
    "ProviderRequest",
    "ProviderResponse",
    "BudgetStatus",
    "UsageReport",
    "ModelPricing",
    "ModelInfo",
    "ProviderConfig",
    "RawProviderResponse",
    "ProviderProtocol",
    "ProviderManagerProtocol",
]