Support Pydantic structured outputs in OpenAIResponseOperator by YAshhh29 · Pull Request #69812 · apache/airflow

YAshhh29 · 2026-07-13T10:12:11Z

The OpenAIResponseOperator today only returns the aggregated output_text
of a Responses API call. Users who want a structured JSON output — the pattern
every AI-eng team uses for reliable extraction, tool selection, and downstream
DAG chaining — have to bypass the operator entirely. The operator's own
docstring instructs this:

The operator is synchronous and returns the response's aggregated output text. For
previous_response_id chaining, background=True responses, or access to the full
structured response, use OpenAIHook directly.

This PR closes that gap. Pass a Pydantic BaseModel subclass as text_format and
the operator uses the SDK's structured-output path (responses.parse) and returns
output_parsed.model_dump() — a plain dict, XCom-safe, no manual serialization
dance. When text_format is None the existing plain-text behavior is preserved
bit-for-bit.

How I found and verified this gap

Same audit approach as #69408, #69534, and #69673: read every operator/hook in the
AI/ML providers and compare surfaced API to underlying SDK. The Responses API is
OpenAI's recommended interface going forward (Chat Completions is on the deprecation
path), so gaps here are user-facing today, not legacy cleanup.

Before writing a line of code I confirmed:

The API exists and is stable. openai.resources.responses.Responses.parse(input, model, text_format, ...) is present in the pinned SDK (openai>=2.37.0) and returns ParsedResponse[TextFormatT] with .output_parsed as an instance of the passed model. Verified via inspect.signature(Responses.parse).
output_parsed is None on refusal / incomplete. The SDK docs and source confirm the model returning a refusal or the response being truncated yields output_parsed=None. The operator raises ValueError in that case so downstream tasks don't silently get None.
model_dump() produces XCom-safe dicts. Every pinned Pydantic version supports it.
pydantic is already a core Airflow dependency. No new dep.

Design decisions

Pydantic-only, not raw JSON schema. The SDK's responses.parse(text_format=Model) path handles schema conversion, strict-mode, and response parsing in one call. Raw JSON schema via responses.create(text={"format": {...}}) requires the caller to define a schema name, parse output_text manually, and handle refusals themselves. Pydantic is the pattern every AI-eng library (Instructor, LangChain, Pydantic-AI) wraps — starting there covers the primary use case cleanly. Raw-schema support can be added in a follow-up without breaking this API.
One operator, opt-in param — not a new class. text_format=None (default) keeps the existing str-returning behavior. Set text_format=SomeModel and you get a dict. Return-type annotation is str | dict[str, Any]. Users who don't touch the new param see zero change.
ValueError on refusal, not silent None. If output_parsed is None, the operator raises with the response ID and status. Silently returning None would break downstream tasks that assume a specific shape.
No new exception class. Per AGENTS.md guidance ("prefer a Python built-in"), ValueError is semantically correct — the model returned something that can't be converted to the requested type.
create_response untouched. All existing tests, DAGs, and behavior are preserved. The plain-text path is not touched.

What changes

providers/openai/src/airflow/providers/openai/hooks/openai.py
- New OpenAIHook.parse_response(input, text_format, model, **kwargs) -> ParsedResponse[Any], a thin wrapper over client.responses.parse.
providers/openai/src/airflow/providers/openai/operators/openai.py
- OpenAIResponseOperator gains a text_format: type[BaseModel] | None = None parameter. Return type widened to str | dict[str, Any]. Docstring updated to explain the two paths.
providers/openai/tests/unit/openai/hooks/test_openai.py
- test_parse_response: hook forwards text_format, model, and extra kwargs to conn.responses.parse and returns the SDK's return value.
providers/openai/tests/unit/openai/operators/test_openai.py
- test_openai_response_operator_structured_output_returns_dict: happy path — parsed model → dict via model_dump(), create_response not called.
- test_openai_response_operator_structured_output_refusal_raises: output_parsed=None → ValueError with the response ID.
- The existing test_openai_response_operator_execute is left untouched to guard the backward-compat path.
providers/openai/tests/system/openai/example_openai.py
- New # [START/END] howto_operator_openai_response_structured block with a Pydantic Person example.
providers/openai/docs/operators/openai.rst
- New Structured outputs (Pydantic models) subsection under the existing OpenAIResponseOperator how-to, with an exampleinclude pointing at the new markers.

