diff --git a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/MessageConverter.java b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/MessageConverter.java index 71ef570d8..e59b7a6aa 100644 --- a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/MessageConverter.java +++ b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/MessageConverter.java @@ -45,7 +45,6 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; -import java.util.stream.Collectors; import org.jspecify.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -321,7 +320,8 @@ public LlmResponse toLlmResponse(@Nullable ChatResponse chatResponse, boolean is } Content content = Content.builder().role("model").parts(mergedParts).build(); - boolean isPartial = isStreaming && isPartialResponse(mergedParts); + // Streamed chunks are partial; SpringAI emits the aggregated final on stream completion. + boolean isPartial = isStreaming; boolean isTurnComplete = !isStreaming || isTurnCompleteFinishReason(firstFinishReason); responseBuilder.content(content).partial(isPartial).turnComplete(isTurnComplete); @@ -680,19 +680,6 @@ private int nullSafeInt(@Nullable Integer value) { return value != null ? value.intValue() : 0; } - /** - * Determines if the merged parts of a streamed candidate are partial: they carry no function call - * and their text does not end with sentence punctuation. - */ - private static boolean isPartialResponse(List parts) { - if (parts.stream().anyMatch(part -> part.functionCall().isPresent())) { - return false; - } - String text = - parts.stream().map(part -> part.text().orElse("")).collect(Collectors.joining()).trim(); - return !text.isEmpty() && !text.endsWith(".") && !text.endsWith("!") && !text.endsWith("?"); - } - /** Determines if a candidate's finish reason indicates the turn is complete. */ private boolean isTurnCompleteFinishReason(@Nullable String finishReason) { if (finishReason == null) { diff --git a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/SpringAI.java b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/SpringAI.java index ece9b1e2f..5767ad510 100644 --- a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/SpringAI.java +++ b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/SpringAI.java @@ -204,6 +204,10 @@ private Flowable generateStreamingContent(LlmRequest llmRequest) { Prompt prompt = messageConverter.toLlmPrompt(llmRequest, resolveDefaultOptions()); observabilityHandler.logRequest(prompt.toString(), model()); + // Per-stream aggregator: accumulates chunk text / tool calls and produces the + // single final response on stream completion. + StreamingResponseAggregator aggregator = new StreamingResponseAggregator(); + Flux responseFlux = streamingChatModel.stream(prompt); responseFlux @@ -221,6 +225,7 @@ private Flowable generateStreamingContent(LlmRequest llmRequest) { // Use enhanced streaming-aware conversion LlmResponse llmResponse = messageConverter.toLlmResponse(chatResponse, true); + aggregator.processStreamingResponse(llmResponse); emitter.onNext(llmResponse); } catch (Exception e) { observabilityHandler.recordError(context, e); @@ -240,6 +245,11 @@ private Flowable generateStreamingContent(LlmRequest llmRequest) { () -> { // Record success for streaming completion observabilityHandler.recordSuccess(context, 0, 0, 0); + // The final (persisted) response is emitted exactly once, on stream + // completion. Always emit one final response, even an empty one: without + // it every event is partial, and BaseLlmFlow treats the turn as + // unfinished and re-calls the model until maxLlmCalls. + emitter.onNext(aggregator.getFinalResponse()); emitter.onComplete(); }); } catch (Exception e) { diff --git a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/StreamingResponseAggregator.java b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/StreamingResponseAggregator.java index da0710493..d7a272557 100644 --- a/contrib/spring-ai/src/main/java/com/google/adk/models/springai/StreamingResponseAggregator.java +++ b/contrib/spring-ai/src/main/java/com/google/adk/models/springai/StreamingResponseAggregator.java @@ -17,6 +17,8 @@ import com.google.adk.models.LlmResponse; import com.google.genai.types.Content; +import com.google.genai.types.FinishReason; +import com.google.genai.types.GenerateContentResponseUsageMetadata; import com.google.genai.types.Part; import java.util.ArrayList; import java.util.List; @@ -36,11 +38,32 @@ */ public class StreamingResponseAggregator { - private final StringBuffer textAccumulator = new StringBuffer(); + private final List textParts = new CopyOnWriteArrayList<>(); + + /** Tracks the thought flag of the part currently being accumulated (Kotlin parity). */ + private Boolean currentThoughtFlag = null; + + /** + * Latest non-empty usage metadata seen on the stream, carried on the final response. Visibility + * is guaranteed by {@code lock}: written in {@code processStreamingResponse} and read in {@code + * getFinalResponse()}, both under the write lock. + */ + private GenerateContentResponseUsageMetadata usageMetadata; + private final List toolCallParts = new CopyOnWriteArrayList<>(); + + /** Non-text, non-function-call parts (inlineData/fileData) passed through as-is. */ + private final List otherParts = new CopyOnWriteArrayList<>(); + private final ReadWriteLock lock = new ReentrantReadWriteLock(); private volatile boolean isFirstResponse = true; + /** Last finish reason seen on the stream; carried on the final, non-STOP becomes an error. */ + private FinishReason finishReason; + + private String errorMessage; + private String modelVersion; + /** * Processes a streaming LlmResponse and returns the current aggregated state. * @@ -59,21 +82,36 @@ public LlmResponse processStreamingResponse(LlmResponse response) { lock.writeLock().lock(); try { - // Process each part in the response + response.usageMetadata().ifPresent(usage -> this.usageMetadata = usage); + response.finishReason().ifPresent(reason -> this.finishReason = reason); + response.errorMessage().ifPresent(message -> this.errorMessage = message); + response.modelVersion().ifPresent(version -> this.modelVersion = version); + + // Process each part in the response. Text is grouped into parts by its thought flag + // (Kotlin parity): reasoning and answer must not merge into one part, or reasoning is + // persisted as answer text and replayed to the model on the next turn. for (Part part : content.parts().get()) { - if (part.text().isPresent()) { - textAccumulator.append(part.text().get()); - } else if (part.functionCall().isPresent()) { + if (part.functionCall().isPresent()) { // Tool calls are typically complete in each response toolCallParts.add(part); + } else if (part.text().isPresent()) { + boolean thought = part.thought().orElse(false); + if (currentThoughtFlag != null && currentThoughtFlag == thought && !textParts.isEmpty()) { + Part last = textParts.get(textParts.size() - 1); + textParts.set( + textParts.size() - 1, withText(last, last.text().orElse("") + part.text().get())); + } else { + textParts.add(withThought(Part.fromText(part.text().get()), thought)); + currentThoughtFlag = thought; + } + } else { + // inlineData/fileData and other non-text parts pass through as-is + otherParts.add(part); } } // Create aggregated content - List aggregatedParts = new ArrayList<>(); - if (textAccumulator.length() > 0) { - aggregatedParts.add(Part.fromText(textAccumulator.toString())); - } + List aggregatedParts = new ArrayList<>(textParts); aggregatedParts.addAll(toolCallParts); Content aggregatedContent = Content.builder().role("model").parts(aggregatedParts).build(); @@ -104,20 +142,42 @@ public LlmResponse processStreamingResponse(LlmResponse response) { public LlmResponse getFinalResponse() { lock.writeLock().lock(); try { - List finalParts = new ArrayList<>(); - if (textAccumulator.length() > 0) { - finalParts.add(Part.fromText(textAccumulator.toString())); - } + List finalParts = new ArrayList<>(textParts); finalParts.addAll(toolCallParts); + finalParts.addAll(otherParts); Content finalContent = Content.builder().role("model").parts(finalParts).build(); - LlmResponse finalResponse = - LlmResponse.builder().content(finalContent).partial(false).turnComplete(true).build(); + LlmResponse.Builder finalBuilder = + LlmResponse.builder().content(finalContent).partial(false).turnComplete(true); + // Carry the last finish reason: a non-STOP stream is an error turn, not a clean one + // (Kotlin aggregator / Python LiteLLM parity). + if (finishReason != null) { + finalBuilder.finishReason(finishReason); + if (finishReason.knownEnum() != FinishReason.Known.STOP) { + finalBuilder.errorCode(finishReason); + if (errorMessage == null && finishReason.knownEnum() == FinishReason.Known.MAX_TOKENS) { + errorMessage = "The generation ran out of tokens before completing."; + } + } + } + if (errorMessage != null) { + finalBuilder.errorMessage(errorMessage); + } + if (modelVersion != null) { + finalBuilder.modelVersion(modelVersion); + } + LlmResponse finalResponse = finalBuilder.usageMetadata(usageMetadata).build(); // Reset internal state without calling reset() to avoid nested locking - textAccumulator.setLength(0); + textParts.clear(); + currentThoughtFlag = null; + usageMetadata = null; toolCallParts.clear(); + otherParts.clear(); + finishReason = null; + errorMessage = null; + modelVersion = null; isFirstResponse = true; return finalResponse; @@ -130,8 +190,14 @@ public LlmResponse getFinalResponse() { public void reset() { lock.writeLock().lock(); try { - textAccumulator.setLength(0); + textParts.clear(); + currentThoughtFlag = null; + usageMetadata = null; toolCallParts.clear(); + otherParts.clear(); + finishReason = null; + errorMessage = null; + modelVersion = null; isFirstResponse = true; } finally { lock.writeLock().unlock(); @@ -142,7 +208,7 @@ public void reset() { public boolean isEmpty() { lock.readLock().lock(); try { - return textAccumulator.length() == 0 && toolCallParts.isEmpty(); + return textParts.isEmpty() && toolCallParts.isEmpty() && otherParts.isEmpty(); } finally { lock.readLock().unlock(); } @@ -152,9 +218,34 @@ public boolean isEmpty() { public int getAccumulatedTextLength() { lock.readLock().lock(); try { - return textAccumulator.length(); + return textParts.stream().mapToInt(part -> part.text().map(String::length).orElse(0)).sum(); } finally { lock.readLock().unlock(); } } + + /** Replaces a part's text, preserving its thought flag and signature. */ + private static Part withText(Part part, String text) { + Part.Builder builder = Part.builder().text(text); + if (part.thought().isPresent()) { + builder.thought(part.thought().get()); + } + if (part.thoughtSignature().isPresent()) { + builder.thoughtSignature(part.thoughtSignature().get()); + } + return builder.build(); + } + + /** Returns the part with the thought flag set (Kotlin parity: reasoning segments). */ + private static Part withThought(Part part, boolean thought) { + if (!thought) { + return part; + } + Part.Builder builder = Part.builder().text(part.text().orElse("")); + builder.thought(true); + if (part.thoughtSignature().isPresent()) { + builder.thoughtSignature(part.thoughtSignature().get()); + } + return builder.build(); + } } diff --git a/contrib/spring-ai/src/test/java/com/google/adk/models/springai/MessageConverterTest.java b/contrib/spring-ai/src/test/java/com/google/adk/models/springai/MessageConverterTest.java index 404e1bace..7312c7cec 100644 --- a/contrib/spring-ai/src/test/java/com/google/adk/models/springai/MessageConverterTest.java +++ b/contrib/spring-ai/src/test/java/com/google/adk/models/springai/MessageConverterTest.java @@ -532,8 +532,10 @@ void testToLlmResponseNonStreamingMode() { } @Test - void testPartialResponseDetection() { - // Test partial response (no punctuation ending) + void testToLlmResponseStreamingAlwaysMarksChunksPartial() { + // Streamed chunks are always partial regardless of punctuation: the final (persisted) + // response is derived from the stream completion signal, not mid-stream guessing. + // Punctuation-based classification dropped turns ending with CJK terminal punctuation. AssistantMessage partialMessage = new AssistantMessage("I am thinking"); Generation partialGeneration = new Generation(partialMessage); ChatResponse partialResponse = new ChatResponse(List.of(partialGeneration)); @@ -541,13 +543,12 @@ void testPartialResponseDetection() { LlmResponse partialLlmResponse = messageConverter.toLlmResponse(partialResponse, true); assertThat(partialLlmResponse.partial()).contains(true); - // Test complete response (ends with punctuation) AssistantMessage completeMessage = new AssistantMessage("I am done."); Generation completeGeneration = new Generation(completeMessage); ChatResponse completeResponse = new ChatResponse(List.of(completeGeneration)); LlmResponse completeLlmResponse = messageConverter.toLlmResponse(completeResponse, true); - assertThat(completeLlmResponse.partial()).contains(false); + assertThat(completeLlmResponse.partial()).contains(true); } @Test @@ -1068,7 +1069,8 @@ void toLlmResponse_streamingThoughtThenToolCall_isCompleteNotPartial() { LlmResponse response = messageConverter.toLlmResponse(new ChatResponse(List.of(thoughtGen, toolCallGen)), true); - assertThat(response.partial()).contains(false); + // Streamed chunks are always partial; the aggregated final is emitted on completion. + assertThat(response.partial()).contains(true); assertThat(response.turnComplete()).contains(true); assertThat(response.finishReason()).contains(new FinishReason(FinishReason.Known.STOP)); } @@ -1095,7 +1097,8 @@ void toLlmResponse_streamingGenerationWithoutOutput_doesNotThrow() { LlmResponse response = messageConverter.toLlmResponse(new ChatResponse(List.of(new Generation(null))), true); - assertThat(response.partial()).contains(false); + // Streamed chunks are always partial; the aggregated final is emitted on completion. + assertThat(response.partial()).contains(true); assertThat(response.content().orElseThrow().parts().orElseThrow()).isEmpty(); } diff --git a/contrib/spring-ai/src/test/java/com/google/adk/models/springai/SpringAITest.java b/contrib/spring-ai/src/test/java/com/google/adk/models/springai/SpringAITest.java index 51fe60abb..b4674fcef 100644 --- a/contrib/spring-ai/src/test/java/com/google/adk/models/springai/SpringAITest.java +++ b/contrib/spring-ai/src/test/java/com/google/adk/models/springai/SpringAITest.java @@ -24,14 +24,19 @@ import com.google.adk.models.LlmRequest; import com.google.adk.models.LlmResponse; import com.google.genai.types.Content; +import com.google.genai.types.FinishReason; import com.google.genai.types.Part; import io.reactivex.rxjava3.subscribers.TestSubscriber; import java.time.Duration; import java.util.List; +import java.util.Map; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.ai.chat.messages.AssistantMessage; +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.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.model.Generation; @@ -149,16 +154,229 @@ void testGenerateContentStreaming() { testObserver.awaitDone(5, TimeUnit.SECONDS); testObserver.assertComplete(); testObserver.assertNoErrors(); - testObserver.assertValueCount(3); + // 3 partial chunks + 1 aggregated final response on stream completion + testObserver.assertValueCount(4); List responses = testObserver.values(); - assertThat(responses).hasSize(3); + assertThat(responses).hasSize(4); // Verify each streaming response for (LlmResponse response : responses) { assertThat(response.content()).isPresent(); assertThat(response.content().get().parts()).isPresent(); } + + // The first 3 values are partial chunks; the last value is the aggregated final response + for (int i = 0; i < 3; i++) { + assertThat(responses.get(i).partial()).contains(true); + } + LlmResponse finalResponse = responses.get(3); + assertThat(finalResponse.partial()).contains(false); + assertThat(finalResponse.turnComplete()).contains(true); + assertThat(finalResponse.content().get().parts().get().get(0).text()) + .contains("I'm doing well!"); + } + + @Test + void testGenerateContentStreamingFinalCarriesUsageMetadata() { + // Usage metadata from the stream's last chunk must survive onto the aggregated final + // response (partial events are streamed but never persisted by the Runner). + AssistantMessage chunk1 = new AssistantMessage("Hello"); + AssistantMessage chunk2 = new AssistantMessage(" world"); + ChatResponse lastResponse = + new ChatResponse( + List.of(new Generation(chunk2)), + ChatResponseMetadata.builder() + .id("resp-usage") + .usage(new DefaultUsage(10, 4, 6)) + .build()); + Flux responseFlux = + Flux.just(new ChatResponse(List.of(new Generation(chunk1))), lastResponse); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + testObserver.assertValueCount(3); + + LlmResponse finalResponse = testObserver.values().get(2); + assertThat(finalResponse.partial()).contains(false); + // Assert the mapped counts (not just presence) to catch a mapping regression. + assertThat(finalResponse.usageMetadata()).isPresent(); + assertThat(finalResponse.usageMetadata().get().promptTokenCount()).contains(10); + assertThat(finalResponse.usageMetadata().get().candidatesTokenCount()).contains(4); + assertThat(finalResponse.usageMetadata().get().totalTokenCount()).contains(6); + } + + @Test + void testGenerateContentStreamingEmptyChunksEmitSingleFinal() { + // A stream of empty chunks must still yield exactly one partial=false response: + // without a final event the turn looks unfinished and BaseLlmFlow re-calls the + // model with the same history until maxLlmCalls. + AssistantMessage emptyChunk = new AssistantMessage(""); + Flux responseFlux = + Flux.just( + new ChatResponse(List.of(new Generation(emptyChunk))), + new ChatResponse(List.of(new Generation(emptyChunk)))); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + // 2 partial chunks + exactly 1 final + testObserver.assertValueCount(3); + List responses = testObserver.values(); + for (int i = 0; i < 2; i++) { + assertThat(responses.get(i).partial()).contains(true); + } + assertThat(responses.get(2).partial()).contains(false); + assertThat(responses.get(2).turnComplete()).contains(true); + } + + @Test + void testGenerateContentStreamingThoughtThenAnswerKeptAsSeparateParts() { + // Reasoning must stay thought-flagged parts separate from the answer (Kotlin parity): + // merged into one text part, reasoning is persisted as answer text and replayed to the + // model on the next turn. + AssistantMessage thoughtChunk = + AssistantMessage.builder() + .content("reasoning step") + .properties(Map.of("isThought", true)) + .build(); + AssistantMessage answerChunk = new AssistantMessage("The answer."); + Flux responseFlux = + Flux.just( + new ChatResponse(List.of(new Generation(thoughtChunk))), + new ChatResponse(List.of(new Generation(answerChunk)))); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + + LlmResponse finalResponse = testObserver.values().get(testObserver.values().size() - 1); + List finalParts = finalResponse.content().get().parts().get(); + // two separate text parts: thought segment first, answer second + assertThat(finalParts.size()).isGreaterThanOrEqualTo(2); + assertThat(finalParts.get(0).thought()).contains(true); + assertThat(finalParts.get(0).text()).contains("reasoning step"); + assertThat(finalParts.get(1).thought()).isEmpty(); + assertThat(finalParts.get(1).text()).contains("The answer."); + } + + @Test + void testGenerateContentStreamingMaxTokensCarriesErrorOnFinal() { + // A MAX_TOKENS stream must not be persisted as a clean turn: the final carries the + // finish reason and the error code (Kotlin aggregator / Python LiteLLM parity). + AssistantMessage chunk = new AssistantMessage("partial answer"); + ChatGenerationMetadata maxTokens = + ChatGenerationMetadata.builder().finishReason("LENGTH").build(); + Flux responseFlux = + Flux.just(new ChatResponse(List.of(new Generation(chunk, maxTokens)))); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + + LlmResponse finalResponse = testObserver.values().get(testObserver.values().size() - 1); + assertThat(finalResponse.partial()).contains(false); + assertThat(finalResponse.finishReason()).isPresent(); + assertThat(finalResponse.finishReason().get().knownEnum()) + .isEqualTo(FinishReason.Known.MAX_TOKENS); + assertThat(finalResponse.errorCode()).isPresent(); + } + + @Test + void testGenerateContentStreamingMixedTextAndToolCallTurn() { + // A text + tool-call turn: the aggregated final must carry both the full text and the + // accumulated function call (checked by review), with the tool call intact. + AssistantMessage textChunk = new AssistantMessage("Checking "); + AssistantMessage textChunk2 = new AssistantMessage("the logs now."); + AssistantMessage toolCallChunk = + AssistantMessage.builder() + .content("") + .toolCalls( + List.of( + new AssistantMessage.ToolCall( + "call-1", "function", "queryLog", "{\"kw\":\"x\"}"))) + .build(); + Flux responseFlux = + Flux.just( + new ChatResponse(List.of(new Generation(textChunk))), + new ChatResponse(List.of(new Generation(textChunk2))), + new ChatResponse(List.of(new Generation(toolCallChunk)))); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + // 3 partial chunks + 1 aggregated final + testObserver.assertValueCount(4); + + LlmResponse finalResponse = testObserver.values().get(3); + assertThat(finalResponse.partial()).contains(false); + List finalParts = finalResponse.content().get().parts().get(); + // full text first, then the accumulated tool call + assertThat(finalParts.get(0).text()).contains("Checking the logs now."); + assertThat(finalParts.get(1).functionCall()).isPresent(); + assertThat(finalParts.get(1).functionCall().get().name()).contains("queryLog"); + } + + @Test + void testGenerateContentStreamingCjkTerminalPunctuationPersistsFinal() { + // Regression for CJK turns: a response ending with CJK terminal punctuation must still + // produce exactly one final (persisted) event, emitted on stream completion. + Flux responseFlux = + Flux.just( + createStreamingChatResponse("今天天气"), + createStreamingChatResponse("不错，适合出门。"), + createStreamingChatResponse("记得带伞。")); + + when(mockStreamingChatModel.stream(any(Prompt.class))).thenReturn(responseFlux); + + SpringAI springAI = new SpringAI(mockStreamingChatModel); + + TestSubscriber testObserver = springAI.generateContent(testRequest, true).test(); + + testObserver.awaitDone(5, TimeUnit.SECONDS); + testObserver.assertComplete(); + testObserver.assertNoErrors(); + testObserver.assertValueCount(4); + + List responses = testObserver.values(); + for (int i = 0; i < 3; i++) { + assertThat(responses.get(i).partial()).contains(true); + } + LlmResponse finalResponse = responses.get(3); + assertThat(finalResponse.partial()).contains(false); + assertThat(finalResponse.content().get().parts().get().get(0).text()) + .contains("今天天气不错，适合出门。记得带伞。"); } @Test @@ -273,7 +491,8 @@ void testGenerateContentStreamingBackpressure() { testObserver.awaitDone(10, TimeUnit.SECONDS); testObserver.assertComplete(); testObserver.assertNoErrors(); - testObserver.assertValueCount(1000); + // 1000 partial chunks + 1 aggregated final response on stream completion + testObserver.assertValueCount(1001); } private ChatResponse createStreamingChatResponse(String text) {