From fa9ba1b920c513cad5d5d2146b26b6607e31d348 Mon Sep 17 00:00:00 2001 From: Igor Dayen Date: Sun, 19 Jul 2026 18:30:47 -0400 Subject: [PATCH 1/5] Add Streaming Capabiity Tutorial --- embabel-modules/embabel-streaming/README.md | 24 ++ .../embabel-streaming-anthropic/pom.xml | 48 ++++ ...WithThinkingAndToolingIntegrationTest.java | 240 +++++++++++++++++ .../embabel-streaming-openai/pom.xml | 48 ++++ .../StreamingWithThinkingIntegrationTest.java | 242 ++++++++++++++++++ embabel-modules/embabel-streaming/pom.xml | 36 +++ embabel-modules/pom.xml | 1 + 7 files changed, 639 insertions(+) create mode 100644 embabel-modules/embabel-streaming/README.md create mode 100644 embabel-modules/embabel-streaming/embabel-streaming-anthropic/pom.xml create mode 100644 embabel-modules/embabel-streaming/embabel-streaming-anthropic/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java create mode 100644 embabel-modules/embabel-streaming/embabel-streaming-openai/pom.xml create mode 100644 embabel-modules/embabel-streaming/embabel-streaming-openai/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingIntegrationTest.java create mode 100644 embabel-modules/embabel-streaming/pom.xml diff --git a/embabel-modules/embabel-streaming/README.md b/embabel-modules/embabel-streaming/README.md new file mode 100644 index 000000000000..faf8511530a8 --- /dev/null +++ b/embabel-modules/embabel-streaming/README.md @@ -0,0 +1,24 @@ +# embabel-streaming + +Integration tests for Embabel Agent streaming with OpenAI and Anthropic. + +## Setup + +Export your API keys before running: + +```bash +export OPENAI_API_KEY=your-openai-key +export ANTHROPIC_API_KEY=your-anthropic-key +``` + +## Run + +**OpenAI tests** (gpt-4.1-mini): +```bash +mvn test -pl embabel-streaming-openai +``` + +**Anthropic tests** (claude-sonnet-4-5): +```bash +mvn test -pl embabel-streaming-anthropic +``` diff --git a/embabel-modules/embabel-streaming/embabel-streaming-anthropic/pom.xml b/embabel-modules/embabel-streaming/embabel-streaming-anthropic/pom.xml new file mode 100644 index 000000000000..fd394c68eaa3 --- /dev/null +++ b/embabel-modules/embabel-streaming/embabel-streaming-anthropic/pom.xml @@ -0,0 +1,48 @@ + + + 4.0.0 + + com.baeldung + embabel-streaming + 0.0.1 + + + embabel-streaming-anthropic + + + + com.embabel.agent + embabel-agent-starter-anthropic + ${embabel-agent.version} + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + ${maven.compiler.source} + ${maven.compiler.source} + ${maven.compiler.source} + + + + org.springframework.boot + spring-boot-maven-plugin + + + repackage + + true + + + + + + + + diff --git a/embabel-modules/embabel-streaming/embabel-streaming-anthropic/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java b/embabel-modules/embabel-streaming/embabel-streaming-anthropic/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java new file mode 100644 index 000000000000..eef9ac240365 --- /dev/null +++ b/embabel-modules/embabel-streaming/embabel-streaming-anthropic/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java @@ -0,0 +1,240 @@ +/* + * Example got patterned after: + * https://github.com/embabel/embabel-agent/blob/main/embabel-agent-autoconfigure/models/embabel-agent-anthropic-autoconfigure/src/test/java/com/embabel/agent/config/models/anthropic/LLMAnthropicStreamingBuilderIT.java + * + * Original code (see link above) was developed by Embabel Pty Ltd, 2026 + */ +package com.baeldung.embabel.agent.api.streaming; + +import com.embabel.agent.AgentTestApplication; +import com.embabel.agent.api.annotation.LlmTool; +import com.embabel.agent.api.common.Ai; +import com.embabel.agent.api.common.PromptRunner; +import com.embabel.agent.api.streaming.StreamingPromptRunnerBuilder; +import com.embabel.agent.api.tool.callback.LogLevel; +import com.embabel.agent.api.tool.callback.ToolCallLoggingInspector; +import com.embabel.common.ai.model.LlmOptions; +import com.embabel.common.ai.model.Thinking; +import com.embabel.common.core.streaming.StreamingEvent; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.ActiveProfiles; +import reactor.core.publisher.Flux; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.*; + +@SpringBootTest( + classes = AgentTestApplication.class, + properties = { + "embabel.models.cheapest=claude-sonnet-4-5", + "embabel.models.best=claude-sonnet-4-5", + "embabel.models.default-llm=claude-sonnet-4-5", + "embabel.agent.platform.llm-operations.prompts.defaultTimeout=240s", + "embabel.agent.platform.llm-operations.data-binding.fixedBackoffMillis=6000", + "embabel.agent.platform.scanning.annotation=false" + } +) +@ActiveProfiles("tool-reasoning") +class StreamingWithThinkingAndToolingIntegrationTest { + + private static final Logger logger = LoggerFactory.getLogger(StreamingWithThinkingAndToolingIntegrationTest.class); + + @Autowired + private Ai ai; + + @BeforeAll + static void setUp() { + System.setProperty("embabel.agent.shell.interactive.enabled", "false"); + } + + // TODO: move to embabel-tutorial-common once available + public record ParkingRecommendation( + String scenario, + Option chosenOption, + String location, + int estimatedTotalCost, + String summary + ) { + @SuppressWarnings("unused") + public enum Option { + STREET, METER, GARAGE + } + } + + // TODO: move to embabel-tutorial-common once available + static class ParkingTooling { + + @LlmTool(description = "Find free street parking. Uncertain and may take time.") + public String findStreetParking(String location, int maxMinutes) { + boolean found = ThreadLocalRandom.current().nextDouble() < 0.3; + if (found) { + return "Street parking found near " + location + " (free)"; + } + return "No street parking found within " + maxMinutes + " minutes"; + } + + @LlmTool(description = "Find metered parking. Moderate cost and moderate availability. May have time limits.") + public String findMeterParking(String location, int maxMinutes) { + boolean found = ThreadLocalRandom.current().nextDouble() < 0.6; + if (found) { + return "Metered parking found near " + location + " ($5/hour, 2-hour limit)"; + } + return "No metered parking found within " + maxMinutes + " minutes"; + } + + @LlmTool(description = "Reserve guaranteed garage parking near destination.") + public String reserveGarage(String location) { + return "Garage reserved near " + location + " ($30/hour, guaranteed)"; + } + } + + private static final String PARKING_PROMPT = + """ + An advisor is driving to a client meeting in Midtown Manhattan. + Constraints: 30 minutes until the meeting, meeting lasts 3 hours. + Options: street parking (free, uncertain), metered ($5/hr, 2-hour limit), garage ($30/hr, guaranteed). + Recommend the best parking option. + """; + + private static final String TOOLING_PROMPT = + """ + An advisor needs to park in Midtown Manhattan for a 3-hour client meeting starting in 30 minutes. + Use the available tools to probe parking options, then recommend the best one. + Arriving late is not acceptable. + """; + + /** + * Verifies structured streaming output with thinking enabled and no tools. + * + *

