Coverage for src / lexigram / contracts / ai / routing.py: 0%

1"""Routing protocols for multi-provider LLM inference. 

2 

3Defines the protocol interfaces that decouple ``lexigram-ai-llm``'s routing 

4implementation from consumers. All types that live in extension packages 

5are represented as ``Any`` to satisfy the zero-dependency constraint of 

6``lexigram-contracts``. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

12 

13if TYPE_CHECKING: 

14 from datetime import datetime 

15 

16__all__ = [ 

17 "InferenceLoggerProtocol", 

18 "LLMRouterProtocol", 

19 "QuotaBackendProtocol", 

20 "RoutingStrategyProtocol", 

21] 

22 

23 

24@runtime_checkable 

25class RoutingStrategyProtocol(Protocol): 

26 """Protocol for routing strategies used by the LLM router.""" 

27 

28 async def execute( 

29 self, 

30 *, 

31 providers: list[Any], 

32 clients: dict[str, Any], 

33 quota: QuotaBackendProtocol, 

34 config: Any, 

35 messages: list[Any], 

36 kwargs: dict[str, Any], 

37 ) -> tuple[Any | None, list[str], int]: 

38 """Execute routing strategy.""" 

39 ... 

40 

41 

42@runtime_checkable 

43class LLMRouterProtocol(Protocol): 

44 """Protocol for multi-provider LLM routing. 

45 

46 Implementations cascade across a list of free providers (subject to 

47 daily quota tracking) before optionally falling through to a paid 

48 fallback. Returns a ``Result[InferenceLog, InferenceError]`` — 

49 typed as ``Any`` here to avoid importing extension-package types. 

50 

51 Example: 

52 ```python 

53 result = await router.route( 

54 messages=[{"role": "user", "content": "Hello!"}], 

55 temperature=0.2, 

56 ) 

57 if result.is_ok(): 

58 log = result.unwrap() 

59 print(log.completion.content) 

60 ``` 

61 """ 

62 

63 async def route( 

64 self, 

65 messages: list[Any], 

66 **kwargs: Any, 

67 ) -> Any: 

68 """Route a completion request across configured providers. 

69 

70 Args: 

71 messages: OpenAI-compatible message list. 

72 **kwargs: Provider-agnostic generation options (temperature, 

73 max_tokens, model, etc.). 

74 

75 Returns: 

76 ``Result[InferenceLog, InferenceError]`` — ``Ok`` on any 

77 successful completion, ``Err`` when every provider (including 

78 the paid fallback) is exhausted or fails. 

79 """ 

80 ... 

81 

82 async def health_probe(self) -> Any: 

83 """Probe configured providers without issuing a normal inference request. 

84 

85 Returns: 

86 ``Result[InferenceLog, InferenceError]`` — ``Ok`` on the first 

87 healthy configured provider, ``Err`` when no enabled provider can 

88 satisfy a lightweight health check. 

89 """ 

90 ... 

91 

92 

93@runtime_checkable 

94class QuotaBackendProtocol(Protocol): 

95 """Protocol for per-provider daily quota tracking. 

96 

97 Implementations MUST be safe for concurrent async access. All 

98 mutations (increment, mark_exhausted, record_error) are fire-and-forget 

99 from the router's perspective; backends that persist to a database 

100 should absorb their own errors rather than propagating them. 

101 

102 Example: 

103 ```python 

104 if await backend.is_exhausted("groq"): 

105 # Skip this provider 

106 ... 

107 await backend.increment("groq") 

108 ``` 

109 """ 

110 

111 async def is_exhausted(self, provider: str) -> bool: 

112 """Return ``True`` when the provider has exceeded its daily quota. 

113 

114 Args: 

115 provider: Provider name (e.g. ``"groq"``). 

116 

117 Returns: 

118 ``True`` when the provider should be skipped. 

119 """ 

120 ... 

121 

122 async def increment(self, provider: str) -> None: 

123 """Record one successful completion for *provider* today. 

124 

125 Args: 

126 provider: Provider name. 

127 """ 

128 ... 

129 

130 async def mark_exhausted( 

131 self, provider: str, *, until: datetime | None = None 

132 ) -> None: 

133 """Mark *provider* exhausted. 

134 

135 Called when the provider returns HTTP 429 (transient throttle — 

136 pass a short ``until`` cooldown) or 402 (account-level — leave 

137 ``until`` unset). 

138 

139 Args: 

140 provider: Cascade-entry key (``name:model``) or provider name. 

141 until: Exhaustion expiry. ``None`` means the rest of the 

142 current UTC day (legacy behavior). 

143 """ 

144 ... 

145 

146 async def record_error(self, provider: str) -> None: 

147 """Record a non-exhaustion error for *provider*. 

148 

149 Used for observability; does not affect quota state. 

150 

151 Args: 

152 provider: Provider name. 

153 """ 

154 ... 

155 

156 async def get_usage(self, provider: str) -> Any: 

157 """Return today's usage record for *provider*. 

158 

159 Args: 

160 provider: Provider name. 

161 

162 Returns: 

163 ``ProviderUsage`` dataclass or ``None`` when no record exists. 

164 """ 

165 ... 

166 

167 async def get_all_usage(self) -> Any: 

168 """Return all today's usage records. 

169 

170 Returns: 

171 ``list[ProviderUsage]`` for all tracked providers. 

172 """ 

173 ... 

174 

175 

176@runtime_checkable 

177class InferenceLoggerProtocol(Protocol): 

178 """Protocol for recording inference attempt logs. 

179 

180 Every routing run — successful or failed — should be logged for 

181 observability. Implementations MUST NOT raise exceptions; logging 

182 failures are absorbed silently so they never interrupt inference. 

183 

184 Example: 

185 ```python 

186 await logger.log(inference_log) 

187 recent = await logger.get_recent(limit=20) 

188 ``` 

189 """ 

190 

191 async def log(self, entry: Any) -> None: 

192 """Record one ``InferenceLog`` entry. 

193 

194 Args: 

195 entry: ``InferenceLog`` dataclass instance. 

196 """ 

197 ... 

198 

199 async def get_recent(self, limit: int = 100) -> list[Any]: 

200 """Return the most recent *limit* log entries. 

201 

202 Args: 

203 limit: Maximum number of entries (newest-first). 

204 

205 Returns: 

206 List of ``InferenceLog`` dataclass instances. 

207 """ 

208 ...