kaxil commented on code in PR #69812:
URL: https://github.com/apache/airflow/pull/69812#discussion_r4109163162
##########
providers/openai/src/airflow/providers/openai/operators/openai.py:
##########
@@ -21,15 +21,43 @@
from functools import cached_property
from typing import TYPE_CHECKING, Any, ClassVar
+from pydantic import BaseModel, ValidationError
+
from airflow.providers.common.compat.sdk import BaseOperator, conf
from airflow.providers.openai.exceptions import OpenAIBatchJobException
from airflow.providers.openai.hooks.openai import OpenAIHook,
validate_execute_complete_event
from airflow.providers.openai.triggers.openai import OpenAIBatchTrigger
if TYPE_CHECKING:
+ from openai.types.responses import Response
+
from airflow.providers.common.compat.sdk import Context
+def _get_structured_response_details(response: Any) -> str:
Review Comment:
`response: Any` turns off type checking for every attribute read below. With
`ParsedResponse[BaseModel]` (it can go in the existing `TYPE_CHECKING` import),
mypy narrows on `output.type == "message"` and `content.type == "refusal"` and
checks each field against the SDK. I tried it on openai 2.37.0: a typo in
`content.refusal` is an `attr-defined` error with the concrete type and passes
silently with `Any`. The `model_construct` tests can't catch an SDK field
rename, and this would.
##########
providers/openai/src/airflow/providers/openai/operators/openai.py:
##########
@@ -111,8 +140,9 @@ class OpenAIResponseOperator(BaseOperator):
input items.
:param model: The OpenAI model to use.
:param response_kwargs: Additional keyword arguments to pass to the OpenAI
``create_response``
- method (for example ``instructions``, ``tools``, ``conversation`` or
``previous_response_id``).
- Templated, so values (e.g. ``previous_response_id``) may reference
upstream XCom.
+ method, or ``parse_response`` when ``text_format`` is set (for example
``instructions``,
+ ``tools``, ``conversation`` or ``previous_response_id``). Templated,
so values (e.g.
+ ``previous_response_id``) may reference upstream XCom.
Do not set ``background`` or ``stream`` here: ``background=True``
returns before the response
completes, so this operator logs a warning and the returned output
text may be empty, while
Review Comment:
With `text_format` set, `background=True` no longer just logs a warning. The
parsed response comes back `queued` or `in_progress`, so the `status !=
"completed"` check in `execute()` raises `ValueError`, and the background
response is left running on OpenAI's side. A clause here would match what the
structured-outputs section of the rst guide already says about responses still
in progress.
##########
providers/openai/src/airflow/providers/openai/hooks/openai.py:
##########
@@ -273,6 +279,30 @@ def create_response(self, input: Any, model: str =
"gpt-4o-mini", **kwargs: Any)
"""
return self.conn.responses.create(model=model, input=input, **kwargs)
+ def parse_response(
+ self,
+ input: Any,
+ text_format: type[_TextFormatT],
+ model: str = "gpt-4o-mini",
+ **kwargs: Any,
Review Comment:
`text_format` sits where `create_response` has `model`, so
`hook.parse_response(prompt, "gpt-4o")`, written by analogy with
`create_response`, passes the model id as the format. Keeping
`create_response`'s positional order and making `text_format` keyword-only also
matches the SDK, where `responses.parse` takes it keyword-only. Easier to
settle before the hook method ships than after.
```suggestion
input: Any,
model: str = "gpt-4o-mini",
*,
text_format: type[_TextFormatT],
**kwargs: Any,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