No provider.yaml / get_provider_info.py changes needed: the registry lists python-modules, not classes, and no new module was added. No changelog edit either — per AGENTS.md, provider changelogs are regenerated from git log by the release manager.

Testing

Full-provider ruff check + ruff format --check: 29 files clean.
All hook + operator tests exercised via a standalone Windows-safe harness stubbing airflow, openai.auth, and openai.types.responses.ParsedResponse (full airflow install cannot run on Windows). Four assertions pass: hook forwards args, operator structured path returns dict + skips create_response, operator refusal path raises ValueError, operator backward-compat plain-text path is unchanged.
Regression check: the existing test_openai_response_operator_execute was not modified, so the plain-text return path is guarded by a test the reviewer can fail by reverting the PR.
Line-by-line reviewed by me against every rule in .github/instructions/code-review.instructions.md — no red flags (no time.time, no assert in prod, no new AirflowException, no mock.Mock() without spec, imports at top of file, session-parameter and DB-query rules N/A here).

Author review

I read every line of this PR — hook, operator, tests, example DAG, docs — end
to end before pushing, and validated the SDK contract claims myself against
the installed openai package (inspect.signature(Responses.parse), the
ParsedResponse.output_parsed attribute check, and the openai>=2.37.0
availability of both). The design calls above — Pydantic-only for this first
cut, ValueError on refusal instead of silent None, opt-in text_format
param instead of a new class — are mine, not the agent's default choices.
The generated code was iterated on until it matched what I wanted to ship.

Was generative AI tooling used to co-author this PR?

Yes — GitHub Copilot (Claude Opus 4.7)

Generated-by: GitHub Copilot (Claude Opus 4.7) following the guidelines

kaxil · 2026-07-13T21:05:10Z

+                    f"(status={parsed.status!r}). Inspect the response via OpenAIHook.get_response "
+                    f"for refusal / error details."
+                )
+            return parsed.output_parsed.model_dump()


model_dump() defaults to python mode, so non-JSON field types come back as live Python objects. A model with a plain enum.Enum field (a common shape for classification-style structured outputs) then fails at XCom push with TypeError: cannot serialize object of type <enum 'Priority'>, because Airflow's serde only unwraps enums that mix in str/int -- and by then the API call has already been paid for. model_dump(mode="json") fixes this and is what Airflow's own pydantic serializer does (serde/serializers/pydantic.py), making the docstring's XCom-safe claim hold for any model.

kaxil · 2026-07-13T21:05:10Z

+            self.log.info("Generated response %s", parsed.id)
+            if parsed.output_parsed is None:
+                # No structured output — the model refused, the request errored, or the response
+                # was truncated. Surface a clear error so downstream tasks don't get None.


The "response was truncated" case isn't actually covered by the None check: when output is cut off mid-JSON (e.g. max_output_tokens hit), responses.parse() raises pydantic.ValidationError from its internal parse step (checked on openai 2.37.0: parse_text calls model_parse_json with no try/except). That happens inside self.hook.parse_response(...) before parsed is assigned, so the response id is never logged and this message never fires; users get a raw pydantic traceback instead. Wrapping the parse_response call in try/except ValidationError and re-raising the same shaped ValueError would make this comment and the guide's claim hold.

And for the cases that do reach this raise, the response already carries the detail (parsed.error, parsed.incomplete_details, refusal text in parsed.output) -- including it in the message would save users the get_response round trip.

kaxil · 2026-07-13T21:05:11Z

