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

1"""Domain types for LLM multi-provider routing results and errors. 

2 

3All types use the ``@dataclass(init=False)`` + ``DomainModel`` pattern 

4consistent with the rest of ``lexigram-ai-llm``. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass 

10from datetime import UTC, datetime 

11from typing import Any 

12import uuid 

13 

14from lexigram.domain import DomainModel 

15from lexigram.validation import Field 

16 

17__all__ = [ 

18 "InferenceError", 

19 "InferenceLog", 

20 "InferenceResult", 

21 "ProviderUsage", 

22] 

23 

24 

25@dataclass(init=False) 

26class InferenceResult(DomainModel): 

27 """Successful result of a single provider inference attempt. 

28 

29 Example: 

30 >>> result = InferenceResult( 

31 ... provider="groq", 

32 ... model="llama-3.1-70b-versatile", 

33 ... content="Hello!", 

34 ... attempt=1, 

35 ... ) 

36 """ 

37 

38 provider: str = Field(description="Provider that produced this completion.") 

39 model: str = Field(description="Model identifier used for this attempt.") 

40 content: str = Field(description="Generated completion text.") 

41 attempt: int = Field( 

42 default=1, ge=1, description="Attempt number (1 = first provider tried)." 

43 ) 

44 prompt_tokens: int = Field(default=0, ge=0, description="Input token count.") 

45 completion_tokens: int = Field(default=0, ge=0, description="Output token count.") 

46 is_paid: bool = Field( 

47 default=False, description="Whether the request was paid or used free quota." 

48 ) 

49 latency_ms: float = Field( 

50 default=0.0, 

51 ge=0.0, 

52 description="Round-trip latency in milliseconds.", 

53 ) 

54 

55 

56@dataclass(init=False) 

57class InferenceError(DomainModel): 

58 """Terminal failure after all providers were exhausted. 

59 

60 Example: 

61 >>> error = InferenceError( 

62 ... message="All providers exhausted", 

63 ... providers_tried=["groq", "gemini"], 

64 ... last_status_code=429, 

65 ... ) 

66 """ 

67 

68 message: str = Field(description="Human-readable failure summary.") 

69 providers_tried: list[str] = Field( 

70 default_factory=list, 

71 description="Ordered list of providers attempted before giving up.", 

72 ) 

73 last_status_code: int | None = Field( 

74 default=None, 

75 description="HTTP status code from the last failed attempt.", 

76 ) 

77 

78 

79@dataclass(init=False) 

80class InferenceLog(DomainModel): 

81 """Complete record of one routing run — stored by ``InferenceLoggerProtocol``. 

82 

83 Contains the successful result (or a terminal error) together with the 

84 full provider attempt trail for observability. 

85 

86 Example: 

87 >>> log = InferenceLog( 

88 ... routing_id=uuid.uuid4(), 

89 ... result=InferenceResult(provider="groq", model="llama-3.1-70b-versatile", content="Hello!"), 

90 ... providers_tried=["gemini", "groq"], 

91 ... total_attempts=2, 

92 ... ) 

93 """ 

94 

95 routing_id: str = Field( 

96 default_factory=lambda: str(uuid.uuid4()), 

97 description="Unique identifier for this routing run.", 

98 ) 

99 result: InferenceResult | None = Field( 

100 default=None, 

101 description="Successful completion result, or ``None`` on total failure.", 

102 ) 

103 error: InferenceError | None = Field( 

104 default=None, 

105 description="Terminal error when all providers failed.", 

106 ) 

107 providers_tried: list[str] = Field( 

108 default_factory=list, 

109 description="All providers attempted (including those exhausted at quota check).", 

110 ) 

111 total_attempts: int = Field( 

112 default=0, 

113 ge=0, 

114 description="Number of actual HTTP calls made (quota-skipped providers not counted).", 

115 ) 

116 context: dict[str, Any] = Field( 

117 default_factory=dict, 

118 description="Arbitrary context metadata (application-level tags, request IDs, etc.).", 

119 ) 

120 created_at: datetime = Field( 

121 default_factory=lambda: datetime.now(UTC), 

122 description="Timestamp when the routing run completed.", 

123 ) 

124 

125 @property 

126 def succeeded(self) -> bool: 

127 """``True`` when ``result`` is populated (routing produced a completion). 

128 

129 Returns: 

130 Whether the routing run was successful. 

131 """ 

132 return self.result is not None 

133 

134 

135@dataclass(init=False) 

136class ProviderUsage(DomainModel): 

137 """Quota usage record for a single provider on a single calendar day. 

138 

139 Example: 

140 >>> usage = ProviderUsage( 

141 ... provider="groq", 

142 ... usage_date="2026-03-21", 

143 ... success_count=42, 

144 ... error_count=1, 

145 ... is_exhausted=False, 

146 ... ) 

147 """ 

148 

149 provider: str = Field(description="Provider name.") 

150 usage_date: str = Field( 

151 description="Calendar date (ISO 8601, UTC) this record tracks.", 

152 ) 

153 success_count: int = Field( 

154 default=0, 

155 ge=0, 

156 description="Number of successful completions today.", 

157 ) 

158 error_count: int = Field( 

159 default=0, 

160 ge=0, 

161 description="Number of non-exhaustion errors today.", 

162 ) 

163 is_exhausted: bool = Field( 

164 default=False, 

165 description="``True`` when the provider has been quota-exhausted for today.", 

166 ) 

167 exhausted_until: datetime | None = Field( 

168 default=None, 

169 description=( 

170 "Exhaustion expiry. The provider is skipped only while " 

171 "``now < exhausted_until``; ``None`` means not exhausted." 

172 ), 

173 )