Fix streaming tool call merge in MessageAggregator

- Added mergeToolCalls() method to aggregate tool call fragments by ID
- Preserves name from first chunk and concatenates arguments
- Added MessageAggregatorQwenTest to reproduce Qwen streaming pattern

Signed-off-by: kimtaewoong <ktw2172@gmail.com>

Author: ultramancode

spring-ai-model/src/main/java/org/springframework/ai/chat/model/MessageAggregator.java

@@ -18,6 +18,7 @@
 
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
@@ -104,7 +105,7 @@ public Flux<ChatResponse> aggregate(Flux<ChatResponse> fluxChatResponse,
 				}
 				AssistantMessage outputMessage = chatResponse.getResult().getOutput();
 				if (!CollectionUtils.isEmpty(outputMessage.getToolCalls())) {
-					toolCallsRef.get().addAll(outputMessage.getToolCalls());
+					mergeToolCalls(toolCallsRef.get(), outputMessage.getToolCalls());
 				}
 
 			}
@@ -188,6 +189,51 @@ public Flux<ChatResponse> aggregate(Flux<ChatResponse> fluxChatResponse,
 		}).doOnError(e -> logger.error("Aggregation Error", e));
 	}
 
+	/**
+	 * Merge tool calls by id to handle streaming responses where tool call data is split
+	 * across multiple chunks. This is common in OpenAI-compatible APIs like Qwen, where
+	 * the first chunk contains the function name and subsequent chunks contain only
+	 * arguments.
+	 * @param existingToolCalls the list of existing tool calls to merge into
+	 * @param newToolCalls the new tool calls to merge
+	 */
+	private void mergeToolCalls(List<ToolCall> existingToolCalls, List<ToolCall> newToolCalls) {
+		Map<String, ToolCall> toolCallMap = new LinkedHashMap<>();
+
+		// Build map from existing tool calls
+		for (ToolCall existing : existingToolCalls) {
+			toolCallMap.put(existing.id(), existing);
+		}
+
+		// Merge new tool calls
+		for (ToolCall newCall : newToolCalls) {
+			if (toolCallMap.containsKey(newCall.id())) {
+				// Merge with existing tool call
+				ToolCall existing = toolCallMap.get(newCall.id());
+
+				// Use non-empty name, prefer new if both present
+				String mergedName = StringUtils.hasText(newCall.name()) ? newCall.name() : existing.name();
+
+				// Use non-empty type, prefer new if both present
+				String mergedType = StringUtils.hasText(newCall.type()) ? newCall.type() : existing.type();
+
+				// Concatenate arguments
+				String mergedArgs = (existing.arguments() != null ? existing.arguments() : "")
+						+ (newCall.arguments() != null ? newCall.arguments() : "");
+
+				toolCallMap.put(newCall.id(), new ToolCall(newCall.id(), mergedType, mergedName, mergedArgs));
+			}
+			else {
+				// New tool call
+				toolCallMap.put(newCall.id(), newCall);
+			}
+		}
+
+		// Update the existing list
+		existingToolCalls.clear();
+		existingToolCalls.addAll(toolCallMap.values());
+	}
+
 	public record DefaultUsage(Integer promptTokens, Integer completionTokens, Integer totalTokens) implements Usage {
 
 		@Override

spring-ai-model/src/test/java/org/springframework/ai/chat/model/MessageAggregatorQwenTest.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2023-2025 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.ai.chat.model;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+
+import org.springframework.ai.chat.messages.AssistantMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link MessageAggregator} with Qwen streaming tool call pattern.
+ * @author Taewoong Kim
+ */
+class MessageAggregatorQwenTest {
+
+	private final MessageAggregator messageAggregator = new MessageAggregator();
+
+	/**
+	 * Test based on actual Qwen streaming response from OpenRouter. Qwen sends tool name
+	 * only in the first chunk, subsequent chunks have empty name.
+	 */
+	@Test
+	void shouldMergeToolCallsFromQwenStreamingResponse() {
+		// Given: Qwen streaming chunks (actual pattern from curl test)
+		// Chunk 1: {"tool_calls":[{"id":"...","function":{"name":"getCurrentWeather"}}]}
+		ChatResponse chunk1 = new ChatResponse(List.of(new Generation(AssistantMessage.builder()
+			.toolCalls(List.of(new AssistantMessage.ToolCall("chatcmpl-tool-123", "function", "getCurrentWeather", "")))
+			.build())));
+
+		// Chunk 2: {"tool_calls":[{"index":0,"function":{"arguments":"{\"location\":
+		// \""}}]}
+		ChatResponse chunk2 = new ChatResponse(List.of(new Generation(AssistantMessage.builder()
+			.toolCalls(List.of(new AssistantMessage.ToolCall("chatcmpl-tool-123", "function", "", "{\"location\": \"")))
+			.build())));
+
+		// Chunk 3: {"tool_calls":[{"index":0,"function":{"arguments":"Se"}}]}
+		ChatResponse chunk3 = new ChatResponse(List.of(new Generation(AssistantMessage.builder()
+			.toolCalls(List.of(new AssistantMessage.ToolCall("chatcmpl-tool-123", "function", "", "Se")))
+			.build())));
+
+		// Chunk 4: {"tool_calls":[{"index":0,"function":{"arguments":"oul"}}]}
+		ChatResponse chunk4 = new ChatResponse(List.of(new Generation(AssistantMessage.builder()
+			.toolCalls(List.of(new AssistantMessage.ToolCall("chatcmpl-tool-123", "function", "", "oul")))
+			.build())));
+
+		// Chunk 5: {"tool_calls":[{"index":0,"function":{"arguments":"\"}"}}]}
+		ChatResponse chunk5 = new ChatResponse(List.of(new Generation(AssistantMessage.builder()
+			.toolCalls(List.of(new AssistantMessage.ToolCall("chatcmpl-tool-123", "function", "", "\"}")))
+			.build())));
+
+		Flux<ChatResponse> flux = Flux.just(chunk1, chunk2, chunk3, chunk4, chunk5);
+
+		// When: Aggregate the streaming responses
+		AtomicReference<ChatResponse> finalResponse = new AtomicReference<>();
+		this.messageAggregator.aggregate(flux, finalResponse::set).blockLast();
+
+		// Then: Verify the tool call was properly merged
+		assertThat(finalResponse.get()).isNotNull();
+		List<AssistantMessage.ToolCall> toolCalls = finalResponse.get().getResult().getOutput().getToolCalls();
+
+		assertThat(toolCalls).hasSize(1);
+		AssistantMessage.ToolCall mergedToolCall = toolCalls.get(0);
+
+		// The bug was: toolName would be empty string, causing "toolName cannot be null
+		// or empty"
+		// After fix: name should be preserved from first chunk
+		assertThat(mergedToolCall.name()).isEqualTo("getCurrentWeather");
+		assertThat(mergedToolCall.arguments()).isEqualTo("{\"location\": \"Seoul\"}");
+	}
+
+}