This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26069-bedrock-body in repository https://gitbox.apache.org/repos/asf/camel.git
commit d7165c29eea2856f8a379b217f0700ca9c030451 Author: Claus Ibsen <[email protected]> AuthorDate: Mon Nov 2 16:12:59 2026 +0100 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. 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 | 125 ++++-- .../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 +- 16 files changed, 835 insertions(+), 58 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..a9526aed3620 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; @@ -288,13 +311,19 @@ public class BedrockProducer extends DefaultProducer { } 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 +437,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 +501,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,6 +527,22 @@ 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()); } @@ -506,7 +550,9 @@ public class BedrockProducer extends DefaultProducer { 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")); + // 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 { @@ -564,7 +610,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 +619,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 +682,7 @@ public class BedrockProducer extends DefaultProducer { getEndpoint().getBedrockRuntimeAsyncClient().invokeModelWithResponseStream( request, BedrockStreamHandler.createChunksHandler( - getConfiguration().getModelId(), + modelId, metadata, allChunks, null)) @@ -623,7 +698,7 @@ public class BedrockProducer extends DefaultProducer { getEndpoint().getBedrockRuntimeAsyncClient().invokeModelWithResponseStream( request, BedrockStreamHandler.createCompleteHandler( - getConfiguration().getModelId(), + modelId, metadata, fullText)) .join(); @@ -826,17 +901,25 @@ 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 + 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 1bc188419d61..171a074e0578 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 *