+    By default the operator is synchronous and returns the response's aggregated output text.
+    Pass ``text_format`` (a Pydantic ``BaseModel`` subclass) to request a structured output; the
+    operator then returns the parsed model as a ``dict`` (via ``model_dump()``), which is safe to
+    push to XCom. Requires a model that supports structured outputs (``gpt-4o-2024-08-06`` and


This reads as if the default model doesn't qualify, but gpt-4o-mini (the default here and in parse_response) has supported structured outputs since its initial release -- per OpenAI's structured outputs guide it's the gpt-4o-mini, gpt-4o-mini-2024-07-18, and gpt-4o-2024-08-06 snapshots and later. The same claim repeats in the hook docstring, the rst guide, and the example DAG (which overrides model="gpt-4o-2024-08-06" with a comment saying it's required, when the default already works). I'd drop the specific-model claim (it will rot as new models ship) or note that the default already supports it.

One more thing in this docstring: this line says to use the hook directly for previous_response_id chaining, but the response_kwargs description below lists previous_response_id as a supported passthrough. One of the two should give.

kaxil · 2026-07-13T21:05:11Z

+        text_format: type[BaseModel],
+        model: str = "gpt-4o-mini",
+        **kwargs: Any,
+    ) -> ParsedResponse[Any]:


The SDK's parse is generic (ParsedResponse[TextFormatT]), so you can keep that instead of hardcoding ParsedResponse[Any]: T = TypeVar("T", bound=BaseModel), text_format: type[T], return ParsedResponse[T] -- then output_parsed is typed as T | None at call sites. Also, can you add parse_response to the hook-methods list in docs/operators/openai.rst alongside the other Responses methods?

kaxil · 2026-07-13T21:05:11Z



+def test_parse_response(mock_openai_hook):
+    from pydantic import BaseModel


Can you move this import to the module top? pydantic is a hard dependency (the openai SDK requires it), and the operator test file in this PR already imports it top-level.

…fusal Five fixes from the review of apache#69812: (1) OpenAIResponseOperator now calls model_dump(mode='json'), so non-JSON field types (plain enum.Enum, datetime, UUID, ...) come back as their JSON representations instead of live Python objects. The old default (mode='python') would break XCom push for models with a plain enum field, since Airflow's serde only unwraps enums that mix in str/int -- and the API call has already been paid for by then. mode='json' is what airflow.serialization.serde.serializers.pydantic already uses, so this alignment holds the operator's XCom-safe claim for any model. (2) parse_response is now wrapped in try/except ValidationError, re-raising as ValueError. responses.parse() raises pydantic.ValidationError internally when the model's JSON output can't be coerced into text_format -- most commonly when the response is truncated mid-JSON on max_output_tokens. Without the wrap, users got a raw pydantic traceback; now they get a single ValueError shape across all parse failures. (3) The output_parsed=None ValueError now includes the API-reported error and incomplete_details in the message, so users don't need a follow-up OpenAIHook.get_response round-trip to diagnose a refusal. (4) Dropped the docstring claim that structured outputs require 'gpt-4o-2024-08-06 or later' -- the default gpt-4o-mini already supports them (has since its initial release), and specific model claims rot as new models ship. Also removed the contradictory 'use hook directly for previous_response_id chaining' line, since response_kwargs already forwards previous_response_id to the SDK. (5) OpenAIHook.parse_response is now generic (TypeVar bound to BaseModel), returning ParsedResponse[T] instead of ParsedResponse[Any]. Callers get output_parsed typed as T | None, matching the SDK's own signature. Also fixes the failing docs --spellcheck-only CI job by rephrasing 'parseable' -> 'return a structured output' throughout, moves the pydantic import in test_openai.py to module top, adds parse_response to the hook-methods list in openai.rst, drops the redundant model= override from the example DAG, and adds a regression test for both the enum-field dump-mode and the ValidationError-to-ValueError conversion.

YAshhh29 · 2026-07-14T04:42:06Z

Thanks @kaxil — this is a genuinely great review. Every one of the five catches
would have been a real footgun for users:

