Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/clients/openai.py: 24%
116 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""OpenAI LLM client implementation.
3Production-ready implementation with OpenAI API integration.
4"""
6from __future__ import annotations
8from collections.abc import AsyncGenerator, AsyncIterator
9from datetime import UTC, datetime
10from typing import Any
12from lexigram.ai.llm.clients._tools_utils import (
13 parse_openai_tool_calls,
14 serialize_message_for_openai,
15 tool_to_openai_format,
16)
17from lexigram.ai.llm.clients.base import AbstractLLMClient
18from lexigram.ai.llm.config import ClientConfig
19from lexigram.ai.llm.exceptions import (
20 LLMAuthenticationError,
21 LLMContentFilterError,
22 LLMError,
23 LLMModelNotFoundError,
24 LLMQuotaExceededError,
25 LLMRateLimitError,
26)
27from lexigram.ai.llm.types import (
28 AIError,
29 ChatMessage,
30 Completion,
31 StreamChunk,
32 ThinkingResult,
33 TokenUsage,
34)
35from lexigram.contracts.core import HealthCheckResult, HealthStatus
36from lexigram.result import Err, Ok, Result
39class OpenAIClient(AbstractLLMClient):
40 """OpenAI LLM client implementation.
42 Conforms to: :class:`~lexigram.contracts.ai.LLMClientProtocol` protocol via structural typing.
44 Supports GPT-4, GPT-3.5-Turbo, and other OpenAI models with:
45 - Streaming responses
46 - Function/tool calling
47 - Vision models
48 - Automatic retry with exponential backoff
49 - Error handling and rate limit management
51 Example:
52 >>> from lexigram.ai import ClientConfig
53 >>> config = ClientConfig(provider="openai", model="gpt-4-turbo")
54 >>> client = OpenAIClient(config)
55 >>> completion = await client.complete([
56 ... ChatMessage(role="user", content="Hello!")
57 ... ])
58 """
60 def __init__(self, config: ClientConfig):
61 """Initialize OpenAI client.
63 Args:
64 config: LLM configuration
66 Raises:
67 ImportError: If openai package is not installed
68 """
69 self.config = config
71 try:
72 from openai import AsyncOpenAI
73 except ImportError as e:
74 raise ImportError(
75 "OpenAI client requires 'openai' package. "
76 "Install with: pip install lexigram-ai-llm[openai]",
77 ) from e
79 api_key = config.api_key.get_secret_value() if config.api_key else None
80 self.client = AsyncOpenAI(
81 api_key=api_key,
82 base_url=config.api_base,
83 timeout=config.timeout,
84 )
85 super().__init__(config=config)
87 async def _do_complete(
88 self,
89 messages: list[ChatMessage],
90 **kwargs: Any,
91 ) -> Result[Completion, LLMError]:
92 """Generate completion from messages.
94 Args:
95 messages: Chat messages
96 **kwargs: Additional OpenAI API parameters (temperature, max_tokens, etc.)
98 Returns:
99 ``Ok(Completion)`` on success.
100 ``Err(LLMRateLimitError | LLMQuotaExceededError | LLMContentFilterError
101 | LLMModelNotFoundError)`` for recoverable domain failures.
103 Raises:
104 LLMAuthenticationError: If API key is invalid.
105 AIError: For unexpected infrastructure failures.
106 """
107 try:
108 # Convert messages to OpenAI format
109 openai_messages = [self._convert_message(msg) for msg in messages]
111 # Pop positional overrides so **kwargs doesn't contain duplicates
112 _model = kwargs.pop("model", self.config.model)
113 _max_tokens = kwargs.pop("max_tokens", self.config.max_tokens)
114 _temperature = kwargs.pop("temperature", self.config.temperature)
115 _tools = kwargs.pop("tools", None)
117 params: dict[str, Any] = {
118 "model": _model,
119 "messages": openai_messages,
120 **kwargs,
121 }
122 if _max_tokens is not None:
123 params["max_tokens"] = _max_tokens
124 if _tools:
125 params["tools"] = [
126 converted
127 for tool in _tools
128 if (converted := tool_to_openai_format(tool)) is not None
129 ]
131 self._apply_thinking(params)
132 if "reasoning_effort" not in params:
133 params["temperature"] = _temperature
135 # Make API call
136 response = await self.client.chat.completions.create(**params)
138 # Validate response structure
139 if not hasattr(response, "choices") or not response.choices:
140 msg = "OpenAI returned an invalid response structure"
141 raise AIError(msg)
143 # Convert to our Completion type
144 choice = response.choices[0]
145 tool_calls = parse_openai_tool_calls(
146 getattr(choice.message, "tool_calls", None)
147 )
148 # DeepSeek-style providers surface reasoning on the message object
149 reasoning_content: str | None = getattr(
150 choice.message, "reasoning_content", None
151 )
152 # OpenAI o-series: track reasoning token count if available
153 reasoning_tokens: int | None = None
154 if response.usage:
155 details = getattr(response.usage, "completion_tokens_details", None)
156 if details and hasattr(details, "reasoning_tokens"):
157 reasoning_tokens = details.reasoning_tokens
158 thinking: ThinkingResult | None = (
159 ThinkingResult(content=reasoning_content or "", tokens=reasoning_tokens)
160 if (reasoning_content or reasoning_tokens is not None)
161 else None
162 )
164 return Ok(
165 Completion(
166 content=choice.message.content or "",
167 model=response.model,
168 finish_reason=choice.finish_reason,
169 tool_calls=tool_calls,
170 thinking=thinking,
171 usage=(
172 TokenUsage(
173 prompt_tokens=response.usage.prompt_tokens,
174 completion_tokens=response.usage.completion_tokens,
175 total_tokens=response.usage.total_tokens,
176 )
177 if response.usage
178 else None
179 ),
180 metadata={
181 "id": response.id,
182 "created": response.created,
183 "system_fingerprint": response.system_fingerprint,
184 },
185 timestamp=datetime.now(UTC),
186 )
187 )
189 except (ValueError, ConnectionError, TimeoutError, OSError) as e:
190 return self._handle_error_as_result(e)
192 async def _do_stream_chat(
193 self,
194 messages: list[ChatMessage],
195 **kwargs: Any,
196 ) -> Result[AsyncIterator[StreamChunk], LLMError]:
197 """Start a streaming completion.
199 Args:
200 messages: Chat messages
201 **kwargs: Additional OpenAI API parameters
203 Returns:
204 ``Ok(AsyncIterator[StreamChunk])`` on successful connection.
205 ``Err(LLMError)`` if the connection could not be established.
206 Mid-stream errors propagate as exceptions from the iterator.
208 Raises:
209 LLMAuthenticationError: If API key is invalid.
210 AIError: For unexpected infrastructure failures.
211 """
212 try:
213 # Convert messages to OpenAI format
214 openai_messages = [self._convert_message(msg) for msg in messages]
216 # Pop positional overrides so **kwargs doesn't contain duplicates
217 _model = kwargs.pop("model", self.config.model)
218 _max_tokens = kwargs.pop("max_tokens", self.config.max_tokens)
219 _temperature = kwargs.pop("temperature", self.config.temperature)
221 params: dict[str, Any] = {
222 "model": _model,
223 "messages": openai_messages,
224 "stream": True,
225 **kwargs,
226 }
227 if _max_tokens is not None:
228 params["max_tokens"] = _max_tokens
230 self._apply_thinking(params)
231 if "reasoning_effort" not in params:
232 params["temperature"] = _temperature
234 # Establish stream — this is where connection-level errors surface
235 stream = await self.client.chat.completions.create(**params)
236 return Ok(self._stream_impl(stream))
238 except (ValueError, ConnectionError, TimeoutError, OSError) as e:
239 return self._handle_error_as_result(e)
241 async def _stream_impl(self, stream: Any) -> AsyncGenerator[StreamChunk, None]:
242 """Yield StreamChunk objects from an established OpenAI stream.
244 Handles both standard text deltas and DeepSeek-style
245 ``reasoning_content`` deltas, emitting thinking chunks first.
246 """
247 index = 0
248 async for chunk in stream:
249 if not chunk.choices:
250 continue
251 choice = chunk.choices[0]
252 # DeepSeek-style providers emit reasoning_content on the delta
253 thinking = getattr(choice.delta, "reasoning_content", None)
254 if thinking:
255 yield StreamChunk(
256 thinking_delta=thinking,
257 is_thinking=True,
258 model=chunk.model,
259 finish_reason=choice.finish_reason,
260 index=index,
261 )
262 index += 1
263 elif choice.delta.content:
264 yield StreamChunk(
265 delta=choice.delta.content,
266 model=chunk.model,
267 finish_reason=choice.finish_reason,
268 index=index,
269 )
270 index += 1
272 async def _do_chat(
273 self,
274 messages: list[ChatMessage],
275 tools: list[Any] | None = None,
276 **kwargs: Any,
277 ) -> Result[Completion, LLMError]:
278 """Generate completion with optional tool/function calling.
280 Tool calling is handled by :meth:`_do_complete` via
281 ``complete(..., tools=...)``; this method forwards the tool
282 descriptors to keep the ``chat`` code path consistent.
284 Args:
285 messages: Chat messages
286 tools: Optional list of tools the LLM can call
287 **kwargs: Additional OpenAI API parameters
289 Returns:
290 ``Ok(Completion)`` on success. ``Err(LLMError)`` for recoverable
291 failures.
292 """
293 return await self._do_complete(messages, tools=tools, **kwargs)
295 def _apply_thinking(self, params: dict[str, Any]) -> None:
296 """Inject thinking/reasoning parameters into the API payload.
298 For suppression (``ThinkingConfig.suppress=True``): injects
299 ``enable_thinking: false`` and ``chat_template_kwargs.enable_thinking: false``
300 into ``extra_body`` so LM Studio, vLLM, SGLang and similar OpenAI-compatible
301 backends skip chain-of-thought generation.
303 For OpenAI o-series effort (``ThinkingConfig.effort``): sets
304 ``reasoning_effort`` and removes ``temperature`` (OpenAI rejects that
305 combination for reasoning models).
307 Args:
308 params: Mutable API payload dict modified in-place.
309 """
310 if self.config.thinking is None:
311 return
312 if self.config.thinking.suppress:
313 existing = params.get("extra_body") or {}
314 params["extra_body"] = {
315 **existing,
316 "enable_thinking": False,
317 "chat_template_kwargs": {"enable_thinking": False},
318 }
319 return
320 if self.config.thinking.effort:
321 params["reasoning_effort"] = self.config.thinking.effort
322 params.pop("temperature", None)
324 def _convert_message(self, msg: ChatMessage) -> dict[str, Any]:
325 """Convert ChatMessage to OpenAI message format.
327 Args:
328 msg: ChatMessage to convert
330 Returns:
331 OpenAI message dict
332 """
333 return serialize_message_for_openai(msg)
335 def _handle_error_as_result(self, error: Exception) -> Result[Any, LLMError]:
336 """Map a caught exception to ``Err`` (recoverable) or re-raise (infra).
338 Recoverable failures → ``Err(LLMRecoverableError)``.
339 Infrastructure failures → raised directly.
340 """
341 err_str = str(error).lower()
342 if "authentication" in err_str or "api key" in err_str:
343 raise LLMAuthenticationError(
344 f"OpenAI authentication failed: {error}"
345 ) from error
346 if "rate limit" in err_str:
347 return Err(LLMRateLimitError(f"OpenAI rate limit exceeded: {error}"))
348 if (
349 "quota" in err_str
350 or "billing" in err_str
351 or "insufficient_quota" in err_str
352 ):
353 return Err(LLMQuotaExceededError(f"OpenAI quota exceeded: {error}"))
354 if "content" in err_str and ("filter" in err_str or "policy" in err_str):
355 return Err(LLMContentFilterError(f"OpenAI content filter: {error}"))
356 if "model" in err_str and (
357 "not found" in err_str or "does not exist" in err_str
358 ):
359 return Err(LLMModelNotFoundError(f"OpenAI model not found: {error}"))
360 raise AIError(f"OpenAI infrastructure error: {error}") from error
362 async def close(self) -> None:
363 """Close the OpenAI client and cleanup resources."""
364 if not self._closed:
365 await self.client.close()
366 await super().close()
368 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
369 """Perform health check.
371 Returns:
372 Structured health check result.
373 """
374 if self._closed:
375 return HealthCheckResult(
376 component="llm.openai",
377 status=HealthStatus.UNHEALTHY,
378 error="Client is closed",
379 )
381 return HealthCheckResult(
382 component="llm.openai",
383 status=HealthStatus.HEALTHY,
384 details={
385 "provider": "openai",
386 "model": self.config.model,
387 },
388 )