Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/clients/groq.py: 27%

1"""Groq provider for ultra-fast LLM inference. 

2 

3Groq provides blazing-fast inference speeds with their custom LPU (Language Processing Unit) 

4hardware, offering up to 10x faster token generation than traditional GPUs. 

5 

6Supported Models: 

7- llama-3.1-405b-reasoning: Meta's largest Llama 3.1 model 

8- llama-3.1-70b-versatile: 70B parameter balanced model 

9- llama-3.1-8b-instant: Fast 8B parameter model 

10- mixtral-8x7b-32768: Mistral's MoE model with 32k context 

11- gemma-7b-it: Google's Gemma 7B instruction-tuned 

12 

13Features: 

14- Ultra-fast inference (100+ tokens/second) 

15- Low latency (<1 second time-to-first-token) 

16- OpenAI-compatible API 

17- Function calling support 

18- Streaming support 

19 

20Example: 

21 >>> from lexigram.ai.llm import GroqClient 

22 >>> 

23 >>> async with GroqClient(api_key="gsk_...") as client: 

24 ... response = await client.complete( 

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

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

27 ... ) 

28 ... print(response.content) 

29 

30API Documentation: https://console.groq.com/docs 

31 

32""" 

33 

34from __future__ import annotations 

35 

36from collections.abc import AsyncIterator 

37import types 

38from typing import Any, cast 

39 

40import aiohttp 

41 

42from lexigram.ai.llm.clients._message_utils import serialize_content_for_openai 

43from lexigram.ai.llm.clients._tools_utils import ( 

44 parse_openai_tool_calls, 

45 serialize_message_for_openai, 

46 tool_to_openai_format, 

47) 

48from lexigram.ai.llm.clients.base import AbstractLLMClient 

49from lexigram.ai.llm.config import ClientConfig 

50from lexigram.ai.llm.exceptions import ( 

51 LLMAuthenticationError, 

52 LLMError, 

53 LLMModelNotFoundError, 

54 LLMQuotaExceededError, 

55 LLMRateLimitError, 

56) 

57from lexigram.ai.llm.http.client import ResilientHTTPClient 

58from lexigram.ai.llm.types import ( 

59 AIError, 

60 ChatMessage, 

61 Completion, 

62 StreamChunk, 

63 TokenUsage, 

64 ToolCall, 

65) 

66from lexigram.contracts.core.health import HealthCheckResult, HealthStatus 

67from lexigram.logging import ( 

68 get_logger, 

69) 

70from lexigram.result import Err, Ok, Result 

71from lexigram.serialization import loads 

72from lexigram.validation import SecretStr 

73 

74logger = get_logger(__name__) 

75 

76 

77class GroqClient(AbstractLLMClient): 

78 """Client for Groq's ultra-fast LLM inference API. 

79 

80 Conforms to: :class:`~lexigram.contracts.ai.LLMClientProtocol` protocol via structural typing. 

81 

82 Supports Chat, Stream, and Vision with: 

83 - Ultra-fast LPU hardware synergy 

84 - OpenAI-compatible API surface 

85 - Blazing-fast token generation 

86 """ 

87 

88 def __init__(self, config: ClientConfig): 

89 """Initialize Groq client. 

90 

91 Args: 

92 config: LLM configuration 

93 """ 

94 super().__init__(config=config) 

95 self._client: ResilientHTTPClient | None = None 

96 

97 @property 

98 def api_key(self) -> SecretStr: 

99 """Get API key from config.""" 

100 return self.config.api_key or SecretStr("") 

101 

102 @property 

103 def base_url(self) -> str: 

104 """Get base URL from config.""" 

105 return self.config.api_base or "https://api.groq.com/openai/v1" 

106 

107 async def __aenter__(self) -> Any: 

108 """Async context manager entry.""" 

109 self._client = ResilientHTTPClient( 

110 base_url=self.base_url, 

111 headers={ 

112 "Authorization": f"Bearer {self.api_key.get_secret_value()}", 

113 "Content-Type": "application/json", 

114 }, 

115 timeout=self.config.timeout, 

116 name="groq-client", 

117 ) 

118 return self 

119 

120 async def __aexit__( 

121 self, 

122 exc_type: type[BaseException] | None, 

123 exc_val: BaseException | None, 

124 exc_tb: types.TracebackType | None, 

125 ) -> Any: 

126 """Async context manager exit.""" 

127 if self._client: 

128 await self._client.close() 

129 self._client = None 

130 

131 def _get_client(self) -> ResilientHTTPClient: 

132 """Get or create HTTP client. 

133 

134 Returns: 

135 HTTP client instance. 

136 """ 

137 if self._client is None: 

138 self._client = ResilientHTTPClient( 

139 base_url=self.base_url, 

140 headers={ 

141 "Authorization": f"Bearer {self.api_key.get_secret_value()}", 

142 "Content-Type": "application/json", 

143 }, 

144 timeout=self.config.timeout, 

145 name="groq-client", 

146 ) 

