This is an automated email from the ASF dual-hosted git repository.
davsclaus 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 d56ad66487bc CAMEL-26069: camel-aws-bedrock - send the body as bytes,
and fail the streaming of image and embedding models (#28892)
d56ad66487bc is described below
commit d56ad66487bc9f06744c54ade02f31ad9a8a5830
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:35:46 2026 +0100
CAMEL-26069: camel-aws-bedrock - send the body as bytes, and fail the
streaming of image and embedding models (#28892)
* CAMEL-26069: camel-aws-bedrock - send the body as bytes, and fail the…
… streaming of image and embedding models
The invoke operations sent the body as its toString(), so a byte[], stream,
file or Map was sent as garbage and no body as "null". The body is now
converted to bytes, and a Map or JsonNode is marshalled to JSON.
The streaming operations now fail for image and embedding models (they do
not stream) instead of giving an empty text, and read the chunks of Nova and
Jamba. In POJO mode the model of the request is used to read the response,
and streaming resolves an inference profile to its model. Also: titan-embed
v2 is rejected by invokeTextModel, no operation fails with a message,
converseStream takes a GuardrailConfiguration header, asks for and sets the
guardrail trace, Jamba invokeTextModel gives its choices, and the header
types in the catalog are corrected.
* CAMEL-26069: camel-aws-bedrock - reuse the static ObjectMapper and fi…
…x the validator test
Reuse the static MAPPER when reading the model responses instead of a new
ObjectMapper per call. Note why a GuardrailConfiguration header given to
converseStream has no streamProcessingMode. The YAML validator no longer
flags a Map body before an invoke operation of aws-bedrock, as it is now
marshalled to JSON: update TextBodyFlowTest.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../camel/catalog/components/aws-bedrock.json | 12 +-
.../camel/catalog/docs/aws-bedrock-component.adoc | 24 +-
.../aws2/bedrock/runtime/aws-bedrock.json | 12 +-
.../src/main/docs/aws-bedrock-component.adoc | 24 +-
.../aws2/bedrock/runtime/BedrockConstants.java | 9 +-
.../aws2/bedrock/runtime/BedrockEndpoint.java | 10 +-
.../aws2/bedrock/runtime/BedrockProducer.java | 148 ++++++--
.../bedrock/runtime/stream/Ai21StreamParser.java | 86 +++++
.../runtime/stream/ConverseStreamHandler.java | 16 +-
.../bedrock/runtime/stream/NovaStreamParser.java | 74 ++++
.../runtime/stream/StreamParserFactory.java | 8 +-
...BedrockInvokeTextModelUnsupportedModelTest.java | 1 +
.../runtime/BedrockProducerRequestTest.java | 417 +++++++++++++++++++++
.../runtime/stream/ConverseStreamHandlerTest.java | 43 +++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 21 ++
.../dsl/BedrockEndpointBuilderFactory.java | 11 +-
.../camel/dsl/yaml/validator/TextBodyFlowTest.java | 12 +-
17 files changed, 850 insertions(+), 78 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-bedrock.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-bedrock.json
index 27287f9994c7..2f94d59671b3 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-bedrock.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/aws-bedrock.json
@@ -69,19 +69,19 @@
"CamelAwsBedrockConverseAdditionalFields": { "index": 11, "kind":
"header", "displayName": "", "group": "producer", "label": "", "required":
false, "javaType": "software.amazon.awssdk.core.document.Document",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The additional model request fields for Converse API",
"constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_ADDITIONAL_MODEL_REQUEST_FIELDS"
},
"CamelAwsBedrockConverseStopReason": { "index": 12, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The stop reason from Converse API
response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_STOP_REASON"
},
"CamelAwsBedrockConverseUsage": { "index": 13, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "TokenUsage", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The usage metrics from
Converse API response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_USAGE"
},
- "CamelAwsBedrockConverseOutputMessage": { "index": 14, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "Message", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The output message from Converse API
response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_OUTPUT_MESSAGE"
},
- "CamelAwsBedrockGuardrailConfig": { "index": 15, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailConfiguration", "deprecated": false, "deprecationNote":
"", "autowired": false, "secret": false, "description": "The guardrail
configuration to apply to the request", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_CONFIG"
},
+ "CamelAwsBedrockConverseOutputMessage": { "index": 14, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "Message", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The output message from Converse API
response (converse only)", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_OUTPUT_MESSAGE"
},
+ "CamelAwsBedrockGuardrailConfig": { "index": 15, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailConfiguration", "deprecated": false, "deprecationNote":
"", "autowired": false, "secret": false, "description": "The guardrail
configuration to apply to the request. converseStream also takes a
GuardrailStreamConfiguration", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_CONFIG"
},
"CamelAwsBedrockGuardrailIdentifier": { "index": 16, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The guardrail identifier to use for the
ApplyGuardrail operation", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_IDENTIFIER"
},
"CamelAwsBedrockGuardrailContent": { "index": 17, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "List<GuardrailContentBlock>", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
content blocks for ApplyGuardrail operation", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_CONTENT"
},
"CamelAwsBedrockGuardrailSource": { "index": 18, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The source type for ApplyGuardrail
operation (INPUT or OUTPUT)", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_SOURCE"
},
- "CamelAwsBedrockGuardrailOutput": { "index": 19, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailAssessment", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The guardrail assessment
output from the response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_OUTPUT"
},
- "CamelAwsBedrockGuardrailTrace": { "index": 20, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailTrace", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The trace information from
guardrail evaluation", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_TRACE"
},
+ "CamelAwsBedrockGuardrailOutput": { "index": 19, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "List<GuardrailOutputContent>", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
output content of the ApplyGuardrail response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_OUTPUT"
},
+ "CamelAwsBedrockGuardrailTrace": { "index": 20, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailTraceAssessment", "deprecated": false, "deprecationNote":
"", "autowired": false, "secret": false, "description": "The trace information
from guardrail evaluation", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_TRACE"
},
"CamelAwsBedrockGuardrailAssessments": { "index": 21, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "List<GuardrailAssessment>", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
guardrail assessments from ApplyGuardrail response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_ASSESSMENTS"
},
"CamelAwsBedrockGuardrailUsage": { "index": 22, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailUsage", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The guardrail usage
metrics from ApplyGuardrail response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_USAGE"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text" ], "byOption": [ { "option":
"operation", "value": "converse", "carriers": [ "any" ] }, { "option":
"operation", "value": "converseStream", "carriers": [ "any" ] }, { "option":
"operation", "value": "applyGuardrail", "carriers": [ "any" ] } ],
"description": "The request of the model as JSON text for the invoke
operations, which also need the CamelAwsBedrockContentType and
CamelAwsBedrockAcceptContentType headers. Bytes, a stream or a file are sen
[...]
- "producerOut": { "carriers": [ "text", "list" ], "byOption": [ { "option":
"operation", "value": "invokeTextModel", "carriers": [ "text", "pojo" ] }, {
"option": "operation", "value": "invokeImageModel", "carriers": [ "pojo" ] }, {
"option": "operation", "value": "invokeEmbeddingsModel", "carriers": [ "pojo"
], "javaType":
"software.amazon.awssdk.services.bedrockruntime.model.InvokeModelResponse" }, {
"option": "operation", "value": "converse", "carriers": [ "text" ], "javaType":
"ja [...]
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map" ],
"byOption": [ { "option": "operation", "value": "converse", "carriers": [ "any"
] }, { "option": "operation", "value": "converseStream", "carriers": [ "any" ]
}, { "option": "operation", "value": "applyGuardrail", "carriers": [ "any" ] }
], "description": "The request of the model as JSON text for the invoke
operations, which also need the CamelAwsBedrockContentType and
CamelAwsBedrockAcceptContentType headers. [...]
+ "producerOut": { "carriers": [ "text", "list" ], "byOption": [ { "option":
"operation", "value": "invokeTextModel", "carriers": [ "text", "pojo" ] }, {
"option": "operation", "value": "invokeImageModel", "carriers": [ "pojo" ] }, {
"option": "operation", "value": "invokeEmbeddingsModel", "carriers": [ "pojo"
], "javaType":
"software.amazon.awssdk.services.bedrockruntime.model.InvokeModelResponse" }, {
"option": "operation", "value": "converse", "carriers": [ "text" ], "javaType":
"ja [...]
},
"properties": {
"label": { "index": 0, "kind": "path", "displayName": "Label", "group":
"producer", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration",
"configurationField": "configuration", "description": "Logical name" },
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws-bedrock-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws-bedrock-component.adoc
index d193f3bea8cb..a58d10e8b11c 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws-bedrock-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/aws-bedrock-component.adoc
@@ -89,10 +89,25 @@ models (for example `amazon.titan-image-generator-v1`,
`amazon.nova-canvas-v1:0`
`invokeEmbeddingsModel`. Invoking `invokeTextModel` with an image, embedding,
rerank or video/speech model fails fast
with an `IllegalArgumentException` that names the model and points to the
operation to use instead.
+The operation is set with the `operation` option or the
`CamelAwsBedrockOperation` header: with neither, the producer
+fails with an `IllegalArgumentException`.
+
+The invoke operations send the message body as the request of the model, in
the format the model expects (usually
+JSON), with the `CamelAwsBedrockContentType` and
`CamelAwsBedrockAcceptContentType` headers. A String, a byte[], an
+InputStream or a file is sent as it is, and a `Map` or a Jackson `JsonNode` is
marshalled to JSON. Marshal another POJO
+to JSON first. A message with no body fails with an `InvalidPayloadException`.
+
+With `pojoRequest=true`, the body is the request of the AWS SDK, and the model
of the request is used to read the
+response, so the `modelId` option is not needed.
+
=== Streaming Support
The streaming operations (`invokeTextModelStreaming`,
`invokeImageModelStreaming`, `invokeEmbeddingsModelStreaming`) enable real-time
streaming of model responses, providing lower latency and better user
experience for interactive applications.
+The streaming operations give the generated text. The image generation and
embedding models do not stream their
+response, so the three streaming operations work the same way, and an image or
embedding model fails with an
+`IllegalArgumentException` that points to `invokeImageModel` or
`invokeEmbeddingsModel`.
+
==== Streaming Output Modes
Two streaming modes are supported:
@@ -123,6 +138,10 @@ All text generation models support streaming:
- Mistral AI (all models)
- Cohere Command models
- Amazon Nova models
+- AI21 Labs Jamba models
+
+A cross-region inference profile id (for example
`us.anthropic.claude-3-haiku-20240307-v1:0`) or an inference profile
+ARN can be used as the model id, as with `invokeTextModel`.
=== Converse API Support
@@ -170,7 +189,8 @@ When a conversation completes, the following headers are
set:
- `CamelAwsBedrockConverseStopReason`: Why the model stopped (e.g.,
"end_turn", "max_tokens")
- `CamelAwsBedrockConverseUsage`: TokenUsage object with input/output token
counts
-- `CamelAwsBedrockConverseOutputMessage`: The complete Message object from the
model
+- `CamelAwsBedrockConverseOutputMessage`: The complete Message object from the
model (converse only)
+- `CamelAwsBedrockGuardrailTrace`: The GuardrailTraceAssessment, when a
guardrail with trace enabled is applied
- `CamelAwsBedrockChunkCount`: (streaming only) Number of chunks received
==== Supported Models for Converse API
@@ -207,7 +227,7 @@ The Camel AWS Bedrock component provides comprehensive
support for guardrails th
*Message-level Configuration* (per-message override via headers):
-- `CamelAwsBedrockGuardrailConfig`: GuardrailConfiguration object for converse
operations
+- `CamelAwsBedrockGuardrailConfig`: GuardrailConfiguration object for converse
operations (converseStream also takes a GuardrailStreamConfiguration)
- `CamelAwsBedrockGuardrailIdentifier`: Guardrail identifier String for the
applyGuardrail operation (overrides the endpoint `guardrailIdentifier`)
- `CamelAwsBedrockGuardrailContent`: Content blocks for applyGuardrail
operation
- `CamelAwsBedrockGuardrailSource`: Source type - "INPUT" or "OUTPUT" for
applyGuardrail
diff --git
a/components/camel-aws/camel-aws-bedrock/src/generated/resources/META-INF/org/apache/camel/component/aws2/bedrock/runtime/aws-bedrock.json
b/components/camel-aws/camel-aws-bedrock/src/generated/resources/META-INF/org/apache/camel/component/aws2/bedrock/runtime/aws-bedrock.json
index 27287f9994c7..2f94d59671b3 100644
---
a/components/camel-aws/camel-aws-bedrock/src/generated/resources/META-INF/org/apache/camel/component/aws2/bedrock/runtime/aws-bedrock.json
+++
b/components/camel-aws/camel-aws-bedrock/src/generated/resources/META-INF/org/apache/camel/component/aws2/bedrock/runtime/aws-bedrock.json
@@ -69,19 +69,19 @@
"CamelAwsBedrockConverseAdditionalFields": { "index": 11, "kind":
"header", "displayName": "", "group": "producer", "label": "", "required":
false, "javaType": "software.amazon.awssdk.core.document.Document",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "The additional model request fields for Converse API",
"constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_ADDITIONAL_MODEL_REQUEST_FIELDS"
},
"CamelAwsBedrockConverseStopReason": { "index": 12, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The stop reason from Converse API
response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_STOP_REASON"
},
"CamelAwsBedrockConverseUsage": { "index": 13, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "TokenUsage", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The usage metrics from
Converse API response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_USAGE"
},
- "CamelAwsBedrockConverseOutputMessage": { "index": 14, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "Message", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The output message from Converse API
response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_OUTPUT_MESSAGE"
},
- "CamelAwsBedrockGuardrailConfig": { "index": 15, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailConfiguration", "deprecated": false, "deprecationNote":
"", "autowired": false, "secret": false, "description": "The guardrail
configuration to apply to the request", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_CONFIG"
},
+ "CamelAwsBedrockConverseOutputMessage": { "index": 14, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "Message", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The output message from Converse API
response (converse only)", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#CONVERSE_OUTPUT_MESSAGE"
},
+ "CamelAwsBedrockGuardrailConfig": { "index": 15, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailConfiguration", "deprecated": false, "deprecationNote":
"", "autowired": false, "secret": false, "description": "The guardrail
configuration to apply to the request. converseStream also takes a
GuardrailStreamConfiguration", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_CONFIG"
},
"CamelAwsBedrockGuardrailIdentifier": { "index": 16, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The guardrail identifier to use for the
ApplyGuardrail operation", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_IDENTIFIER"
},
"CamelAwsBedrockGuardrailContent": { "index": 17, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "List<GuardrailContentBlock>", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
content blocks for ApplyGuardrail operation", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_CONTENT"
},
"CamelAwsBedrockGuardrailSource": { "index": 18, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The source type for ApplyGuardrail
operation (INPUT or OUTPUT)", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_SOURCE"
},
- "CamelAwsBedrockGuardrailOutput": { "index": 19, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailAssessment", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The guardrail assessment
output from the response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_OUTPUT"
},
- "CamelAwsBedrockGuardrailTrace": { "index": 20, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailTrace", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The trace information from
guardrail evaluation", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_TRACE"
},
+ "CamelAwsBedrockGuardrailOutput": { "index": 19, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "List<GuardrailOutputContent>", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
output content of the ApplyGuardrail response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_OUTPUT"
},
+ "CamelAwsBedrockGuardrailTrace": { "index": 20, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailTraceAssessment", "deprecated": false, "deprecationNote":
"", "autowired": false, "secret": false, "description": "The trace information
from guardrail evaluation", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_TRACE"
},
"CamelAwsBedrockGuardrailAssessments": { "index": 21, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "List<GuardrailAssessment>", "deprecated": false,
"deprecationNote": "", "autowired": false, "secret": false, "description": "The
guardrail assessments from ApplyGuardrail response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_ASSESSMENTS"
},
"CamelAwsBedrockGuardrailUsage": { "index": 22, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "GuardrailUsage", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "The guardrail usage
metrics from ApplyGuardrail response", "constantName":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConstants#GUARDRAIL_USAGE"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text" ], "byOption": [ { "option":
"operation", "value": "converse", "carriers": [ "any" ] }, { "option":
"operation", "value": "converseStream", "carriers": [ "any" ] }, { "option":
"operation", "value": "applyGuardrail", "carriers": [ "any" ] } ],
"description": "The request of the model as JSON text for the invoke
operations, which also need the CamelAwsBedrockContentType and
CamelAwsBedrockAcceptContentType headers. Bytes, a stream or a file are sen
[...]
- "producerOut": { "carriers": [ "text", "list" ], "byOption": [ { "option":
"operation", "value": "invokeTextModel", "carriers": [ "text", "pojo" ] }, {
"option": "operation", "value": "invokeImageModel", "carriers": [ "pojo" ] }, {
"option": "operation", "value": "invokeEmbeddingsModel", "carriers": [ "pojo"
], "javaType":
"software.amazon.awssdk.services.bedrockruntime.model.InvokeModelResponse" }, {
"option": "operation", "value": "converse", "carriers": [ "text" ], "javaType":
"ja [...]
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map" ],
"byOption": [ { "option": "operation", "value": "converse", "carriers": [ "any"
] }, { "option": "operation", "value": "converseStream", "carriers": [ "any" ]
}, { "option": "operation", "value": "applyGuardrail", "carriers": [ "any" ] }
], "description": "The request of the model as JSON text for the invoke
operations, which also need the CamelAwsBedrockContentType and
CamelAwsBedrockAcceptContentType headers. [...]
+ "producerOut": { "carriers": [ "text", "list" ], "byOption": [ { "option":
"operation", "value": "invokeTextModel", "carriers": [ "text", "pojo" ] }, {
"option": "operation", "value": "invokeImageModel", "carriers": [ "pojo" ] }, {
"option": "operation", "value": "invokeEmbeddingsModel", "carriers": [ "pojo"
], "javaType":
"software.amazon.awssdk.services.bedrockruntime.model.InvokeModelResponse" }, {
"option": "operation", "value": "converse", "carriers": [ "text" ], "javaType":
"ja [...]
},
"properties": {
"label": { "index": 0, "kind": "path", "displayName": "Label", "group":
"producer", "label": "", "required": true, "type": "string", "javaType":
"java.lang.String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "configurationClass":
"org.apache.camel.component.aws2.bedrock.runtime.BedrockConfiguration",
"configurationField": "configuration", "description": "Logical name" },
diff --git
a/components/camel-aws/camel-aws-bedrock/src/main/docs/aws-bedrock-component.adoc
b/components/camel-aws/camel-aws-bedrock/src/main/docs/aws-bedrock-component.adoc
index d193f3bea8cb..a58d10e8b11c 100644
---
a/components/camel-aws/camel-aws-bedrock/src/main/docs/aws-bedrock-component.adoc
+++
b/components/camel-aws/camel-aws-bedrock/src/main/docs/aws-bedrock-component.adoc
@@ -89,10 +89,25 @@ models (for example `amazon.titan-image-generator-v1`,
`amazon.nova-canvas-v1:0`
`invokeEmbeddingsModel`. Invoking `invokeTextModel` with an image, embedding,
rerank or video/speech model fails fast
with an `IllegalArgumentException` that names the model and points to the
operation to use instead.
+The operation is set with the `operation` option or the
`CamelAwsBedrockOperation` header: with neither, the producer
+fails with an `IllegalArgumentException`.
+
+The invoke operations send the message body as the request of the model, in
the format the model expects (usually
+JSON), with the `CamelAwsBedrockContentType` and
`CamelAwsBedrockAcceptContentType` headers. A String, a byte[], an
+InputStream or a file is sent as it is, and a `Map` or a Jackson `JsonNode` is
marshalled to JSON. Marshal another POJO
+to JSON first. A message with no body fails with an `InvalidPayloadException`.
+
+With `pojoRequest=true`, the body is the request of the AWS SDK, and the model
of the request is used to read the
+response, so the `modelId` option is not needed.
+
=== Streaming Support
The streaming operations (`invokeTextModelStreaming`,
`invokeImageModelStreaming`, `invokeEmbeddingsModelStreaming`) enable real-time
streaming of model responses, providing lower latency and better user
experience for interactive applications.
+The streaming operations give the generated text. The image generation and
embedding models do not stream their
+response, so the three streaming operations work the same way, and an image or
embedding model fails with an
+`IllegalArgumentException` that points to `invokeImageModel` or
`invokeEmbeddingsModel`.
+
==== Streaming Output Modes
Two streaming modes are supported:
@@ -123,6 +138,10 @@ All text generation models support streaming:
- Mistral AI (all models)
- Cohere Command models
- Amazon Nova models
+- AI21 Labs Jamba models
+
+A cross-region inference profile id (for example
`us.anthropic.claude-3-haiku-20240307-v1:0`) or an inference profile
+ARN can be used as the model id, as with `invokeTextModel`.
=== Converse API Support
@@ -170,7 +189,8 @@ When a conversation completes, the following headers are
set:
- `CamelAwsBedrockConverseStopReason`: Why the model stopped (e.g.,
"end_turn", "max_tokens")
- `CamelAwsBedrockConverseUsage`: TokenUsage object with input/output token
counts
-- `CamelAwsBedrockConverseOutputMessage`: The complete Message object from the
model
+- `CamelAwsBedrockConverseOutputMessage`: The complete Message object from the
model (converse only)
+- `CamelAwsBedrockGuardrailTrace`: The GuardrailTraceAssessment, when a
guardrail with trace enabled is applied
- `CamelAwsBedrockChunkCount`: (streaming only) Number of chunks received
==== Supported Models for Converse API
@@ -207,7 +227,7 @@ The Camel AWS Bedrock component provides comprehensive
support for guardrails th
*Message-level Configuration* (per-message override via headers):
-- `CamelAwsBedrockGuardrailConfig`: GuardrailConfiguration object for converse
operations
+- `CamelAwsBedrockGuardrailConfig`: GuardrailConfiguration object for converse
operations (converseStream also takes a GuardrailStreamConfiguration)
- `CamelAwsBedrockGuardrailIdentifier`: Guardrail identifier String for the
applyGuardrail operation (overrides the endpoint `guardrailIdentifier`)
- `CamelAwsBedrockGuardrailContent`: Content blocks for applyGuardrail
operation
- `CamelAwsBedrockGuardrailSource`: Source type - "INPUT" or "OUTPUT" for
applyGuardrail
diff --git
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockConstants.java
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockConstants.java
index 0ab3692e5775..b214d56eb46f 100644
---
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockConstants.java
+++
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockConstants.java
@@ -51,9 +51,10 @@ public interface BedrockConstants {
String CONVERSE_STOP_REASON = "CamelAwsBedrockConverseStopReason";
@Metadata(description = "The usage metrics from Converse API response",
javaType = "TokenUsage")
String CONVERSE_USAGE = "CamelAwsBedrockConverseUsage";
- @Metadata(description = "The output message from Converse API response",
javaType = "Message")
+ @Metadata(description = "The output message from Converse API response
(converse only)", javaType = "Message")
String CONVERSE_OUTPUT_MESSAGE = "CamelAwsBedrockConverseOutputMessage";
- @Metadata(description = "The guardrail configuration to apply to the
request", javaType = "GuardrailConfiguration")
+ @Metadata(description = "The guardrail configuration to apply to the
request. converseStream also takes a GuardrailStreamConfiguration",
+ javaType = "GuardrailConfiguration")
String GUARDRAIL_CONFIG = "CamelAwsBedrockGuardrailConfig";
@Metadata(description = "The guardrail identifier to use for the
ApplyGuardrail operation", javaType = "String")
String GUARDRAIL_IDENTIFIER = "CamelAwsBedrockGuardrailIdentifier";
@@ -61,9 +62,9 @@ public interface BedrockConstants {
String GUARDRAIL_CONTENT = "CamelAwsBedrockGuardrailContent";
@Metadata(description = "The source type for ApplyGuardrail operation
(INPUT or OUTPUT)", javaType = "String")
String GUARDRAIL_SOURCE = "CamelAwsBedrockGuardrailSource";
- @Metadata(description = "The guardrail assessment output from the
response", javaType = "GuardrailAssessment")
+ @Metadata(description = "The output content of the ApplyGuardrail
response", javaType = "List<GuardrailOutputContent>")
String GUARDRAIL_OUTPUT = "CamelAwsBedrockGuardrailOutput";
- @Metadata(description = "The trace information from guardrail evaluation",
javaType = "GuardrailTrace")
+ @Metadata(description = "The trace information from guardrail evaluation",
javaType = "GuardrailTraceAssessment")
String GUARDRAIL_TRACE = "CamelAwsBedrockGuardrailTrace";
@Metadata(description = "The guardrail assessments from ApplyGuardrail
response", javaType = "List<GuardrailAssessment>")
String GUARDRAIL_ASSESSMENTS = "CamelAwsBedrockGuardrailAssessments";
diff --git
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockEndpoint.java
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockEndpoint.java
index 4c017a1c6f3e..8007cb65dbc9 100644
---
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockEndpoint.java
+++
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockEndpoint.java
@@ -42,7 +42,7 @@ import
software.amazon.awssdk.services.bedrockruntime.BedrockRuntimeClient;
@UriEndpoint(firstVersion = "4.5.0", scheme = "aws-bedrock", title = "AWS
Bedrock",
syntax = "aws-bedrock:label", producerOnly = true, category = {
Category.AI, Category.CLOUD },
headersClass = BedrockConstants.class)
-@BodyTypes(producerIn = @BodyType(value = Carrier.TEXT,
+@BodyTypes(producerIn = @BodyType(value = { Carrier.TEXT, Carrier.BYTES,
Carrier.STREAM, Carrier.FILE, Carrier.MAP },
byOption = {
@BodyType.When(option = "operation",
value = "converse", carriers = Carrier.ANY),
@BodyType.When(option = "operation",
value = "converseStream",
@@ -51,8 +51,9 @@ import
software.amazon.awssdk.services.bedrockruntime.BedrockRuntimeClient;
carriers =
Carrier.ANY) },
description = "The request of the model as
JSON text for the invoke operations, which also"
+ " need the
CamelAwsBedrockContentType and CamelAwsBedrockAcceptContentType"
- + " headers. Bytes, a stream
or a file are sent as their toString(): convert"
- + " them to a String first,
and marshal a Map or a POJO to JSON. converse,"
+ + " headers. Bytes, a stream
or a file are sent as they are, and a Map or a"
+ + " Jackson JsonNode is
marshalled to JSON: marshal another POJO to JSON first."
+ + " No body fails. converse,"
+ " converseStream and
applyGuardrail take their input from headers, such as"
+ "
CamelAwsBedrockConverseMessages. With pojoRequest=true an"
+ " InvokeModelRequest of the
AWS SDK: another body fails"),
@@ -71,7 +72,8 @@ import
software.amazon.awssdk.services.bedrockruntime.BedrockRuntimeClient;
carriers =
Carrier.TEXT,
javaType =
"java.lang.String") },
description = "The streaming operations and
converseStream give the text, or a List of the"
- + " chunks with
streamOutputMode=chunks. invokeTextModel gives the JSON"
+ + " chunks with
streamOutputMode=chunks: an image or embedding model fails, as"
+ + " it does not stream.
invokeTextModel gives the JSON"
+ " response as text for the
Amazon Titan models, and a Jackson JsonNode for"
+ " the others: the response,
or the generated text for Llama, Cohere and"
+ " Claude 2.
invokeImageModel gives a JsonNode array of the images as Base64"
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 0db965a4a4b9..66681709f276 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
@@ -18,6 +18,7 @@ package org.apache.camel.component.aws2.bedrock.runtime;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
@@ -69,6 +70,7 @@ public class BedrockProducer extends DefaultProducer {
* 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 static final ObjectMapper MAPPER = new ObjectMapper();
private transient String bedrockProducerToString;
public BedrockProducer(Endpoint endpoint) {
@@ -115,9 +117,29 @@ public class BedrockProducer extends DefaultProducer {
if (ObjectHelper.isEmpty(operation)) {
operation = getConfiguration().getOperation();
}
+ if (operation == null) {
+ throw new IllegalArgumentException(
+ "Operation must be specified with the operation option or
the " + BedrockConstants.OPERATION + " header");
+ }
return operation;
}
+ /**
+ * The request of the model: the message body converted to bytes, so a
String, byte[], InputStream or File body is
+ * sent as is. A Map or a Jackson JsonNode is marshalled to JSON.
+ */
+ private static SdkBytes requestBody(Exchange exchange) throws
InvalidPayloadException {
+ Object body = exchange.getMessage().getBody();
+ if (body instanceof Map || body instanceof JsonNode) {
+ try {
+ return SdkBytes.fromByteArray(MAPPER.writeValueAsBytes(body));
+ } catch (JsonProcessingException e) {
+ throw new InvalidPayloadException(exchange, byte[].class,
exchange.getMessage(), e);
+ }
+ }
+ return
SdkBytes.fromByteArray(exchange.getMessage().getMandatoryBody(byte[].class));
+ }
+
protected BedrockConfiguration getConfiguration() {
return getEndpoint().getConfiguration();
}
@@ -147,7 +169,8 @@ public class BedrockProducer extends DefaultProducer {
throw ase;
}
Message message = getMessageForResponse(exchange);
- setResponseText(result, message);
+ // the request carries the model, which the endpoint may not
set
+ setResponseText(invokeModelRequest.modelId(), result, message);
} else {
throw new IllegalArgumentException(
"InvokeTextModel operation requires InvokeModelRequest
in POJO mode");
@@ -167,7 +190,7 @@ public class BedrockProducer extends DefaultProducer {
throw new IllegalArgumentException("Model Accept Content Type
must be specified");
}
InvokeModelRequest request = builder
-
.body(SdkBytes.fromUtf8String(String.valueOf(exchange.getMessage().getBody())))
+ .body(requestBody(exchange))
.modelId(getConfiguration().getModelId())
.build();
InvokeModelResponse result;
@@ -178,7 +201,7 @@ public class BedrockProducer extends DefaultProducer {
throw ase;
}
Message message = getMessageForResponse(exchange);
- setResponseText(result, message);
+ setResponseText(request.modelId(), result, message);
}
}
@@ -214,7 +237,7 @@ public class BedrockProducer extends DefaultProducer {
throw new IllegalArgumentException("Model Accept Content Type
must be specified");
}
InvokeModelRequest request = builder
-
.body(SdkBytes.fromUtf8String(String.valueOf(exchange.getMessage().getBody())))
+ .body(requestBody(exchange))
.modelId(getConfiguration().getModelId())
.build();
InvokeModelResponse result;
@@ -266,7 +289,7 @@ public class BedrockProducer extends DefaultProducer {
throw new IllegalArgumentException("Model Accept Content Type
must be specified");
}
InvokeModelRequest request = builder
-
.body(SdkBytes.fromUtf8String(String.valueOf(exchange.getMessage().getBody())))
+ .body(requestBody(exchange))
.modelId(getConfiguration().getModelId())
.build();
InvokeModelResponse result;
@@ -282,19 +305,24 @@ public class BedrockProducer extends DefaultProducer {
}
private static void setBase64Image(InvokeModelResponse result, Message
message) throws JsonProcessingException {
- ObjectMapper mapper = new ObjectMapper();
- JsonNode jsonString = mapper.readTree(result.body().asUtf8String());
+ JsonNode jsonString = MAPPER.readTree(result.body().asUtf8String());
message.setBody(jsonString.get("images"));
}
protected void setResponseText(InvokeModelResponse result, Message
message) {
- String modelId =
resolveFoundationModelId(getConfiguration().getModelId());
+ setResponseText(getConfiguration().getModelId(), result, message);
+ }
+
+ protected void setResponseText(String requestModelId, InvokeModelResponse
result, Message message) {
+ String modelId = resolveFoundationModelId(requestModelId);
+ if (modelId == null) {
+ throw new IllegalArgumentException("Model id must be specified
with the modelId option");
+ }
switch (modelId) {
// Amazon Titan Models
case "amazon.titan-text-express-v1":
case "amazon.titan-text-lite-v1":
case "amazon.titan-text-premier-v1:0":
- case "amazon.titan-embed-text-v2:0":
setTitanText(result, message);
break;
@@ -408,6 +436,7 @@ public class BedrockProducer extends DefaultProducer {
// Embedding models — use the invokeEmbeddingsModel operation
instead
case "amazon.titan-embed-text-v1":
+ case "amazon.titan-embed-text-v2:0":
case "amazon.titan-embed-image-v1":
case "cohere.embed-english-v3":
case "cohere.embed-multilingual-v3":
@@ -471,9 +500,7 @@ public class BedrockProducer extends DefaultProducer {
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")) {
+ if (nonTextModelKind(modelId) != null) {
throw new IllegalArgumentException(
"Model " + modelId
+ " is not a text generation
model and cannot be used with the invokeTextModel operation.");
@@ -499,43 +526,55 @@ public class BedrockProducer extends DefaultProducer {
}
}
+ /**
+ * The kind of model, by its vendor/family name, when the model does not
generate text, or null for a text model.
+ */
+ private static String nonTextModelKind(String modelId) {
+ if (modelId.startsWith("stability.") ||
modelId.contains("image-generator") || modelId.contains("-canvas")) {
+ return "an image generation model";
+ } else if (modelId.contains("embed")) {
+ return "an embedding model";
+ } else if (modelId.contains("rerank")) {
+ return "a rerank model";
+ } else if (modelId.contains("-reel") || modelId.contains("-sonic")) {
+ return "a video/speech generation model";
+ }
+ return null;
+ }
+
private void setTitanText(InvokeModelResponse result, Message message) {
message.setBody(result.body().asUtf8String());
}
private void setAi21Text(InvokeModelResponse result, Message message)
throws JsonProcessingException {
- ObjectMapper mapper = new ObjectMapper();
- JsonNode jsonString = mapper.readTree(result.body().asUtf8String());
- message.setBody(jsonString.get("completions"));
+ JsonNode jsonString = MAPPER.readTree(result.body().asUtf8String());
+ // Jurassic-2 answers with completions, Jamba with choices
+ JsonNode completions = jsonString.get("completions");
+ message.setBody(completions != null ? completions :
jsonString.get("choices"));
}
private void setAnthropicText(InvokeModelResponse result, Message message)
throws JsonProcessingException {
- ObjectMapper mapper = new ObjectMapper();
- JsonNode jsonString = mapper.readTree(result.body().asUtf8String());
+ JsonNode jsonString = MAPPER.readTree(result.body().asUtf8String());
message.setBody(jsonString.get("completion"));
}
private void setAnthropicV3Text(InvokeModelResponse result, Message
message) throws JsonProcessingException {
- ObjectMapper mapper = new ObjectMapper();
- JsonNode jsonString = mapper.readTree(result.body().asUtf8String());
+ JsonNode jsonString = MAPPER.readTree(result.body().asUtf8String());
message.setBody(jsonString);
}
private void setMistralText(InvokeModelResponse result, Message message)
throws JsonProcessingException {
- ObjectMapper mapper = new ObjectMapper();
- JsonNode jsonString = mapper.readTree(result.body().asUtf8String());
+ JsonNode jsonString = MAPPER.readTree(result.body().asUtf8String());
message.setBody(jsonString);
}
private void setCohereText(InvokeModelResponse result, Message message)
throws JsonProcessingException {
- ObjectMapper mapper = new ObjectMapper();
- JsonNode jsonString = mapper.readTree(result.body().asUtf8String());
+ JsonNode jsonString = MAPPER.readTree(result.body().asUtf8String());
message.setBody(jsonString.get("text"));
}
private void setLlamaText(InvokeModelResponse result, Message message)
throws JsonProcessingException {
- ObjectMapper mapper = new ObjectMapper();
- JsonNode jsonString = mapper.readTree(result.body().asUtf8String());
+ JsonNode jsonString = MAPPER.readTree(result.body().asUtf8String());
message.setBody(jsonString.get("generation"));
}
@@ -564,7 +603,7 @@ public class BedrockProducer extends DefaultProducer {
throw new IllegalArgumentException("Model Accept Content Type
must be specified");
}
InvokeModelWithResponseStreamRequest request = builder
-
.body(SdkBytes.fromUtf8String(String.valueOf(exchange.getMessage().getBody())))
+ .body(requestBody(exchange))
.modelId(getConfiguration().getModelId())
.build();
processStreamingRequest(request, exchange);
@@ -573,20 +612,49 @@ public class BedrockProducer extends DefaultProducer {
private void invokeImageModelStreaming(BedrockRuntimeClient
bedrockRuntimeClient, Exchange exchange)
throws InvalidPayloadException {
- // Image streaming works the same way as text streaming
+ // the image and embedding models do not stream their response, so
this is the same as text streaming,
+ // and an image or embedding model fails with a message that points to
the operation to use
invokeTextModelStreaming(bedrockRuntimeClient, exchange);
}
private void invokeEmbeddingsModelStreaming(BedrockRuntimeClient
bedrockRuntimeClient, Exchange exchange)
throws InvalidPayloadException {
- // Embeddings streaming works the same way as text streaming
+ // the image and embedding models do not stream their response, so
this is the same as text streaming,
+ // and an image or embedding model fails with a message that points to
the operation to use
invokeTextModelStreaming(bedrockRuntimeClient, exchange);
}
+ /**
+ * The model of a streaming request, resolved to its foundation model to
choose the parser of its chunks. A model
+ * that does not generate text fails: its chunks have no text to give.
+ */
+ private static String
streamingModelId(InvokeModelWithResponseStreamRequest request) {
+ String modelId = resolveFoundationModelId(request.modelId());
+ if (modelId == null) {
+ throw new IllegalArgumentException("Model id must be specified
with the modelId option");
+ }
+ String kind = nonTextModelKind(modelId);
+ if (kind != null) {
+ String hint = "";
+ if (kind.startsWith("an image")) {
+ hint = " Use the invokeImageModel operation instead.";
+ } else if (kind.startsWith("an embedding")) {
+ hint = " Use the invokeEmbeddingsModel operation instead.";
+ }
+ throw new IllegalArgumentException(
+ "Model " + modelId + " is " + kind
+ + " and cannot be used with the
streaming operations, which give the generated text."
+ + hint);
+ }
+ return modelId;
+ }
+
private void processStreamingRequest(
InvokeModelWithResponseStreamRequest request,
Exchange exchange) {
+ // the request carries the model, which the endpoint may not set in
POJO mode
+ String modelId = streamingModelId(request);
try {
String streamOutputMode = getConfiguration().getStreamOutputMode();
if (streamOutputMode == null) {
@@ -607,7 +675,7 @@ public class BedrockProducer extends DefaultProducer {
getEndpoint().getBedrockRuntimeAsyncClient().invokeModelWithResponseStream(
request,
BedrockStreamHandler.createChunksHandler(
- getConfiguration().getModelId(),
+ modelId,
metadata,
allChunks,
null))
@@ -623,7 +691,7 @@ public class BedrockProducer extends DefaultProducer {
getEndpoint().getBedrockRuntimeAsyncClient().invokeModelWithResponseStream(
request,
BedrockStreamHandler.createCompleteHandler(
- getConfiguration().getModelId(),
+ modelId,
metadata,
fullText))
.join();
@@ -826,17 +894,27 @@ public class BedrockProducer extends DefaultProducer {
}
// Optional: Guardrail configuration (use
GuardrailStreamConfiguration for streaming)
- GuardrailStreamConfiguration guardrailConfig
- =
exchange.getMessage().getHeader(BedrockConstants.GUARDRAIL_CONFIG,
- GuardrailStreamConfiguration.class);
- if (ObjectHelper.isNotEmpty(guardrailConfig)) {
- builder.guardrailConfig(guardrailConfig);
+ Object guardrailConfig =
exchange.getMessage().getHeader(BedrockConstants.GUARDRAIL_CONFIG);
+ if (guardrailConfig instanceof GuardrailStreamConfiguration
streamConfig) {
+ builder.guardrailConfig(streamConfig);
+ } else if (guardrailConfig instanceof GuardrailConfiguration
config) {
+ // the GuardrailConfiguration of the converse operation, as
the docs name it. It has no
+ // streamProcessingMode, so Bedrock uses its default (sync):
pass a GuardrailStreamConfiguration
+ // to choose async
+ builder.guardrailConfig(GuardrailStreamConfiguration.builder()
+ .guardrailIdentifier(config.guardrailIdentifier())
+ .guardrailVersion(config.guardrailVersion())
+ .trace(config.traceAsString())
+ .build());
} else if
(ObjectHelper.isNotEmpty(getConfiguration().getGuardrailIdentifier())) {
// Build from endpoint configuration
builder.guardrailConfig(GuardrailStreamConfiguration
.builder()
.guardrailIdentifier(getConfiguration().getGuardrailIdentifier())
.guardrailVersion(getConfiguration().getGuardrailVersion())
+ .trace(getConfiguration().isGuardrailTrace()
+ ? GuardrailTrace.ENABLED
+ : GuardrailTrace.DISABLED)
.streamProcessingMode(getConfiguration().isGuardrailTrace()
? GuardrailStreamProcessingMode.ASYNC
: GuardrailStreamProcessingMode.SYNC)
diff --git
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/Ai21StreamParser.java
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/Ai21StreamParser.java
new file mode 100644
index 000000000000..f362cf6547a5
--- /dev/null
+++
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/Ai21StreamParser.java
@@ -0,0 +1,86 @@
+/*
+ * 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 com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Parser for AI21 Labs Jamba model streaming responses
+ * <p>
+ * Jamba streaming response format, as the chat completions: {"choices":
[{"index": 0, "delta": {"role": "assistant",
+ * "content": "chunk"}, "finish_reason": null}]} {"choices": [{"index": 0,
"delta": {"content": ""}, "finish_reason":
+ * "stop"}], "usage": {"prompt_tokens": 10, "completion_tokens": 150,
"total_tokens": 160}}
+ */
+public class Ai21StreamParser implements StreamResponseParser {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ @Override
+ public String extractText(String chunk) throws JsonProcessingException {
+ JsonNode choice = firstChoice(chunk);
+ if (choice != null) {
+ JsonNode delta = choice.get("delta");
+ if (ObjectHelper.isNotEmpty(delta)) {
+ JsonNode content = delta.get("content");
+ return ObjectHelper.isNotEmpty(content) && !content.isNull() ?
content.asText() : "";
+ }
+ }
+ return "";
+ }
+
+ @Override
+ public String extractCompletionReason(String chunk) throws
JsonProcessingException {
+ JsonNode choice = firstChoice(chunk);
+ if (choice != null) {
+ JsonNode finishReason = choice.get("finish_reason");
+ return ObjectHelper.isNotEmpty(finishReason) &&
!finishReason.isNull() ? finishReason.asText() : null;
+ }
+ return null;
+ }
+
+ @Override
+ public Integer extractTokenCount(String chunk) throws
JsonProcessingException {
+ if (chunk == null || chunk.trim().isEmpty()) {
+ return null;
+ }
+ JsonNode usage = MAPPER.readTree(chunk).get("usage");
+ if (ObjectHelper.isNotEmpty(usage)) {
+ JsonNode completionTokens = usage.get("completion_tokens");
+ return ObjectHelper.isNotEmpty(completionTokens) &&
!completionTokens.isNull() ? completionTokens.asInt() : null;
+ }
+ return null;
+ }
+
+ @Override
+ public boolean isFinalChunk(String chunk) throws JsonProcessingException {
+ return ObjectHelper.isNotEmpty(extractCompletionReason(chunk));
+ }
+
+ private static JsonNode firstChoice(String chunk) throws
JsonProcessingException {
+ if (chunk == null || chunk.trim().isEmpty()) {
+ return null;
+ }
+ JsonNode choices = MAPPER.readTree(chunk).get("choices");
+ if (ObjectHelper.isNotEmpty(choices) && choices.isArray() &&
!choices.isEmpty()) {
+ return choices.get(0);
+ }
+ return null;
+ }
+}
diff --git
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandler.java
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandler.java
index 3b978996dd15..902f07054c2a 100644
---
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandler.java
+++
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandler.java
@@ -23,7 +23,7 @@ import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import
software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamResponseHandler;
-import software.amazon.awssdk.services.bedrockruntime.model.GuardrailTrace;
+import
software.amazon.awssdk.services.bedrockruntime.model.GuardrailTraceAssessment;
import software.amazon.awssdk.services.bedrockruntime.model.TokenUsage;
/**
@@ -62,6 +62,10 @@ public final class ConverseStreamHandler {
if
(ObjectHelper.isNotEmpty(metadataEvent.usage())) {
metadata.setUsage(metadataEvent.usage());
}
+ if (ObjectHelper.isNotEmpty(metadataEvent.trace())
+ &&
ObjectHelper.isNotEmpty(metadataEvent.trace().guardrail())) {
+
metadata.setGuardrailTrace(metadataEvent.trace().guardrail());
+ }
})
.onMessageStop(stop -> {
if (ObjectHelper.isNotEmpty(stop.stopReason())) {
@@ -107,6 +111,10 @@ public final class ConverseStreamHandler {
if
(ObjectHelper.isNotEmpty(metadataEvent.usage())) {
metadata.setUsage(metadataEvent.usage());
}
+ if (ObjectHelper.isNotEmpty(metadataEvent.trace())
+ &&
ObjectHelper.isNotEmpty(metadataEvent.trace().guardrail())) {
+
metadata.setGuardrailTrace(metadataEvent.trace().guardrail());
+ }
})
.onMessageStop(stop -> {
if (ObjectHelper.isNotEmpty(stop.stopReason())) {
@@ -130,7 +138,7 @@ public final class ConverseStreamHandler {
private String stopReason;
private TokenUsage usage;
private int chunkCount;
- private GuardrailTrace guardrailTrace;
+ private GuardrailTraceAssessment guardrailTrace;
public String getFullText() {
return fullText;
@@ -172,11 +180,11 @@ public final class ConverseStreamHandler {
this.chunkCount = chunkCount;
}
- public GuardrailTrace getGuardrailTrace() {
+ public GuardrailTraceAssessment getGuardrailTrace() {
return guardrailTrace;
}
- public void setGuardrailTrace(GuardrailTrace guardrailTrace) {
+ public void setGuardrailTrace(GuardrailTraceAssessment guardrailTrace)
{
this.guardrailTrace = guardrailTrace;
}
}
diff --git
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/NovaStreamParser.java
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/NovaStreamParser.java
new file mode 100644
index 000000000000..39176e7d08a4
--- /dev/null
+++
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/NovaStreamParser.java
@@ -0,0 +1,74 @@
+/*
+ * 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 com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * Parser for Amazon Nova model streaming responses
+ * <p>
+ * Nova streaming response format: {"messageStart": {"role": "assistant"}}
{"contentBlockDelta": {"delta": {"text":
+ * "chunk"}, "contentBlockIndex": 0}} {"contentBlockStop":
{"contentBlockIndex": 0}} {"messageStop": {"stopReason":
+ * "end_turn"}} {"metadata": {"usage": {"inputTokens": 10, "outputTokens":
150}}}
+ */
+public class NovaStreamParser implements StreamResponseParser {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ @Override
+ public String extractText(String chunk) throws JsonProcessingException {
+ JsonNode text = path(chunk, "contentBlockDelta", "delta", "text");
+ return text != null ? text.asText() : "";
+ }
+
+ @Override
+ public String extractCompletionReason(String chunk) throws
JsonProcessingException {
+ JsonNode stopReason = path(chunk, "messageStop", "stopReason");
+ return stopReason != null ? stopReason.asText() : null;
+ }
+
+ @Override
+ public Integer extractTokenCount(String chunk) throws
JsonProcessingException {
+ JsonNode outputTokens = path(chunk, "metadata", "usage",
"outputTokens");
+ return outputTokens != null ? outputTokens.asInt() : null;
+ }
+
+ @Override
+ public boolean isFinalChunk(String chunk) throws JsonProcessingException {
+ return path(chunk, "messageStop") != null;
+ }
+
+ /**
+ * The node at the given path of the chunk, or null when the chunk has no
such node.
+ */
+ private static JsonNode path(String chunk, String... names) throws
JsonProcessingException {
+ if (chunk == null || chunk.trim().isEmpty()) {
+ return null;
+ }
+ JsonNode node = MAPPER.readTree(chunk);
+ for (String name : names) {
+ node = node.get(name);
+ if (ObjectHelper.isEmpty(node) || node.isNull()) {
+ return null;
+ }
+ }
+ return node;
+ }
+}
diff --git
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/StreamParserFactory.java
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/StreamParserFactory.java
index 19b1ebc0dffb..a7167dd7a7f8 100644
---
a/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/StreamParserFactory.java
+++
b/components/camel-aws/camel-aws-bedrock/src/main/java/org/apache/camel/component/aws2/bedrock/runtime/stream/StreamParserFactory.java
@@ -61,14 +61,14 @@ public final class StreamParserFactory {
return new CohereStreamParser();
}
- // Amazon Nova Models (use Claude-compatible format)
+ // Amazon Nova Models
if (modelId.startsWith("amazon.nova")) {
- return new ClaudeStreamParser();
+ return new NovaStreamParser();
}
- // AI21 Models (use Titan-compatible format for simplicity)
+ // AI21 Jamba Models (the Jurassic-2 models do not stream)
if (modelId.startsWith("ai21.")) {
- return new TitanStreamParser();
+ return new Ai21StreamParser();
}
throw new IllegalArgumentException("Unsupported model for streaming: "
+ modelId);
diff --git
a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockInvokeTextModelUnsupportedModelTest.java
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockInvokeTextModelUnsupportedModelTest.java
index 63e0bfcf34bf..02d795a491ab 100644
---
a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockInvokeTextModelUnsupportedModelTest.java
+++
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockInvokeTextModelUnsupportedModelTest.java
@@ -96,6 +96,7 @@ public class BedrockInvokeTextModelUnsupportedModelTest {
Arguments.of(BedrockModels.STABLE_IMAGE_CORE.model,
"invokeImageModel"),
// Embedding models -> invokeEmbeddingsModel
Arguments.of(BedrockModels.TITAN_EMBEDDINGS_G1.model,
"invokeEmbeddingsModel"),
+ Arguments.of(BedrockModels.TITAN_TEXT_EMBEDDINGS_V2.model,
"invokeEmbeddingsModel"),
Arguments.of(BedrockModels.TITAN_MULTIMODAL_EMBEDDINGS_G1.model,
"invokeEmbeddingsModel"),
Arguments.of(BedrockModels.COHERE_EMBED_ENGLISH_V3.model,
"invokeEmbeddingsModel"),
Arguments.of(BedrockModels.COHERE_EMBED_MULTILINGUAL_V3.model,
"invokeEmbeddingsModel"),
diff --git
a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducerRequestTest.java
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducerRequestTest.java
new file mode 100644
index 000000000000..5a2d2d0ad653
--- /dev/null
+++
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/BedrockProducerRequestTest.java
@@ -0,0 +1,417 @@
+/*
+ * 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 java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.aws2.bedrock.BedrockModels;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.SimpleRegistry;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.core.async.SdkPublisher;
+import
software.amazon.awssdk.services.bedrockruntime.BedrockRuntimeAsyncClient;
+import software.amazon.awssdk.services.bedrockruntime.BedrockRuntimeClient;
+import software.amazon.awssdk.services.bedrockruntime.model.ContentBlock;
+import software.amazon.awssdk.services.bedrockruntime.model.ConversationRole;
+import
software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamRequest;
+import
software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamResponseHandler;
+import
software.amazon.awssdk.services.bedrockruntime.model.GuardrailConfiguration;
+import software.amazon.awssdk.services.bedrockruntime.model.GuardrailTrace;
+import software.amazon.awssdk.services.bedrockruntime.model.InvokeModelRequest;
+import
software.amazon.awssdk.services.bedrockruntime.model.InvokeModelResponse;
+import
software.amazon.awssdk.services.bedrockruntime.model.InvokeModelWithResponseStreamRequest;
+import
software.amazon.awssdk.services.bedrockruntime.model.InvokeModelWithResponseStreamResponse;
+import
software.amazon.awssdk.services.bedrockruntime.model.InvokeModelWithResponseStreamResponseHandler;
+import software.amazon.awssdk.services.bedrockruntime.model.Message;
+import software.amazon.awssdk.services.bedrockruntime.model.ResponseStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The request the producer sends to the model, and how it reads the model of
the request (CAMEL-26069). Mocked clients
+ * stand in for AWS, so no credentials or network access are needed.
+ */
+@ExtendWith(MockitoExtension.class)
+public class BedrockProducerRequestTest {
+
+ private static final String REQUEST = "{\"inputText\":\"Hello\"}";
+ private static final String TITAN =
BedrockModels.TITAN_TEXT_EXPRESS_V1.model;
+ private static final String CLAUDE =
BedrockModels.ANTROPHIC_CLAUDE_HAIKU_V3.model;
+
+ @Mock
+ private BedrockRuntimeClient client;
+
+ @Mock
+ private BedrockRuntimeAsyncClient asyncClient;
+
+ @TempDir
+ Path tempDir;
+
+ private CamelContext camelContext;
+ private ProducerTemplate template;
+
+ @BeforeEach
+ public void setup() {
+ lenient().when(client.invokeModel(any(InvokeModelRequest.class)))
+
.thenReturn(InvokeModelResponse.builder().body(SdkBytes.fromUtf8String("{\"results\":[]}")).build());
+
+ SimpleRegistry registry = new SimpleRegistry();
+ registry.bind("bedrockClient", client);
+ registry.bind("bedrockAsyncClient", asyncClient);
+ camelContext = new DefaultCamelContext(registry);
+ camelContext.start();
+ template = camelContext.createProducerTemplate();
+ }
+
+ @AfterEach
+ public void teardown() {
+ if (template != null) {
+ template.stop();
+ }
+ if (camelContext != null) {
+ camelContext.stop();
+ }
+ }
+
+ @Test
+ public void stringBodyIsSent() {
+ assertEquals(REQUEST, sentRequest(REQUEST));
+ }
+
+ @Test
+ public void bytesBodyIsSentAsTheBytes() {
+ assertEquals(REQUEST,
sentRequest(REQUEST.getBytes(StandardCharsets.UTF_8)));
+ }
+
+ @Test
+ public void streamBodyIsSentAsItsContent() {
+ assertEquals(REQUEST, sentRequest(new
ByteArrayInputStream(REQUEST.getBytes(StandardCharsets.UTF_8))));
+ }
+
+ @Test
+ public void fileBodyIsSentAsItsContent() throws Exception {
+ Path file = Files.writeString(tempDir.resolve("request.json"),
REQUEST);
+ assertEquals(REQUEST, sentRequest(file.toFile()));
+ }
+
+ @Test
+ public void mapBodyIsSentAsJson() {
+ assertEquals(REQUEST, sentRequest(Map.of("inputText", "Hello")));
+ }
+
+ @Test
+ public void jsonNodeBodyIsSentAsJson() throws Exception {
+ assertEquals(REQUEST, sentRequest(new
ObjectMapper().readTree(REQUEST)));
+ }
+
+ @Test
+ public void noBodyFails() {
+ Exchange exchange = send(invokeUri("invokeTextModel", TITAN), null);
+
+ assertInstanceOf(InvalidPayloadException.class,
exchange.getException());
+ }
+
+ @Test
+ public void imageAndEmbeddingsOperationsSendTheBodyToo() {
+ ArgumentCaptor<InvokeModelRequest> request =
ArgumentCaptor.forClass(InvokeModelRequest.class);
+ when(client.invokeModel(request.capture()))
+
.thenReturn(InvokeModelResponse.builder().body(SdkBytes.fromUtf8String("{\"images\":[]}")).build());
+
+ byte[] body = REQUEST.getBytes(StandardCharsets.UTF_8);
+ assertNull(send(invokeUri("invokeImageModel",
BedrockModels.TITAN_IMAGE_GENERATOR_V1.model), body).getException());
+ assertNull(send(invokeUri("invokeEmbeddingsModel",
BedrockModels.TITAN_EMBEDDINGS_G1.model), body).getException());
+
+ assertEquals(2, request.getAllValues().size());
+ request.getAllValues().forEach(r -> assertEquals(REQUEST,
r.body().asUtf8String()));
+ }
+
+ @Test
+ public void noOperationFailsWithAMessage() {
+ Exchange exchange =
send("aws-bedrock://label?bedrockRuntimeClient=#bedrockClient&modelId=" + TITAN
+ +
"®ion=us-east-1&accessKey=unused&secretKey=unused",
+ REQUEST);
+
+ assertInstanceOf(IllegalArgumentException.class,
exchange.getException());
+ assertEquals("Operation must be specified with the operation option or
the CamelAwsBedrockOperation header",
+ exchange.getException().getMessage());
+ }
+
+ @Test
+ public void titanEmbeddingsV2IsRejectedByInvokeTextModel() {
+ Exchange exchange = send(invokeUri("invokeTextModel",
BedrockModels.TITAN_TEXT_EMBEDDINGS_V2.model), REQUEST);
+
+ assertInstanceOf(IllegalArgumentException.class,
exchange.getException());
+
assertTrue(exchange.getException().getMessage().contains("invokeEmbeddingsModel"),
+ exchange.getException().getMessage());
+ }
+
+ @Test
+ public void jambaResponseGivesTheChoices() {
+
when(client.invokeModel(any(InvokeModelRequest.class))).thenReturn(InvokeModelResponse.builder()
+
.body(SdkBytes.fromUtf8String("{\"choices\":[{\"message\":{\"content\":\"Hi\"}}]}")).build());
+
+ Exchange exchange = send(invokeUri("invokeTextModel",
BedrockModels.JAMBA_1_5_MINI.model), REQUEST);
+
+ assertNull(exchange.getException());
+ JsonNode body = assertInstanceOf(JsonNode.class,
exchange.getMessage().getBody());
+ assertEquals("Hi", body.get(0).get("message").get("content").asText());
+ }
+
+ @Test
+ public void pojoRequestReadsTheResponseWithTheModelOfTheRequest() {
+
when(client.invokeModel(any(InvokeModelRequest.class))).thenReturn(InvokeModelResponse.builder()
+
.body(SdkBytes.fromUtf8String("{\"content\":[{\"type\":\"text\",\"text\":\"Hi\"}]}")).build());
+ InvokeModelRequest request =
InvokeModelRequest.builder().modelId(CLAUDE)
+ .body(SdkBytes.fromUtf8String(REQUEST)).build();
+
+ // no modelId option on the endpoint: the request carries the model
+ Exchange exchange = send(pojoUri("invokeTextModel"), request);
+
+ assertNull(exchange.getException());
+ JsonNode body = assertInstanceOf(JsonNode.class,
exchange.getMessage().getBody());
+ assertEquals("Hi", body.get("content").get(0).get("text").asText());
+ }
+
+ @Test
+ public void streamingPojoRequestUsesTheModelOfTheRequest() {
+ stream("{\"outputText\":\"Hello\"}", "{\"outputText\":\"
world\",\"completionReason\":\"FINISH\"}");
+ InvokeModelWithResponseStreamRequest request =
InvokeModelWithResponseStreamRequest.builder().modelId(TITAN)
+ .body(SdkBytes.fromUtf8String(REQUEST)).build();
+
+ Exchange exchange = send(pojoUri("invokeTextModelStreaming"), request);
+
+ assertNull(exchange.getException());
+ assertEquals("Hello world", exchange.getMessage().getBody());
+ }
+
+ @Test
+ public void streamingResolvesAnInferenceProfileToTheModel() {
+
stream("{\"type\":\"content_block_delta\",\"delta\":{\"type\":\"text_delta\",\"text\":\"Hi\"}}");
+
+ Exchange exchange = send(invokeUri("invokeTextModelStreaming", "us." +
CLAUDE), REQUEST);
+
+ assertNull(exchange.getException());
+ assertEquals("Hi", exchange.getMessage().getBody());
+ }
+
+ @Test
+ public void streamingSendsTheBodyAsTheBytes() {
+ ArgumentCaptor<InvokeModelWithResponseStreamRequest> request
+ =
ArgumentCaptor.forClass(InvokeModelWithResponseStreamRequest.class);
+ stream("{\"outputText\":\"Hi\"}");
+
+ Exchange exchange = send(invokeUri("invokeTextModelStreaming", TITAN),
REQUEST.getBytes(StandardCharsets.UTF_8));
+
+ assertNull(exchange.getException());
+ verify(asyncClient).invokeModelWithResponseStream(request.capture(),
+ any(InvokeModelWithResponseStreamResponseHandler.class));
+ assertEquals(REQUEST, request.getValue().body().asUtf8String());
+ }
+
+ @Test
+ public void imageModelStreamingFailsInsteadOfGivingAnEmptyText() {
+ Exchange exchange = send(invokeUri("invokeImageModelStreaming",
BedrockModels.TITAN_IMAGE_GENERATOR_V1.model),
+ REQUEST);
+
+ assertInstanceOf(IllegalArgumentException.class,
exchange.getException());
+
assertTrue(exchange.getException().getMessage().contains("invokeImageModel
operation"),
+ exchange.getException().getMessage());
+ }
+
+ @Test
+ public void embeddingsModelStreamingFailsInsteadOfGivingAnEmptyText() {
+ Exchange exchange = send(invokeUri("invokeEmbeddingsModelStreaming",
BedrockModels.TITAN_EMBEDDINGS_G1.model),
+ REQUEST);
+
+ assertInstanceOf(IllegalArgumentException.class,
exchange.getException());
+
assertTrue(exchange.getException().getMessage().contains("invokeEmbeddingsModel
operation"),
+ exchange.getException().getMessage());
+ }
+
+ @Test
+ public void novaStreamingGivesTheText() {
+ stream("{\"messageStart\":{\"role\":\"assistant\"}}",
+
"{\"contentBlockDelta\":{\"delta\":{\"text\":\"Hello\"},\"contentBlockIndex\":0}}",
+ "{\"contentBlockDelta\":{\"delta\":{\"text\":\"
world\"},\"contentBlockIndex\":0}}",
+ "{\"messageStop\":{\"stopReason\":\"end_turn\"}}",
+
"{\"metadata\":{\"usage\":{\"inputTokens\":5,\"outputTokens\":2}}}");
+
+ Exchange exchange = send(invokeUri("invokeTextModelStreaming",
BedrockModels.NOVA_LITE_V1.model), REQUEST);
+
+ assertNull(exchange.getException());
+ assertEquals("Hello world", exchange.getMessage().getBody());
+ assertEquals("end_turn",
exchange.getMessage().getHeader(BedrockConstants.STREAMING_COMPLETION_REASON));
+ assertEquals(2,
exchange.getMessage().getHeader(BedrockConstants.STREAMING_TOKEN_COUNT));
+ }
+
+ @Test
+ public void jambaStreamingGivesTheText() {
+
stream("{\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"Hello\"},\"finish_reason\":null}]}",
+ "{\"choices\":[{\"index\":0,\"delta\":{\"content\":\"
world\"},\"finish_reason\":\"stop\"}],"
+
+
"\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":2,\"total_tokens\":7}}");
+
+ Exchange exchange = send(invokeUri("invokeTextModelStreaming",
BedrockModels.JAMBA_1_5_MINI.model), REQUEST);
+
+ assertNull(exchange.getException());
+ assertEquals("Hello world", exchange.getMessage().getBody());
+ assertEquals("stop",
exchange.getMessage().getHeader(BedrockConstants.STREAMING_COMPLETION_REASON));
+ assertEquals(2,
exchange.getMessage().getHeader(BedrockConstants.STREAMING_TOKEN_COUNT));
+ }
+
+ @Test
+ public void converseStreamTakesTheGuardrailConfigurationOfTheDocs() {
+ ArgumentCaptor<ConverseStreamRequest> request =
ArgumentCaptor.forClass(ConverseStreamRequest.class);
+ when(asyncClient.converseStream(request.capture(),
any(ConverseStreamResponseHandler.class)))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ Exchange exchange =
template.send("aws-bedrock://label?bedrockRuntimeAsyncClient=#bedrockAsyncClient"
+ +
"&operation=converseStream&modelId=" + CLAUDE
+ +
"®ion=us-east-1&accessKey=unused&secretKey=unused",
+ ex -> {
+
ex.getMessage().setHeader(BedrockConstants.CONVERSE_MESSAGES,
List.of(userMessage()));
+
ex.getMessage().setHeader(BedrockConstants.GUARDRAIL_CONFIG,
GuardrailConfiguration.builder()
+
.guardrailIdentifier("my-guardrail").guardrailVersion("1").trace(GuardrailTrace.ENABLED)
+ .build());
+ });
+
+ assertNull(exchange.getException());
+ assertNotNull(request.getValue().guardrailConfig());
+ assertEquals("my-guardrail",
request.getValue().guardrailConfig().guardrailIdentifier());
+ assertEquals("1",
request.getValue().guardrailConfig().guardrailVersion());
+ assertEquals(GuardrailTrace.ENABLED,
request.getValue().guardrailConfig().trace());
+ }
+
+ @Test
+ public void converseStreamEnablesTheGuardrailTraceOfTheEndpoint() {
+ ArgumentCaptor<ConverseStreamRequest> request =
ArgumentCaptor.forClass(ConverseStreamRequest.class);
+ when(asyncClient.converseStream(request.capture(),
any(ConverseStreamResponseHandler.class)))
+ .thenReturn(CompletableFuture.completedFuture(null));
+
+ Exchange exchange =
template.send("aws-bedrock://label?bedrockRuntimeAsyncClient=#bedrockAsyncClient"
+ +
"&operation=converseStream&modelId=" + CLAUDE
+ +
"&guardrailIdentifier=my-guardrail&guardrailTrace=true"
+ +
"®ion=us-east-1&accessKey=unused&secretKey=unused",
+ ex ->
ex.getMessage().setHeader(BedrockConstants.CONVERSE_MESSAGES,
List.of(userMessage())));
+
+ assertNull(exchange.getException());
+ assertEquals(GuardrailTrace.ENABLED,
request.getValue().guardrailConfig().trace());
+ }
+
+ private static Message userMessage() {
+ return
Message.builder().role(ConversationRole.USER).content(ContentBlock.fromText("Hello")).build();
+ }
+
+ private String sentRequest(Object body) {
+ ArgumentCaptor<InvokeModelRequest> request =
ArgumentCaptor.forClass(InvokeModelRequest.class);
+ Exchange exchange = send(invokeUri("invokeTextModel", TITAN), body);
+
+ assertNull(exchange.getException());
+ verify(client).invokeModel(request.capture());
+ return request.getValue().body().asUtf8String();
+ }
+
+ private Exchange send(String uri, Object body) {
+ return template.send(uri, ex -> {
+ ex.getMessage().setHeader(BedrockConstants.MODEL_CONTENT_TYPE,
"application/json");
+
ex.getMessage().setHeader(BedrockConstants.MODEL_ACCEPT_CONTENT_TYPE,
"application/json");
+ ex.getMessage().setBody(body);
+ });
+ }
+
+ /**
+ * The async client streams the given chunks to the response handler of
the producer.
+ */
+ private void stream(String... chunks) {
+
when(asyncClient.invokeModelWithResponseStream(any(InvokeModelWithResponseStreamRequest.class),
+
any(InvokeModelWithResponseStreamResponseHandler.class))).thenAnswer(invocation
-> {
+ InvokeModelWithResponseStreamResponseHandler handler =
invocation.getArgument(1);
+ List<ResponseStream> events = Arrays.stream(chunks)
+ .map(c -> (ResponseStream)
ResponseStream.chunkBuilder().bytes(SdkBytes.fromUtf8String(c)).build())
+ .toList();
+
handler.responseReceived(InvokeModelWithResponseStreamResponse.builder().build());
+
handler.onEventStream(SdkPublisher.adapt(publisherOf(events)));
+ handler.complete();
+ return CompletableFuture.completedFuture(null);
+ });
+ }
+
+ private static <T> Publisher<T> publisherOf(List<T> events) {
+ return subscriber -> subscriber.onSubscribe(new Subscription() {
+ private int next;
+ private boolean completed;
+
+ @Override
+ public void request(long n) {
+ for (long i = 0; i < n && next < events.size(); i++) {
+ subscriber.onNext(events.get(next++));
+ }
+ if (next == events.size() && !completed) {
+ completed = true;
+ subscriber.onComplete();
+ }
+ }
+
+ @Override
+ public void cancel() {
+ completed = true;
+ }
+ });
+ }
+
+ private static String invokeUri(String operation, String modelId) {
+ return
"aws-bedrock://label?bedrockRuntimeClient=#bedrockClient&bedrockRuntimeAsyncClient=#bedrockAsyncClient"
+ + "&operation=" + operation + "&modelId=" + modelId
+ + "®ion=us-east-1&accessKey=unused&secretKey=unused";
+ }
+
+ private static String pojoUri(String operation) {
+ return
"aws-bedrock://label?bedrockRuntimeClient=#bedrockClient&bedrockRuntimeAsyncClient=#bedrockAsyncClient"
+ + "&operation=" + operation + "&pojoRequest=true"
+ + "®ion=us-east-1&accessKey=unused&secretKey=unused";
+ }
+}
diff --git
a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandlerTest.java
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandlerTest.java
index a07aa1a728ee..15d2613290b1 100644
---
a/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandlerTest.java
+++
b/components/camel-aws/camel-aws-bedrock/src/test/java/org/apache/camel/component/aws2/bedrock/runtime/stream/ConverseStreamHandlerTest.java
@@ -20,7 +20,13 @@ import java.util.ArrayList;
import java.util.List;
import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscription;
+import software.amazon.awssdk.core.async.SdkPublisher;
+import
software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamOutput;
+import
software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamResponse;
import
software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamResponseHandler;
+import
software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamTrace;
+import
software.amazon.awssdk.services.bedrockruntime.model.GuardrailTraceAssessment;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -76,4 +82,41 @@ class ConverseStreamHandlerTest {
// Test usage is null initially
assertEquals(null, metadata.getUsage());
}
+
+ @Test
+ void testGuardrailTraceOfTheMetadataEvent() {
+ GuardrailTraceAssessment trace =
GuardrailTraceAssessment.builder().modelOutput("Hi").build();
+ ConverseStreamOutput event = ConverseStreamOutput.metadataBuilder()
+
.trace(ConverseStreamTrace.builder().guardrail(trace).build()).build();
+ ConverseStreamHandler.StreamMetadata complete = new
ConverseStreamHandler.StreamMetadata();
+ ConverseStreamHandler.StreamMetadata chunks = new
ConverseStreamHandler.StreamMetadata();
+
+ emit(ConverseStreamHandler.createCompleteHandler(complete, new
StringBuilder()), event);
+ emit(ConverseStreamHandler.createChunksHandler(chunks, new
ArrayList<>(), null), event);
+
+ assertEquals(trace, complete.getGuardrailTrace());
+ assertEquals(trace, chunks.getGuardrailTrace());
+ }
+
+ private static void emit(ConverseStreamResponseHandler handler,
ConverseStreamOutput event) {
+ handler.responseReceived(ConverseStreamResponse.builder().build());
+ handler.onEventStream(SdkPublisher.adapt(subscriber ->
subscriber.onSubscribe(new Subscription() {
+ private boolean done;
+
+ @Override
+ public void request(long n) {
+ if (!done) {
+ done = true;
+ subscriber.onNext(event);
+ subscriber.onComplete();
+ }
+ }
+
+ @Override
+ public void cancel() {
+ done = true;
+ }
+ })));
+ handler.complete();
+ }
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index ecb5b66e8081..661e3e3d2b8d 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -675,6 +675,27 @@ If you replaced these dependencies to use another HTTP
client implementation, na
`software.amazon.awssdk.http.service.impl` (synchronous clients) or
`software.amazon.awssdk.http.async.service.impl`
(asynchronous clients): Camel then leaves the choice of the HTTP client to the
AWS SDK.
+=== camel-aws-bedrock - the invoke operations send the body, not its toString()
+
+The invoke operations of `aws-bedrock` now convert the message body to bytes,
so a byte[], an InputStream or a file is
+sent as its content, and a `Map` or a Jackson `JsonNode` is marshalled to
JSON. Before, the body was sent as its
+`toString()`, so only a String body worked. A message with no body now fails
with an `InvalidPayloadException`, where
+before the text `null` was sent to the model.
+
+Other changes of `aws-bedrock`:
+
+* `invokeTextModel` with `amazon.titan-embed-text-v2:0` now fails with an
`IllegalArgumentException` that points to
+ `invokeEmbeddingsModel`, as with the other embedding models. Before, it gave
the JSON of the embedding as text.
+* The streaming operations fail with an `IllegalArgumentException` for an
image generation or embedding model, which
+ do not stream. Before, `invokeImageModelStreaming` and
`invokeEmbeddingsModelStreaming` gave an empty text.
+* With no `operation` option and no `CamelAwsBedrockOperation` header, the
producer fails with an
+ `IllegalArgumentException` that says so, instead of a `NullPointerException`.
+* The `CamelAwsBedrockGuardrailTrace` header is a `GuardrailTraceAssessment`,
as `converse` always set it, and
+ `converseStream` now sets it too. `ConverseStreamHandler.StreamMetadata`
keeps it as a `GuardrailTraceAssessment`
+ instead of a `GuardrailTrace`.
+* `converseStream` with `guardrailTrace=true` now asks for the guardrail trace.
+* `invokeTextModel` with an AI21 Jamba model gives the `choices` of the
response, which was null before.
+
=== camel-weather - the CamelWeatherLocation header is the location that is
queried
The latitude and longitude of the current geolocation (a location of
`current`, or no location) are no longer written
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BedrockEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BedrockEndpointBuilderFactory.java
index 1cb83e8e392e..b5c168375220 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BedrockEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/BedrockEndpointBuilderFactory.java
@@ -920,7 +920,7 @@ public interface BedrockEndpointBuilderFactory {
return "CamelAwsBedrockConverseUsage";
}
/**
- * The output message from Converse API response.
+ * The output message from Converse API response (converse only).
*
* The option is a: {@code Message} type.
*
@@ -933,7 +933,8 @@ public interface BedrockEndpointBuilderFactory {
return "CamelAwsBedrockConverseOutputMessage";
}
/**
- * The guardrail configuration to apply to the request.
+ * The guardrail configuration to apply to the request. converseStream
+ * also takes a GuardrailStreamConfiguration.
*
* The option is a: {@code GuardrailConfiguration} type.
*
@@ -981,9 +982,9 @@ public interface BedrockEndpointBuilderFactory {
return "CamelAwsBedrockGuardrailSource";
}
/**
- * The guardrail assessment output from the response.
+ * The output content of the ApplyGuardrail response.
*
- * The option is a: {@code GuardrailAssessment} type.
+ * The option is a: {@code List<GuardrailOutputContent>} type.
*
* Group: producer
*
@@ -995,7 +996,7 @@ public interface BedrockEndpointBuilderFactory {
/**
* The trace information from guardrail evaluation.
*
- * The option is a: {@code GuardrailTrace} type.
+ * The option is a: {@code GuardrailTraceAssessment} type.
*
* Group: producer
*
diff --git
a/dsl/camel-yaml-dsl/camel-yaml-dsl-validator/src/test/java/org/apache/camel/dsl/yaml/validator/TextBodyFlowTest.java
b/dsl/camel-yaml-dsl/camel-yaml-dsl-validator/src/test/java/org/apache/camel/dsl/yaml/validator/TextBodyFlowTest.java
index 2d9afd617374..0174a24bb55c 100644
---
a/dsl/camel-yaml-dsl/camel-yaml-dsl-validator/src/test/java/org/apache/camel/dsl/yaml/validator/TextBodyFlowTest.java
+++
b/dsl/camel-yaml-dsl/camel-yaml-dsl-validator/src/test/java/org/apache/camel/dsl/yaml/validator/TextBodyFlowTest.java
@@ -1923,10 +1923,10 @@ public class TextBodyFlowTest {
@Test
public void testTheBodyOfAwsAiServices() {
- // translate, comprehend, the speech of polly and the question of
retrieveAndGenerate are text, and the invoke
- // operations of aws-bedrock send the body as its toString(), so a Map
is not the JSON of the request; the
- // translation is text and a cloudtrail event is JSON in bytes.
converse, rekognition, textract and timestream
- // take their input from headers, and polly without an operation is
not known, as the component may set it
+ // translate, comprehend, the speech of polly and the question of
retrieveAndGenerate are text; the invoke
+ // operations of aws-bedrock marshal a Map to the JSON of the request;
the translation is text and a cloudtrail
+ // event is JSON in bytes. converse, rekognition, textract and
timestream take their input from headers, and
+ // polly without an operation is not known, as the component may set it
String yaml = """
- route:
id: translate
@@ -2030,11 +2030,11 @@ public class TextBodyFlowTest {
json: {}
- to: aws2-polly:p?voiceId=Joanna
""";
- assertThat(lines(yaml)).hasSize(7)
+ assertThat(lines(yaml)).hasSize(6)
.anySatisfy(m -> assertThat(m).contains("route translate: to:
aws2-translate needs the body as text"))
.anySatisfy(m -> assertThat(m).contains("route analyze: to:
aws2-comprehend needs the body as text"))
.anySatisfy(m -> assertThat(m).contains("route speak: to:
aws2-polly needs the body as text"))
- .anySatisfy(m -> assertThat(m).contains("route prompt: to:
aws-bedrock needs the body as text"))
+ .noneSatisfy(m -> assertThat(m).contains("route prompt:"))
.anySatisfy(m -> assertThat(m).contains("route ask: to:
aws-bedrock-agent-runtime needs the body as text"))
.anySatisfy(m -> assertThat(m).contains("route translated:
simple reads a field of the body (${body[text]})"))
.anySatisfy(m -> assertThat(m).contains("route trail: simple
reads a field of the body (${body[eventName]})"));
