[integrations][anthropic] Stop sending rejected sampling parameters

docs/content/docs/development/chat_models.md

@@ -291,9 +291,8 @@ Anthropic provides cloud-based chat models featuring the Claude family, known fo
 | `prompt` | Prompt \| str | None | Prompt template or reference to prompt resource |
 | `tools` | List[str] | None | List of tool names available to the model |
 | `max_tokens` | int | `1024` | Maximum number of tokens to generate |
-| `temperature` | float | `0.1` | Sampling temperature (0.0 to 1.0) |
+| `temperature` | float | `0.1` | Sampling temperature (0.0 to 1.0) (not sent on Claude 4.7 and later, which reject a non-default sampling parameter) |
 | `json_prefill` | bool | `False` | Prefill assistant response with "{" to enforce JSON output (applies only on models that accept assistant-message prefilling; disabled when tools are used, or when the request carries an `output_config`) |
-| `additional_kwargs` | dict | `{}` | Additional Anthropic API parameters |
 
 {{< /tab >}}
 
@@ -306,10 +305,10 @@ Anthropic provides cloud-based chat models featuring the Claude family, known fo
 | `prompt` | Prompt \| String | None | Prompt template or reference to prompt resource |
 | `tools` | List<String> | None | List of tool names available to the model |
 | `max_tokens` | long | `1024` | Maximum number of tokens to generate |
-| `temperature` | double | `0.1` | Sampling temperature (0.0 to 1.0) |
+| `temperature` | double | `0.1` | Sampling temperature (0.0 to 1.0) (not sent on Claude 4.7 and later, which reject a non-default sampling parameter) |
 | `json_prefill` | boolean | `false` | Prefill assistant response with "{" to enforce JSON output (applies only on models that accept assistant-message prefilling; disabled when tools are used, or when the request carries an `output_config`) |
 | `strict_tools` | boolean | `false` | Enable strict mode for tool calling schemas |
-| `additional_kwargs` | Map<String, Object> | `{}` | Additional Anthropic API parameters (top_k, top_p, stop_sequences); an `output_config` supplied here takes precedence over one derived from an output schema |
+| `additional_kwargs` | Map<String, Object> | `{}` | Additional Anthropic API parameters (`temperature`, `top_k`, `top_p`, `stop_sequences`); none of the three sampling parameters is sent on Claude 4.7 and later. A `temperature` supplied here takes precedence over the top-level one, and an `output_config` supplied here takes precedence over one derived from an output schema |
 
 {{< /tab >}}
 

integrations/chat-models/anthropic/pom.xml

@@ -43,6 +43,12 @@ under the License.
             <artifactId>anthropic-java</artifactId>
             <version>${anthropic.version}</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>${slf4j.version}</version>
+        </dependency>
     </dependencies>
 
 </project>

integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelConnection.java

