app.ai_providers.openai_provider

  1"""OpenAI API provider implementation.
  2
  3Uses the ``openai`` Python SDK to communicate with the OpenAI Chat
  4Completions and Responses APIs. Supports synchronous and streaming
  5generation, CSV file attachments via the Responses API, and automatic
  6fallback between ``max_completion_tokens`` and ``max_tokens`` parameters.
  7
  8Attributes:
  9    logger: Module-level logger for OpenAI provider operations.
 10"""
 11
 12from __future__ import annotations
 13
 14import logging
 15from typing import Any, Callable, Iterator, Mapping
 16
 17from .base import (
 18    AIProvider,
 19    AIProviderError,
 20    DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
 21    DEFAULT_MAX_TOKENS,
 22    _is_attachment_unsupported_error,
 23    _is_context_length_error,
 24    _is_unsupported_parameter_error,
 25    _normalize_api_key_value,
 26    _resolve_completion_token_retry_limit,
 27    _resolve_timeout_seconds,
 28    _run_with_rate_limit_retries,
 29    _T,
 30)
 31from .utils import (
 32    _extract_openai_delta_text,
 33    _extract_openai_text,
 34    _inline_attachment_data_into_prompt,
 35    upload_and_request_via_responses_api,
 36)
 37
 38logger = logging.getLogger(__name__)
 39
 40DEFAULT_OPENAI_MODEL = "gpt-5.4"
 41
 42
 43class OpenAIProvider(AIProvider):
 44    """OpenAI API provider implementation.
 45
 46    Attributes:
 47        api_key (str): The OpenAI API key.
 48        model (str): The OpenAI model identifier.
 49        attach_csv_as_file (bool): Whether to upload CSV artifacts as
 50            file attachments via the Responses API.
 51        request_timeout_seconds (float): HTTP timeout in seconds.
 52        client: The ``openai.OpenAI`` SDK client instance.
 53    """
 54
 55    def __init__(
 56        self,
 57        api_key: str,
 58        model: str = DEFAULT_OPENAI_MODEL,
 59        attach_csv_as_file: bool = True,
 60        request_timeout_seconds: float = DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
 61    ) -> None:
 62        """Initialize the OpenAI provider.
 63
 64        Args:
 65            api_key: OpenAI API key. Must be non-empty.
 66            model: OpenAI model identifier.
 67            attach_csv_as_file: If ``True``, attempt file uploads via
 68                the Responses API.
 69            request_timeout_seconds: HTTP timeout in seconds.
 70
 71        Raises:
 72            AIProviderError: If the ``openai`` SDK is not installed or
 73                the API key is empty.
 74        """
 75        try:
 76            import openai
 77        except ImportError as error:
 78            raise AIProviderError(
 79                "openai SDK is not installed. Install it with `pip install openai`."
 80            ) from error
 81
 82        normalized_api_key = _normalize_api_key_value(api_key)
 83        if not normalized_api_key:
 84            raise AIProviderError(
 85                "OpenAI API key is not configured. "
 86                "Set `ai.openai.api_key` in config.yaml or the OPENAI_API_KEY environment variable."
 87            )
 88
 89        self._openai = openai
 90        self.api_key = normalized_api_key
 91        self.model = model
 92        self.attach_csv_as_file = bool(attach_csv_as_file)
 93        self._csv_attachment_supported: bool | None = None
 94        self.request_timeout_seconds = _resolve_timeout_seconds(
 95            request_timeout_seconds,
 96            DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
 97        )
 98        self.client = openai.OpenAI(
 99            api_key=normalized_api_key,
100            timeout=self.request_timeout_seconds,
101        )
102        logger.info("Initialized OpenAI provider with model %s (timeout %.1fs)", model, self.request_timeout_seconds)
103
104    def analyze(
105        self,
106        system_prompt: str,
107        user_prompt: str,
108        max_tokens: int = DEFAULT_MAX_TOKENS,
109    ) -> str:
110        """Send a prompt to OpenAI and return the generated text.
111
112        Args:
113            system_prompt: The system-level instruction text.
114            user_prompt: The user-facing prompt with investigation context.
115            max_tokens: Maximum completion tokens.
116
117        Returns:
118            The generated analysis text.
119
120        Raises:
121            AIProviderError: On any API or network failure.
122        """
123        return self.analyze_with_attachments(
124            system_prompt=system_prompt,
125            user_prompt=user_prompt,
126            attachments=None,
127            max_tokens=max_tokens,
128        )
129
130    def analyze_stream(
131        self,
132        system_prompt: str,
133        user_prompt: str,
134        max_tokens: int = DEFAULT_MAX_TOKENS,
135    ) -> Iterator[str]:
136        """Stream generated text chunks from OpenAI.
137
138        Args:
139            system_prompt: The system-level instruction text.
140            user_prompt: The user-facing prompt with investigation context.
141            max_tokens: Maximum completion tokens.
142
143        Yields:
144            Text chunk strings as they are generated.
145
146        Raises:
147            AIProviderError: On empty response or API failure.
148        """
149        def _stream() -> Iterator[str]:
150            messages = [
151                {"role": "system", "content": system_prompt},
152                {"role": "user", "content": user_prompt},
153            ]
154            stream = self._run_openai_request(
155                lambda: self._create_chat_completion(
156                    messages=messages,
157                    max_tokens=max_tokens,
158                    stream=True,
159                )
160            )
161            emitted = False
162            try:
163                for chunk in stream:
164                    choices = getattr(chunk, "choices", None)
165                    if not choices:
166                        continue
167                    choice = choices[0]
168                    delta = getattr(choice, "delta", None)
169                    if delta is None and isinstance(choice, dict):
170                        delta = choice.get("delta")
171                    chunk_text = _extract_openai_delta_text(
172                        delta,
173                        ("content", "reasoning_content", "reasoning", "refusal"),
174                    )
175                    if not chunk_text:
176                        continue
177                    emitted = True
178                    yield chunk_text
179            except AIProviderError:
180                raise
181            except self._openai.APIConnectionError as error:
182                raise AIProviderError(
183                    "Unable to connect to OpenAI API. Check network access and endpoint configuration."
184                ) from error
185            except self._openai.AuthenticationError as error:
186                raise AIProviderError(
187                    "OpenAI authentication failed. Check `ai.openai.api_key` or OPENAI_API_KEY."
188                ) from error
189            except self._openai.BadRequestError as error:
190                if _is_context_length_error(error):
191                    raise AIProviderError(
192                        "OpenAI request exceeded the model context length. Reduce prompt size and retry."
193                    ) from error
194                raise AIProviderError(f"OpenAI request was rejected: {error}") from error
195            except self._openai.APIError as error:
196                raise AIProviderError(f"OpenAI API error: {error}") from error
197            except Exception as error:
198                raise AIProviderError(f"Unexpected OpenAI provider error: {error}") from error
199
200            if not emitted:
201                raise AIProviderError("OpenAI returned an empty response.")
202
203        return _stream()
204
205    def analyze_with_attachments(
206        self,
207        system_prompt: str,
208        user_prompt: str,
209        attachments: list[Mapping[str, str]] | None,
210        max_tokens: int = DEFAULT_MAX_TOKENS,
211    ) -> str:
212        """Analyze with optional CSV file attachments via the Responses API.
213
214        Args:
215            system_prompt: The system-level instruction text.
216            user_prompt: The user-facing prompt with investigation context.
217            attachments: Optional list of attachment descriptors.
218            max_tokens: Maximum completion tokens.
219
220        Returns:
221            The generated analysis text.
222
223        Raises:
224            AIProviderError: On any API or network failure.
225        """
226        def _request() -> str:
227            return self._request_non_stream(
228                system_prompt=system_prompt,
229                user_prompt=user_prompt,
230                max_tokens=max_tokens,
231                attachments=attachments,
232            )
233
234        return self._run_openai_request(_request)
235
236    def _run_openai_request(self, request_fn: Callable[[], _T]) -> _T:
237        """Execute an OpenAI request with rate-limit retries and error mapping.
238
239        Args:
240            request_fn: A zero-argument callable that performs the request.
241
242        Returns:
243            The return value of ``request_fn`` on success.
244
245        Raises:
246            AIProviderError: On any OpenAI SDK error.
247        """
248        try:
249            return _run_with_rate_limit_retries(
250                request_fn=request_fn,
251                rate_limit_error_type=self._openai.RateLimitError,
252                provider_name="OpenAI",
253            )
254        except AIProviderError:
255            raise
256        except self._openai.APIConnectionError as error:
257            raise AIProviderError(
258                "Unable to connect to OpenAI API. Check network access and endpoint configuration."
259            ) from error
260        except self._openai.AuthenticationError as error:
261            raise AIProviderError(
262                "OpenAI authentication failed. Check `ai.openai.api_key` or OPENAI_API_KEY."
263            ) from error
264        except self._openai.BadRequestError as error:
265            if _is_context_length_error(error):
266                raise AIProviderError(
267                    "OpenAI request exceeded the model context length. Reduce prompt size and retry."
268                ) from error
269            raise AIProviderError(f"OpenAI request was rejected: {error}") from error
270        except self._openai.APIError as error:
271            raise AIProviderError(f"OpenAI API error: {error}") from error
272        except Exception as error:
273            raise AIProviderError(f"Unexpected OpenAI provider error: {error}") from error
274
275    def _request_non_stream(
276        self,
277        system_prompt: str,
278        user_prompt: str,
279        max_tokens: int,
280        attachments: list[Mapping[str, str]] | None = None,
281    ) -> str:
282        """Perform a non-streaming OpenAI request with attachment handling.
283
284        Tries file-attachment mode first, then falls back to inlining
285        attachment data, and finally issues a plain Chat Completions request.
286
287        Args:
288            system_prompt: The system-level instruction text.
289            user_prompt: The user-facing prompt text.
290            max_tokens: Maximum completion tokens.
291            attachments: Optional list of attachment descriptors.
292
293        Returns:
294            The generated analysis text.
295
296        Raises:
297            AIProviderError: If the response is empty.
298        """
299        attachment_response = self._request_with_csv_attachments(
300            system_prompt=system_prompt,
301            user_prompt=user_prompt,
302            max_tokens=max_tokens,
303            attachments=attachments,
304        )
305        if attachment_response:
306            return attachment_response
307
308        prompt_for_completion = user_prompt
309        if attachments:
310            prompt_for_completion, inlined_attachment_data = _inline_attachment_data_into_prompt(
311                user_prompt=user_prompt,
312                attachments=attachments,
313            )
314            if inlined_attachment_data:
315                logger.info("OpenAI attachment fallback inlined attachment data into prompt.")
316
317        messages = [
318            {"role": "system", "content": system_prompt},
319            {"role": "user", "content": prompt_for_completion},
320        ]
321        response = self._create_chat_completion(
322            messages=messages,
323            max_tokens=max_tokens,
324        )
325        text = _extract_openai_text(response)
326        if not text:
327            raise AIProviderError("OpenAI returned an empty response.")
328        return text
329
330    def _create_chat_completion(
331        self,
332        messages: list[dict[str, str]],
333        max_tokens: int,
334        stream: bool = False,
335    ) -> Any:
336        """Create a Chat Completions request with token parameter fallback.
337
338        Tries ``max_completion_tokens`` first, then falls back to
339        ``max_tokens`` if the endpoint reports the parameter as unsupported.
340        Also retries with a reduced token count when the provider rejects
341        the requested maximum.
342
343        Args:
344            messages: The conversation messages list.
345            max_tokens: Maximum completion tokens.
346            stream: If ``True``, return a streaming response iterator.
347
348        Returns:
349            The OpenAI ``ChatCompletion`` response or streaming iterator.
350        """
351        def _create_with_token_parameter(token_parameter: str, token_count: int) -> Any:
352            """Try creating with a specific token parameter, retrying on token limit."""
353            request_kwargs: dict[str, Any] = {
354                "model": self.model,
355                "messages": messages,
356                token_parameter: token_count,
357            }
358            if stream:
359                request_kwargs["stream"] = True
360            try:
361                return self.client.chat.completions.create(**request_kwargs)
362            except self._openai.BadRequestError as error:
363                retry_token_count = _resolve_completion_token_retry_limit(
364                    error=error,
365                    requested_tokens=token_count,
366                )
367                if retry_token_count is None:
368                    raise
369                logger.warning(
370                    "OpenAI rejected %s=%d; retrying with %s=%d.",
371                    token_parameter,
372                    token_count,
373                    token_parameter,
374                    retry_token_count,
375                )
376                request_kwargs[token_parameter] = retry_token_count
377                return self.client.chat.completions.create(**request_kwargs)
378
379        try:
380            return _create_with_token_parameter(
381                token_parameter="max_completion_tokens",
382                token_count=max_tokens,
383            )
384        except self._openai.BadRequestError as error:
385            if not _is_unsupported_parameter_error(error, "max_completion_tokens"):
386                raise
387            return _create_with_token_parameter(
388                token_parameter="max_tokens",
389                token_count=max_tokens,
390            )
391
392    def _request_with_csv_attachments(
393        self,
394        system_prompt: str,
395        user_prompt: str,
396        max_tokens: int,
397        attachments: list[Mapping[str, str]] | None,
398    ) -> str | None:
399        """Attempt to send a request with CSV files via the Responses API.
400
401        Args:
402            system_prompt: The system-level instruction text.
403            user_prompt: The user-facing prompt text.
404            max_tokens: Maximum completion tokens.
405            attachments: Optional list of attachment descriptors.
406
407        Returns:
408            The generated text if succeeded, or ``None`` if skipped.
409        """
410        normalized_attachments = self._prepare_csv_attachments(
411            attachments,
412            supports_file_attachments=hasattr(self.client, "files") and hasattr(self.client, "responses"),
413        )
414        if not normalized_attachments:
415            return None
416
417        try:
418            text = upload_and_request_via_responses_api(
419                client=self.client,
420                openai_module=self._openai,
421                model=self.model,
422                normalized_attachments=normalized_attachments,
423                system_prompt=system_prompt,
424                user_prompt=user_prompt,
425                max_tokens=max_tokens,
426                provider_name="OpenAI",
427                upload_purpose="assistants",
428                convert_csv_to_txt=True,
429            )
430            self._csv_attachment_supported = True
431            return text
432        except Exception as error:
433            if _is_attachment_unsupported_error(error):
434                self._csv_attachment_supported = False
435                logger.info(
436                    "OpenAI endpoint does not support CSV attachments via /files + /responses; "
437                    "falling back to chat.completions text mode."
438                )
439                return None
440            raise
441
442    def get_model_info(self) -> dict[str, str]:
443        """Return OpenAI provider and model metadata.
444
445        Returns:
446            A dictionary with ``"provider"`` and ``"model"`` keys.
447        """
448        return {"provider": "openai", "model": self.model}