147 return self._client 

148 

149 async def _do_complete( 

150 self, 

151 messages: list[ChatMessage], 

152 **kwargs: Any, 

153 ) -> Result[Completion, LLMError]: 

154 """Generate chat completion. 

155 

156 Args: 

157 model: Model ID (e.g., "llama-3.1-70b-versatile"). 

158 messages: List of messages (ChatMessage or dict). 

159 temperature: Sampling temperature (0-2, default: 0.7). 

160 max_tokens: Maximum tokens to generate. 

161 stream: Whether to stream the response. 

162 tools: Function calling tools (OpenAI format). 

163 **kwargs: Additional parameters. 

164 

165 Returns: 

166 ``Ok(Completion)`` for non-streaming success. 

167 ``Ok(AsyncIterator[StreamChunk])`` for streaming success. 

168 ``Err(LLMError)`` for recoverable failures. 

169 

170 Raises: 

171 LLMAuthenticationError: If API key is invalid. 

172 AIError: For unexpected infrastructure failures. 

173 """ 

174 try: 

175 client = self._get_client() 

176 model = kwargs.pop("model", self.config.model) 

177 temperature = kwargs.pop("temperature", self.config.temperature) 

178 max_tokens = kwargs.pop("max_tokens", self.config.max_tokens) 

179 tools = kwargs.pop("tools", None) 

180 

181 # Convert ChatMessage to dict 

182 message_dicts: list[dict[str, Any]] = [] 

183 for msg in cast("list[ChatMessage | dict[str, Any]]", messages): 

184 if isinstance(msg, ChatMessage): 

185 message_dicts.append(serialize_message_for_openai(msg)) 

186 else: 

187 message_dicts.append(msg) 

188 

189 # Build request payload 

190 payload = { 

191 "model": model, 

192 "messages": message_dicts, 

193 "temperature": temperature, 

194 "stream": False, 

195 **kwargs, 

196 } 

197 

198 if max_tokens: 

199 payload["max_tokens"] = max_tokens 

200 

201 if tools: 

202 payload["tools"] = [ 

203 converted 

204 for tool in tools 

205 if (converted := tool_to_openai_format(tool)) is not None 

206 ] 

207 

208 # Non-streaming request 

209 response = await client.post("/chat/completions", json=payload) 

210 response.raise_for_status() 

211 data = response.json() 

212 

213 # Parse response 

214 choice = data["choices"][0] 

215 message = choice["message"] 

216 

217 # Handle function calling 

218 tool_calls = parse_openai_tool_calls(message.get("tool_calls")) 

219 

220 return Ok( 

221 Completion( 

222 content=message.get("content", ""), 

223 model=model, 

224 finish_reason=choice.get("finish_reason", "stop"), 

225 usage=TokenUsage( 

226 prompt_tokens=data["usage"]["prompt_tokens"], 

227 completion_tokens=data["usage"]["completion_tokens"], 

228 total_tokens=data["usage"]["total_tokens"], 

229 ), 

230 tool_calls=tool_calls, 

231 ) 

232 ) 

233 except (OSError, ConnectionError, TimeoutError, RuntimeError, ValueError) as e: 

234 return self._handle_error_as_result(e) 

235 

236 async def _do_stream_chat( 

237 self, 

238 messages: list[ChatMessage], 

239 **kwargs: Any, 

240 ) -> Result[AsyncIterator[StreamChunk], LLMError]: 

241 """Start a streaming completion (protocol-aligned method). 

242 

243 Args: 

244 messages: List of chat messages (ChatMessage or dict). 

245 **kwargs: Additional Groq API parameters. 

246 

247 Returns: 

248 ``Ok(AsyncIterator[StreamChunk])`` on success. 

249 ``Err(LLMError)`` for recoverable failures. 

250 """ 

251 try: 

252 client = self._get_client() 

253 model = kwargs.pop("model", self.config.model) 

254 temperature = kwargs.pop("temperature", self.config.temperature) 

255 max_tokens = kwargs.pop("max_tokens", self.config.max_tokens) 

256 

257 # Convert ChatMessage to dict 

258 message_dicts: list[dict[str, Any]] = [] 

259 for msg in cast("list[ChatMessage | dict[str, Any]]", messages): 

260 if isinstance(msg, ChatMessage): 

261 message_dicts.append( 

262 { 

263 "role": msg.role.value, 

264 "content": serialize_content_for_openai(msg.content), 

265 } 

266 ) 

267 else: 

268 message_dicts.append(msg) 

269 

270 payload = { 

271 "model": model, 

272 "messages": message_dicts, 

273 "temperature": temperature, 

274 "stream": True, 

275 **kwargs, 

276 } 

