This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 732191e16f76 CAMEL-24224: camel-aws-bedrock - capture streaming
metadata for Claude/Nova and accept inference profile ids (#24982)
732191e16f76 is described below
commit 732191e16f76356c2e0c9c6be39ec66da0c6f1e2
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Jul 21 18:03:28 2026 +0200
CAMEL-24224: camel-aws-bedrock - capture streaming metadata for Claude/Nova
and accept inference profile ids (#24982)
Inspect every streaming chunk for metadata instead of gating on the final
chunk; resolve inference-profile ids/ARNs and fall back to family-prefix
dispatch.
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../aws2/bedrock/runtime/BedrockProducer.java | 71 +++++++++++++++++++-
.../runtime/stream/BedrockStreamHandler.java | 47 +++++++-------
.../runtime/BedrockProducerModelIdTest.java | 55 ++++++++++++++++
.../stream/BedrockStreamHandlerMetadataTest.java | 75 ++++++++++++++++++++++
4 files changed, 223 insertions(+), 25 deletions(-)
diff --git
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducer.java
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducer.java
index 75e0cacb9c41..05ffe8fe0f6f 100644
---
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducer.java
+++
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducer.java
@@ -63,6 +63,11 @@ import
software.amazon.awssdk.services.bedrockruntime.model.ToolConfiguration;
public class BedrockProducer extends DefaultProducer {
private static final Logger LOG =
LoggerFactory.getLogger(BedrockProducer.class);
+
+ /**
+ * Region prefixes used by Bedrock cross-region inference profile ids,
e.g. {@code us.anthropic.claude-...}.
+ */
+ private static final List<String> INFERENCE_PROFILE_REGION_PREFIXES =
List.of("us.", "eu.", "apac.", "us-gov.");
private transient String bedrockProducerToString;
public BedrockProducer(Endpoint endpoint) {
@@ -282,7 +287,7 @@ public class BedrockProducer extends DefaultProducer {
}
protected void setResponseText(InvokeModelResponse result, Message
message) {
- String modelId = getConfiguration().getModelId();
+ String modelId =
resolveFoundationModelId(getConfiguration().getModelId());
switch (modelId) {
// Amazon Titan Models
case "amazon.titan-text-express-v1":
@@ -425,7 +430,71 @@ public class BedrockProducer extends DefaultProducer {
+ " is a video/speech
generation model and cannot be used with the invokeTextModel operation.");
default:
+ setResponseTextByModelFamily(modelId, result, message);
+ }
+ }
+
+ /**
+ * Resolves the underlying foundation-model id for a model reference.
Bedrock also accepts cross-region inference
+ * profile ids (for example {@code
us.anthropic.claude-3-5-sonnet-20241022-v2:0}) and inference-profile or
+ * provisioned-throughput ARNs in place of the plain model id. They all
address the same foundation model, so they
+ * must be handled the same way.
+ */
+ protected static String resolveFoundationModelId(String modelId) {
+ if (modelId == null) {
+ return null;
+ }
+ String resolved = modelId;
+ // an ARN carries the profile or model id in its last path segment
+ if (resolved.startsWith("arn:")) {
+ int slash = resolved.lastIndexOf('/');
+ if (slash >= 0 && slash < resolved.length() - 1) {
+ resolved = resolved.substring(slash + 1);
+ }
+ }
+ for (String prefix : INFERENCE_PROFILE_REGION_PREFIXES) {
+ if (resolved.startsWith(prefix)) {
+ return resolved.substring(prefix.length());
+ }
+ }
+ return resolved;
+ }
+
+ /**
+ * Fallback for model ids that are not listed explicitly, typically a
model released after this component was built.
+ * Dispatching on the vendor/family prefix lets a new model of a known
family work instead of failing, since the
+ * response format is defined by the family rather than the individual
model.
+ */
+ private void setResponseTextByModelFamily(String modelId,
InvokeModelResponse result, Message message) {
+ if (modelId == null) {
+ throw new IllegalStateException("Model id must be specified");
+ }
+ // families that do not produce a text completion must keep failing
rather than being parsed as text
+ if (modelId.startsWith("stability.") ||
modelId.contains("image-generator") || modelId.contains("-canvas")
+ || modelId.contains("embed") || modelId.contains("rerank") ||
modelId.contains("-reel")
+ || modelId.contains("-sonic")) {
+ throw new IllegalArgumentException(
+ "Model " + modelId
+ + " is not a text generation
model and cannot be used with the invokeTextModel operation.");
+ }
+ try {
+ if (modelId.startsWith("amazon.titan-text")) {
+ setTitanText(result, message);
+ } else if (modelId.startsWith("anthropic.claude") ||
modelId.startsWith("amazon.nova")) {
+ setAnthropicV3Text(result, message);
+ } else if (modelId.startsWith("meta.llama")) {
+ setLlamaText(result, message);
+ } else if (modelId.startsWith("mistral.")) {
+ setMistralText(result, message);
+ } else if (modelId.startsWith("cohere.command")) {
+ setCohereText(result, message);
+ } else if (modelId.startsWith("ai21.")) {
+ setAi21Text(result, message);
+ } else {
throw new IllegalStateException("Unexpected model: " +
modelId);
+ }
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
}
}
diff --git
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/BedrockStreamHandler.java
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/BedrockStreamHandler.java
index cf0feaafdbd5..5b6bdc14ff39 100644
---
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/BedrockStreamHandler.java
+++
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/BedrockStreamHandler.java
@@ -36,6 +36,27 @@ public final class BedrockStreamHandler {
// Utility class
}
+ /**
+ * Records the completion reason and token count carried by a streaming
chunk, if any.
+ * <p>
+ * Metadata is not always carried by the final chunk: Claude and Nova
report {@code stop_reason} and {@code usage}
+ * on the {@code message_delta} chunk, which precedes the final {@code
message_stop}, while Titan, Llama, Mistral
+ * and Cohere report them on their final chunk. Every chunk is therefore
inspected and whatever is present is kept;
+ * the parsers return {@code null} for chunks that do not carry the values.
+ */
+ static void applyChunkMetadata(StreamResponseParser parser, String
chunkJson, StreamMetadata metadata)
+ throws JsonProcessingException {
+ String completionReason = parser.extractCompletionReason(chunkJson);
+ if (ObjectHelper.isNotEmpty(completionReason)) {
+ metadata.setCompletionReason(completionReason);
+ }
+
+ Integer tokenCount = parser.extractTokenCount(chunkJson);
+ if (ObjectHelper.isNotEmpty(tokenCount)) {
+ metadata.setTokenCount(tokenCount);
+ }
+ }
+
/**
* Create a response handler for complete mode (accumulates all chunks)
*
@@ -62,18 +83,7 @@ public final class BedrockStreamHandler {
fullText.append(text);
}
- // Extract metadata from final chunk
- if (parser.isFinalChunk(chunkJson)) {
- String completionReason =
parser.extractCompletionReason(chunkJson);
- if
(ObjectHelper.isNotEmpty(completionReason)) {
-
metadata.setCompletionReason(completionReason);
- }
-
- Integer tokenCount =
parser.extractTokenCount(chunkJson);
- if (ObjectHelper.isNotEmpty(tokenCount)) {
- metadata.setTokenCount(tokenCount);
- }
- }
+ applyChunkMetadata(parser, chunkJson,
metadata);
chunkCount[0]++;
} catch (JsonProcessingException e) {
LOG.warn("Error parsing streaming chunk: {}",
e.getMessage(), e);
@@ -119,18 +129,7 @@ public final class BedrockStreamHandler {
}
}
- // Extract metadata from final chunk
- if (parser.isFinalChunk(chunkJson)) {
- String completionReason =
parser.extractCompletionReason(chunkJson);
- if
(ObjectHelper.isNotEmpty(completionReason)) {
-
metadata.setCompletionReason(completionReason);
- }
-
- Integer tokenCount =
parser.extractTokenCount(chunkJson);
- if (ObjectHelper.isNotEmpty(tokenCount)) {
- metadata.setTokenCount(tokenCount);
- }
- }
+ applyChunkMetadata(parser, chunkJson,
metadata);
chunkCount[0]++;
} catch (JsonProcessingException e) {
LOG.warn("Error parsing streaming chunk: {}",
e.getMessage(), e);
diff --git
a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducerModelIdTest.java
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducerModelIdTest.java
new file mode 100644
index 000000000000..5d5eb2855bca
--- /dev/null
+++
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducerModelIdTest.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.bedrock.runtime;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Bedrock accepts cross-region inference profile ids and inference-profile
ARNs in place of a plain foundation-model
+ * id. They must resolve to the same foundation model so the response is
parsed with the right model family.
+ */
+class BedrockProducerModelIdTest {
+
+ private static final String CLAUDE =
"anthropic.claude-3-5-sonnet-20241022-v2:0";
+
+ @Test
+ void plainModelIdIsUnchanged() {
+ assertEquals(CLAUDE, BedrockProducer.resolveFoundationModelId(CLAUDE));
+ }
+
+ @Test
+ void crossRegionInferenceProfileIdsResolveToTheFoundationModel() {
+ assertEquals(CLAUDE, BedrockProducer.resolveFoundationModelId("us." +
CLAUDE));
+ assertEquals(CLAUDE, BedrockProducer.resolveFoundationModelId("eu." +
CLAUDE));
+ assertEquals(CLAUDE, BedrockProducer.resolveFoundationModelId("apac."
+ CLAUDE));
+ assertEquals(CLAUDE,
BedrockProducer.resolveFoundationModelId("us-gov." + CLAUDE));
+ }
+
+ @Test
+ void inferenceProfileArnResolvesToTheFoundationModel() {
+ String arn =
"arn:aws:bedrock:eu-west-1:123456789012:inference-profile/eu." + CLAUDE;
+ assertEquals(CLAUDE, BedrockProducer.resolveFoundationModelId(arn));
+ }
+
+ @Test
+ void nullModelIdIsTolerated() {
+ assertNull(BedrockProducer.resolveFoundationModelId(null));
+ }
+}
diff --git
a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/BedrockStreamHandlerMetadataTest.java
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/BedrockStreamHandlerMetadataTest.java
new file mode 100644
index 000000000000..2afd7672b004
--- /dev/null
+++
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/BedrockStreamHandlerMetadataTest.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.bedrock.runtime.stream;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * Claude and Nova report the completion reason and token usage on the {@code
message_delta} chunk, which is not the
+ * final chunk ({@code message_stop}). Metadata extraction must therefore not
be gated on the final chunk.
+ */
+class BedrockStreamHandlerMetadataTest {
+
+ private static final String MESSAGE_DELTA
+ =
"{\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":150}}";
+ private static final String MESSAGE_STOP = "{\"type\":\"message_stop\"}";
+ private static final String CONTENT_DELTA
+ =
"{\"type\":\"content_block_delta\",\"delta\":{\"type\":\"text_delta\",\"text\":\"Hello\"}}";
+
+ @Test
+ void claudeMetadataIsCapturedFromTheNonFinalMessageDeltaChunk() throws
Exception {
+ ClaudeStreamParser parser = new ClaudeStreamParser();
+ BedrockStreamHandler.StreamMetadata metadata = new
BedrockStreamHandler.StreamMetadata();
+
+ // the chunk carrying the metadata is explicitly not the final chunk
+ assertFalse(parser.isFinalChunk(MESSAGE_DELTA));
+
+ BedrockStreamHandler.applyChunkMetadata(parser, MESSAGE_DELTA,
metadata);
+
+ assertEquals("end_turn", metadata.getCompletionReason());
+ assertEquals(150, metadata.getTokenCount());
+ }
+
+ @Test
+ void chunksWithoutMetadataLeaveEarlierValuesIntact() throws Exception {
+ ClaudeStreamParser parser = new ClaudeStreamParser();
+ BedrockStreamHandler.StreamMetadata metadata = new
BedrockStreamHandler.StreamMetadata();
+
+ // a realistic ordering: text chunk, then the metadata-bearing delta,
then the final stop chunk
+ BedrockStreamHandler.applyChunkMetadata(parser, CONTENT_DELTA,
metadata);
+ BedrockStreamHandler.applyChunkMetadata(parser, MESSAGE_DELTA,
metadata);
+ BedrockStreamHandler.applyChunkMetadata(parser, MESSAGE_STOP,
metadata);
+
+ assertEquals("end_turn", metadata.getCompletionReason());
+ assertEquals(150, metadata.getTokenCount());
+ }
+
+ @Test
+ void titanMetadataOnTheFinalChunkStillWorks() throws Exception {
+ TitanStreamParser parser = new TitanStreamParser();
+ BedrockStreamHandler.StreamMetadata metadata = new
BedrockStreamHandler.StreamMetadata();
+ String finalChunk =
"{\"outputText\":\"done\",\"completionReason\":\"FINISH\",\"totalOutputTextTokenCount\":42}";
+
+ BedrockStreamHandler.applyChunkMetadata(parser, finalChunk, metadata);
+
+ assertEquals("FINISH", metadata.getCompletionReason());
+ assertEquals(42, metadata.getTokenCount());
+ }
+}
