Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/clients/groq.py: 27%
133 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Groq provider for ultra-fast LLM inference.
3Groq provides blazing-fast inference speeds with their custom LPU (Language Processing Unit)
4hardware, offering up to 10x faster token generation than traditional GPUs.
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
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
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)
30API Documentation: https://console.groq.com/docs
32"""
34from __future__ import annotations
36from collections.abc import AsyncIterator
37import types
38from typing import Any, cast
40import aiohttp
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
74logger = get_logger(__name__)
77class GroqClient(AbstractLLMClient):
78 """Client for Groq's ultra-fast LLM inference API.
80 Conforms to: :class:`~lexigram.contracts.ai.LLMClientProtocol` protocol via structural typing.
82 Supports Chat, Stream, and Vision with:
83 - Ultra-fast LPU hardware synergy
84 - OpenAI-compatible API surface
85 - Blazing-fast token generation
86 """
88 def __init__(self, config: ClientConfig):
89 """Initialize Groq client.
91 Args:
92 config: LLM configuration
93 """
94 super().__init__(config=config)
95 self._client: ResilientHTTPClient | None = None
97 @property
98 def api_key(self) -> SecretStr:
99 """Get API key from config."""
100 return self.config.api_key or SecretStr("")
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"
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
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
131 def _get_client(self) -> ResilientHTTPClient:
132 """Get or create HTTP client.
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
149 async def _do_complete(
150 self,
151 messages: list[ChatMessage],
152 **kwargs: Any,
153 ) -> Result[Completion, LLMError]:
154 """Generate chat completion.
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.
165 Returns:
166 ``Ok(Completion)`` for non-streaming success.
167 ``Ok(AsyncIterator[StreamChunk])`` for streaming success.
168 ``Err(LLMError)`` for recoverable failures.
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)
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)
189 # Build request payload
190 payload = {
191 "model": model,
192 "messages": message_dicts,
193 "temperature": temperature,
194 "stream": False,
195 **kwargs,
196 }
198 if max_tokens:
199 payload["max_tokens"] = max_tokens
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 ]
208 # Non-streaming request
209 response = await client.post("/chat/completions", json=payload)
210 response.raise_for_status()
211 data = response.json()
213 # Parse response
214 choice = data["choices"][0]
215 message = choice["message"]
217 # Handle function calling
218 tool_calls = parse_openai_tool_calls(message.get("tool_calls"))
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)
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).
243 Args:
244 messages: List of chat messages (ChatMessage or dict).
245 **kwargs: Additional Groq API parameters.
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)
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)
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
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)
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)
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 )
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", []))
317 async def _stream_completion(
318 self,
319 client: ResilientHTTPClient,
320 payload: dict[str, Any],
321 ) -> AsyncIterator[StreamChunk]:
322 """Stream chat completion.
324 Args:
325 client: HTTP client.
326 payload: Request payload.
328 Yields:
329 StreamChunk objects.
331 """
332 try:
333 async with client.stream(
334 "POST",
335 "/chat/completions",
336 json=payload,
337 ) as response:
338 response.raise_for_status()
340 async for line in response.aiter_lines():
341 if not line or line == "data: [DONE]":
342 continue
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", {})
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 )
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
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
379 async def close(self) -> None:
380 """Close the HTTP client.
382 Example:
383 >>> await client.close()
385 """
386 if self._client:
387 await self._client.close()
388 self._client = None
389 await super().close()
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}