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 b70f8a63 [integrations][java] List Jackson enum values in OpenAI,
Azure and Anthropic native schemas (#1247)
b70f8a63 is described below
commit b70f8a6336019719c970974d4f30dd8074283c7d
Author: Weiqing Yang <[email protected]>
AuthorDate: Tue Oct 13 19:56:57 2026 -0700
[integrations][java] List Jackson enum values in OpenAI, Azure and
Anthropic native schemas (#1247)
Generated-by: GitHub Copilot CLI 1.0.95 (claude-opus-5.5)
---
integrations/chat-models/anthropic/pom.xml | 12 ++
.../anthropic/AnthropicChatModelConnection.java | 56 ++++-----
.../AnthropicChatModelConnectionTest.java | 114 +++++++++++++++++-
.../chatmodels/common/PojoJsonSchemaGenerator.java | 35 +++++-
.../common/PojoJsonSchemaGeneratorTest.java | 32 ++++++
integrations/chat-models/openai/pom.xml | 12 ++
.../openai/AzureOpenAIChatModelConnection.java | 24 +---
.../openai/OpenAIChatCompletionsUtils.java | 38 +++++-
.../openai/OpenAICompletionsConnection.java | 25 +---
.../openai/AzureOpenAIChatModelConnectionTest.java | 68 ++++++++++-
.../openai/OpenAICompletionsConnectionTest.java | 127 ++++++++++++++++++++-
11 files changed, 448 insertions(+), 95 deletions(-)
diff --git a/integrations/chat-models/anthropic/pom.xml
b/integrations/chat-models/anthropic/pom.xml
index 07d0b592..aad13568 100644
--- a/integrations/chat-models/anthropic/pom.xml
+++ b/integrations/chat-models/anthropic/pom.xml
@@ -44,6 +44,18 @@ under the License.
<version>${anthropic.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+
<artifactId>flink-agents-integrations-chat-models-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- Versions managed by the victools BOM imported in the root pom. -->
+ <dependency>
+ <groupId>com.github.victools</groupId>
+ <artifactId>jsonschema-generator</artifactId>
+ </dependency>
+
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
diff --git
a/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java
b/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java
index d4c3ecee..7331e82a 100644
---
a/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java
+++
b/integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java
@@ -19,7 +19,8 @@ package
org.apache.flink.agents.integrations.chatmodels.anthropic;
import com.anthropic.client.AnthropicClient;
import com.anthropic.client.okhttp.AnthropicOkHttpClient;
-import com.anthropic.core.JsonSchemaLocalValidation;
+import com.anthropic.core.JsonValue;
+import com.anthropic.models.messages.JsonOutputFormat;
import com.anthropic.models.messages.Message;
import com.anthropic.models.messages.MessageCreateParams;
import com.anthropic.models.messages.MessageParam;
@@ -27,6 +28,7 @@ import com.anthropic.models.messages.Model;
import com.anthropic.models.messages.OutputConfig;
import com.anthropic.models.messages.TextBlockParam;
import com.anthropic.models.messages.Tool;
+import com.github.victools.jsonschema.generator.Option;
import org.apache.flink.agents.api.chat.messages.ChatMessage;
import org.apache.flink.agents.api.chat.messages.ChatResult;
import org.apache.flink.agents.api.chat.messages.MessageRole;
@@ -35,11 +37,13 @@ import
org.apache.flink.agents.api.chat.model.BaseChatModelConnection;
import org.apache.flink.agents.api.chat.model.NativeStructuredOutputSupport;
import org.apache.flink.agents.api.resource.ResourceContext;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import
org.apache.flink.agents.integrations.chatmodels.common.PojoJsonSchemaGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -402,34 +406,32 @@ public class AnthropicChatModelConnection extends
BaseChatModelConnection {
}
/**
- * Derives the native {@code output_config} for a POJO class through the
SDK's typed
- * structured-output builder.
+ * Derives the native {@code output_config} for a POJO output schema class.
*
- * <p>The Kotlin facade {@code StructuredOutputsKt.outputFormatFromClass}
would produce this
- * directly, but it is compiled {@code ACC_SYNTHETIC} and so cannot be
named from Java. The
- * typed builder generates the same schema; the config is extracted from
the throwaway request
- * it produces and reattached to the real one, which also avoids that
overload's side effect of
- * retyping the request and the response as {@code
StructuredMessageCreateParams} and {@code
- * StructuredMessage}. The throwaway request is never sent, so its
placeholder model, message
- * and token limit only have to satisfy the builder's required-field check.
- *
- * <p>Local schema validation is off so that the provider, not the client,
is the authority on
- * which schemas it accepts.
+ * <p>The schema comes from the shared generator rather than the SDK's
typed {@code
+ * outputConfig(Class)} builder, whose schema lists an annotated enum
constant by its Java name,
+ * a value the response then fails to deserialize from. It lists every
property as required and
+ * forbids additional properties, as the schema that builder derives does.
*/
- private static <T> OutputConfig toNativeOutputConfig(Class<T> schemaClass)
{
- return MessageCreateParams.builder()
- .model(Model.of(""))
- .addUserMessage("")
- .maxTokens(1)
- .outputConfig(schemaClass, JsonSchemaLocalValidation.NO)
- .build()
- .rawParams()
- .outputConfig()
- .orElseThrow(
- () ->
- new IllegalStateException(
- "Anthropic SDK did not produce an
output_config for schema "
- + schemaClass.getName()));
+ private static OutputConfig toNativeOutputConfig(Class<?> schemaClass) {
+ Map<String, JsonValue> schema = new LinkedHashMap<>();
+ PojoJsonSchemaGenerator.generateAllRequired(
+ schemaClass,
Option.FORBIDDEN_ADDITIONAL_PROPERTIES_BY_DEFAULT)
+ .fields()
+ .forEachRemaining(
+ member ->
+ schema.put(
+ member.getKey(),
+
JsonValue.fromJsonNode(member.getValue())));
+ return OutputConfig.builder()
+ .format(
+ JsonOutputFormat.builder()
+ .schema(
+ JsonOutputFormat.Schema.builder()
+ .additionalProperties(schema)
+ .build())
+ .build())
+ .build();
}
@Override
diff --git
a/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnectionTest.java
b/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnectionTest.java
index 77058ac9..54d9077d 100644
---
a/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnectionTest.java
+++
b/integrations/chat-models/anthropic/src/test/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnectionTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.agents.integrations.chatmodels.anthropic;
+import com.anthropic.core.ObjectMappers;
import com.anthropic.models.messages.ContentBlockParam;
import com.anthropic.models.messages.Message;
import com.anthropic.models.messages.MessageCreateParams;
@@ -27,9 +28,12 @@ import com.anthropic.models.messages.OutputConfig;
import com.anthropic.models.messages.StopReason;
import com.anthropic.models.messages.TextBlock;
import com.anthropic.models.messages.Usage;
+import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.annotation.JsonValue;
import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.agents.api.chat.messages.ChatMessage;
import org.apache.flink.agents.api.chat.messages.ChatResult;
@@ -52,6 +56,7 @@ import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.NullSource;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -310,12 +315,12 @@ class AnthropicChatModelConnectionTest {
"");
}
- /** A POJO the SDK can derive a JSON schema from. */
+ /** A representative POJO output schema. */
public static class Answer {
public String verdict;
}
- /** A polymorphic member, which the SDK renders as a discriminated union.
*/
+ /** A polymorphic member, which the derived schema states as a
discriminated union. */
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "kind")
@JsonSubTypes({
@JsonSubTypes.Type(value = Dog.class, name = "dog"),
@@ -339,6 +344,48 @@ class AnthropicChatModelConnectionTest {
public Pet pet;
}
+ /**
+ * Output schema fixture whose enum constants are deserialized from values
other than their
+ * names, one through {@code @JsonProperty} on the constants and one
through a
+ * {@code @JsonValue} method.
+ */
+ public static class Ticket {
+ public Status status;
+ public Phase phase;
+ }
+
+ /** Enum constants deserialized from their {@code @JsonProperty} values. */
+ public enum Status {
+ @JsonProperty("in-progress")
+ IN_PROGRESS,
+ @JsonProperty("done")
+ DONE
+ }
+
+ /** Enum constants deserialized from their {@code @JsonValue} method. */
+ public enum Phase {
+ STARTED("started"),
+ FINISHED("finished");
+
+ private final String wire;
+
+ Phase(String wire) {
+ this.wire = wire;
+ }
+
+ @JsonValue
+ public String wire() {
+ return wire;
+ }
+ }
+
+ /** Output schema fixture with an {@code Optional} field and a nested
object. */
+ public static class Draft {
+ public String title;
+ public Optional<String> note;
+ public Answer answer;
+ }
+
private static Map<String, Object> paramsWithModel(String model, Object
jsonPrefill) {
Map<String, Object> params = params(jsonPrefill);
params.put("model", model);
@@ -367,6 +414,23 @@ class AnthropicChatModelConnectionTest {
.orElse(Set.of());
}
+ /** The JSON Schema the request carries, as a JSON tree. */
+ private static JsonNode
nativeSchema(AnthropicChatModelConnection.BuiltRequest built) {
+ return ObjectMappers.jsonMapper()
+ .valueToTree(
+ built.params
+ .outputConfig()
+ .flatMap(OutputConfig::format)
+ .orElseThrow()
+ .schema());
+ }
+
+ private static List<String> textValues(JsonNode array) {
+ List<String> values = new ArrayList<>();
+ array.forEach(value -> values.add(value.asText()));
+ return values;
+ }
+
/** The JSON Schema the request carries, as the SDK holds it on the output
config. */
private static String
nativeSchemaPayload(AnthropicChatModelConnection.BuiltRequest built) {
return built.params
@@ -495,10 +559,11 @@ class AnthropicChatModelConnectionTest {
}
@Test
- @DisplayName("a polymorphic member is sent as the discriminated union the
SDK derives")
+ @DisplayName("a polymorphic member is sent as a discriminated union")
void testPolymorphicMemberSchemaIsSent() {
- // Jackson renders this member as an object declaring no properties,
while the SDK derives
- // the full union the provider accepts. Reading a Jackson-rendered
schema here would refuse
+ // Jackson renders this member as an object declaring no properties,
while the schema
+ // generator derives the full union the provider accepts. Reading a
Jackson-rendered schema
+ // here would refuse
// a request that works.
assertThat(nativeSchemaPayload(build(CAPABLE_MODEL, Owner.class,
null)))
.contains("bark={type=string}")
@@ -506,6 +571,45 @@ class AnthropicChatModelConnectionTest {
.contains("kind={const=dog}");
}
+ @Test
+ @DisplayName("the derived schema lists enum constants the way Jackson
deserializes them")
+ void derivedSchemaFollowsJacksonEnumValues() throws Exception {
+ JsonNode properties =
+ nativeSchema(build(CAPABLE_MODEL, Ticket.class,
null)).path("properties");
+
+ // Structured outputs confine the model to the listed values, so each
has to deserialize
+ // into the enum. Listed by constant name instead, a plain mapper
refuses every value the
+ // schema allows.
+ List<Status> statuses = new ArrayList<>();
+ for (JsonNode value : properties.path("status").path("enum")) {
+ statuses.add(MAPPER.treeToValue(value, Status.class));
+ }
+ assertThat(statuses).containsExactlyInAnyOrder(Status.values());
+
+ List<Phase> phases = new ArrayList<>();
+ for (JsonNode value : properties.path("phase").path("enum")) {
+ phases.add(MAPPER.treeToValue(value, Phase.class));
+ }
+ assertThat(phases).containsExactlyInAnyOrder(Phase.values());
+ }
+
+ @Test
+ @DisplayName("the derived schema requires every field and forbids extras
at every level")
+ void derivedSchemaRequiresEveryFieldAndForbidsExtras() {
+ JsonNode schema = nativeSchema(build(CAPABLE_MODEL, Draft.class,
null));
+
+ // A response can neither drop a declared field nor add an undeclared
one; the Optional
+ // field is required too and admits null instead.
+ assertThat(textValues(schema.path("required")))
+ .containsExactlyInAnyOrder("title", "note", "answer");
+
assertThat(textValues(schema.path("properties").path("note").path("type")))
+ .contains("null");
+
assertThat(schema.path("additionalProperties").asBoolean(true)).isFalse();
+ JsonNode answer = schema.path("properties").path("answer");
+
assertThat(textValues(answer.path("required"))).containsExactly("verdict");
+
assertThat(answer.path("additionalProperties").asBoolean(true)).isFalse();
+ }
+
@Test
@DisplayName("a caller-supplied output_config wins and the schema falls
back")
void testCallerOutputConfigWinsOverSchema() {
diff --git
a/integrations/chat-models/common/src/main/java/org/apache/flink/agents/integrations/chatmodels/common/PojoJsonSchemaGenerator.java
b/integrations/chat-models/common/src/main/java/org/apache/flink/agents/integrations/chatmodels/common/PojoJsonSchemaGenerator.java
index e36322e9..08a973ff 100644
---
a/integrations/chat-models/common/src/main/java/org/apache/flink/agents/integrations/chatmodels/common/PojoJsonSchemaGenerator.java
+++
b/integrations/chat-models/common/src/main/java/org/apache/flink/agents/integrations/chatmodels/common/PojoJsonSchemaGenerator.java
@@ -18,6 +18,7 @@
package org.apache.flink.agents.integrations.chatmodels.common;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.github.victools.jsonschema.generator.FieldScope;
import com.github.victools.jsonschema.generator.Option;
import com.github.victools.jsonschema.generator.OptionPreset;
import com.github.victools.jsonschema.generator.SchemaGenerator;
@@ -28,6 +29,7 @@ import
com.github.victools.jsonschema.module.jackson.JacksonModule;
import com.github.victools.jsonschema.module.jackson.JacksonOption;
import java.util.Optional;
+import java.util.function.Predicate;
/** Derives the JSON Schema that chat model connections send as a native
output schema. */
public final class PojoJsonSchemaGenerator {
@@ -44,9 +46,12 @@ public final class PojoJsonSchemaGenerator {
// e.g. "getDerived()".
// - Sorting fields before methods and applying no further comparison
leaves properties in
// declaration order, which is the order pydantic emits.
- // - The required check marks every field required except an Optional
one. The default marks
- // nothing required, which lets a model omit fields at will, while
marking everything
- // required would force the fields a caller declared omissible.
+ // - The required check of generate marks every field required except an
Optional one. The
+ // default marks nothing required, which lets a model omit fields at
will, while marking
+ // everything required would force the fields a caller declared
omissible.
+ // generateAllRequired marks every field required for callers whose
provider expects every
+ // property in the required set, as OpenAI's strict mode does. An
Optional field stays
+ // nullable there, so the schema still admits null for it.
// - The Jackson module names properties the way an ObjectMapper reads
them, honoring
// @JsonProperty renames and dropping @JsonIgnore fields, which the
required check would
// otherwise mark required. It lists enum constants mapped by
@JsonProperty or by a
@@ -64,6 +69,27 @@ public final class PojoJsonSchemaGenerator {
* @return the schema document
*/
public static ObjectNode generate(Class<?> type, Option... options) {
+ return generate(
+ type, field ->
!Optional.class.equals(field.getRawMember().getType()), options);
+ }
+
+ /**
+ * Generates the JSON Schema for {@code type} with every field required.
+ *
+ * <p>The schema differs from the one {@link #generate} derives only in
its required set, which
+ * also lists an {@code Optional} field. That field's schema admits null,
so a response can
+ * still send null for it.
+ *
+ * @param type the class a response is deserialized into
+ * @param options additional generator options enabled on top of the fixed
settings
+ * @return the schema document
+ */
+ public static ObjectNode generateAllRequired(Class<?> type, Option...
options) {
+ return generate(type, field -> true, options);
+ }
+
+ private static ObjectNode generate(
+ Class<?> type, Predicate<FieldScope> requiredCheck, Option...
options) {
SchemaGeneratorConfigBuilder builder =
new SchemaGeneratorConfigBuilder(
SchemaVersion.DRAFT_2020_12, OptionPreset.PLAIN_JSON);
@@ -76,8 +102,7 @@ public final class PojoJsonSchemaGenerator {
JacksonOption.FLATTENED_ENUMS_FROM_JSONVALUE));
builder.forTypesInGeneral()
.withPropertySorter(PropertySortUtils.SORT_PROPERTIES_FIELDS_BEFORE_METHODS);
- builder.forFields()
- .withRequiredCheck(field ->
!Optional.class.equals(field.getRawMember().getType()));
+ builder.forFields().withRequiredCheck(requiredCheck);
return new SchemaGenerator(builder.build()).generateSchema(type);
}
}
diff --git
a/integrations/chat-models/common/src/test/java/org/apache/flink/agents/integrations/chatmodels/common/PojoJsonSchemaGeneratorTest.java
b/integrations/chat-models/common/src/test/java/org/apache/flink/agents/integrations/chatmodels/common/PojoJsonSchemaGeneratorTest.java
index 838336a7..1d567f6e 100644
---
a/integrations/chat-models/common/src/test/java/org/apache/flink/agents/integrations/chatmodels/common/PojoJsonSchemaGeneratorTest.java
+++
b/integrations/chat-models/common/src/test/java/org/apache/flink/agents/integrations/chatmodels/common/PojoJsonSchemaGeneratorTest.java
@@ -22,6 +22,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonValue;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
import com.github.victools.jsonschema.generator.Option;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
@@ -145,6 +146,37 @@ class PojoJsonSchemaGeneratorTest {
.containsExactlyInAnyOrder("summary", "counts", "total");
}
+ @Test
+ @DisplayName("The all-required schema requires an Optional field and
admits null for it")
+ void allRequiredRequiresOptionalFieldsAsNullable() {
+ JsonNode schema =
PojoJsonSchemaGenerator.generateAllRequired(Report.class);
+
+ // A provider expecting every property in the required set rejects a
schema that leaves
+ // the Optional field out, so the field stays omissible only by taking
null.
+ assertThat(textValues(schema.path("required")))
+ .containsExactlyInAnyOrder("summary", "counts", "note",
"total");
+
assertThat(textValues(schema.path("properties").path("note").path("type")))
+ .contains("null");
+ }
+
+ @Test
+ @DisplayName("The all-required schema differs from the default one only in
its required set")
+ void allRequiredChangesOnlyTheRequiredSet() {
+ for (Class<?> type : List.of(Report.class, Profile.class,
Ticket.class)) {
+ ObjectNode allRequired =
+ PojoJsonSchemaGenerator.generateAllRequired(
+ type,
Option.FORBIDDEN_ADDITIONAL_PROPERTIES_BY_DEFAULT);
+ ObjectNode partlyRequired =
+ PojoJsonSchemaGenerator.generate(
+ type,
Option.FORBIDDEN_ADDITIONAL_PROPERTIES_BY_DEFAULT);
+ allRequired.remove("required");
+ partlyRequired.remove("required");
+
+ // Compared as text, so property order, enum values and the option
count too.
+
assertThat(allRequired.toString()).isEqualTo(partlyRequired.toString());
+ }
+ }
+
@Test
@DisplayName("The generated schema keeps declaration order and emits no
getter property")
void keepsDeclarationOrderWithoutGetters() {
diff --git a/integrations/chat-models/openai/pom.xml
b/integrations/chat-models/openai/pom.xml
index 361ed301..3f4badcd 100644
--- a/integrations/chat-models/openai/pom.xml
+++ b/integrations/chat-models/openai/pom.xml
@@ -44,6 +44,18 @@ under the License.
<version>${openai.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+
<artifactId>flink-agents-integrations-chat-models-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- Versions managed by the victools BOM imported in the root pom. -->
+ <dependency>
+ <groupId>com.github.victools</groupId>
+ <artifactId>jsonschema-generator</artifactId>
+ </dependency>
+
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
diff --git
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java
index 2db309aa..9c69cc5e 100644
---
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java
+++
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnection.java
@@ -25,12 +25,10 @@ import com.openai.azure.AzureUrlPathMode;
import com.openai.azure.credential.AzureApiKeyCredential;
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.ResponseFormatJsonSchema;
import com.openai.models.chat.completions.ChatCompletion;
import com.openai.models.chat.completions.ChatCompletionCreateParams;
import com.openai.models.chat.completions.ChatCompletionFunctionTool;
@@ -420,7 +418,7 @@ public class AzureOpenAIChatModelConnection extends
BaseChatModelConnection {
String nativeSchemaName = null;
if (canApplyNativeStructuredOutput(outputSchema, tools,
rawModelParams)) {
Class<?> schemaClass = (Class<?>) outputSchema;
- builder.responseFormat(toNativeResponseFormat(schemaClass));
+
builder.responseFormat(OpenAIChatCompletionsUtils.toNativeResponseFormat(schemaClass));
nativeSchemaName = schemaClass.getSimpleName();
}
@@ -471,26 +469,6 @@ public class AzureOpenAIChatModelConnection 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();
- }
-
@Override
public void close() throws Exception {
this.client.close();
diff --git
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtils.java
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtils.java
index 04712f30..c2a96e36 100644
---
a/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtils.java
+++
b/integrations/chat-models/openai/src/main/java/org/apache/flink/agents/integrations/chatmodels/openai/OpenAIChatCompletionsUtils.java
@@ -20,8 +20,10 @@ package
org.apache.flink.agents.integrations.chatmodels.openai;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.github.victools.jsonschema.generator.Option;
import com.openai.core.JsonValue;
import com.openai.core.Timeout;
+import com.openai.models.ResponseFormatJsonSchema;
import com.openai.models.chat.completions.ChatCompletionAssistantMessageParam;
import com.openai.models.chat.completions.ChatCompletionContentPart;
import com.openai.models.chat.completions.ChatCompletionContentPartImage;
@@ -48,6 +50,7 @@ import
org.apache.flink.agents.api.chat.messages.ToolResultBlock;
import
org.apache.flink.agents.api.chat.messages.UnsupportedContentBlockException;
import org.apache.flink.agents.api.chat.messages.UrlSource;
import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import
org.apache.flink.agents.integrations.chatmodels.common.PojoJsonSchemaGenerator;
import java.math.BigDecimal;
import java.math.BigInteger;
@@ -64,8 +67,8 @@ import java.util.stream.Collectors;
/**
* Static helpers for converting between Flink Agents {@link ChatMessage} and
OpenAI Chat
* Completions API message types, plus shared parsing/validation of common
connection arguments
- * ({@code timeout}, {@code max_retries}). No tool-definition conversion —
that stays
- * per-connection.
+ * ({@code timeout}, {@code max_retries}) and the native response format for
an output schema. No
+ * tool-definition conversion — that stays per-connection.
*
* <p>Used by both {@code OpenAICompletionsConnection} (OpenAI /
OpenAI-compatible providers) and
* {@code AzureOpenAIChatModelConnection} (Azure OpenAI). Both rely on the
same openai-java SDK
@@ -175,6 +178,37 @@ final class OpenAIChatCompletionsUtils {
}
}
+ /**
+ * Derives the strict {@code json_schema} response format for a POJO
output schema class.
+ *
+ * <p>The schema comes from the shared generator rather than the SDK's
typed {@code
+ * responseFormat(Class)} builder, whose schema lists an annotated enum
constant by its Java
+ * name, a value the response then fails to deserialize from. Strict mode
expects every property
+ * in the required set and no additional properties, so the schema states
both.
+ */
+ static ResponseFormatJsonSchema toNativeResponseFormat(Class<?>
schemaClass) {
+ Map<String, JsonValue> schema = new LinkedHashMap<>();
+ PojoJsonSchemaGenerator.generateAllRequired(
+ schemaClass,
Option.FORBIDDEN_ADDITIONAL_PROPERTIES_BY_DEFAULT)
+ .fields()
+ .forEachRemaining(
+ member ->
+ schema.put(
+ member.getKey(),
+
JsonValue.fromJsonNode(member.getValue())));
+ return ResponseFormatJsonSchema.builder()
+ .jsonSchema(
+ ResponseFormatJsonSchema.JsonSchema.builder()
+ .name("json-schema-from-" +
schemaClass.getSimpleName())
+ .schema(
+
ResponseFormatJsonSchema.JsonSchema.Schema.builder()
+ .additionalProperties(schema)
+ .build())
+ .strict(true)
+ .build())
+ .build();
+ }
+
/** Convert a list of Flink Agents ChatMessages to OpenAI
ChatCompletionMessageParams. */
public static List<ChatCompletionMessageParam> convertToOpenAIMessages(
List<ChatMessage> messages) {
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 2724829b..bc3cf9fd 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,13 +22,11 @@ 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;
@@ -307,7 +305,8 @@ public class OpenAICompletionsConnection extends
BaseChatModelConnection {
// the branch applies whenever the query answers other than
INFEASIBLE. Model capability is
// not asked: a caller that hands this connection a schema has already
chosen to send it.
if (canApplyNativeStructuredOutput(outputSchema, tools,
rawModelParams)) {
- builder.responseFormat(toNativeResponseFormat((Class<?>)
outputSchema));
+ builder.responseFormat(
+
OpenAIChatCompletionsUtils.toNativeResponseFormat((Class<?>) outputSchema));
}
Object temperature = modelParams.remove("temperature");
@@ -348,26 +347,6 @@ 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/AzureOpenAIChatModelConnectionTest.java
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnectionTest.java
index d4db5351..a4bd0bfd 100644
---
a/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnectionTest.java
+++
b/integrations/chat-models/openai/src/test/java/org/apache/flink/agents/integrations/chatmodels/openai/AzureOpenAIChatModelConnectionTest.java
@@ -18,12 +18,16 @@
package org.apache.flink.agents.integrations.chatmodels.openai;
+import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
import com.openai.core.JsonField;
import com.openai.core.JsonMissing;
import com.openai.core.JsonValue;
+import com.openai.core.ObjectMappers;
import com.openai.errors.BadRequestException;
import com.openai.models.ChatModel;
import com.openai.models.ResponseFormatJsonSchema;
@@ -54,6 +58,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.io.IOException;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -84,6 +89,8 @@ class AzureOpenAIChatModelConnectionTest {
private static final Map<String, Object> CALLER_RESPONSE_FORMAT =
Map.of("type", "json_object");
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
private static ResourceDescriptor.Builder connectionDescriptor() {
return ResourceDescriptor.Builder.newBuilder(
AzureOpenAIChatModelConnection.class.getName());
@@ -95,7 +102,7 @@ class AzureOpenAIChatModelConnectionTest {
public int age;
}
- /** A polymorphic member, which the SDK renders as a discriminated union.
*/
+ /** A polymorphic member, which the derived schema states as a
discriminated union. */
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "kind")
@JsonSubTypes({
@JsonSubTypes.Type(value = Dog.class, name = "dog"),
@@ -119,6 +126,30 @@ class AzureOpenAIChatModelConnectionTest {
public Pet pet;
}
+ /** Output schema fixture whose enum constants are deserialized from
values other than names. */
+ public static class Ticket {
+ public Status status;
+ }
+
+ /** Enum constants deserialized from their {@code @JsonProperty} values. */
+ public enum Status {
+ @JsonProperty("in-progress")
+ IN_PROGRESS,
+ @JsonProperty("done")
+ DONE
+ }
+
+ /** The JSON Schema the request carries, as a JSON tree. */
+ private static JsonNode nativeSchema(ChatCompletionCreateParams params) {
+ return ObjectMappers.jsonMapper()
+ .valueToTree(
+ params.responseFormat()
+ .orElseThrow()
+ .asJsonSchema()
+ .jsonSchema()
+ ._schema());
+ }
+
/** The JSON Schema the request carries, as the SDK holds it on the
response format. */
private static String nativeSchemaPayload(ChatCompletionCreateParams
params) {
return params.responseFormat()
@@ -311,8 +342,8 @@ class AzureOpenAIChatModelConnectionTest {
assertThat(request.responseFormat()).isPresent();
ResponseFormatJsonSchema jsonSchema =
request.responseFormat().get().asJsonSchema();
- // The SDK derives the wire name from the class, so it identifies the
schema without being
- // equal to the class name.
+ // The wire name is derived from the class, so it identifies the
schema without being equal
+ // to the class name.
assertThat(jsonSchema.jsonSchema().name()).contains("Person");
assertThat(jsonSchema.jsonSchema().strict()).contains(true);
// Asserting the members rather than only the flags: a schema
declaring no properties
@@ -323,10 +354,11 @@ class AzureOpenAIChatModelConnectionTest {
}
@Test
- @DisplayName("A polymorphic member is sent as the discriminated union the
SDK derives")
+ @DisplayName("A polymorphic member is sent as a discriminated union")
void testPolymorphicMemberSchemaIsSent() {
- // Jackson renders this member as an object declaring no properties,
while the SDK derives
- // the full union the provider accepts. Reading a Jackson-rendered
schema here would refuse
+ // Jackson renders this member as an object declaring no properties,
while the schema
+ // generator derives the full union the provider accepts. Reading a
Jackson-rendered schema
+ // here would refuse
// a request that works.
ChatCompletionCreateParams request =
connection()
@@ -338,6 +370,30 @@ class AzureOpenAIChatModelConnectionTest {
.contains("kind={const=dog}");
}
+ @Test
+ @DisplayName("The derived schema lists enum constants the way Jackson
deserializes them")
+ void derivedSchemaFollowsJacksonEnumValues() throws Exception {
+ JsonNode status =
+ nativeSchema(
+ connection()
+ .buildRequest(
+ userMessage(),
+ List.of(),
+ params("gpt-4o-mini"),
+ Ticket.class))
+ .path("properties")
+ .path("status");
+
+ // Strict mode confines the model to the listed values, so each has to
deserialize into
+ // the enum. Listed by constant name instead, a plain mapper refuses
every value the schema
+ // allows.
+ List<Status> statuses = new ArrayList<>();
+ for (JsonNode value : status.path("enum")) {
+ statuses.add(MAPPER.treeToValue(value, Status.class));
+ }
+ assertThat(statuses).containsExactlyInAnyOrder(Status.values());
+ }
+
@Test
@DisplayName("A native request still targets the deployment, not the
backing model")
void testCapableNativeRequestStillTargetsTheDeployment() {
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
index 58a75604..05cf7889 100644
---
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
@@ -18,8 +18,13 @@
package org.apache.flink.agents.integrations.chatmodels.openai;
+import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.annotation.JsonValue;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.openai.core.ObjectMappers;
import com.openai.errors.BadRequestException;
import com.openai.models.ChatModel;
import com.openai.models.ResponseFormatJsonSchema;
@@ -41,10 +46,12 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
@@ -60,13 +67,15 @@ class OpenAICompletionsConnectionTest {
private static final ResourceContext NOOP =
ResourceContext.fromGetResource((a, b) -> null);
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
/** A representative POJO output schema. */
public static class Person {
public String name;
public int age;
}
- /** A polymorphic member, which the SDK renders as a discriminated union.
*/
+ /** A polymorphic member, which the derived schema states as a
discriminated union. */
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "kind")
@JsonSubTypes({
@JsonSubTypes.Type(value = Dog.class, name = "dog"),
@@ -90,6 +99,65 @@ class OpenAICompletionsConnectionTest {
public Pet pet;
}
+ /**
+ * Output schema fixture whose enum constants are deserialized from values
other than their
+ * names, one through {@code @JsonProperty} on the constants and one
through a
+ * {@code @JsonValue} method.
+ */
+ public static class Ticket {
+ public Status status;
+ public Phase phase;
+ }
+
+ /** Enum constants deserialized from their {@code @JsonProperty} values. */
+ public enum Status {
+ @JsonProperty("in-progress")
+ IN_PROGRESS,
+ @JsonProperty("done")
+ DONE
+ }
+
+ /** Enum constants deserialized from their {@code @JsonValue} method. */
+ public enum Phase {
+ STARTED("started"),
+ FINISHED("finished");
+
+ private final String wire;
+
+ Phase(String wire) {
+ this.wire = wire;
+ }
+
+ @JsonValue
+ public String wire() {
+ return wire;
+ }
+ }
+
+ /** Output schema fixture with an {@code Optional} field and a nested
object. */
+ public static class Draft {
+ public String title;
+ public Optional<String> note;
+ public Person author;
+ }
+
+ /** The JSON Schema the request carries, as a JSON tree. */
+ private static JsonNode nativeSchema(ChatCompletionCreateParams params) {
+ return ObjectMappers.jsonMapper()
+ .valueToTree(
+ params.responseFormat()
+ .orElseThrow()
+ .asJsonSchema()
+ .jsonSchema()
+ ._schema());
+ }
+
+ private static List<String> textValues(JsonNode array) {
+ List<String> values = new ArrayList<>();
+ array.forEach(value -> values.add(value.asText()));
+ return values;
+ }
+
/** The JSON Schema the request carries, as the SDK holds it on the
response format. */
private static String nativeSchemaPayload(ChatCompletionCreateParams
params) {
return params.responseFormat()
@@ -302,10 +370,11 @@ class OpenAICompletionsConnectionTest {
}
@Test
- @DisplayName("A polymorphic member is sent as the discriminated union the
SDK derives")
+ @DisplayName("A polymorphic member is sent as a discriminated union")
void testPolymorphicMemberSchemaIsSent() {
- // Jackson renders this member as an object declaring no properties,
while the SDK derives
- // the full union the provider accepts. Reading a Jackson-rendered
schema here would refuse
+ // Jackson renders this member as an object declaring no properties,
while the schema
+ // generator derives the full union the provider accepts. Reading a
Jackson-rendered schema
+ // here would refuse
// a request that works.
ChatCompletionCreateParams request =
connection().buildRequest(userMessage(), List.of(),
params("gpt-4o"), Owner.class);
@@ -316,6 +385,56 @@ class OpenAICompletionsConnectionTest {
.contains("kind={const=dog}");
}
+ @Test
+ @DisplayName("The derived schema lists enum constants the way Jackson
deserializes them")
+ void derivedSchemaFollowsJacksonEnumValues() throws Exception {
+ JsonNode properties =
+ nativeSchema(
+ connection()
+ .buildRequest(
+ userMessage(),
+ List.of(),
+ params("gpt-4o"),
+ Ticket.class))
+ .path("properties");
+
+ // Strict mode confines the model to the listed values, so each has to
deserialize into
+ // the enum. Listed by constant name instead, a plain mapper refuses
every value the schema
+ // allows.
+ List<Status> statuses = new ArrayList<>();
+ for (JsonNode value : properties.path("status").path("enum")) {
+ statuses.add(MAPPER.treeToValue(value, Status.class));
+ }
+ assertThat(statuses).containsExactlyInAnyOrder(Status.values());
+
+ List<Phase> phases = new ArrayList<>();
+ for (JsonNode value : properties.path("phase").path("enum")) {
+ phases.add(MAPPER.treeToValue(value, Phase.class));
+ }
+ assertThat(phases).containsExactlyInAnyOrder(Phase.values());
+ }
+
+ @Test
+ @DisplayName("The derived schema requires every field and forbids extras
at every level")
+ void derivedSchemaRequiresEveryFieldAndForbidsExtras() {
+ JsonNode schema =
+ nativeSchema(
+ connection()
+ .buildRequest(
+ userMessage(), List.of(),
params("gpt-4o"), Draft.class));
+
+ // Strict mode rejects a schema that leaves a property out of the
required set or admits
+ // additional properties, so the Optional field is required too and
admits null instead.
+ assertThat(textValues(schema.path("required")))
+ .containsExactlyInAnyOrder("title", "note", "author");
+
assertThat(textValues(schema.path("properties").path("note").path("type")))
+ .contains("null");
+
assertThat(schema.path("additionalProperties").asBoolean(true)).isFalse();
+ JsonNode author = schema.path("properties").path("author");
+
assertThat(textValues(author.path("required"))).containsExactlyInAnyOrder("name",
"age");
+
assertThat(author.path("additionalProperties").asBoolean(true)).isFalse();
+ }
+
@Test
@DisplayName("Native NOT applied when no output schema is supplied")
void testNativeNotAppliedWhenSchemaNull() {
