Coverage for src / lexigram / contracts / ai / providers.py: 0%
47 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-15 18:57 +0800
1"""Provider registry contracts for multi-provider LLM selection and fallback."""
3from __future__ import annotations
5from dataclasses import dataclass, field
6from datetime import datetime
7from enum import StrEnum
8from typing import TYPE_CHECKING, Any
10from typing_extensions import Protocol, runtime_checkable
12from lexigram.contracts.ai.types import ModelCapability
14if TYPE_CHECKING:
15 from lexigram.contracts.ai.llm import LLMClientProtocol
18class SelectionStrategy(StrEnum):
19 """Strategy for selecting which model/provider to use.
21 Attributes:
22 COST_OPTIMAL: Choose cheapest option that meets requirements
23 LATENCY_OPTIMAL: Choose fastest provider
24 CAPABILITY_MATCH: Choose first model supporting all required capabilities
25 ROUND_ROBIN: Rotate through healthy providers
26 PREFERRED: Try preferred provider, fallback to others
27 """
29 COST_OPTIMAL = "cost_optimal"
30 LATENCY_OPTIMAL = "latency_optimal"
31 CAPABILITY_MATCH = "capability_match"
32 ROUND_ROBIN = "round_robin"
33 PREFERRED = "preferred"
36@dataclass(frozen=True)
37class ModelInfo:
38 """Information about an available LLM model.
40 Attributes:
41 model_id: Unique model identifier (e.g., "gpt-4o")
42 provider: Provider name (e.g., "openai")
43 display_name: Human-readable name
44 capabilities: Set of capabilities this model has
45 context_window: Input context window size in tokens
46 max_output_tokens: Maximum output tokens this model can generate
47 input_cost_per_million: Cost per million input tokens (USD)
48 output_cost_per_million: Cost per million output tokens (USD)
49 is_available: Whether model is currently available
50 metadata: Additional provider-specific metadata
51 """
53 model_id: str
54 provider: str
55 display_name: str
56 capabilities: frozenset[ModelCapability]
57 context_window: int
58 max_output_tokens: int
59 input_cost_per_million: float
60 output_cost_per_million: float
61 is_available: bool = True
62 metadata: dict[str, Any] = field(default_factory=dict)
65@dataclass(frozen=True)
66class ProviderHealth:
67 """Health status of a provider.
69 Mutable dataclass tracking provider health metrics including
70 latency, error rate, and availability status.
72 Attributes:
73 provider: Provider name
74 is_healthy: Whether provider is considered healthy
75 latency_ms: Average response latency in milliseconds
76 error_rate: Error rate as a fraction (0-1)
77 last_check: When health was last checked
78 details: Additional health detail information
79 """
81 provider: str
82 is_healthy: bool
83 latency_ms: float
84 error_rate: float
85 last_check: datetime
86 details: dict[str, Any] = field(default_factory=dict)
89@runtime_checkable
90class ProviderRegistryProtocol(Protocol):
91 """Protocol for registering and discovering LLM providers.
93 Maintains a registry of supported providers and their models,
94 with health tracking and capability-based filtering.
95 """
97 async def register_provider(
98 self, name: str, client: LLMClientProtocol, models: list[ModelInfo]
99 ) -> None:
100 """Register a new provider with its available models.
102 Args:
103 name: Provider name (e.g., "openai", "anthropic")
104 client: The LLM client for this provider
105 models: List of available models from this provider
106 """
107 ...
109 async def get_client(self, provider: str) -> LLMClientProtocol | None:
110 """Get the LLM client for a provider.
112 Args:
113 provider: Provider name
115 Returns:
116 The LLM client, or None if not registered
117 """
118 ...
120 def list_providers(self) -> list[str]:
121 """List all registered provider names.
123 Returns:
124 List of provider names
125 """
126 ...
128 def list_models(
129 self, capabilities: set[ModelCapability] | None = None
130 ) -> list[ModelInfo]:
131 """List all available models with optional capability filtering.
133 Args:
134 capabilities: If provided, return only models with all these capabilities
136 Returns:
137 List of model information
138 """
139 ...
141 def get_model_info(self, model_id: str) -> ModelInfo | None:
142 """Get information about a specific model.
144 Args:
145 model_id: The model identifier
147 Returns:
148 Model information, or None if not found
149 """
150 ...
153@runtime_checkable
154class ModelSelectorProtocol(Protocol):
155 """Protocol for selecting models based on criteria.
157 Implements various selection strategies including cost optimization,
158 latency optimization, and capability matching.
159 """
161 async def select(
162 self,
163 capabilities: set[ModelCapability],
164 preferred_provider: str | None = None,
165 max_cost_per_million: float | None = None,
166 strategy: SelectionStrategy = SelectionStrategy.COST_OPTIMAL,
167 ) -> ModelInfo | None:
168 """Select a model based on the given criteria.
170 Args:
171 capabilities: Required capabilities for the model
172 preferred_provider: Prefer this provider if possible
173 max_cost_per_million: Maximum cost threshold (if relevant to strategy)
174 strategy: Selection strategy to use
176 Returns:
177 Selected model info, or None if no suitable model found
178 """
179 ...
182@runtime_checkable
183class FallbackChainProtocol(Protocol):
184 """Protocol for executing requests with provider fallback.
186 Handles cascading requests across multiple providers with
187 automatic fallback on failure.
188 """
190 async def execute(self, request: Any, providers: list[str]) -> Any:
191 """Execute request with automatic fallback to other providers.
193 Tries providers in order until one succeeds.
195 Args:
196 request: The request to execute
197 providers: List of provider names to try in order
199 Returns:
200 The response from the first successful provider
202 Raises:
203 AllProvidersExhaustedError: If all providers failed
204 """
205 ...
208__all__ = [
209 "FallbackChainProtocol",
210 "ModelInfo",
211 "ModelSelectorProtocol",
212 "ProviderHealth",
213 "ProviderRegistryProtocol",
214 "SelectionStrategy",
215]