[anthropic] fix issue #1370 with tool call duplication

Without this fix during the stream event handling when `EventType.MESSAGE_STOP` occurs, the latest content block was resent again and it caused to the additional tool call(if it was the latest event)

[anthropic] Replace `switchMap` -> `flatMap` to avoid cancellation of the original request

Previously, internalStream used switchMap to process ChatCompletionResponses,
which caused the active stream (including potential recursive calls) to be
canceled whenever a new response arrived. This led to incomplete processing
of streaming tool calls and unexpected behavior when handling tool_use events.

Replaced switchMap with flatMap to ensure that each response is fully processed
without being interrupted, allowing recursive internalStream calls to complete
as expected.

Without this fix during the stream event handling when `EventType.MESSAGE_STOP` occurs, the latest content block was resent again and it caused to the additional tool call(if it was the latest event)

Signed-off-by: Mikhail Mazurkevich <Mikhail.Mazurkevich@jetbrains.com>
This commit is contained in:
Mikhail Mazurkevich
2025-04-08 15:59:44 +04:00
committed by Ilayaperumal Gopinathan
parent b5c3216b76
commit e6fbc28a5b
3 changed files with 70 additions and 2 deletions

View File

@@ -253,7 +253,7 @@ public class AnthropicChatModel implements ChatModel {
this.getAdditionalHttpHeaders(prompt));
// @formatter:off
Flux<ChatResponse> chatResponseFlux = response.switchMap(chatCompletionResponse -> {
Flux<ChatResponse> chatResponseFlux = response.flatMap(chatCompletionResponse -> {
AnthropicApi.Usage usage = chatCompletionResponse.usage();
Usage currentChatResponseUsage = usage != null ? this.getDefaultUsage(chatCompletionResponse.usage()) : new EmptyUsage();
Usage accumulatedUsage = UsageUtils.getCumulativeUsage(currentChatResponseUsage, previousChatResponse);

View File

@@ -179,7 +179,14 @@ public class StreamHelper {
}
}
else if (event.type().equals(EventType.MESSAGE_STOP)) {
// pass through
// Don't return the latest Content block as it was before. Instead, return it
// with an updated event type and general information like: model, message
// type, id and usage
contentBlockReference.get()
.withType(event.type().name())
.withContent(List.of())
.withStopReason(null)
.withStopSequence(null);
}
else {
contentBlockReference.get().withType(event.type().name()).withContent(List.of());

View File

@@ -16,6 +16,7 @@
package org.springframework.ai.anthropic.api;
import java.util.ArrayList;
import java.util.List;
import org.junit.jupiter.api.Test;
@@ -27,6 +28,7 @@ import org.springframework.ai.anthropic.api.AnthropicApi.ChatCompletionRequest;
import org.springframework.ai.anthropic.api.AnthropicApi.ChatCompletionResponse;
import org.springframework.ai.anthropic.api.AnthropicApi.ContentBlock;
import org.springframework.ai.anthropic.api.AnthropicApi.Role;
import org.springframework.ai.model.ModelOptionsUtils;
import org.springframework.http.ResponseEntity;
import static org.assertj.core.api.Assertions.assertThat;
@@ -42,6 +44,24 @@ public class AnthropicApiIT {
AnthropicApi anthropicApi = AnthropicApi.builder().apiKey(System.getenv("ANTHROPIC_API_KEY")).build();
List<AnthropicApi.Tool> tools = List.of(new AnthropicApi.Tool("getCurrentWeather",
"Get the weather in location. Return temperature in 30°F or 30°C format.", ModelOptionsUtils.jsonToMap("""
{
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state e.g. San Francisco, CA"
},
"unit": {
"type": "string",
"enum": ["C", "F"]
}
},
"required": ["location", "unit"]
}
""")));
@Test
void chatCompletionEntity() {
@@ -106,6 +126,47 @@ public class AnthropicApiIT {
bla.stream().forEach(r -> System.out.println(r));
}
@Test
void chatCompletionStreamWithToolCall() {
List<AnthropicMessage> messageConversation = new ArrayList<>();
AnthropicMessage chatCompletionMessage = new AnthropicMessage(
List.of(new ContentBlock("What's the weather like in San Francisco? Show the temperature in Celsius.")),
Role.USER);
messageConversation.add(chatCompletionMessage);
ChatCompletionRequest chatCompletionRequest = ChatCompletionRequest.builder()
.model(AnthropicApi.ChatModel.CLAUDE_3_OPUS)
.messages(messageConversation)
.maxTokens(1500)
.stream(true)
.temperature(0.8)
.tools(tools)
.build();
List<ChatCompletionResponse> responses = this.anthropicApi.chatCompletionStream(chatCompletionRequest)
.collectList()
.block();
// Check that tool uses response returned only once
List<ChatCompletionResponse> toolCompletionResponses = responses.stream()
.filter(r -> r.stopReason() != null && r.stopReason().equals(ContentBlock.Type.TOOL_USE.value))
.toList();
assertThat(toolCompletionResponses).size().isEqualTo(1);
List<ContentBlock> toolContentBlocks = toolCompletionResponses.get(0).content();
assertThat(toolContentBlocks).size().isEqualTo(1);
ContentBlock toolContentBlock = toolContentBlocks.get(0);
assertThat(toolContentBlock.type()).isEqualTo(ContentBlock.Type.TOOL_USE);
assertThat(toolContentBlock.name()).isEqualTo("getCurrentWeather");
// Check that message stop response also returned
List<ChatCompletionResponse> messageStopEvents = responses.stream()
.filter(r -> r.type().equals(AnthropicApi.EventType.MESSAGE_STOP.name()))
.toList();
assertThat(messageStopEvents).size().isEqualTo(1);
}
@Test
void chatCompletionStreamError() {
AnthropicMessage chatCompletionMessage = new AnthropicMessage(List.of(new ContentBlock("Tell me a Joke?")),