277 if max_tokens: 

278 payload["max_tokens"] = max_tokens 

279 

280 return Ok(self._stream_completion(client, payload)) 

281 except (OSError, ConnectionError, TimeoutError, RuntimeError, ValueError) as e: 

282 return self._handle_error_as_result(e) 

283 

284 async def _do_chat( 

285 self, 

286 messages: list[ChatMessage], 

287 tools: list[ToolCall] | None = None, 

288 **kwargs: Any, 

289 ) -> Result[Completion, LLMError]: 

290 return await self._do_complete(messages, tools=tools, **kwargs) 

291 

292 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

293 """Run a lightweight provider health probe.""" 

294 try: 

295 _ = await self.list_models() 

296 return HealthCheckResult( 

297 component="llm.groq", 

298 status=HealthStatus.HEALTHY, 

299 details={"model": self.config.model}, 

300 ) 

301 except (OSError, aiohttp.ClientError, RuntimeError, ValueError) as exc: 

302 return HealthCheckResult( 

303 component="llm.groq", 

304 status=HealthStatus.UNHEALTHY, 

305 error=str(exc), 

306 details={"model": self.config.model}, 

307 ) 

308 

309 async def list_models(self) -> list[dict[str, Any]]: 

310 """List models available from the Groq API.""" 

311 client = self._get_client() 

312 response = await client.get("/models") 

313 response.raise_for_status() 

314 data = response.json() 

315 return cast("list[dict[str, Any]]", data.get("data", [])) 

316 

317 async def _stream_completion( 

318 self, 

319 client: ResilientHTTPClient, 

320 payload: dict[str, Any], 

321 ) -> AsyncIterator[StreamChunk]: 

322 """Stream chat completion. 

323 

324 Args: 

325 client: HTTP client. 

326 payload: Request payload. 

327 

328 Yields: 

329 StreamChunk objects. 

330 

331 """ 

332 try: 

333 async with client.stream( 

334 "POST", 

335 "/chat/completions", 

336 json=payload, 

337 ) as response: 

338 response.raise_for_status() 

339 

340 async for line in response.aiter_lines(): 

341 if not line or line == "data: [DONE]": 

342 continue 

343 

344 if line.startswith("data: "): 

345 try: 

346 data = loads(line[6:].encode("utf-8")) 

347 choice = data["choices"][0] 

348 delta = choice.get("delta", {}) 

349 

350 content = delta.get("content", "") 

351 if content: 

352 yield StreamChunk( 

353 delta=content, 

354 model=data["model"], 

355 finish_reason=choice.get("finish_reason"), 

356 ) 

357 

358 except (ValueError, TypeError) as e: 

359 logger.debug("Error parsing stream chunk: %s", e) 

360 continue 

361 except (aiohttp.ClientError, OSError, ValueError, RuntimeError) as e: 

362 raise AIError(f"Groq streaming error: {e}") from e 

363 

364 def _handle_error_as_result(self, error: Exception) -> Result[Any, LLMError]: 

365 """Map a caught exception to ``Err`` (recoverable) or re-raise (infra).""" 

366 if isinstance(error, aiohttp.ClientResponseError): 

367 if error.status == 401: 

368 raise LLMAuthenticationError( 

369 f"Groq authentication failed: {error}" 

370 ) from error 

371 if error.status == 429: 

372 return Err(LLMRateLimitError(f"Groq rate limit exceeded: {error}")) 

373 if error.status == 402: 

374 return Err(LLMQuotaExceededError(f"Groq quota exceeded: {error}")) 

375 if error.status == 404: 

376 return Err(LLMModelNotFoundError(f"Groq model not found: {error}")) 

377 raise AIError(f"Groq infrastructure error: {error}") from error 

378 

379 async def close(self) -> None: 

380 """Close the HTTP client. 

381 

382 Example: 

383 >>> await client.close() 

384 

385 """ 

386 if self._client: 

387 await self._client.close() 

388 self._client = None 

389 await super().close() 

390 

391 

392# Common model configurations for convenience 

393GROQ_MODELS = { 

394 "llama-3.1-405b-reasoning": { 

395 "context_window": 8192, 

396 "description": "Meta's largest Llama 3.1 model - best reasoning", 

397 }, 

398 "llama-3.1-70b-versatile": { 

399 "context_window": 8192, 

400 "description": "Balanced 70B model - good for most tasks", 

401 }, 

402 "llama-3.1-8b-instant": { 

403 "context_window": 8192, 

404 "description": "Fast 8B model - instant responses", 

405 }, 

406 "mixtral-8x7b-32768": { 

407 "context_window": 32768, 

408 "description": "Mistral's MoE model - large context window", 

409 }, 

410 "gemma-7b-it": { 

411 "context_window": 8192, 

412 "description": "Google's Gemma 7B - instruction tuned", 

413 }, 

414}