Fix converse streaming issues:

- Correct finish reason when stop reason is not tool_use
- Populate finish reason for non-tool_use cases
- Ensure multiple tool calls are output in ChatResponse

Closes gh-4374, gh-4126, gh-3251

Signed-off-by: Jared Rufer <[email protected]>

Author: emopti-jrufer

models/spring-ai-bedrock-converse/src/main/java/org/springframework/ai/bedrock/converse/BedrockProxyChatModel.java

@@ -34,8 +34,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Sinks;
-import reactor.core.publisher.Sinks.EmitFailureHandler;
 import reactor.core.scheduler.Schedulers;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.core.SdkBytes;
@@ -51,9 +49,7 @@
 import software.amazon.awssdk.services.bedrockruntime.model.ConverseMetrics;
 import software.amazon.awssdk.services.bedrockruntime.model.ConverseRequest;
 import software.amazon.awssdk.services.bedrockruntime.model.ConverseResponse;
-import software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamOutput;
 import software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamRequest;
-import software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamResponseHandler;
 import software.amazon.awssdk.services.bedrockruntime.model.DocumentBlock;
 import software.amazon.awssdk.services.bedrockruntime.model.DocumentSource;
 import software.amazon.awssdk.services.bedrockruntime.model.ImageBlock;
@@ -76,6 +72,7 @@
 
 import org.springframework.ai.bedrock.converse.api.BedrockMediaFormat;
 import org.springframework.ai.bedrock.converse.api.ConverseApiUtils;
+import org.springframework.ai.bedrock.converse.api.ConverseChatResponseStream;
 import org.springframework.ai.bedrock.converse.api.URLValidator;
 import org.springframework.ai.chat.messages.AssistantMessage;
 import org.springframework.ai.chat.messages.MessageType;
@@ -84,6 +81,7 @@
 import org.springframework.ai.chat.metadata.ChatGenerationMetadata;
 import org.springframework.ai.chat.metadata.ChatResponseMetadata;
 import org.springframework.ai.chat.metadata.DefaultUsage;
+import org.springframework.ai.chat.metadata.Usage;
 import org.springframework.ai.chat.model.ChatModel;
 import org.springframework.ai.chat.model.ChatResponse;
 import org.springframework.ai.chat.model.Generation;
@@ -680,9 +678,14 @@ private Flux<ChatResponse> internalStream(Prompt prompt, ChatResponse perviousCh
 				.toolConfig(converseRequest.toolConfig())
 				.build();
 
-			Flux<ConverseStreamOutput> response = converseStream(converseStreamRequest);
+			Usage accumulatedUsage = null;
+			if (perviousChatResponse != null && perviousChatResponse.getMetadata() != null) {
+				accumulatedUsage = perviousChatResponse.getMetadata().getUsage();
+			}
 
-			Flux<ChatResponse> chatResponses = ConverseApiUtils.toChatResponse(response, perviousChatResponse);
+			Flux<ChatResponse> chatResponses = new ConverseChatResponseStream(this.bedrockRuntimeAsyncClient,
+					converseStreamRequest, accumulatedUsage)
+				.stream();
 
 			Flux<ChatResponse> chatResponseFlux = chatResponses.switchMap(chatResponse -> {
 
@@ -729,48 +732,6 @@ private Flux<ChatResponse> internalStream(Prompt prompt, ChatResponse perviousCh
 		});
 	}
 
-	public static final EmitFailureHandler DEFAULT_EMIT_FAILURE_HANDLER = EmitFailureHandler
-		.busyLooping(Duration.ofSeconds(10));
-
-	/**
-	 * Invoke the model and return the response stream.
-	 *
-	 * https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters.html
-	 * https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_Converse.html
-	 * https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/services/bedrockruntime/BedrockRuntimeAsyncClient.html#converseStream
-	 * @param converseStreamRequest Model invocation request.
-	 * @return The model invocation response stream.
-	 */
-	public Flux<ConverseStreamOutput> converseStream(ConverseStreamRequest converseStreamRequest) {
-		Assert.notNull(converseStreamRequest, "'converseStreamRequest' must not be null");
-
-		Sinks.Many<ConverseStreamOutput> eventSink = Sinks.many().multicast().onBackpressureBuffer();
-
-		ConverseStreamResponseHandler.Visitor visitor = ConverseStreamResponseHandler.Visitor.builder()
-			.onDefault(output -> {
-				logger.debug("Received converse stream output:{}", output);
-				eventSink.emitNext(output, DEFAULT_EMIT_FAILURE_HANDLER);
-			})
-			.build();
-
-		ConverseStreamResponseHandler responseHandler = ConverseStreamResponseHandler.builder()
-			.onEventStream(stream -> stream.subscribe(e -> e.accept(visitor)))
-			.onComplete(() -> {
-				eventSink.emitComplete(DEFAULT_EMIT_FAILURE_HANDLER);
-				logger.info("Completed streaming response.");
-			})
-			.onError(error -> {
-				logger.error("Error handling Bedrock converse stream response", error);
-				eventSink.emitError(error, DEFAULT_EMIT_FAILURE_HANDLER);
-			})
-			.build();
-
-		this.bedrockRuntimeAsyncClient.converseStream(converseStreamRequest, responseHandler);
-
-		return eventSink.asFlux();
-
-	}
-
 	/**
 	 * Use the provided convention for reporting observation data
 	 * @param observationConvention The provided convention