Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/routing/strategies/sequential.py: 34%

1"""Sequential Cascade Routing Strategy.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.ai.llm.routing.config import LLMConfig, ProviderConfig 

8from lexigram.ai.llm.routing.strategies.base import ( 

9 _attempt_provider, 

10 _gen_defaults, 

11 _handle_free_failure, 

12) 

13from lexigram.ai.llm.routing.types import InferenceResult 

14from lexigram.ai.llm.types import AIError 

15from lexigram.contracts.ai import LLMClientProtocol 

16from lexigram.contracts.ai.routing import QuotaBackendProtocol 

17from lexigram.logging import ( 

18 get_logger, 

19) 

20 

21logger = get_logger(__name__) 

22 

23 

24class SequentialCascadeStrategy: 

25 """Try providers one at a time in configuration order. 

26 

27 For each enabled, non-exhausted provider the strategy attempts the 

28 configured model once. 

29 """ 

30 

31 async def execute( 

32 self, 

33 *, 

34 providers: list[ProviderConfig], 

35 clients: dict[str, LLMClientProtocol], 

36 quota: QuotaBackendProtocol, 

37 config: LLMConfig, 

38 messages: list[Any], 

39 kwargs: dict[str, Any], 

40 ) -> tuple[InferenceResult | None, list[str], int]: 

41 temperature, max_tokens = _gen_defaults(config, kwargs) 

42 providers_tried: list[str] = [] 

43 total_attempts = 0 

44 

45 for pcfg in providers: 

46 if not pcfg.enabled: 

47 continue 

48 client = clients.get(pcfg.key) 

49 if client is None: 

50 continue 

51 

52 providers_tried.append(pcfg.name) 

53 

54 if await quota.is_exhausted(pcfg.key): 

55 logger.debug("sequential: skipping exhausted %s", pcfg.key) 

56 continue 

57 

58 model = pcfg.model 

59 total_attempts += 1 

60 try: 

61 result = await _attempt_provider( 

62 client=client, 

63 provider_name=pcfg.name, 

64 model=model, 

65 messages=messages, 

66 temperature=temperature, 

67 max_tokens=max_tokens, 

68 ) 

69 await quota.increment(pcfg.key) 

70 logger.info( 

71 "sequential: success provider=%s model=%s", 

72 pcfg.name, 

73 model, 

74 ) 

75 return result, providers_tried, total_attempts 

76 except AIError as exc: 

77 await _handle_free_failure( 

78 exc=exc, 

79 provider_key=pcfg.key, 

80 model=model, 

81 quota=quota, 

82 cooldown_seconds=config.quota.cooldown_seconds, 

83 ) 

84 

85 return None, providers_tried, total_attempts