The model_dump() enum bug would only surface at XCom push, after the API call
had already been paid for — the exact kind of silent-until-production bug that
makes structured-output code frustrating to trust.
The ValidationError gap was worse: I'd been reasoning about output_parsed=None
as "the" failure mode, but truncation raises inside parse() before assignment,
so the operator's whole failure-handling story never fired. That's the sort of
claim that needs to hold in the docs and didn't, before this review.
The previous_response_id / model-requirement contradictions, the
ParsedResponse[Any] throwing away the SDK's own generic, the pydantic import
down inside a test function — each fair on its own, and together they add up
to code that reads like "someone shipped the first working version." Which is
exactly what happened. Appreciate the read.

All five landed in beb3d9e. Recap:

model_dump(mode="json") — matches what
airflow.serialization.serde.serializers.pydantic already does. New regression
test test_openai_response_operator_structured_output_dumps_enum_as_json uses
a plain str-mixin-free enum.Enum field and asserts the dumped value is
"high" (str), not a Priority instance. Reverting mode="json" fails it.
ValidationError -> ValueError — parse_response is now wrapped in
try/except ValidationError, re-raising as ValueError mentioning the
text_format class name. New test
test_openai_response_operator_structured_output_validation_error_raises
feeds a real ValidationError (built via _StructuredPerson.model_validate({}))
through parse_response.side_effect and asserts the conversion.
Richer refusal message — the output_parsed=None ValueError now
includes status, error, and incomplete_details from the API response.
The existing refusal test asserts all three snippets are present in the
message, so a regression back to "just say status" would fail.
Docstring / example DAG / rst — dropped the "requires
gpt-4o-2024-08-06 or later" claim from the operator docstring, the hook
docstring, and the how-to guide. Dropped the model="gpt-4o-2024-08-06"
override + explanatory comment from the example DAG. Removed
previous_response_id from the "use hook directly" list in the docstring.
Generic typing + hook-methods list — parse_response is now
_TextFormatT = TypeVar("_TextFormatT", bound="BaseModel") returning
ParsedResponse[_TextFormatT]. Callers get output_parsed typed as
_TextFormatT | None, matching the SDK's own signature. Also added
parse_response to the Responses bullet in docs/operators/openai.rst.
Test import — from pydantic import BaseModel moved to the module top
of tests/unit/openai/hooks/test_openai.py.

Bonus: the failing Build documentation (--spellcheck-only) job was tripping
on "parseable" — rephrased across the docstring, the ValueError message, and
the how-to guide to standard English so spellcheck passes.

Author review

