From 4c7449d3c0469d9eb70add56e80b84795e10321d Mon Sep 17 00:00:00 2001 From: Weiqing Yang Date: Tue, 13 Oct 2026 11:18:54 -0700 Subject: [PATCH 1/2] [integrations][java] Let the shared POJO schema generator require every field Adds PojoJsonSchemaGenerator.generateAllRequired, which derives the same schema as generate except that every field, including an Optional one, is listed as required. It is the required set OpenAI's strict mode expects, and it lets the OpenAI, Azure OpenAI and Anthropic connections move off the SDK's schema builder without changing their required set. Generated-by: GitHub Copilot CLI 1.0.95 (claude-opus-5.5) --- .../common/PojoJsonSchemaGenerator.java | 35 ++++++++++++++++--- .../common/PojoJsonSchemaGeneratorTest.java | 32 +++++++++++++++++ 2 files changed, 62 insertions(+), 5 deletions(-) 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 e36322e9a7..08a973ffbd 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.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 @@ private 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 @@ private 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. + * + *

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 requiredCheck, Option... options) { SchemaGeneratorConfigBuilder builder = new SchemaGeneratorConfigBuilder( SchemaVersion.DRAFT_2020_12, OptionPreset.PLAIN_JSON); @@ -76,8 +102,7 @@ public static ObjectNode generate(Class type, Option... options) { 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 838336a725..1d567f6e17 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.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 @@ void requiresEveryFieldExceptOptional() { .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() { From 6a5be318119bf1ec06795748e2cb9204dc966268 Mon Sep 17 00:00:00 2001 From: Weiqing Yang Date: Tue, 13 Oct 2026 11:25:39 -0700 Subject: [PATCH 2/2] [integrations][openai][anthropic] Derive native schemas with the shared generator The OpenAI Chat Completions, Azure OpenAI and Anthropic connections derived their native structured-output schema through the SDK's typed builders. Those schemas list an enum constant by its Java name even when @JsonProperty or @JsonValue names it differently, so strict mode confines the model to values the response then fails to deserialize from (#1246). Derive the schema with PojoJsonSchemaGenerator.generateAllRequired instead, which lists enum constants the way Jackson deserializes them and keeps the strict-mode shape the SDK schemas had: every property required (an Optional field admits null) and no additional properties. The two OpenAI copies of the helper become one in OpenAIChatCompletionsUtils; vLLM inherits the fix through OpenAICompletionsConnection. The schema otherwise matches the SDK's, except that properties now follow declaration order instead of alphabetical order, and Swagger 2 @Schema annotations are no longer read. Generated-by: GitHub Copilot CLI 1.0.95 (claude-opus-5.5) --- integrations/chat-models/anthropic/pom.xml | 12 ++ .../AnthropicChatModelConnection.java | 56 ++++---- .../AnthropicChatModelConnectionTest.java | 114 +++++++++++++++- integrations/chat-models/openai/pom.xml | 12 ++ .../AzureOpenAIChatModelConnection.java | 24 +--- .../openai/OpenAIChatCompletionsUtils.java | 38 +++++- .../openai/OpenAICompletionsConnection.java | 25 +--- .../AzureOpenAIChatModelConnectionTest.java | 68 +++++++++- .../OpenAICompletionsConnectionTest.java | 127 +++++++++++++++++- 9 files changed, 386 insertions(+), 90 deletions(-) diff --git a/integrations/chat-models/anthropic/pom.xml b/integrations/chat-models/anthropic/pom.xml index 07d0b592bc..aad135687b 100644 --- a/integrations/chat-models/anthropic/pom.xml +++ b/integrations/chat-models/anthropic/pom.xml @@ -44,6 +44,18 @@ under the License. ${anthropic.version} + + org.apache.flink + flink-agents-integrations-chat-models-common + ${project.version} + + + + + com.github.victools + jsonschema-generator + + org.slf4j slf4j-api 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 d4c3ecee51..7331e82a12 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 @@ 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.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.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 @@ static Object effectiveTemperature(Object topLevel, Map addition } /** - * 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. * - *

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. - * - *

Local schema validation is off so that the provider, not the client, is the authority on - * which schemas it accepts. + *

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 OutputConfig toNativeOutputConfig(Class 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 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 77058ac9de..54d9077dd1 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.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.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 @@ private static Stream incapableModels() { ""); } - /** 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 @@ public static class Owner { 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 note; + public Answer answer; + } + private static Map paramsWithModel(String model, Object jsonPrefill) { Map params = params(jsonPrefill); params.put("model", model); @@ -367,6 +414,23 @@ private static Set nativeSchemaProperties( .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 textValues(JsonNode array) { + List 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 @@ void testNamedModelIgnoresTheConfiguredModel() { } @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 @@ void testPolymorphicMemberSchemaIsSent() { .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 statuses = new ArrayList<>(); + for (JsonNode value : properties.path("status").path("enum")) { + statuses.add(MAPPER.treeToValue(value, Status.class)); + } + assertThat(statuses).containsExactlyInAnyOrder(Status.values()); + + List 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/openai/pom.xml b/integrations/chat-models/openai/pom.xml index 361ed30181..3f4badcd3c 100644 --- a/integrations/chat-models/openai/pom.xml +++ b/integrations/chat-models/openai/pom.xml @@ -44,6 +44,18 @@ under the License. ${openai.version} + + org.apache.flink + flink-agents-integrations-chat-models-common + ${project.version} + + + + + com.github.victools + jsonschema-generator + + org.slf4j slf4j-api 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 2db309aae3..9c69cc5eca 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.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 @@ ChatCompletionCreateParams buildRequest( 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 @@ ChatCompletionCreateParams buildRequest( 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 ResponseFormatJsonSchema toNativeResponseFormat(Class 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 04712f30ac..c2a96e3672 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 @@ 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.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 @@ /** * 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. * *

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 @@ private static BigDecimal toBigDecimal(Number raw, String argumentName) { } } + /** + * Derives the strict {@code json_schema} response format for a POJO output schema class. + * + *

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 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 convertToOpenAIMessages( List 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 2724829b7c..bc3cf9fd5e 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.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 @@ ChatCompletionCreateParams buildRequest( // 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 @@ ChatCompletionCreateParams buildRequest( 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 ResponseFormatJsonSchema toNativeResponseFormat(Class 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 convertTools(List tools, boolean strictMode) { List 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 d4db5351db..a4bd0bfd5e 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 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 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 @@ public static class Person { 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 @@ public static class Owner { 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 @@ void testNativeAppliedForCapableDeploymentModel() { 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 @@ void testNativeAppliedForCapableDeploymentModel() { } @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 @@ void testPolymorphicMemberSchemaIsSent() { .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 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 58a756049e..05cf78890c 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 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 @@ public static class Owner { 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 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 textValues(JsonNode array) { + List 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 @@ void testNativeAppliedForPojoCapableModel() { } @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 @@ void testPolymorphicMemberSchemaIsSent() { .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 statuses = new ArrayList<>(); + for (JsonNode value : properties.path("status").path("enum")) { + statuses.add(MAPPER.treeToValue(value, Status.class)); + } + assertThat(statuses).containsExactlyInAnyOrder(Status.values()); + + List 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() {