This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new c1570eba [integrations][openai] Apply OpenAI native structured output
(#919)
c1570eba is described below
commit c1570ebad00ce324ac368924ce573b5a10f6cfba
Author: Weiqing Yang <[email protected]>
AuthorDate: Sun Aug 2 02:18:20 2026 -0700
[integrations][openai] Apply OpenAI native structured output (#919)
---
.../openai/OpenAICompletionsConnection.java | 106 +++++++++-
.../openai/OpenAICompletionsConnectionTest.java | 218 +++++++++++++++++++++
.../chat_models/openai/openai_chat_model.py | 108 +++++++++-
.../tests/test_openai_native_structured_output.py | 204 +++++++++++++++++++
.../tests/test_output_schema_param_declared.py | 24 ++-
5 files changed, 650 insertions(+), 10 deletions(-)
diff --git
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java
index 29d0dcf7..92683b9c 100644
---
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java
+++
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnection.java
@@ -22,11 +22,13 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.openai.client.OpenAIClient;
import com.openai.client.okhttp.OpenAIOkHttpClient;
+import com.openai.core.JsonSchemaLocalValidation;
import com.openai.core.JsonValue;
import com.openai.models.ChatModel;
import com.openai.models.FunctionDefinition;
import com.openai.models.FunctionParameters;
import com.openai.models.ReasoningEffort;
+import com.openai.models.ResponseFormatJsonSchema;
import com.openai.models.chat.completions.ChatCompletion;
import com.openai.models.chat.completions.ChatCompletionCreateParams;
import com.openai.models.chat.completions.ChatCompletionFunctionTool;
@@ -43,6 +45,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* A chat model integration for the OpenAI Chat Completions service using the
official Java SDK.
@@ -119,11 +122,70 @@ public class OpenAICompletionsConnection extends
BaseChatModelConnection {
this.client = builder.build();
}
+ // Models for which OpenAI documents json_schema strict Structured Outputs
support.
+ // Source of truth:
https://platform.openai.com/docs/guides/structured-outputs
+ //
+ // A name carrying a non-text modality marker is rejected before anything
else. Audio, realtime,
+ // speech and transcription variants expose no json_schema response format
even though they
+ // share the name prefix of a capable text family, so the marker check has
to win over a prefix
+ // match rather than merely coexist with it.
+ //
+ // A text family whose entire lifetime post-dates the Structured Outputs
cutoff is matched by
+ // name prefix, so its dated snapshots and size variants resolve without
enumerating each one.
+ //
+ // Two cases cannot use a prefix and are matched exactly instead. The
gpt-4o family straddles
+ // the cutoff — gpt-4o-2024-05-13 predates it and is not capable — so the
boundary there is
+ // temporal rather than nominal. The o1 family is not uniform: o1 is
capable while o1-mini is
+ // not, so an "o1" prefix would admit an incapable sibling.
+ //
+ // A name outside every listed family reports not-capable and degrades to
the prompt fallback
+ // rather than failing at the provider. Within a listed family the prefix
assumes capability,
+ // so a family variant that ships without json_schema support has to be
excluded explicitly,
+ // either by a marker that appears in no capable name or by replacing the
family prefix with
+ // exact names.
+ private static final Set<String> NON_TEXT_MODALITY_MARKERS =
+ Set.of("-audio", "-realtime", "-tts", "-transcribe");
+ private static final Set<String> NATIVE_STRUCTURED_OUTPUT_FAMILY_PREFIXES =
+ Set.of("gpt-4o-mini", "gpt-4o-search-preview", "gpt-4.1", "gpt-5",
"o3", "o4-mini");
+ private static final Set<String> NATIVE_STRUCTURED_OUTPUT_MODELS =
+ Set.of("gpt-4o", "gpt-4o-2024-08-06", "gpt-4o-2024-11-20", "o1",
"o1-2024-12-17");
+
+ @Override
+ protected boolean supportsNativeStructuredOutput(String effectiveModel) {
+ if (effectiveModel == null) {
+ return false;
+ }
+ if
(NON_TEXT_MODALITY_MARKERS.stream().anyMatch(effectiveModel::contains)) {
+ return false;
+ }
+ return NATIVE_STRUCTURED_OUTPUT_FAMILY_PREFIXES.stream()
+ .anyMatch(effectiveModel::startsWith)
+ || NATIVE_STRUCTURED_OUTPUT_MODELS.contains(effectiveModel);
+ }
+
@Override
public ChatMessage chat(
List<ChatMessage> messages, List<Tool> tools, Map<String, Object>
modelParams) {
+ return doChat(messages, tools, modelParams, null);
+ }
+
+ @Override
+ public ChatMessage chat(
+ List<ChatMessage> messages,
+ List<Tool> tools,
+ Map<String, Object> modelParams,
+ Object outputSchema) {
+ return doChat(messages, tools, modelParams, outputSchema);
+ }
+
+ private ChatMessage doChat(
+ List<ChatMessage> messages,
+ List<Tool> tools,
+ Map<String, Object> modelParams,
+ Object outputSchema) {
try {
- ChatCompletionCreateParams params = buildRequest(messages, tools,
modelParams);
+ ChatCompletionCreateParams params =
+ buildRequest(messages, tools, modelParams, outputSchema);
ChatCompletion completion =
client.chat().completions().create(params);
ChatMessage response =
OpenAIChatCompletionsUtils.convertFromOpenAIMessage(
@@ -150,8 +212,13 @@ public class OpenAICompletionsConnection extends
BaseChatModelConnection {
}
}
- private ChatCompletionCreateParams buildRequest(
- List<ChatMessage> messages, List<Tool> tools, Map<String, Object>
rawModelParams) {
+ // Package-private so the request body (including the native
response_format) can be asserted
+ // without issuing a live API call through the final OpenAI client.
+ ChatCompletionCreateParams buildRequest(
+ List<ChatMessage> messages,
+ List<Tool> tools,
+ Map<String, Object> rawModelParams,
+ Object outputSchema) {
Map<String, Object> modelParams =
rawModelParams != null ? new HashMap<>(rawModelParams) : new
HashMap<>();
@@ -170,6 +237,19 @@ public class OpenAICompletionsConnection extends
BaseChatModelConnection {
builder.tools(convertTools(tools, strictMode));
}
+ // Native structured output applies only for a POJO Class schema on a
model the provider
+ // documents as capable; a RowTypeInfo (wrapped in OutputSchema) or an
incapable model keeps
+ // the prompt-engineering fallback.
+ //
+ // TODO(#912): the requested strategy is not visible here, so this
re-check cannot tell an
+ // explicit NATIVE request apart from one that merely resolved to
native. A caller asking
+ // for NATIVE on a model this predicate rejects therefore gets an
unconstrained response
+ // instead of an error. Once strategy resolution is wired up, NATIVE
must either bypass
+ // this capability re-check or fail explicitly.
+ if (outputSchema instanceof Class &&
supportsNativeStructuredOutput(modelName)) {
+ builder.responseFormat(toNativeResponseFormat((Class<?>)
outputSchema));
+ }
+
Object temperature = modelParams.remove("temperature");
if (temperature instanceof Number) {
builder.temperature(((Number) temperature).doubleValue());
@@ -208,6 +288,26 @@ public class OpenAICompletionsConnection extends
BaseChatModelConnection {
return builder.build();
}
+ // Derives the strict json_schema response format from a POJO class via
the SDK's typed
+ // structured-output builder. The Kotlin-facade
StructuredOutputsKt.responseFormatFromClass is
+ // not callable from Java, so the response format is extracted through the
typed builder, which
+ // generates the same strict draft-2020-12 schema, and then reattached to
the standard builder.
+ private static <T> ResponseFormatJsonSchema
toNativeResponseFormat(Class<T> schemaClass) {
+ return ChatCompletionCreateParams.builder()
+ .model(ChatModel.of(""))
+ .addUserMessage("")
+ .responseFormat(schemaClass, JsonSchemaLocalValidation.NO)
+ .build()
+ .rawParams()
+ .responseFormat()
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "OpenAI SDK did not produce a
response_format for schema "
+ + schemaClass.getName()))
+ .asJsonSchema();
+ }
+
private List<ChatCompletionTool> convertTools(List<Tool> tools, boolean
strictMode) {
List<ChatCompletionTool> openaiTools = new ArrayList<>(tools.size());
for (Tool tool : tools) {
diff --git
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnectionTest.java
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnectionTest.java
new file mode 100644
index 00000000..be7a966e
--- /dev/null
+++
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAICompletionsConnectionTest.java
@@ -0,0 +1,218 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.agents.integrations.chatmodels.openai;
+
+import com.openai.models.ResponseFormatJsonSchema;
+import com.openai.models.chat.completions.ChatCompletionCreateParams;
+import org.apache.flink.agents.api.chat.messages.ChatMessage;
+import org.apache.flink.agents.api.chat.messages.MessageRole;
+import org.apache.flink.agents.api.resource.ResourceContext;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.tools.Tool;
+import org.apache.flink.agents.api.tools.ToolMetadata;
+import org.apache.flink.agents.api.tools.ToolParameters;
+import org.apache.flink.agents.api.tools.ToolResponse;
+import org.apache.flink.agents.api.tools.ToolType;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Unit tests for {@link OpenAICompletionsConnection}'s native
structured-output behavior. These
+ * assert the built request body without a live API call by inspecting {@code
buildRequest}, and
+ * exercise the model-dependent capability predicate directly.
+ */
+class OpenAICompletionsConnectionTest {
+
+ private static final ResourceContext NOOP =
ResourceContext.fromGetResource((a, b) -> null);
+
+ /** A representative POJO output schema. */
+ public static class Person {
+ public String name;
+ public int age;
+ }
+
+ private static OpenAICompletionsConnection connection() {
+ ResourceDescriptor desc =
+
ResourceDescriptor.Builder.newBuilder(OpenAICompletionsConnection.class.getName())
+ .addInitialArgument("api_key", "test-key")
+ .addInitialArgument("model", "gpt-4o")
+ .build();
+ return new OpenAICompletionsConnection(desc, NOOP);
+ }
+
+ private static Map<String, Object> params(String model) {
+ Map<String, Object> params = new HashMap<>();
+ params.put("model", model);
+ return params;
+ }
+
+ private static List<ChatMessage> userMessage() {
+ return List.of(new ChatMessage(MessageRole.USER, "hi"));
+ }
+
+ @Test
+ @DisplayName("Native response_format json_schema strict applied for a POJO
on a capable model")
+ void testNativeAppliedForPojoCapableModel() {
+ ChatCompletionCreateParams params =
+ connection().buildRequest(userMessage(), List.of(),
params("gpt-4o"), Person.class);
+
+ assertThat(params.responseFormat()).isPresent();
+ ResponseFormatJsonSchema jsonSchema =
params.responseFormat().get().asJsonSchema();
+ assertThat(jsonSchema.jsonSchema().strict()).contains(true);
+ }
+
+ @Test
+ @DisplayName("Native NOT applied for a POJO on an incapable model (prompt
fallback)")
+ void testNativeNotAppliedForIncapableModel() {
+ ChatCompletionCreateParams params =
+ connection()
+ .buildRequest(
+ userMessage(), List.of(),
params("gpt-3.5-turbo"), Person.class);
+
+ assertThat(params.responseFormat()).isEmpty();
+ }
+
+ @Test
+ @DisplayName("Native NOT applied for a pre-cutoff same-family gpt-4o
snapshot")
+ void testNativeNotAppliedForPreCutoffSnapshot() {
+ // gpt-4o-2024-05-13 predates the Structured Outputs cutoff even
though it shares the gpt-4o
+ // prefix; treating it as capable would fail silently at the provider.
+ ChatCompletionCreateParams params =
+ connection()
+ .buildRequest(
+ userMessage(),
+ List.of(),
+ params("gpt-4o-2024-05-13"),
+ Person.class);
+
+ assertThat(params.responseFormat()).isEmpty();
+ }
+
+ @Test
+ @DisplayName("Native NOT applied when no output schema is supplied")
+ void testNativeNotAppliedWhenSchemaNull() {
+ ChatCompletionCreateParams params =
+ connection().buildRequest(userMessage(), List.of(),
params("gpt-4o"), null);
+
+ assertThat(params.responseFormat()).isEmpty();
+ }
+
+ @Test
+ @DisplayName("Native NOT applied for a non-POJO schema form (POJO-only
scope)")
+ void testNativeNotAppliedForNonPojoSchema() {
+ // A RowTypeInfo schema arrives wrapped in OutputSchema (not a bare
POJO Class), so it must
+ // not activate native structured output; any non-Class schema object
exercises the same
+ // instanceof gate.
+ Object nonClassSchema = "row<name STRING>";
+
+ ChatCompletionCreateParams params =
+ connection()
+ .buildRequest(userMessage(), List.of(),
params("gpt-4o"), nonClassSchema);
+
+ assertThat(params.responseFormat()).isEmpty();
+ }
+
+ @Test
+ @DisplayName("Native applied for a POJO even when tools are bound (no
empty-tools gate)")
+ void testNativeAppliedEvenWhenToolsBound() {
+ ChatCompletionCreateParams params =
+ connection()
+ .buildRequest(
+ userMessage(),
+ List.of(new StubTool()),
+ params("gpt-4o"),
+ Person.class);
+
+ assertThat(params.responseFormat()).isPresent();
+ }
+
+ @Test
+ @DisplayName("Capability predicate accepts the documented capable models")
+ void testCapabilityPredicateAcceptsCapableModels() {
+ OpenAICompletionsConnection connection = connection();
+
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-2024-08-06")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-2024-11-20")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-mini")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-mini-2024-07-18")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-search-preview")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-search-preview-2025-03-11"))
+ .isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-mini-search-preview"))
+ .isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4.1")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4.1-mini")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-5")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-5-mini")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-5-chat-latest")).isTrue();
+ assertThat(connection.supportsNativeStructuredOutput("o1")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("o1-2024-12-17")).isTrue();
+ assertThat(connection.supportsNativeStructuredOutput("o3")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("o3-mini")).isTrue();
+
assertThat(connection.supportsNativeStructuredOutput("o4-mini")).isTrue();
+ }
+
+ @Test
+ @DisplayName(
+ "Capability predicate rejects non-text modality, incapable,
pre-cutoff, unknown, empty,"
+ + " and null models")
+ void testCapabilityPredicateRejectsIncapableModels() {
+ OpenAICompletionsConnection connection = connection();
+
+
assertThat(connection.supportsNativeStructuredOutput("gpt-3.5-turbo")).isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4")).isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4-turbo")).isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-2024-05-13")).isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-audio-preview")).isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-mini-audio-preview"))
+ .isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-mini-realtime-preview"))
+ .isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-mini-tts")).isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("gpt-4o-mini-transcribe")).isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("o1-mini")).isFalse();
+
assertThat(connection.supportsNativeStructuredOutput("some-unknown-model")).isFalse();
+ assertThat(connection.supportsNativeStructuredOutput("")).isFalse();
+ assertThat(connection.supportsNativeStructuredOutput(null)).isFalse();
+ }
+
+ /** Minimal tool stub; only its presence in the tools list matters. */
+ private static class StubTool extends Tool {
+ StubTool() {
+ super(new ToolMetadata("add", "adds", "{\"type\":\"object\"}"));
+ }
+
+ @Override
+ public ToolType getToolType() {
+ return ToolType.FUNCTION;
+ }
+
+ @Override
+ public ToolResponse call(ToolParameters parameters) {
+ return ToolResponse.success(null);
+ }
+ }
+}
diff --git
a/python/flink_agents/integrations/chat_models/openai/openai_chat_model.py
b/python/flink_agents/integrations/chat_models/openai/openai_chat_model.py
index 68c2414b..381098b0 100644
--- a/python/flink_agents/integrations/chat_models/openai/openai_chat_model.py
+++ b/python/flink_agents/integrations/chat_models/openai/openai_chat_model.py
@@ -19,7 +19,13 @@ from typing import Any, Dict, List, Literal, Sequence
import httpx
from openai import NOT_GIVEN, OpenAI
-from pydantic import Field, PrivateAttr
+
+# Private SDK module (leading underscore): the openai client itself uses this
helper to
+# build the strict json_schema for response_format, and there is no public
re-export. It
+# has existed at this path since the structured-output support in openai
1.66.3 (the
+# pinned minimum). A future openai bump that moves it will fail loudly on
import here.
+from openai.lib._pydantic import to_strict_json_schema
+from pydantic import BaseModel, Field, PrivateAttr
from typing_extensions import override
from flink_agents.api.agents.types import OutputSchema
@@ -38,6 +44,65 @@ from
flink_agents.integrations.chat_models.openai.openai_utils import (
DEFAULT_OPENAI_MODEL = "gpt-4o-mini"
+# Models with documented json_schema strict Structured Outputs support. Source
of
+# truth: https://platform.openai.com/docs/guides/structured-outputs
+#
+# A name carrying a non-text modality marker is rejected before anything else.
Audio,
+# realtime, speech and transcription variants expose no json_schema response
format
+# even though they share the name prefix of a capable text family, so the
marker check
+# has to win over a prefix match rather than merely coexist with it.
+#
+# A text family whose entire lifetime post-dates the Structured Outputs cutoff
is
+# matched by name prefix, so its dated snapshots and size variants resolve
without
+# enumerating each one.
+#
+# Two cases cannot use a prefix and are matched exactly instead. The gpt-4o
family
+# straddles the cutoff -- gpt-4o-2024-05-13 predates it and is not capable --
so the
+# boundary there is temporal rather than nominal. The o1 family is not
uniform: o1 is
+# capable while o1-mini is not, so an "o1" prefix would admit an incapable
sibling.
+#
+# A name outside every listed family reports not-capable and degrades to the
prompt
+# fallback rather than failing at the provider. Within a listed family the
prefix
+# assumes capability, so a family variant that ships without json_schema
support has
+# to be excluded explicitly, either by a marker that appears in no capable
name or by
+# replacing the family prefix with exact names.
+_NON_TEXT_MODALITY_MARKERS = ("-audio", "-realtime", "-tts", "-transcribe")
+_NATIVE_STRUCTURED_OUTPUT_FAMILY_PREFIXES = (
+ "gpt-4o-mini",
+ "gpt-4o-search-preview",
+ "gpt-4.1",
+ "gpt-5",
+ "o3",
+ "o4-mini",
+)
+_NATIVE_STRUCTURED_OUTPUT_MODELS = frozenset(
+ {"gpt-4o", "gpt-4o-2024-08-06", "gpt-4o-2024-11-20", "o1", "o1-2024-12-17"}
+)
+
+
+def _native_response_format(output_schema: Any) -> Dict[str, Any] | None:
+ """Build the OpenAI ``response_format`` for a native structured-output
request.
+
+ Returns ``None`` (leaving behavior unchanged) unless the schema is a
``BaseModel``
+ subclass. A ``RowTypeInfo`` schema is skipped so it keeps the
prompt-engineering
+ fallback.
+ """
+ if output_schema is None:
+ return None
+ model = (
+ output_schema.output_schema if isinstance(output_schema, OutputSchema)
else None
+ )
+ if not (isinstance(model, type) and issubclass(model, BaseModel)):
+ return None
+ return {
+ "type": "json_schema",
+ "json_schema": {
+ "name": model.__name__,
+ "schema": to_strict_json_schema(model),
+ "strict": True,
+ },
+ }
+
class OpenAIChatModelConnection(BaseChatModelConnection):
"""The connection to the OpenAI LLM.
@@ -135,6 +200,25 @@ class OpenAIChatModelConnection(BaseChatModelConnection):
"http_client": self._http_client,
}
+ @override
+ def supports_native_structured_output(self, effective_model: str | None)
-> bool:
+ """Whether OpenAI documents json_schema strict support for
``effective_model``.
+
+ See the module-level allowlist for the source of truth and the
rationale for
+ rejecting non-text modality variants, matching capable text families
by prefix,
+ and matching the gpt-4o snapshots and the o1 names exactly. A name
outside
+ every listed family reports ``False`` so it degrades to the
prompt-engineering
+ fallback rather than failing at the provider.
+ """
+ if not effective_model:
+ return False
+ if any(marker in effective_model for marker in
_NON_TEXT_MODALITY_MARKERS):
+ return False
+ return (
+
effective_model.startswith(_NATIVE_STRUCTURED_OUTPUT_FAMILY_PREFIXES)
+ or effective_model in _NATIVE_STRUCTURED_OUTPUT_MODELS
+ )
+
def chat(
self,
messages: Sequence[ChatMessage],
@@ -151,10 +235,10 @@ class OpenAIChatModelConnection(BaseChatModelConnection):
tools : Optional[List]
List of tools that can be called by the model
output_schema : OutputSchema | None
- Rejected when non-``None``: this connection has no native
structured-output
- translation, so callers stay on the prompt-engineering fallback.
Declaring
- the parameter keeps a caller-supplied schema out of ``**kwargs``,
which is
- forwarded to the provider SDK.
+ The schema the response should conform to, or ``None`` for an
unconstrained
+ response. Native structured output is applied only for a
``BaseModel``
+ schema on a model the provider documents as capable; a
``RowTypeInfo``
+ schema or an incapable model keeps the prompt-engineering fallback.
**kwargs : Any
Additional parameters passed to the model service (e.g.,
temperature,
max_tokens, etc.)
@@ -164,7 +248,6 @@ class OpenAIChatModelConnection(BaseChatModelConnection):
ChatMessage
Model response message
"""
- self._reject_unsupported_output_schema(output_schema)
tool_specs = None
if tools is not None:
tool_specs = [to_openai_tool(metadata=tool.metadata) for tool in
tools]
@@ -174,6 +257,19 @@ class OpenAIChatModelConnection(BaseChatModelConnection):
tool_spec["function"]["strict"] = strict
tool_spec["function"]["parameters"]["additionalProperties"] = False
+ # TODO(#912): the requested strategy is not visible here, so this check
+ # cannot tell an explicit NATIVE request apart from one that merely
+ # resolved to native. A caller asking for NATIVE on a model this
+ # predicate rejects therefore gets an unconstrained response instead of
+ # an error. Once strategy resolution is wired up, NATIVE must either
+ # bypass this capability check or fail explicitly.
+ if output_schema is not None and
self.supports_native_structured_output(
+ kwargs.get("model")
+ ):
+ response_format = _native_response_format(output_schema)
+ if response_format is not None:
+ kwargs["response_format"] = response_format
+
response = self.client.chat.completions.create(
messages=convert_to_openai_messages(messages),
tools=tool_specs or NOT_GIVEN,
diff --git
a/python/flink_agents/integrations/chat_models/openai/tests/test_openai_native_structured_output.py
b/python/flink_agents/integrations/chat_models/openai/tests/test_openai_native_structured_output.py
new file mode 100644
index 00000000..f3149249
--- /dev/null
+++
b/python/flink_agents/integrations/chat_models/openai/tests/test_openai_native_structured_output.py
@@ -0,0 +1,204 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#################################################################################
+from typing import Any
+from unittest.mock import MagicMock
+
+import pytest
+from pydantic import BaseModel
+from pyflink.common.typeinfo import Types
+
+from flink_agents.api.agents.types import OutputSchema
+from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.integrations.chat_models.openai.openai_chat_model import (
+ OpenAIChatModelConnection,
+)
+from flink_agents.plan.function import PythonFunction
+from flink_agents.plan.tools.function_tool import FunctionTool
+
+
+class Person(BaseModel):
+ """A representative BaseModel output schema."""
+
+ name: str
+ age: int
+
+
+def _connection() -> OpenAIChatModelConnection:
+ conn = OpenAIChatModelConnection(
+ name="openai", api_key="test-key", api_base_url="http://localhost";
+ )
+ mock_client = MagicMock()
+ mock_message = MagicMock()
+ mock_message.role = "assistant"
+ mock_message.content = "ok"
+ mock_message.tool_calls = None
+ mock_client.chat.completions.create.return_value.choices = [
+ MagicMock(message=mock_message)
+ ]
+ mock_client.chat.completions.create.return_value.usage = None
+ conn._client = mock_client
+ return conn
+
+
+def _create_call_kwargs(conn: OpenAIChatModelConnection) -> dict[str, Any]:
+ return conn.client.chat.completions.create.call_args.kwargs
+
+
+def _add(a: int, b: int) -> int:
+ """Add two integers.
+
+ Parameters
+ ----------
+ a : int
+ first
+ b : int
+ second
+
+ Returns:
+ -------
+ int
+ sum
+ """
+ return a + b
+
+
+def test_native_applied_for_basemodel_capable_model() -> None:
+ """response_format json_schema strict applied for a BaseModel on a capable
model."""
+ conn = _connection()
+ conn.chat(
+ [ChatMessage(role=MessageRole.USER, content="hi")],
+ model="gpt-4o",
+ output_schema=OutputSchema(output_schema=Person),
+ )
+ response_format = _create_call_kwargs(conn)["response_format"]
+ assert response_format["type"] == "json_schema"
+ assert response_format["json_schema"]["strict"] is True
+ assert response_format["json_schema"]["schema"]["additionalProperties"] is
False
+
+
+def test_native_not_applied_for_incapable_model() -> None:
+ """Native NOT applied for a BaseModel on an incapable model (prompt
fallback)."""
+ conn = _connection()
+ conn.chat(
+ [ChatMessage(role=MessageRole.USER, content="hi")],
+ model="gpt-3.5-turbo",
+ output_schema=OutputSchema(output_schema=Person),
+ )
+ assert "response_format" not in _create_call_kwargs(conn)
+
+
+def test_native_not_applied_for_pre_cutoff_snapshot() -> None:
+ """Native NOT applied for a pre-cutoff same-family gpt-4o snapshot.
+
+ gpt-4o-2024-05-13 predates the Structured Outputs cutoff even though it
shares the
+ gpt-4o prefix; treating it as capable would fail silently at the provider.
+ """
+ conn = _connection()
+ conn.chat(
+ [ChatMessage(role=MessageRole.USER, content="hi")],
+ model="gpt-4o-2024-05-13",
+ output_schema=OutputSchema(output_schema=Person),
+ )
+ assert "response_format" not in _create_call_kwargs(conn)
+
+
+def test_native_not_applied_when_schema_none() -> None:
+ """Native NOT applied when no output schema is supplied."""
+ conn = _connection()
+ conn.chat(
+ [ChatMessage(role=MessageRole.USER, content="hi")],
+ model="gpt-4o",
+ output_schema=None,
+ )
+ assert "response_format" not in _create_call_kwargs(conn)
+
+
+def test_native_not_applied_for_row_type_info() -> None:
+ """Native NOT applied for a RowTypeInfo schema (BaseModel-only scope)."""
+ conn = _connection()
+ row_type = Types.ROW_NAMED(["name"], [Types.STRING()])
+ conn.chat(
+ [ChatMessage(role=MessageRole.USER, content="hi")],
+ model="gpt-4o",
+ output_schema=OutputSchema(output_schema=row_type),
+ )
+ assert "response_format" not in _create_call_kwargs(conn)
+
+
+def test_native_applied_even_when_tools_bound() -> None:
+ """Native applied for a BaseModel even when tools are bound (no
empty-tools gate)."""
+ conn = _connection()
+ tool = FunctionTool(func=PythonFunction.from_callable(_add))
+ conn.chat(
+ [ChatMessage(role=MessageRole.USER, content="hi")],
+ tools=[tool],
+ model="gpt-4o",
+ output_schema=OutputSchema(output_schema=Person),
+ )
+ assert "response_format" in _create_call_kwargs(conn)
+
+
[email protected](
+ "model",
+ [
+ "gpt-4o",
+ "gpt-4o-2024-08-06",
+ "gpt-4o-2024-11-20",
+ "gpt-4o-mini",
+ "gpt-4o-mini-2024-07-18",
+ "gpt-4o-search-preview",
+ "gpt-4o-search-preview-2025-03-11",
+ "gpt-4o-mini-search-preview",
+ "gpt-4.1",
+ "gpt-4.1-mini",
+ "gpt-5",
+ "gpt-5-mini",
+ "gpt-5-chat-latest",
+ "o1",
+ "o1-2024-12-17",
+ "o3",
+ "o3-mini",
+ "o4-mini",
+ ],
+)
+def test_capability_predicate_accepts_capable_models(model: str) -> None:
+ """The capability predicate accepts the documented capable models."""
+ assert _connection().supports_native_structured_output(model) is True
+
+
[email protected](
+ "model",
+ [
+ "gpt-3.5-turbo",
+ "gpt-4",
+ "gpt-4-turbo",
+ "gpt-4o-2024-05-13",
+ "gpt-4o-audio-preview",
+ "gpt-4o-mini-audio-preview",
+ "gpt-4o-mini-realtime-preview",
+ "gpt-4o-mini-tts",
+ "gpt-4o-mini-transcribe",
+ "o1-mini",
+ "some-unknown-model",
+ "",
+ None,
+ ],
+)
+def test_capability_predicate_rejects_incapable_models(model: str | None) ->
None:
+ """The predicate rejects modality variants, incapable, unknown, and empty
models."""
+ assert _connection().supports_native_structured_output(model) is False
diff --git
a/python/flink_agents/integrations/chat_models/tests/test_output_schema_param_declared.py
b/python/flink_agents/integrations/chat_models/tests/test_output_schema_param_declared.py
index e31d3753..9b814c3c 100644
---
a/python/flink_agents/integrations/chat_models/tests/test_output_schema_param_declared.py
+++
b/python/flink_agents/integrations/chat_models/tests/test_output_schema_param_declared.py
@@ -87,6 +87,21 @@ def _connections_implementing_chat() ->
List[Type[BaseChatModelConnection]]:
]
+def _translates_schema_natively(cls: Type[BaseChatModelConnection]) -> bool:
+ """Whether ``cls`` applies an output schema through a native provider
parameter.
+
+ Overriding ``supports_native_structured_output`` is how a connection
reports that
+ capability, so the same override marks the connections that must accept a
schema
+ instead of rejecting it. Such a connection owns the decision of what to do
with a
+ schema it cannot apply natively for the effective model, and its own tests
pin
+ that behavior.
+ """
+ return (
+ cls.supports_native_structured_output
+ is not BaseChatModelConnection.supports_native_structured_output
+ )
+
+
def test_every_connection_declares_output_schema_param() -> None:
"""Every connection in the tree must declare ``output_schema`` defaulting
to None.
@@ -147,13 +162,20 @@ def
test_every_connection_rejects_an_output_schema_it_cannot_translate() -> None
own it lets a connection silently return an unconstrained response that
the caller
would treat as schema-conforming. Rejecting turns that into an error at
the call.
+ A connection that does translate a schema natively is exempt, since
rejecting is
+ exactly what it must not do.
+
The tree is walked rather than hand-listed, so a connection added to a
module that
is already imported is held to the contract for free. Reach still stops at
those
imports: ``__subclasses__()`` only sees classes that have been imported,
so a
connection in a brand-new module needs that module added at the top of
this file.
"""
schema = OutputSchema(output_schema=_Answer)
- connections = _connections_implementing_chat()
+ connections = [
+ cls
+ for cls in _connections_implementing_chat()
+ if not _translates_schema_natively(cls)
+ ]
assert connections, "the connection walk found nothing to check"
offenders = [