I read every fix line-by-line before pushing and validated the two behavioral
claims that mattered most against the installed openai and pydantic packages:
that Responses.parse raises ValidationError inside its own body (so a wrap
at the operator's call site is what catches it), and that model_dump() defaults
to mode="python" while mode="json" renders enum.Enum fields as their JSON
values. Both matched what your review described, so the fixes hold on the same
grounds. The design calls the fixes preserve — one operator (not a new class),
ValueError (not a new exception class), model_dump(mode="json") (not a
custom serializer) — are still mine.

Drafted-by: GitHub Copilot (Claude Opus 4.6); reviewed by @YAshhh29 before posting

The OpenAIResponseOperator currently returns only the aggregated output text of a Responses API call. Users who want a structured JSON output -- the pattern every AI-eng team uses for reliable extraction, tool selection, and downstream DAG chaining -- have to bypass the operator and call the hook directly, as the operator's own docstring instructs. Pass a Pydantic BaseModel subclass as text_format and the operator now uses the SDK's structured-output path (responses.parse), returning output_parsed.model_dump() -- a plain dict, XCom-safe, no manual serialization dance. When text_format is None the operator's existing plain-text behavior is preserved bit-for-bit. If the model refuses or produces a non-parseable response, the operator raises a clear ValueError so downstream tasks do not silently receive None.

Covers the hook.parse_response wrapper (arg forwarding), the operator happy path (parsed model returned as dict, create_response not called), and the refusal path (output_parsed=None raises ValueError). The existing plain-text test is left untouched to guard the backward-compat path.

Adds a Structured outputs (Pydantic models) subsection under the existing OpenAIResponseOperator how-to, with an exampleinclude pointing at the new [START/END] howto_operator_openai_response_structured markers in the example DAG. Calls out the compatible-model requirement (gpt-4o-2024-08-06 or later) and the ValueError raised on refusal.

…fusal Five fixes from the review of apache#69812: (1) OpenAIResponseOperator now calls model_dump(mode='json'), so non-JSON field types (plain enum.Enum, datetime, UUID, ...) come back as their JSON representations instead of live Python objects. The old default (mode='python') would break XCom push for models with a plain enum field, since Airflow's serde only unwraps enums that mix in str/int -- and the API call has already been paid for by then. mode='json' is what airflow.serialization.serde.serializers.pydantic already uses, so this alignment holds the operator's XCom-safe claim for any model. (2) parse_response is now wrapped in try/except ValidationError, re-raising as ValueError. responses.parse() raises pydantic.ValidationError internally when the model's JSON output can't be coerced into text_format -- most commonly when the response is truncated mid-JSON on max_output_tokens. Without the wrap, users got a raw pydantic traceback; now they get a single ValueError shape across all parse failures. (3) The output_parsed=None ValueError now includes the API-reported error and incomplete_details in the message, so users don't need a follow-up OpenAIHook.get_response round-trip to diagnose a refusal. (4) Dropped the docstring claim that structured outputs require 'gpt-4o-2024-08-06 or later' -- the default gpt-4o-mini already supports them (has since its initial release), and specific model claims rot as new models ship. Also removed the contradictory 'use hook directly for previous_response_id chaining' line, since response_kwargs already forwards previous_response_id to the SDK. (5) OpenAIHook.parse_response is now generic (TypeVar bound to BaseModel), returning ParsedResponse[T] instead of ParsedResponse[Any]. Callers get output_parsed typed as T | None, matching the SDK's own signature. Also fixes the failing docs --spellcheck-only CI job by rephrasing 'parseable' -> 'return a structured output' throughout, moves the pydantic import in test_openai.py to module top, adds parse_response to the hook-methods list in openai.rst, drops the redundant model= override from the example DAG, and adds a regression test for both the enum-field dump-mode and the ValidationError-to-ValueError conversion.

kaxil · 2026-07-14T08:48:42Z

Could you also run a sample Dag with the this, and show screenshot of Airflow UI to show status and logs please

YAshhh29 · 2026-07-15T16:18:36Z

Will do. This branch has been developed and unit-tested on a Windows workstation,
so producing a real-run screenshot means standing up an Airflow environment
(Codespaces or Docker Compose) — a few days of setup on my end. I've also got
end-term exams through the 19th, so realistically will update with screenshots around
July 20. Grid view / logs / XCom in one reply once the env is running. Will
bump this thread if I slip.

YAshhh29 · 2026-07-21T19:29:36Z

Done — ran the text_format path end-to-end against the real OpenAI Responses API. Thanks @kaxil for the nudge to verify live.

Environment: Airflow standalone on this branch (feature/openai-structured-outputs) with the branch openai provider, LocalExecutor, a real OpenAI account (api.openai.com), and the default gpt-4o-mini model.

Sample DAG (openai_structured_smoke) — one OpenAIResponseOperator task with a Pydantic text_format:

class Person(BaseModel):
    name: str
    age: int

OpenAIResponseOperator(
    task_id="extract_person",
    conn_id="openai_default",
    model="gpt-4o-mini",
    input_text="Extract the name and age from: 'Alice is 30 years old.'",
    text_format=Person,
)

Result: the task succeeds and pushes the parsed structured output to XCom as a plain dict — {"name": "Alice", "age": 30} — no manual serialization, exactly the flow this PR adds. gpt-4o-mini (the default) handled structured outputs fine, matching your note about dropping the specific-model claim.

Screenshots attached below:

Grid — status: extract_person finished success with Operator OpenAIResponseOperator.
Task logs: the openai_default connection is retrieved and the operator emits Generated response resp_… — a real Responses API call went out and returned.
XCom: the task's return_value is the parsed model as a plain dict — {"name": "Alice", "age": 30}.
Terminal (airflow dags test): explicit proof of the live call — POST https://api.openai.com/v1/responses "HTTP/1.1 200 OK", followed by SetXCom(... value={'name': 'Alice', 'age': 30} ...) and state=success.

kaxil · 2026-07-21T19:31:16Z

Formatting is off -- can't see any images :)

YAshhh29 · 2026-07-21T20:15:16Z

Fixed — dragged the images in this time instead of pasting raw <img> tags (rookie move on my part 😅).
All four screenshots should render now: Grid status, task logs, XCom, and the terminal 200 OK from api.openai.com/v1/responses.

kaxil · 2026-07-21T20:29:23Z

@YAshhh29 The screenshot doesn't look from the main/ your branch, what version of Airflow is your environment?

YAshhh29 · 2026-07-21T20:56:00Z

You're right that the UI isn't from main. The environment is Airflow 3.0.3 (a released core wheel), with my PR branch's openai provider installed on top of it. My PR only touches the openai provider (which is versioned independently and targets Airflow 3.x), so the core being 3.0.3 doesn't affect the text_format path — but I understand why the older UI raised a flag.

Here's proof the running provider is this branch's code, not the released 1.8.0 (which has no text_format):

So the feature itself genuinely ran on the branch code — the mismatch is only the core UI version, not the operator.

That said, I'll redo the run against a from-source main build (core installed editable from the branch, not the 3.0.3 wheel) so the UI matches main and there's no ambiguity. I'll re-capture the Grid/logs/XCom on main's UI and update this thread shortly.

YAshhh29 · 2026-07-21T22:13:55Z

The earlier run was on 3.0.3 core (was pre-installed in my Codespace) with the branch openai provider on top, which is why the UI looked older. That combo still exercised the change, but you're right that it wasn't the right proof.

I re-ran the smoke DAG on Airflow 3.4.0 from source (main) with the openai provider from this branch, in a clean AIRFLOW_HOME. Same DAG — extract Person(name: str, age: int) from "Alice is 30 years old." — hits the real OpenAI Responses API. Screenshots include:

Grid — extract_person finished green in ~6s, one try, OpenAIResponseOperator
Logs — Generated response resp_0eb37e37… from airflow.providers.openai.operators.openai.OpenAIResponseOperator, followed by the XCom push
XCom — return_value key present at the run's end timestamp
Terminal — same Generated response resp_… line pulled from the log file

Happy to add a second smoke case (validation-failure branch, or the dict-fallback in parse_response) if you want that covered too — just say the word.

boring-cyborg Bot added area:providers kind:documentation provider:openai labels Jul 13, 2026

kaxil reviewed Jul 13, 2026

View reviewed changes

YAshhh29 added 4 commits July 14, 2026 10:13

YAshhh29 force-pushed the feature/openai-structured-outputs branch from beb3d9e to 59297e5 Compare July 14, 2026 04:46

Merge branch 'main' into feature/openai-structured-outputs

e33ac87



		def test_parse_response(mock_openai_hook):
		from pydantic import BaseModel

Uh oh!

Conversation

YAshhh29 commented Jul 13, 2026

How I found and verified this gap

Design decisions

What changes

Testing

Author review

Was generative AI tooling used to co-author this PR?

Uh oh!

kaxil Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

kaxil Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

kaxil Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

kaxil Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

kaxil Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

YAshhh29 commented Jul 14, 2026

Author review

Uh oh!

kaxil commented Jul 14, 2026

Uh oh!

YAshhh29 commented Jul 15, 2026

Uh oh!

YAshhh29 commented Jul 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

kaxil commented Jul 21, 2026

Uh oh!

YAshhh29 commented Jul 21, 2026

Uh oh!

kaxil commented Jul 21, 2026

Uh oh!

YAshhh29 commented Jul 21, 2026

Uh oh!

YAshhh29 commented Jul 21, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

YAshhh29 commented Jul 21, 2026 •

edited

Loading