orchestrata_ultimate.providers.router

Reference › orchestrata_ultimate › providers › router

router selects the provider and model for each request (specification 011, user story US-002). RequestRouter is built over a ProviderRegistry and the manager's dictionaries of circuit breakers and rate limiters; provider_manager is its only importer.

_get_candidates builds the candidate list: if the request names a model it resolves that one model; otherwise, if the request lists capabilities, it asks the registry for matches; otherwise every priced model of every provider is a candidate. route drops candidates whose breaker is open or whose limiter has no headroom, ranks the rest with _rank and returns the first (provider, model) pair, or None. _rank follows the request's RoutingStrategy: cost-optimised sorts by combined input and output price per million tokens, quality-optimised by quality rank, latency-optimised by the p95 latency from the health data, and round-robin rotates the list. get_fallback_chain returns the ordered alternatives excluding providers already tried, get_rejection_reasons explains why candidates were rejected, and get_rate_limit_backoff returns the seconds until the most restrictive exhausted limiter resets.

At a glance

Class diagram

Dependency diagram

API

Classes

RequestRouter

Decorators: traces(spec='011', story='US-002', reqs=['REQ-011-011', 'REQ-011-012', 'REQ-011-013', 'REQ-011-014', 'REQ-011-015', 'RE… · Line 34

Strategy-based request router.

Source

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

"""Request routing engine.

SPEC-011 FR-003: Selects optimal provider based on configurable strategy.
Filters by availability, rate limit headroom, capability match, and budget.
"""

import logging

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    ModelPricing,
    ProviderRequest,
    RoutingStrategy,
)
from orchestrata_ultimate.providers.circuit_breaker import CircuitBreaker
from orchestrata_ultimate.providers.rate_limiter import RateLimiter
from orchestrata_ultimate.providers.registry import ProviderRegistry

logger = logging.getLogger(__name__)


@traces(
    spec="011",
    story="US-002",
    reqs=[
        "REQ-011-011",
        "REQ-011-012",
        "REQ-011-013",
        "REQ-011-014",
        "REQ-011-015",
        "REQ-011-016",
    ],
)
class RequestRouter:
    """Strategy-based request router.

    Selects the best provider+model for a request by:
    1. Finding eligible providers (capability match or explicit model)
    2. Filtering by availability (circuit breaker, rate limit)
    3. Ranking by strategy (cost, latency, quality, round_robin)
    """

    def __init__(
        self,
        registry: ProviderRegistry,
        circuit_breakers: dict[str, CircuitBreaker],
        rate_limiters: dict[str, RateLimiter],
    ):
        self._registry = registry
        self._breakers = circuit_breakers
        self._limiters = rate_limiters
        self._round_robin_idx: int = 0

    @traces(spec="011", story="US-002", reqs=["REQ-011-015", "REQ-011-016"])
    def route(
        self,
        request: ProviderRequest,
        health_data: dict[str, dict] | None = None,
    ) -> tuple | None:
        """Select provider and model for a request.

        Returns (provider_name, model_name) or None if no eligible provider.
        """
        candidates = self._get_candidates(request)
        if not candidates:
            return None
        # Filter by availability
        available = []
        for provider_name, model_name, pricing in candidates:
            breaker = self._breakers.get(provider_name)
            if breaker and not breaker.is_available:
                continue
            limiter = self._limiters.get(provider_name)
            if limiter and not limiter.is_available:
                continue
            available.append((provider_name, model_name, pricing))
        if not available:
            return None
        # Rank by strategy
        strategy = request.routing_strategy
        ranked = self._rank(available, strategy, health_data)
        if not ranked:
            return None
        provider_name, model_name, _ = ranked[0]
        return (provider_name, model_name)

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-036"])
    def get_fallback_chain(
        self,
        request: ProviderRequest,
        exclude: list[str] | None = None,
    ) -> list[tuple]:
        """Get ordered fallback chain excluding already-tried providers.

        Returns list of (provider_name, model_name) tuples.
        """
        exclude = exclude or []
        candidates = self._get_candidates(request)
        chain = []
        for provider_name, model_name, pricing in candidates:
            if provider_name in exclude:
                continue
            breaker = self._breakers.get(provider_name)
            if breaker and not breaker.is_available:
                continue
            chain.append((provider_name, model_name))
        return chain

    @traces(spec="011", story="US-011-001", reqs=["REQ-011-007"])
    def _get_candidates(self, request: ProviderRequest) -> list[tuple]:
        """Get candidate providers+models for a request."""
        # If explicit model specified, resolve it
        if request.model:
            resolved = self._registry.resolve_model(request.model)
            if resolved:
                provider_name, model_name = resolved
                pricing = self._registry.get_pricing(provider_name, model_name)
                if pricing:
                    return [(provider_name, model_name, pricing)]
            return []
        # Capability-based lookup
        capabilities = request.capabilities or []
        if capabilities:
            return self._registry.find_by_capability(capabilities)
        # No model, no capabilities — all providers eligible
        all_candidates: list[tuple[str, str, ModelPricing]] = []
        for name in self._registry.provider_names:
            config = self._registry.get_config(name)
            if config:
                all_candidates.extend(
                    (name, model_name, pricing)
                    for model_name, pricing in config.pricing.items()
                )
        return all_candidates

    @traces(spec="011", story="US-002", reqs=["REQ-011-015", "REQ-011-016"])
    def get_rejection_reasons(self, request: ProviderRequest) -> list[str]:
        """Return human-readable reasons why each candidate was rejected.

        The error message produced by :class:`ProviderManager` uses these
        strings so operators can tell at a glance whether routing failed
        because of an unknown model, an open circuit breaker, or RPM/TPM
        exhaustion.
        """
        candidates = self._get_candidates(request)
        if not candidates:
            return ["unknown model/capabilities"]
        reasons: list[str] = []
        for provider_name, model_name, _ in candidates:
            breaker = self._breakers.get(provider_name)
            if breaker and not breaker.is_available:
                reasons.append(f"{provider_name}/{model_name}: circuit breaker open")
                continue
            limiter = self._limiters.get(provider_name)
            if limiter and not limiter.is_available:
                reasons.append(
                    f"{provider_name}/{model_name}: "
                    f"{limiter.get_unavailability_reason()}"
                )
                continue
        if not reasons:
            # route() returned None but no candidate hit the breaker or
            # limiter filters — rejected by ranking or a residual filter.
            # Never return an empty list: ProviderManager splices these
            # strings into its error and an empty tail read as
            # "No eligible provider found for request: " with no cause
            # (review finding, 2026-07-14).
            reasons.append(
                "candidates existed but none ranked as available "
                "(no breaker/limiter cause recorded)"
            )
        return reasons

    @traces(spec="011", story="US-002", reqs=["REQ-011-015", "REQ-011-016"])
    def get_rate_limit_backoff(self, request: ProviderRequest) -> float:
        """Seconds to wait for the most restrictive exhausted limiter to reset.

        Returns 0.0 when no candidate is currently rate-limited locally.
        """
        candidates = self._get_candidates(request)
        max_backoff = 0.0
        for provider_name, _, _ in candidates:
            limiter = self._limiters.get(provider_name)
            if limiter and not limiter.is_available:
                max_backoff = max(max_backoff, limiter.seconds_until_reset)
        return max_backoff

    @traces(
        spec="011",
        story="US-011-001",
        reqs=["REQ-011-011", "REQ-011-012", "REQ-011-013", "REQ-011-014"],
    )
    def _rank(
        self,
        candidates: list[tuple],
        strategy: RoutingStrategy,
        health_data: dict[str, dict] | None = None,
    ) -> list[tuple]:
        """Rank candidates by routing strategy."""
        if strategy == RoutingStrategy.COST_OPTIMIZED:
            return sorted(
                candidates,
                key=lambda c: c[2].input_per_1m + c[2].output_per_1m,
            )
        elif strategy == RoutingStrategy.QUALITY_OPTIMIZED:
            return sorted(
                candidates,
                key=lambda c: c[2].quality_rank,
            )
        elif strategy == RoutingStrategy.LATENCY_OPTIMIZED:
            if health_data:
                return sorted(
                    candidates,
                    key=lambda c: health_data.get(c[0], {}).get(
                        "latency_p95_ms", float("inf")
                    ),
                )
            return candidates
        elif strategy == RoutingStrategy.ROUND_ROBIN:
            if not candidates:
                return candidates
            idx = self._round_robin_idx % len(candidates)
            self._round_robin_idx += 1
            return candidates[idx:] + candidates[:idx]
        return candidates