Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions integrations/chat-models/anthropic/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,16 @@

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;
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;
Expand All @@ -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;
Expand Down Expand Up @@ -402,34 +406,32 @@ static Object effectiveTemperature(Object topLevel, Map<String, Object> 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.
*
* <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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -310,12 +315,12 @@ private static Stream<String> 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"),
Expand All @@ -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<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);
Expand Down Expand Up @@ -367,6 +414,23 @@ private static Set<String> 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<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
Expand Down Expand Up @@ -495,17 +559,57 @@ 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}")
.contains("meow={type=string}")
.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() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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 {
Expand All @@ -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
Expand All @@ -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.
*
* <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);
Expand All @@ -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);
}
}
Loading
Loading