{@link Thinking#withTokenBudget(int)} sets {@code thinkingEnabled=true} in the streaming + * converter, sending {@code } format instructions in the system prompt. + * Budget must be less than {@code max_tokens}=8192 for {@code claude-sonnet-4-5}. + * + *

The streaming pipeline emits one {@link com.embabel.common.core.streaming.StreamingEvent.Thinking} + * event per line, so expect multiple reasoning events rather than a single block. + */ + @Nested + class StreamingWithThinkingNoTools { + + @Test + void whenStreamingWithThinking_thenReceivesReasoningAndRecommendation() { + // budget_tokens must be < max_tokens (8192). This also enables format instructions. + LlmOptions thinkingOptions = new LlmOptions().withThinking(Thinking.withTokenBudget(8000)); + PromptRunner runner = ai.withDefaultLlm().withLlm(thinkingOptions); + assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); + + List received = new CopyOnWriteArrayList<>(); + List reasoning = new CopyOnWriteArrayList<>(); + AtomicReference errorOccurred = new AtomicReference<>(); + AtomicBoolean completionCalled = new AtomicBoolean(false); + + Flux> stream = new StreamingPromptRunnerBuilder(runner) + .streaming() + .withPrompt(PARKING_PROMPT) + .createObjectStreamWithThinking(ParkingRecommendation.class); + + stream + .timeout(Duration.ofSeconds(120)) + .doOnNext(event -> { + if (event.isObject()) { + ParkingRecommendation rec = event.getObject(); + if (rec != null) { + received.add(rec); + logger.info("Received recommendation: option={}, cost={}, summary={}", + rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); + } + } else if (event.isThinking()) { + reasoning.add(event.getThinking()); + logger.info("Received reasoning: {}", event.getThinking()); + } + }) + .doOnError(error -> { + errorOccurred.set(error); + logger.error("Stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> { + completionCalled.set(true); + logger.info("Stream completed: {} recommendations, {} reasoning blocks", + received.size(), reasoning.size()); + }) + .blockLast(Duration.ofSeconds(240)); + + assertNull(errorOccurred.get(), "Streaming should not produce errors"); + assertTrue(completionCalled.get(), "Stream should complete successfully"); + assertFalse(received.isEmpty(), "Should receive at least one parking recommendation"); + assertFalse(reasoning.isEmpty(), "Should receive reasoning blocks"); + } + } + + @Nested + class StreamingWithThinkingAndTooling { + + /** + * Verifies that streaming with tools produces reasoning and a recommendation informed by tool results. + * + *

Reasoning events are emitted only from the final LLM iteration — after all tool calls + * complete. Intermediate reasoning (the model's thinking while deciding which tools to call) + * is not surfaced to the subscriber, as each tool-loop iteration replaces the previous stream. + */ + @Test + void whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning() { + PromptRunner runner = ai.withDefaultLlm() + .withToolObject(new ParkingTooling()) + .withToolCallInspectors(new ToolCallLoggingInspector(LogLevel.INFO, logger)); + assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); + + List received = new CopyOnWriteArrayList<>(); + List reasoning = new CopyOnWriteArrayList<>(); + AtomicReference errorOccurred = new AtomicReference<>(); + AtomicBoolean completionCalled = new AtomicBoolean(false); + + Flux> stream = new StreamingPromptRunnerBuilder(runner) + .streaming() + .withPrompt(TOOLING_PROMPT) + .createObjectStreamWithThinking(ParkingRecommendation.class); + + stream + .timeout(Duration.ofSeconds(120)) + .doOnNext(event -> { + if (event.isObject()) { + ParkingRecommendation rec = event.getObject(); + if (rec != null) { + received.add(rec); + logger.info("Received recommendation: option={}, cost={}, summary={}", + rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); + } + } else if (event.isThinking()) { + reasoning.add(event.getThinking()); + logger.info("Received reasoning: {}", event.getThinking()); + } + }) + .doOnError(error -> { + errorOccurred.set(error); + logger.error("Stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> { + completionCalled.set(true); + logger.info("Stream completed: {} recommendations, {} reasoning blocks", + received.size(), reasoning.size()); + }) + .blockLast(Duration.ofSeconds(240)); + + assertNull(errorOccurred.get(), "Streaming should not produce errors"); + assertTrue(completionCalled.get(), "Stream should complete successfully"); + assertFalse(received.isEmpty(), "Should receive at least one parking recommendation"); + assertFalse(reasoning.isEmpty(), "Should receive reasoning blocks alongside tool results"); + } + } +} diff --git a/embabel-modules/embabel-streaming/embabel-streaming-openai/pom.xml b/embabel-modules/embabel-streaming/embabel-streaming-openai/pom.xml new file mode 100644 index 000000000000..5df83595ee52 --- /dev/null +++ b/embabel-modules/embabel-streaming/embabel-streaming-openai/pom.xml @@ -0,0 +1,48 @@ + + + 4.0.0 + + com.baeldung + embabel-streaming + 0.0.1 + + + embabel-streaming-openai + + + + com.embabel.agent + embabel-agent-starter-openai + ${embabel-agent.version} + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + ${maven.compiler.source} + ${maven.compiler.source} + ${maven.compiler.source} + + + + org.springframework.boot + spring-boot-maven-plugin + + + repackage + + true + + + + + + + + diff --git a/embabel-modules/embabel-streaming/embabel-streaming-openai/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingIntegrationTest.java b/embabel-modules/embabel-streaming/embabel-streaming-openai/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingIntegrationTest.java new file mode 100644 index 000000000000..ba27ec4be071 --- /dev/null +++ b/embabel-modules/embabel-streaming/embabel-streaming-openai/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingIntegrationTest.java @@ -0,0 +1,242 @@ +/* + * Example got patterned after: + * https://github.com/embabel/embabel-agent/blob/main/embabel-agent-autoconfigure/models/embabel-agent-openai-autoconfigure/src/test/java/com/embabel/agent/config/models/openai/LLMOpenAiStreamingBuilderIT.java + * + * Original code (see link above) was developed by Embabel Pty Ltd, 2026 + */ +package com.baeldung.embabel.agent.api.streaming; + +import com.embabel.agent.AgentTestApplication; +import com.embabel.agent.api.common.Ai; +import com.embabel.agent.api.common.PromptRunner; +import com.embabel.agent.api.streaming.StreamingPromptRunnerBuilder; +import com.embabel.common.ai.model.LlmOptions; +import com.embabel.common.ai.model.Thinking; +import com.embabel.common.core.streaming.StreamingEvent; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.ActiveProfiles; +import reactor.core.publisher.Flux; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.*; + +@SpringBootTest( + classes = AgentTestApplication.class, + properties = { + "embabel.models.cheapest=gpt-4.1-mini", + "embabel.models.best=gpt-4.1-mini", + "embabel.models.default-llm=gpt-4.1-mini", + "embabel.agent.platform.llm-operations.prompts.defaultTimeout=240s", + "embabel.agent.platform.llm-operations.data-binding.fixedBackoffMillis=6000", + "embabel.agent.platform.scanning.annotation=false" + } +) +@ActiveProfiles("tool-reasoning") +class StreamingWithThinkingIntegrationTest { + + private static final Logger logger = LoggerFactory.getLogger(StreamingWithThinkingIntegrationTest.class); + + @Autowired + private Ai ai; + + @BeforeAll + static void setUp() { + System.setProperty("embabel.agent.shell.interactive.enabled", "false"); + } + + // TODO: move to embabel-tutorial-common once available + public record ParkingRecommendation( + String scenario, + Option chosenOption, + String location, + int estimatedTotalCost, + String summary + ) { + @SuppressWarnings("unused") + public enum Option { + STREET, METER, GARAGE + } + } + + private static final String PARKING_PROMPT = + """ + An advisor is driving to a client meeting in Midtown Manhattan. + Constraints: 30 minutes until the meeting, meeting lasts 3 hours. + Options: street parking (free, uncertain), metered ($5/hr, 2-hour limit), garage ($30/hr, guaranteed). + Recommend the best parking option. + """; + + private static final String TIMED_PARKING_PROMPT = + """ + Provide parking recommendations for an advisor visiting Midtown Manhattan across three time scenarios: + 1. Early morning (before 8am): street meters are free before 8am + 2. Business hours (9am-5pm): all paid options apply, 30-minute window, 3-hour stay + 3. Evening (after 6pm): meters are free after 6pm, but garages may close at 9pm (3-hour stay at risk) + + Return all three recommendations. + """; + + @Nested + class SimpleStreaming { + + /** Verifies that a single parking recommendation is received via streaming. */ + @Test + void whenStreaming_thenReceivesParkingRecommendation() { + PromptRunner runner = ai.withDefaultLlm(); + assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); + + List received = new CopyOnWriteArrayList<>(); + AtomicReference errorOccurred = new AtomicReference<>(); + AtomicBoolean completionCalled = new AtomicBoolean(false); + + Flux stream = new StreamingPromptRunnerBuilder(runner) + .streaming() + .withPrompt(PARKING_PROMPT) + .createObjectStream(ParkingRecommendation.class); + + stream + .timeout(Duration.ofSeconds(120)) + .doOnNext(rec -> { + received.add(rec); + logger.info("Received parking recommendation: option={}, cost={}, summary={}", + rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); + }) + .doOnError(error -> { + errorOccurred.set(error); + logger.error("Stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> { + completionCalled.set(true); + logger.info("Stream completed successfully"); + }) + .blockLast(Duration.ofSeconds(240)); + + assertNull(errorOccurred.get(), "Streaming should not produce errors"); + assertTrue(completionCalled.get(), "Stream should complete successfully"); + assertFalse(received.isEmpty(), "Should receive at least one parking recommendation"); + } + } + + /** + * Verifies text-instructed thinking on OpenAI via {@code } format tags. + * + *

{@link Thinking#withTokenBudget(int)} sets {@code thinkingEnabled=true} in the streaming + * converter, which includes {@code } format instructions in the system prompt. + * OpenAI ignores the budget parameter — there is no native thinking API call. + * + *

{@code gpt-4.1-mini} follows the instructions and outputs reasoning wrapped in + * {@code ...} tags. When emitted as a single line, tags are stripped and + * one clean reasoning event is produced. When emitted as multiple lines, {@code } + * appears in the first event and {@code } in the last — tags are not stripped in + * that case. The number of events and whether tags appear depends on how the model chunks + * its output. + * + *

Note: {@code gpt-5.4-mini} does not follow {@code } format instructions and + * produces 0 reasoning blocks. + */ + @Nested + class StreamingWithInstructedThinking { + + @Test + void whenStreamingWithInstructedThinking_thenReceivesReasoningBlocks() { + // withTokenBudget sets thinkingEnabled=true → sends format instructions. + // OpenAI ignores the budget; gpt-4.1-mini outputs multi-line reasoning inside tags. + LlmOptions thinkingOptions = new LlmOptions().withThinking(Thinking.withTokenBudget(10000)); + PromptRunner runner = ai.withDefaultLlm().withLlm(thinkingOptions); + assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); + + List received = new CopyOnWriteArrayList<>(); + List reasoning = new CopyOnWriteArrayList<>(); + AtomicReference errorOccurred = new AtomicReference<>(); + AtomicBoolean completionCalled = new AtomicBoolean(false); + + Flux> stream = new StreamingPromptRunnerBuilder(runner) + .streaming() + .withPrompt(PARKING_PROMPT) + .createObjectStreamWithThinking(ParkingRecommendation.class); + + stream + .timeout(Duration.ofSeconds(120)) + .doOnNext(event -> { + if (event.isObject()) { + ParkingRecommendation rec = event.getObject(); + if (rec != null) { + received.add(rec); + logger.info("Received recommendation: option={}, cost={}, summary={}", + rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); + } + } else if (event.isThinking()) { + reasoning.add(event.getThinking()); + logger.info("Received reasoning: {}", event.getThinking()); + } + }) + .doOnError(error -> { + errorOccurred.set(error); + logger.error("Stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> { + completionCalled.set(true); + logger.info("Stream completed: {} recommendations, {} reasoning blocks", + received.size(), reasoning.size()); + }) + .blockLast(Duration.ofSeconds(240)); + + assertNull(errorOccurred.get(), "Streaming should not produce errors"); + assertTrue(completionCalled.get(), "Stream should complete successfully"); + assertFalse(received.isEmpty(), "Should receive at least one parking recommendation"); + assertFalse(reasoning.isEmpty(), "Should receive reasoning blocks"); + } + } + + @Nested + class StreamingCollection { + + /** Verifies that multiple parking recommendations stream in, one per time scenario. */ + @Test + void whenStreamingMultipleScenarios_thenReceivesRecommendationPerScenario() { + PromptRunner runner = ai.withDefaultLlm(); + assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); + + List received = new CopyOnWriteArrayList<>(); + AtomicReference errorOccurred = new AtomicReference<>(); + AtomicBoolean completionCalled = new AtomicBoolean(false); + + Flux stream = new StreamingPromptRunnerBuilder(runner) + .streaming() + .withPrompt(TIMED_PARKING_PROMPT) + .createObjectStream(ParkingRecommendation.class); + + stream + .timeout(Duration.ofSeconds(120)) + .doOnNext(rec -> { + received.add(rec); + logger.info("Received recommendation: scenario={}, option={}, cost={}", + rec.scenario(), rec.chosenOption(), rec.estimatedTotalCost()); + }) + .doOnError(error -> { + errorOccurred.set(error); + logger.error("Stream error: {}", error.getMessage()); + }) + .doOnComplete(() -> { + completionCalled.set(true); + logger.info("Stream completed with {} recommendations", received.size()); + }) + .blockLast(Duration.ofSeconds(240)); + + assertNull(errorOccurred.get(), "Streaming should not produce errors"); + assertTrue(completionCalled.get(), "Stream should complete successfully"); + assertTrue(received.size() >= 2, "Should receive multiple scenario recommendations, got: " + received.size()); + } + } +} diff --git a/embabel-modules/embabel-streaming/pom.xml b/embabel-modules/embabel-streaming/pom.xml new file mode 100644 index 000000000000..f4e99e093fbb --- /dev/null +++ b/embabel-modules/embabel-streaming/pom.xml @@ -0,0 +1,36 @@ + + + 4.0.0 + + com.baeldung + embabel-modules + 0.0.1 + + + embabel-streaming + pom + + + 21 + 21 + UTF-8 + 1.0.0-SNAPSHOT + + + + + com.embabel.agent + embabel-agent-test-internal + ${embabel-agent.version} + test + + + + + embabel-streaming-openai + embabel-streaming-anthropic + + + diff --git a/embabel-modules/pom.xml b/embabel-modules/pom.xml index 258f429366b5..241720457454 100644 --- a/embabel-modules/pom.xml +++ b/embabel-modules/pom.xml @@ -18,6 +18,7 @@ embabel-quiz-generator embabel-tool-reasoning + embabel-streaming \ No newline at end of file From d4b3c34f02864011d9154a7f26ac58ae50735f40 Mon Sep 17 00:00:00 2001 From: Igor Dayen Date: Mon, 20 Jul 2026 08:59:47 -0400 Subject: [PATCH 2/5] Release upgrade --- embabel-modules/embabel-streaming/pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/embabel-modules/embabel-streaming/pom.xml b/embabel-modules/embabel-streaming/pom.xml index f4e99e093fbb..6a2fede93db0 100644 --- a/embabel-modules/embabel-streaming/pom.xml +++ b/embabel-modules/embabel-streaming/pom.xml @@ -16,7 +16,7 @@ 21 21 UTF-8 - 1.0.0-SNAPSHOT + 1.0.0 From 8accc31c831d2e186940cbbf6852a591986b883e Mon Sep 17 00:00:00 2001 From: Igor Dayen Date: Sat, 25 Jul 2026 23:18:39 -0400 Subject: [PATCH 3/5] Implement reviewer feedback --- embabel-modules/embabel-streaming/README.md | 9 +- .../embabel-streaming-anthropic/pom.xml | 48 ---- .../embabel-streaming-openai/pom.xml | 48 ---- .../StreamingWithThinkingIntegrationTest.java | 242 ------------------ embabel-modules/embabel-streaming/pom.xml | 36 ++- ...WithThinkingAndToolingIntegrationTest.java | 158 +++++++----- 6 files changed, 131 insertions(+), 410 deletions(-) delete mode 100644 embabel-modules/embabel-streaming/embabel-streaming-anthropic/pom.xml delete mode 100644 embabel-modules/embabel-streaming/embabel-streaming-openai/pom.xml delete mode 100644 embabel-modules/embabel-streaming/embabel-streaming-openai/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingIntegrationTest.java rename embabel-modules/embabel-streaming/{embabel-streaming-anthropic => }/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java (62%) diff --git a/embabel-modules/embabel-streaming/README.md b/embabel-modules/embabel-streaming/README.md index faf8511530a8..e6562324183a 100644 --- a/embabel-modules/embabel-streaming/README.md +++ b/embabel-modules/embabel-streaming/README.md @@ -7,18 +7,13 @@ Integration tests for Embabel Agent streaming with OpenAI and Anthropic. Export your API keys before running: ```bash -export OPENAI_API_KEY=your-openai-key export ANTHROPIC_API_KEY=your-anthropic-key ``` ## Run -**OpenAI tests** (gpt-4.1-mini): -```bash -mvn test -pl embabel-streaming-openai -``` - **Anthropic tests** (claude-sonnet-4-5): ```bash -mvn test -pl embabel-streaming-anthropic +mvn test -pl embabel-streaming -P integration + ``` diff --git a/embabel-modules/embabel-streaming/embabel-streaming-anthropic/pom.xml b/embabel-modules/embabel-streaming/embabel-streaming-anthropic/pom.xml deleted file mode 100644 index fd394c68eaa3..000000000000 --- a/embabel-modules/embabel-streaming/embabel-streaming-anthropic/pom.xml +++ /dev/null @@ -1,48 +0,0 @@ - - - 4.0.0 - - com.baeldung - embabel-streaming - 0.0.1 - - - embabel-streaming-anthropic - - - - com.embabel.agent - embabel-agent-starter-anthropic - ${embabel-agent.version} - - - - - - - org.apache.maven.plugins - maven-compiler-plugin - - ${maven.compiler.source} - ${maven.compiler.source} - ${maven.compiler.source} - - - - org.springframework.boot - spring-boot-maven-plugin - - - repackage - - true - - - - - - - - diff --git a/embabel-modules/embabel-streaming/embabel-streaming-openai/pom.xml b/embabel-modules/embabel-streaming/embabel-streaming-openai/pom.xml deleted file mode 100644 index 5df83595ee52..000000000000 --- a/embabel-modules/embabel-streaming/embabel-streaming-openai/pom.xml +++ /dev/null @@ -1,48 +0,0 @@ - - - 4.0.0 - - com.baeldung - embabel-streaming - 0.0.1 - - - embabel-streaming-openai - - - - com.embabel.agent - embabel-agent-starter-openai - ${embabel-agent.version} - - - - - - - org.apache.maven.plugins - maven-compiler-plugin - - ${maven.compiler.source} - ${maven.compiler.source} - ${maven.compiler.source} - - - - org.springframework.boot - spring-boot-maven-plugin - - - repackage - - true - - - - - - - - diff --git a/embabel-modules/embabel-streaming/embabel-streaming-openai/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingIntegrationTest.java b/embabel-modules/embabel-streaming/embabel-streaming-openai/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingIntegrationTest.java deleted file mode 100644 index ba27ec4be071..000000000000 --- a/embabel-modules/embabel-streaming/embabel-streaming-openai/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingIntegrationTest.java +++ /dev/null @@ -1,242 +0,0 @@ -/* - * Example got patterned after: - * https://github.com/embabel/embabel-agent/blob/main/embabel-agent-autoconfigure/models/embabel-agent-openai-autoconfigure/src/test/java/com/embabel/agent/config/models/openai/LLMOpenAiStreamingBuilderIT.java - * - * Original code (see link above) was developed by Embabel Pty Ltd, 2026 - */ -package com.baeldung.embabel.agent.api.streaming; - -import com.embabel.agent.AgentTestApplication; -import com.embabel.agent.api.common.Ai; -import com.embabel.agent.api.common.PromptRunner; -import com.embabel.agent.api.streaming.StreamingPromptRunnerBuilder; -import com.embabel.common.ai.model.LlmOptions; -import com.embabel.common.ai.model.Thinking; -import com.embabel.common.core.streaming.StreamingEvent; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Nested; -import org.junit.jupiter.api.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.test.context.ActiveProfiles; -import reactor.core.publisher.Flux; - -import java.time.Duration; -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; - -import static org.junit.jupiter.api.Assertions.*; - -@SpringBootTest( - classes = AgentTestApplication.class, - properties = { - "embabel.models.cheapest=gpt-4.1-mini", - "embabel.models.best=gpt-4.1-mini", - "embabel.models.default-llm=gpt-4.1-mini", - "embabel.agent.platform.llm-operations.prompts.defaultTimeout=240s", - "embabel.agent.platform.llm-operations.data-binding.fixedBackoffMillis=6000", - "embabel.agent.platform.scanning.annotation=false" - } -) -@ActiveProfiles("tool-reasoning") -class StreamingWithThinkingIntegrationTest { - - private static final Logger logger = LoggerFactory.getLogger(StreamingWithThinkingIntegrationTest.class); - - @Autowired - private Ai ai; - - @BeforeAll - static void setUp() { - System.setProperty("embabel.agent.shell.interactive.enabled", "false"); - } - - // TODO: move to embabel-tutorial-common once available - public record ParkingRecommendation( - String scenario, - Option chosenOption, - String location, - int estimatedTotalCost, - String summary - ) { - @SuppressWarnings("unused") - public enum Option { - STREET, METER, GARAGE - } - } - - private static final String PARKING_PROMPT = - """ - An advisor is driving to a client meeting in Midtown Manhattan. - Constraints: 30 minutes until the meeting, meeting lasts 3 hours. - Options: street parking (free, uncertain), metered ($5/hr, 2-hour limit), garage ($30/hr, guaranteed). - Recommend the best parking option. - """; - - private static final String TIMED_PARKING_PROMPT = - """ - Provide parking recommendations for an advisor visiting Midtown Manhattan across three time scenarios: - 1. Early morning (before 8am): street meters are free before 8am - 2. Business hours (9am-5pm): all paid options apply, 30-minute window, 3-hour stay - 3. Evening (after 6pm): meters are free after 6pm, but garages may close at 9pm (3-hour stay at risk) - - Return all three recommendations. - """; - - @Nested - class SimpleStreaming { - - /** Verifies that a single parking recommendation is received via streaming. */ - @Test - void whenStreaming_thenReceivesParkingRecommendation() { - PromptRunner runner = ai.withDefaultLlm(); - assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); - - List received = new CopyOnWriteArrayList<>(); - AtomicReference errorOccurred = new AtomicReference<>(); - AtomicBoolean completionCalled = new AtomicBoolean(false); - - Flux stream = new StreamingPromptRunnerBuilder(runner) - .streaming() - .withPrompt(PARKING_PROMPT) - .createObjectStream(ParkingRecommendation.class); - - stream - .timeout(Duration.ofSeconds(120)) - .doOnNext(rec -> { - received.add(rec); - logger.info("Received parking recommendation: option={}, cost={}, summary={}", - rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); - }) - .doOnError(error -> { - errorOccurred.set(error); - logger.error("Stream error: {}", error.getMessage()); - }) - .doOnComplete(() -> { - completionCalled.set(true); - logger.info("Stream completed successfully"); - }) - .blockLast(Duration.ofSeconds(240)); - - assertNull(errorOccurred.get(), "Streaming should not produce errors"); - assertTrue(completionCalled.get(), "Stream should complete successfully"); - assertFalse(received.isEmpty(), "Should receive at least one parking recommendation"); - } - } - - /** - * Verifies text-instructed thinking on OpenAI via {@code } format tags. - * - *

{@link Thinking#withTokenBudget(int)} sets {@code thinkingEnabled=true} in the streaming - * converter, which includes {@code } format instructions in the system prompt. - * OpenAI ignores the budget parameter — there is no native thinking API call. - * - *

{@code gpt-4.1-mini} follows the instructions and outputs reasoning wrapped in - * {@code ...} tags. When emitted as a single line, tags are stripped and - * one clean reasoning event is produced. When emitted as multiple lines, {@code } - * appears in the first event and {@code } in the last — tags are not stripped in - * that case. The number of events and whether tags appear depends on how the model chunks - * its output. - * - *

Note: {@code gpt-5.4-mini} does not follow {@code } format instructions and - * produces 0 reasoning blocks. - */ - @Nested - class StreamingWithInstructedThinking { - - @Test - void whenStreamingWithInstructedThinking_thenReceivesReasoningBlocks() { - // withTokenBudget sets thinkingEnabled=true → sends format instructions. - // OpenAI ignores the budget; gpt-4.1-mini outputs multi-line reasoning inside tags. - LlmOptions thinkingOptions = new LlmOptions().withThinking(Thinking.withTokenBudget(10000)); - PromptRunner runner = ai.withDefaultLlm().withLlm(thinkingOptions); - assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); - - List received = new CopyOnWriteArrayList<>(); - List reasoning = new CopyOnWriteArrayList<>(); - AtomicReference errorOccurred = new AtomicReference<>(); - AtomicBoolean completionCalled = new AtomicBoolean(false); - - Flux> stream = new StreamingPromptRunnerBuilder(runner) - .streaming() - .withPrompt(PARKING_PROMPT) - .createObjectStreamWithThinking(ParkingRecommendation.class); - - stream - .timeout(Duration.ofSeconds(120)) - .doOnNext(event -> { - if (event.isObject()) { - ParkingRecommendation rec = event.getObject(); - if (rec != null) { - received.add(rec); - logger.info("Received recommendation: option={}, cost={}, summary={}", - rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); - } - } else if (event.isThinking()) { - reasoning.add(event.getThinking()); - logger.info("Received reasoning: {}", event.getThinking()); - } - }) - .doOnError(error -> { - errorOccurred.set(error); - logger.error("Stream error: {}", error.getMessage()); - }) - .doOnComplete(() -> { - completionCalled.set(true); - logger.info("Stream completed: {} recommendations, {} reasoning blocks", - received.size(), reasoning.size()); - }) - .blockLast(Duration.ofSeconds(240)); - - assertNull(errorOccurred.get(), "Streaming should not produce errors"); - assertTrue(completionCalled.get(), "Stream should complete successfully"); - assertFalse(received.isEmpty(), "Should receive at least one parking recommendation"); - assertFalse(reasoning.isEmpty(), "Should receive reasoning blocks"); - } - } - - @Nested - class StreamingCollection { - - /** Verifies that multiple parking recommendations stream in, one per time scenario. */ - @Test - void whenStreamingMultipleScenarios_thenReceivesRecommendationPerScenario() { - PromptRunner runner = ai.withDefaultLlm(); - assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); - - List received = new CopyOnWriteArrayList<>(); - AtomicReference errorOccurred = new AtomicReference<>(); - AtomicBoolean completionCalled = new AtomicBoolean(false); - - Flux stream = new StreamingPromptRunnerBuilder(runner) - .streaming() - .withPrompt(TIMED_PARKING_PROMPT) - .createObjectStream(ParkingRecommendation.class); - - stream - .timeout(Duration.ofSeconds(120)) - .doOnNext(rec -> { - received.add(rec); - logger.info("Received recommendation: scenario={}, option={}, cost={}", - rec.scenario(), rec.chosenOption(), rec.estimatedTotalCost()); - }) - .doOnError(error -> { - errorOccurred.set(error); - logger.error("Stream error: {}", error.getMessage()); - }) - .doOnComplete(() -> { - completionCalled.set(true); - logger.info("Stream completed with {} recommendations", received.size()); - }) - .blockLast(Duration.ofSeconds(240)); - - assertNull(errorOccurred.get(), "Streaming should not produce errors"); - assertTrue(completionCalled.get(), "Stream should complete successfully"); - assertTrue(received.size() >= 2, "Should receive multiple scenario recommendations, got: " + received.size()); - } - } -} diff --git a/embabel-modules/embabel-streaming/pom.xml b/embabel-modules/embabel-streaming/pom.xml index 6a2fede93db0..89434fae76e8 100644 --- a/embabel-modules/embabel-streaming/pom.xml +++ b/embabel-modules/embabel-streaming/pom.xml @@ -10,9 +10,9 @@ embabel-streaming - pom + 21 21 21 UTF-8 @@ -20,6 +20,11 @@ + + com.embabel.agent + embabel-agent-starter-anthropic + ${embabel-agent.version} + com.embabel.agent embabel-agent-test-internal @@ -28,9 +33,30 @@ - - embabel-streaming-openai - embabel-streaming-anthropic - + + + + org.apache.maven.plugins + maven-compiler-plugin + + ${maven.compiler.source} + ${maven.compiler.source} + ${maven.compiler.source} + + + + org.springframework.boot + spring-boot-maven-plugin + + + repackage + + true + + + + + + diff --git a/embabel-modules/embabel-streaming/embabel-streaming-anthropic/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java b/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java similarity index 62% rename from embabel-modules/embabel-streaming/embabel-streaming-anthropic/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java rename to embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java index eef9ac240365..3fa32a9a3075 100644 --- a/embabel-modules/embabel-streaming/embabel-streaming-anthropic/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java +++ b/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java @@ -1,9 +1,3 @@ -/* - * Example got patterned after: - * https://github.com/embabel/embabel-agent/blob/main/embabel-agent-autoconfigure/models/embabel-agent-anthropic-autoconfigure/src/test/java/com/embabel/agent/config/models/anthropic/LLMAnthropicStreamingBuilderIT.java - * - * Original code (see link above) was developed by Embabel Pty Ltd, 2026 - */ package com.baeldung.embabel.agent.api.streaming; import com.embabel.agent.AgentTestApplication; @@ -25,14 +19,12 @@ import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.ActiveProfiles; import reactor.core.publisher.Flux; - import java.time.Duration; +import java.util.ArrayList; import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ThreadLocalRandom; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.*; @SpringBootTest( @@ -59,7 +51,6 @@ static void setUp() { System.setProperty("embabel.agent.shell.interactive.enabled", "false"); } - // TODO: move to embabel-tutorial-common once available public record ParkingRecommendation( String scenario, Option chosenOption, @@ -73,7 +64,6 @@ public enum Option { } } - // TODO: move to embabel-tutorial-common once available static class ParkingTooling { @LlmTool(description = "Find free street parking. Uncertain and may take time.") @@ -108,7 +98,17 @@ public String reserveGarage(String location) { Recommend the best parking option. """; - private static final String TOOLING_PROMPT = + private static final String TIMED_PARKING_PROMPT = + """ + Provide parking recommendations for an advisor visiting Midtown Manhattan across three time scenarios: + 1. Early morning (before 8am): street meters are free before 8am + 2. Business hours (9am-5pm): all paid options apply, 30-minute window, 3-hour stay + 3. Evening (after 6pm): meters are free after 6pm, but garages may close at 9pm (3-hour stay at risk) + + Return all three recommendations. + """; + + private static final String TOOLING_PARKING_PROMPT = """ An advisor needs to park in Midtown Manhattan for a 3-hour client meeting starting in 30 minutes. Use the available tools to probe parking options, then recommend the best one. @@ -116,14 +116,76 @@ public String reserveGarage(String location) { """; /** - * Verifies structured streaming output with thinking enabled and no tools. - * - *

{@link Thinking#withTokenBudget(int)} sets {@code thinkingEnabled=true} in the streaming - * converter, sending {@code } format instructions in the system prompt. - * Budget must be less than {@code max_tokens}=8192 for {@code claude-sonnet-4-5}. + * Verifies basic structured streaming: a single object is emitted and completed without errors. + */ + @Nested + class SimpleStreaming { + + @Test + void whenStreaming_thenReceivesParkingRecommendation() { + PromptRunner runner = ai.withDefaultLlm(); + assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); + + List received = new ArrayList<>(); + + Flux stream = new StreamingPromptRunnerBuilder(runner) + .streaming() + .withPrompt(PARKING_PROMPT) + .createObjectStream(ParkingRecommendation.class); + + stream + .timeout(Duration.ofSeconds(120)) + .doOnNext(rec -> { + received.add(rec); + logger.info("Received parking recommendation: option={}, cost={}, summary={}", + rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); + }) + .doOnError(error -> logger.error("Stream error: {}", error.getMessage())) + .doOnComplete(() -> logger.info("Stream completed successfully")) + .blockLast(Duration.ofSeconds(240)); + + assertThat(received).hasSize(1); + assertThat(received.getFirst().chosenOption()).isNotNull(); + assertThat(received.getFirst().summary()).isNotBlank(); + } + } + + /** + * Verifies that a prompt requesting multiple results streams them as individual objects, + * one per scenario, rather than a single batched response. + */ + @Nested + class StreamingCollection { + + @Test + void whenStreamingMultipleScenarios_thenReceivesRecommendationPerScenario() { + PromptRunner runner = ai.withDefaultLlm(); + assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); + + List received = new ArrayList<>(); + + new StreamingPromptRunnerBuilder(runner) + .streaming() + .withPrompt(TIMED_PARKING_PROMPT) + .createObjectStream(ParkingRecommendation.class) + .timeout(Duration.ofSeconds(120)) + .doOnNext(rec -> { + received.add(rec); + logger.info("Received recommendation: scenario={}, option={}, cost={}", + rec.scenario(), rec.chosenOption(), rec.estimatedTotalCost()); + }) + .blockLast(Duration.ofSeconds(240)); + + assertThat(received).hasSizeGreaterThanOrEqualTo(2); + } + } + + /** + * Verifies structured streaming output with thinking enabled and with no tools. * - *

The streaming pipeline emits one {@link com.embabel.common.core.streaming.StreamingEvent.Thinking} - * event per line, so expect multiple reasoning events rather than a single block. + *

Token budget must stay below {@code max_tokens} (8192 for {@code claude-sonnet-4-5}). + * Reasoning arrives as multiple {@link com.embabel.common.core.streaming.StreamingEvent.Thinking} + * events — one per line, not a single block. */ @Nested class StreamingWithThinkingNoTools { @@ -135,10 +197,8 @@ void whenStreamingWithThinking_thenReceivesReasoningAndRecommendation() { PromptRunner runner = ai.withDefaultLlm().withLlm(thinkingOptions); assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); - List received = new CopyOnWriteArrayList<>(); - List reasoning = new CopyOnWriteArrayList<>(); - AtomicReference errorOccurred = new AtomicReference<>(); - AtomicBoolean completionCalled = new AtomicBoolean(false); + List received = new ArrayList<>(); + List reasoning = new ArrayList<>(); Flux> stream = new StreamingPromptRunnerBuilder(runner) .streaming() @@ -160,21 +220,12 @@ void whenStreamingWithThinking_thenReceivesReasoningAndRecommendation() { logger.info("Received reasoning: {}", event.getThinking()); } }) - .doOnError(error -> { - errorOccurred.set(error); - logger.error("Stream error: {}", error.getMessage()); - }) - .doOnComplete(() -> { - completionCalled.set(true); - logger.info("Stream completed: {} recommendations, {} reasoning blocks", - received.size(), reasoning.size()); - }) .blockLast(Duration.ofSeconds(240)); - assertNull(errorOccurred.get(), "Streaming should not produce errors"); - assertTrue(completionCalled.get(), "Stream should complete successfully"); - assertFalse(received.isEmpty(), "Should receive at least one parking recommendation"); - assertFalse(reasoning.isEmpty(), "Should receive reasoning blocks"); + assertThat(received).isNotEmpty(); + assertThat(received.getFirst().chosenOption()).isNotNull(); + assertThat(received.getFirst().summary()).isNotBlank(); + assertThat(reasoning).isNotEmpty(); } } @@ -186,7 +237,7 @@ class StreamingWithThinkingAndTooling { * *

Reasoning events are emitted only from the final LLM iteration — after all tool calls * complete. Intermediate reasoning (the model's thinking while deciding which tools to call) - * is not surfaced to the subscriber, as each tool-loop iteration replaces the previous stream. + * is not surfaced to the subscriber, as each Spring AI-managed tool-loop iteration starts a new stream. */ @Test void whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning() { @@ -195,17 +246,13 @@ void whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning( .withToolCallInspectors(new ToolCallLoggingInspector(LogLevel.INFO, logger)); assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); - List received = new CopyOnWriteArrayList<>(); - List reasoning = new CopyOnWriteArrayList<>(); - AtomicReference errorOccurred = new AtomicReference<>(); - AtomicBoolean completionCalled = new AtomicBoolean(false); + List received = new ArrayList<>(); + List reasoning = new ArrayList<>(); - Flux> stream = new StreamingPromptRunnerBuilder(runner) + new StreamingPromptRunnerBuilder(runner) .streaming() - .withPrompt(TOOLING_PROMPT) - .createObjectStreamWithThinking(ParkingRecommendation.class); - - stream + .withPrompt(TOOLING_PARKING_PROMPT) + .createObjectStreamWithThinking(ParkingRecommendation.class) .timeout(Duration.ofSeconds(120)) .doOnNext(event -> { if (event.isObject()) { @@ -220,21 +267,12 @@ void whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning( logger.info("Received reasoning: {}", event.getThinking()); } }) - .doOnError(error -> { - errorOccurred.set(error); - logger.error("Stream error: {}", error.getMessage()); - }) - .doOnComplete(() -> { - completionCalled.set(true); - logger.info("Stream completed: {} recommendations, {} reasoning blocks", - received.size(), reasoning.size()); - }) .blockLast(Duration.ofSeconds(240)); - assertNull(errorOccurred.get(), "Streaming should not produce errors"); - assertTrue(completionCalled.get(), "Stream should complete successfully"); - assertFalse(received.isEmpty(), "Should receive at least one parking recommendation"); - assertFalse(reasoning.isEmpty(), "Should receive reasoning blocks alongside tool results"); + assertThat(received).isNotEmpty(); + assertThat(received.getFirst().chosenOption()).isNotNull(); + assertThat(received.getFirst().summary()).isNotBlank(); + assertThat(reasoning).isNotEmpty(); } } } From f9ac6f0bd897442608355077b604bd4b5837c8ae Mon Sep 17 00:00:00 2001 From: Igor Dayen Date: Mon, 27 Jul 2026 21:55:46 -0400 Subject: [PATCH 4/5] Strenthen asserts --- ...ingWithThinkingAndToolingIntegrationTest.java | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java b/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java index 3fa32a9a3075..3f1633baeb53 100644 --- a/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java +++ b/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java @@ -140,8 +140,6 @@ void whenStreaming_thenReceivesParkingRecommendation() { logger.info("Received parking recommendation: option={}, cost={}, summary={}", rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); }) - .doOnError(error -> logger.error("Stream error: {}", error.getMessage())) - .doOnComplete(() -> logger.info("Stream completed successfully")) .blockLast(Duration.ofSeconds(240)); assertThat(received).hasSize(1); @@ -177,6 +175,12 @@ void whenStreamingMultipleScenarios_thenReceivesRecommendationPerScenario() { .blockLast(Duration.ofSeconds(240)); assertThat(received).hasSizeGreaterThanOrEqualTo(2); + assertThat(received).allSatisfy(r -> { + assertThat(r.scenario()).isNotBlank(); + assertThat(r.chosenOption()).isNotNull(); + assertThat(r.summary()).isNotBlank(); + }); + assertThat(received).map(ParkingRecommendation::scenario).doesNotHaveDuplicates(); } } @@ -222,10 +226,10 @@ void whenStreamingWithThinking_thenReceivesReasoningAndRecommendation() { }) .blockLast(Duration.ofSeconds(240)); - assertThat(received).isNotEmpty(); + assertThat(received).hasSize(1); assertThat(received.getFirst().chosenOption()).isNotNull(); assertThat(received.getFirst().summary()).isNotBlank(); - assertThat(reasoning).isNotEmpty(); + assertThat(reasoning).allSatisfy(r -> assertThat(r).isNotBlank()); } } @@ -269,10 +273,10 @@ void whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning( }) .blockLast(Duration.ofSeconds(240)); - assertThat(received).isNotEmpty(); + assertThat(received).hasSize(1); assertThat(received.getFirst().chosenOption()).isNotNull(); assertThat(received.getFirst().summary()).isNotBlank(); - assertThat(reasoning).isNotEmpty(); + assertThat(reasoning).allSatisfy(r -> assertThat(r).isNotBlank()); } } } From 6d85107aad304b8f33d99d4336ea04f5d0703486 Mon Sep 17 00:00:00 2001 From: Igor Dayen Date: Tue, 4 Aug 2026 19:53:57 -0400 Subject: [PATCH 5/5] Remove asserts --- ...WithThinkingAndToolingIntegrationTest.java | 38 ------------------- 1 file changed, 38 deletions(-) diff --git a/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java b/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java index 3f1633baeb53..bfaa5367d05b 100644 --- a/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java +++ b/embabel-modules/embabel-streaming/src/test/java/com/baeldung/embabel/agent/api/streaming/StreamingWithThinkingAndToolingIntegrationTest.java @@ -126,8 +126,6 @@ void whenStreaming_thenReceivesParkingRecommendation() { PromptRunner runner = ai.withDefaultLlm(); assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); - List received = new ArrayList<>(); - Flux stream = new StreamingPromptRunnerBuilder(runner) .streaming() .withPrompt(PARKING_PROMPT) @@ -136,15 +134,10 @@ void whenStreaming_thenReceivesParkingRecommendation() { stream .timeout(Duration.ofSeconds(120)) .doOnNext(rec -> { - received.add(rec); logger.info("Received parking recommendation: option={}, cost={}, summary={}", rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); }) .blockLast(Duration.ofSeconds(240)); - - assertThat(received).hasSize(1); - assertThat(received.getFirst().chosenOption()).isNotNull(); - assertThat(received.getFirst().summary()).isNotBlank(); } } @@ -160,27 +153,16 @@ void whenStreamingMultipleScenarios_thenReceivesRecommendationPerScenario() { PromptRunner runner = ai.withDefaultLlm(); assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); - List received = new ArrayList<>(); - new StreamingPromptRunnerBuilder(runner) .streaming() .withPrompt(TIMED_PARKING_PROMPT) .createObjectStream(ParkingRecommendation.class) .timeout(Duration.ofSeconds(120)) .doOnNext(rec -> { - received.add(rec); logger.info("Received recommendation: scenario={}, option={}, cost={}", rec.scenario(), rec.chosenOption(), rec.estimatedTotalCost()); }) .blockLast(Duration.ofSeconds(240)); - - assertThat(received).hasSizeGreaterThanOrEqualTo(2); - assertThat(received).allSatisfy(r -> { - assertThat(r.scenario()).isNotBlank(); - assertThat(r.chosenOption()).isNotNull(); - assertThat(r.summary()).isNotBlank(); - }); - assertThat(received).map(ParkingRecommendation::scenario).doesNotHaveDuplicates(); } } @@ -201,9 +183,6 @@ void whenStreamingWithThinking_thenReceivesReasoningAndRecommendation() { PromptRunner runner = ai.withDefaultLlm().withLlm(thinkingOptions); assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); - List received = new ArrayList<>(); - List reasoning = new ArrayList<>(); - Flux> stream = new StreamingPromptRunnerBuilder(runner) .streaming() .withPrompt(PARKING_PROMPT) @@ -215,21 +194,14 @@ void whenStreamingWithThinking_thenReceivesReasoningAndRecommendation() { if (event.isObject()) { ParkingRecommendation rec = event.getObject(); if (rec != null) { - received.add(rec); logger.info("Received recommendation: option={}, cost={}, summary={}", rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); } } else if (event.isThinking()) { - reasoning.add(event.getThinking()); logger.info("Received reasoning: {}", event.getThinking()); } }) .blockLast(Duration.ofSeconds(240)); - - assertThat(received).hasSize(1); - assertThat(received.getFirst().chosenOption()).isNotNull(); - assertThat(received.getFirst().summary()).isNotBlank(); - assertThat(reasoning).allSatisfy(r -> assertThat(r).isNotBlank()); } } @@ -250,9 +222,6 @@ void whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning( .withToolCallInspectors(new ToolCallLoggingInspector(LogLevel.INFO, logger)); assertTrue(runner.supportsStreaming(), "Default LLM must support streaming"); - List received = new ArrayList<>(); - List reasoning = new ArrayList<>(); - new StreamingPromptRunnerBuilder(runner) .streaming() .withPrompt(TOOLING_PARKING_PROMPT) @@ -262,21 +231,14 @@ void whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning( if (event.isObject()) { ParkingRecommendation rec = event.getObject(); if (rec != null) { - received.add(rec); logger.info("Received recommendation: option={}, cost={}, summary={}", rec.chosenOption(), rec.estimatedTotalCost(), rec.summary()); } } else if (event.isThinking()) { - reasoning.add(event.getThinking()); logger.info("Received reasoning: {}", event.getThinking()); } }) .blockLast(Duration.ofSeconds(240)); - - assertThat(received).hasSize(1); - assertThat(received.getFirst().chosenOption()).isNotNull(); - assertThat(received.getFirst().summary()).isNotBlank(); - assertThat(reasoning).allSatisfy(r -> assertThat(r).isNotBlank()); } } }