@@ -42,6 +42,8 @@
 import org.apache.flink.agents.api.resource.ResourceContext;
 import org.apache.flink.agents.api.resource.ResourceDescriptor;
 import org.apache.flink.agents.api.tools.ToolMetadata;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -51,6 +53,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 
 /**
@@ -81,6 +84,8 @@
  */
 public class AnthropicChatModelConnection extends BaseChatModelConnection {
 
+    private static final Logger LOG = LoggerFactory.getLogger(AnthropicChatModelConnection.class);
+
     private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() {};
 
     private final ObjectMapper mapper = new ObjectMapper();
@@ -223,6 +228,128 @@ static boolean supportsJsonPrefill(String effectiveModel) {
         return !PREFILL_UNSUPPORTED_MODELS.contains(effectiveModel);
     }
 
+    // Models that reject a non-default sampling parameter. Source of truth:
+    // https://platform.claude.com/docs/en/about-claude/models/migration-guide
+    //
+    // The documented rule is that setting temperature, top_p or top_k to any non-default value on
+    // Claude Opus 4.7 or later returns a 400, and the Claude Sonnet 5 release notes carry the same
+    // rule for the Sonnet line while stating it is new for Sonnet-class models. Sending the
+    // provider default, or omitting the parameter, stays acceptable on every model, so the way to
+    // honour this is to drop the parameter rather than to substitute a value.
+    //
+    // This is the third boundary encoded in this class and it lines up with neither of the others.
+    // Structured output starts at the 4.5 generation and prefill rejection at 4.6, while sampling
+    // rejection starts at 4.7. Claude 4.6 therefore rejects a prefill while still accepting a
+    // temperature, so the prefill list above cannot be reused here even though the two overlap.
+    //
+    // Claude Fable 5, Claude Mythos 5 and Claude Mythos Preview are listed without a matching
+    // sentence in the migration guide. That guide records each release's deltas, and these models
+    // succeed Claude Opus 4.8, which already rejects sampling parameters, so there was no delta to
+    // record. They are treated as rejecting because they inherit the constraint, not because a
+    // release note restates it.
+    private static final Set<String> SAMPLING_UNSUPPORTED_MODELS =
+            Set.of(
+                    "claude-opus-4-7",
+                    "claude-opus-4-8",
+                    "claude-opus-5",
+                    "claude-sonnet-5",
+                    "claude-fable-5",
+                    "claude-mythos-5",
+                    "claude-mythos-preview");
+
+    // The {model name, parameter name} pairs already reported through the warning in
+    // sendSamplingParam, so a rejected sampling parameter is surfaced once per model and parameter
+    // instead of on every request. The parameter belongs in the key because a model withdraws three
+    // of them at once: keying on the name alone would report whichever was dropped first and leave
+    // every later one silent.
+    private static final Set<List<String>> SAMPLING_WARNED_PARAMS = ConcurrentHashMap.newKeySet();
+
+    /**
+     * Whether {@code effectiveModel} accepts non-default sampling parameters, meaning {@code
+     * temperature}, {@code top_p} and {@code top_k}.
+     *
+     * <p>See the list above for the source of truth and for why it is kept apart from the prefill
+     * and structured-output lists. An unrecognized name reports {@code true}, matching the
+     * documented rule: sampling parameters are the long-standing behaviour and only the listed
+     * names withdraw them. That default carries the same cost as {@link #supportsJsonPrefill}: a
+     * rejecting model this list has not caught up with is sent a sampling parameter and answered
+     * with a 400.
+     */
+    static boolean supportsSamplingParams(String effectiveModel) {
+        // Load-bearing: the list is an immutable Set, whose contains(null) throws rather than
+        // reporting absence.
+        if (effectiveModel == null) {
+            return true;
+        }
+        return !SAMPLING_UNSUPPORTED_MODELS.contains(effectiveModel);
+    }
+
+    /**
+     * Whether the sampling parameter {@code param}, carrying {@code value}, may be put on a request
+     * for {@code effectiveModel}, warning the first time it may not.
+     *
+     * <p>A rejected parameter is dropped rather than clamped: the provider accepts an omitted
+     * parameter but answers a non-default one with a 400, and substituting the provider default
+     * would quietly change sampling behaviour instead of leaving it to the provider.
+     */
+    private static boolean sendSamplingParam(String effectiveModel, String param, Object value) {
+        if (supportsSamplingParams(effectiveModel)) {
+            return true;
+        }
+        samplingWarning(effectiveModel, param, value).ifPresent(LOG::warn);
+        return false;
+    }
+
+    /**
+     * The warning owed for dropping {@code param}, which carried {@code value}, on {@code
+     * effectiveModel}: the message on the first call for that model and parameter, and empty on
+     * every call after.
+     *
+     * <p>The message is owed against the pair rather than against the model name alone. A model
+     * withdraws three sampling parameters at once, so a debt held against the name would be settled
+     * by whichever parameter happened to be dropped first, leaving every later one dropped in
+     * silence.
+     *
+     * <p>Building the message here rather than at the call site keeps it on the same side of the
+     * once-per-pair decision as the decision itself, so there is no arrangement of the caller that
+     * reports a parameter on every request.
+     *
+     * <p>{@code effectiveModel} must not be null, which holds because only a name on the list above
+     * ever reaches this method.
+     */
+    static Optional<String> samplingWarning(String effectiveModel, String param, Object value) {
+        if (!SAMPLING_WARNED_PARAMS.add(List.of(effectiveModel, param))) {
+            return Optional.empty();
+        }
+        return Optional.of(
+                String.format(
+                        "Model %s rejects non-default sampling parameters, so the configured %s %s"
+                                + " was not sent. Steer the model through its prompt instead.",
+                        effectiveModel, param, value));
+    }
+
+    /**
+     * The temperature a request would carry, given the top-level {@code temperature} and an {@code
+     * additional_kwargs} map that may hold one of its own.
+     *
+     * <p>The map wins when it holds a {@link Number} under that key: an entry naming the parameter
+     * outright is the more specific setting, so it takes the top-level parameter's place. An absent
+     * key, or one holding anything other than a {@link Number}, leaves {@code topLevel} in place,
+     * because such a value reaches the request by neither route.
+     *
+     * <p>Resolving here rather than gating each route separately is what keeps the
+     * dropped-parameter warning honest. That warning is owed once per model and parameter, so
+     * gating the top-level value first would spend the one report on a value the map was about to
+     * override, naming a setting that was never going to be sent.
+     */
+    static Object effectiveTemperature(Object topLevel, Map<String, Object> additionalKwargs) {
+        if (additionalKwargs == null) {
+            return topLevel;
+        }
+        Object override = additionalKwargs.get("temperature");
+        return override instanceof Number ? override : topLevel;
+    }
+
     /**
      * Derives the native {@code output_config} for a POJO class through the SDK's typed
      * structured-output builder.
@@ -365,16 +492,19 @@ BuiltRequest buildRequest(
             builder.maxTokens(((Number) maxTokens).longValue());
         }
 
-        Object temperature = modelParams.remove("temperature");
-        if (temperature instanceof Number) {
-            builder.temperature(((Number) temperature).doubleValue());
-        }
-
         @SuppressWarnings("unchecked")
         Map<String, Object> additionalKwargs =
                 (Map<String, Object>) modelParams.remove("additional_kwargs");
+
+        Object temperature =
+                effectiveTemperature(modelParams.remove("temperature"), additionalKwargs);
+        if (temperature instanceof Number
+                && sendSamplingParam(modelName, "temperature", temperature)) {
+            builder.temperature(((Number) temperature).doubleValue());
+        }
+
         if (additionalKwargs != null) {
-            applyAdditionalKwargs(builder, additionalKwargs);
+            applyAdditionalKwargs(builder, additionalKwargs, modelName);
         }
 
         // Read here rather than inside the native structured-output branch below because it governs
@@ -687,22 +817,29 @@ private String extractJsonFromMarkdown(String content) {
     }
 
     private void applyAdditionalKwargs(
-            MessageCreateParams.Builder builder, Map<String, Object> kwargs) {
+            MessageCreateParams.Builder builder, Map<String, Object> kwargs, String modelName) {
         for (Map.Entry<String, Object> entry : kwargs.entrySet()) {
             String key = entry.getKey();
             Object value = entry.getValue();
 
             switch (key) {
                 case "top_k":
-                    if (value instanceof Number) {
+                    if (value instanceof Number && sendSamplingParam(modelName, key, value)) {
                         builder.topK(((Number) value).longValue());
                     }
                     break;
                 case "top_p":
-                    if (value instanceof Number) {
+                    if (value instanceof Number && sendSamplingParam(modelName, key, value)) {
                         builder.topP(((Number) value).doubleValue());
                     }
                     break;
+                case "temperature":
+                    // Resolved against the top-level "temperature" by effectiveTemperature and
+                    // gated before this map is applied, so there is nothing left to do with it
+                    // here. The case still has to exist: without a label of its own the key falls
+                    // to the default branch and goes onto the body as a raw property, ungated,
+                    // which the models that withdraw sampling parameters answer with a 400.
+                    break;
                 case "stop_sequences":
                     if (value instanceof List) {
                         @SuppressWarnings("unchecked")

integrations/chat-models/anthropic/src/main/java/org/apache/flink/agents/integrations/chatmodels/anthropic/AnthropicChatModelSetup.java

@@ -48,8 +48,9 @@
  *   <li><b>strict_tools</b> (optional): When true, tool calls adhere strictly to the JSON schema.
  *       (default: false)
  *   <li><b>tools</b> (optional): List of tool names available for the model to use
- *   <li><b>additional_kwargs</b> (optional): Additional parameters (top_k, top_p, stop_sequences).
- *       An output_config supplied here takes precedence over one derived from an output schema.
+ *   <li><b>additional_kwargs</b> (optional): Additional parameters (temperature, top_k, top_p,
+ *       stop_sequences). A temperature supplied here takes precedence over the top-level one, and
+ *       an output_config supplied here takes precedence over one derived from an output schema.
  * </ul>
  *
  * <p>Example usage: