LLM Streaming in the Embabel Agentic AI Framework

Last updated: August 26, 2026

Written by: Igor Dayen

Reviewed by: Milos Simic

Spring AI

Embabel

Refactor Java code safely — and automatically — with OpenRewrite.

Refactoring big codebases by hand is slow, risky, and easy to put off. That’s where OpenRewrite comes in. The open-source framework for large-scale, automated code transformations helps teams modernize safely and consistently.

Each month, the creators and maintainers of OpenRewrite at Moderne run live, hands-on training sessions — one for newcomers and one for experienced users. You’ll see how recipes work, how to apply them across projects, and how to modernize code with confidence.

Join the next session, bring your questions, and learn how to automate the kind of work that usually eats your sprint time.

1. Overview

LLM streaming delivers output token by token as it’s generated. This reduces latency, as users see the first results in the UI within milliseconds instead of waiting seconds for a complete answer.

For agents with structured output and reasoning-aware agents, streaming should enable incremental processing:

parsing responses into Java objects as they arrive, and
processing thinking blocks as they arrive as well

However, current major frameworks don’t support these functionalities.

In this tutorial, we’ll:

briefly review the limitations of these frameworks
and show how to use the Embabel Agentic AI Framework to process objects and thinking blocks incrementally, as they arrive in a stream

We’ll cover four cases:

one object without reasoning
multiple objects without reasoning
multiple objects with reasoning but without tool calls
multiple objects with both reasoning and tool calls

2. Structured Streaming: Framework Limitations

The major Java frameworks for LLM integration, Spring AI and LangChain4j, support streaming at a limited level. We can get a live stream of text as the model generates it, but neither framework turns that stream into a collection of typed objects out of the box.

The Spring AI’s entity() method, which converts model output into a Java object, works with the blocking call() method but not with stream(). The ChatClient API reference spells this out directly: “In the future, we will offer a convenience method that will let you return a Java entity with the reactive stream() method. In the meantime, you should use the Structured Output Converter.”

LangChain4j has the same limitation. Its AiServices allows returning custom POJOs, lists, and enums only for non-streaming calls. For streaming, the return type must be a TokenStream, which gives us raw text as it arrives, plus rich callback hooks, but doesn’t provide typed objects as stream output.

In practice, anyone who wants a stream of structured objects with reasoning has to build that functionality themselves, as in this Baeldung’s article. This means choosing an NL-delimited format like JSONL, buffering tokens until a full line is available, parsing each line as JSON, and filtering out any thinking blocks along the way. Implementing this takes time away from the actual project.

This is where Embabel Agentic AI comes in handy. It natively provides the desired functionality, which we further explore.

3. Setup

First, let’s define the necessary Embabel dependencies:

<dependency>
    <groupId>com.embabel.agent</groupId>
    <artifactId>embabel-agent-starter-anthropic</artifactId>
    <version>${embabel-agent.version}</version>
</dependency>

<dependency>
    <groupId>com.embabel.agent</groupId>
    <artifactId>embabel-agent-test-internal</artifactId>
    <version>${embabel-agent.version}</version>
    <scope>test</scope>
</dependency>

The embabel-agent-starter-anthropic dependency provides the core classes we need to use Anthropic models with Embabel. In the examples in this tutorial, we use the Anthropic model claude-sonnet-4-5.

Additionally, we include embabel-agent-test-internal to use the AgentTestApplication test support class with @SpringBootTest.

3.1. Provider Support in Embabel

Note that model providers differ in how they combine streaming with thinking and tools. Anthropic’s native thinking produces explicit reasoning blocks, but other providers may not surface reasoning in the same way.

Embabel supports over a dozen LLM providers, with examples using OpenAI published on Baeldung in the past.

3.2. ParkingRecommendation

We’ll stream ParkingRecommendation objects in all tests throughout the article:

The record includes four scalar properties: parking scenario (scenario), chosen option cost (estimatedTotalCost), summary, and chosen option (enumeration Option).

4. Streaming a Single Object

First, we show how to stream a single object without reasoning and tool calls. The test method is whenStreaming_thenReceivesParkingRecommendation():

void whenStreaming_thenReceivesParkingRecommendation() {
     streamParkingRecommendations(PARKING_PROMPT);
  }

PARKING_PROMPT asks for a single parking recommendation, and the private method streamParkingRecommendations() does the actual object streaming:

Flux<ParkingRecommendation> 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: scenario={}, option={}, cost={}, summary={}",
      rec.scenario(),
      rec.chosenOption(),
      rec.estimatedTotalCost(),
      rec.summary());
    }).blockLast(Duration.ofSeconds(240));

Here:

We create a Flux<ParkingRecommendation> by calling createObjectStream(). A Flux is a Reactor type representing an asynchronous sequence of zero or more elements that arrive over time rather than all at once. In this example, the model emits a ParkingRecommendation object.
PARKING_PROMPT instructs the LLM to find the optimal parking option out of three available options: street parking, metered parking, and garage parking. The prompt specifies constraints and costs for each alternative.
The pipeline subscribes with doOnNext() to collect and log each object.
blockLast() blocks the calling thread until the stream completes. This is appropriate in a test context, though production code would chain further reactive operators instead of blocking. For example, we would return the Flux directly to a web endpoint that streams the response to the client.

We can run the unit test method whenStreaming_thenReceivesParkingRecommendation() like this:

$ mvn test -pl embabel-streaming -P integration -Dtest=StreamingWithThinkingAndToolingIntegrationTest#whenStreaming_thenReceivesParkingRecommendation

After the stream completes, logs confirm the stream delivers a complete ParkingRecommendation object:

17:34:13.781 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Midtown Manhattan parking for 3-hour client meeting with 30-minute arrival buffer, option=GARAGE, cost=90, summary=Choose garage parking. The meeting duration (3 hours) exceeds the metered parking limit (2 hours), eliminating that option. Street parking in Midtown Manhattan is extremely unreliable and searching could make you late for the meeting. The guaranteed spot and ability to stay for the full meeting duration justifies the $90 cost for this professional context.

The object includes all the properties we expect: scenario, chosenOption, estimatedTotalCost, and summary.

5. Streaming Multiple Objects

The streamParkingRecommendations() method can stream multiple objects. To show it, we request three recommendations with TIMED_PARKING_PROMPT:

void whenStreamingMultipleScenarios_thenReceivesRecommendationPerScenario() {
    streamParkingRecommendations(TIMED_PARKING_PROMPT);
}

When streaming:

doOnNext() collects and logs each object as it arrives
blockLast() waits for the stream to complete

We can run the unit test method whenStreamingMultipleScenarios_thenReceivesRecommendationPerScenario like this:

$ mvn test -pl embabel-streaming -P integration -Dtest=StreamingWithThinkingAndToolingIntegrationTest#whenStreamingMultipleScenarios_thenReceivesRecommendationPerScenario

Instead of obtaining a single batched response with all objects, we get each ParkingRecommendation as a separate object as it is received:

18:31:24.200 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Early morning (before 8am), option=STREET, cost=0, summary=Arrive before 8am to take advantage of free street parking. Meters are not enforced until 8am, providing zero-cost parking. Be prepared to feed the meter or move to a garage if your meeting extends past 8am.
18:31:27.354 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Business hours (9am-5pm), option=GARAGE, cost=45, summary=For a 3-hour business meeting with only a 30-minute arrival window, a parking garage is the most reliable option. Street parking in Midtown is extremely competitive during peak hours, and the time spent searching could cause you to miss your meeting. Garage rates typically range $35-55 for 3 hours in Midtown Manhattan.
18:31:28.889 [HttpClient-4-Worker-0] WARN  StreamHelper - Unhandled event type: CONTENT_BLOCK_STOP
18:31:28.895 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Evening (after 6pm), option=STREET, cost=0, summary=Street meters become free after 6pm, making this the best option if you're confident your 3-hour meeting will end before 9pm. However, have a backup plan: identify a 24-hour garage nearby in case your meeting runs late, as many garages close at 9pm and you could be locked in or locked out.

The logged timing of each object creation certifies that complete objects are obtained sequentially and that the collection is filled incrementally.

Note the entry in the log “Unhandled event type: CONTENT_BLOCK_STOP“. It just signals “this content block (a text block, tool call block, etc.) is finished.” It carries no payload and says that we received a server-side event, but there is no handler wired up for it.

6. Streaming with Reasoning

First, we must enable deep thinking by configuring LlmOptions with a token budget:

void whenStreamingWithThinking_thenReceivesReasoningAndRecommendation() {
    LlmOptions thinkingOptions = new LlmOptions().withThinking(Thinking.withTokenBudget(8000));
    PromptRunner runner = ai.withDefaultLlm().withLlm(thinkingOptions);
    streamParkingRecommendationsWithThinking(runner, TIMED_PARKING_PROMPT);
}

The budget controls how many tokens the model may spend on reasoning before producing the answer, and it must be lower than max_tokens (8192 for claude-sonnet-4-5).

Objects are streamed in the private method streamParkingRecommendationsWithThinking(). It uses createObjectStreamWithThinking() instead of createObjectStream(), and the return type changes accordingly:

Flux<StreamingEvent<ParkingRecommendation>> stream = new StreamingPromptRunnerBuilder(runner)
  .streaming()
  .withPrompt(prompt)
  .createObjectStreamWithThinking(ParkingRecommendation.class);

StreamingEvent wraps typed objects and reasoning fragments, as the stream interleaves both kinds of events. The doOnNext() callback distinguishes them via the event.isObject() and event.isThinking() checks, collecting them into separate lists, received for objects and reasoning for thinking fragments:

.doOnNext(event -> {
  if (event.isObject()) {
      ParkingRecommendation rec = event.getObject();
      if (rec != null) {
          received.add(rec);
          logger.info("Received recommendation: scenario={}, option={}, cost={}, summary={}",
            rec.scenario(), rec.chosenOption(), rec.estimatedTotalCost(), rec.summary());
      }
  } else if (event.isThinking()) {
      reasoning.add(event.getThinking());
      logger.info("Received reasoning: {}", event.getThinking());
  }
})

We can run the unit test whenStreamingWithThinking_thenReceivesReasoningAndRecommendation() as follows:

$ mvn test -pl embabel-streaming -P integration -Dtest=StreamingWithThinkingAndToolingIntegrationTest#whenStreamingWithThinking_thenReceivesReasoningAndRecommendation

Logs demonstrate that recommendations and reasoning fragments are processed as they arrive.

21:34:05.311 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received reasoning: The user is asking for parking recommendations for Midtown Manhattan across three different time scenarios. I need to analyze each scenario and provide recommendations in JSONL format.
21:34:05.311 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Let me think through each scenario:
...
21:34:13.300 [boundedElastic-1] INFO StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Early morning (before 8am), option=STREET, cost=0, summary=Street parking is free before 8am in Midtown. Arrive early to secure a spot while meters are not enforced. No cost advantage to using garage or paid meter.
21:34:14.003 [boundedElastic-1] INFO StreamingWithThinkingAndToolingIntegrationTest - Received reasoning: For business hours, considering reliability vs cost. 30-minute window suggests tight schedule, and 3-hour stay during peak hours means garage offers certainty despite higher cost
21:34:16.867 [boundedElastic-1] INFO StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Business hours (9am-5pm, 3-hour stay), option=GARAGE, cost=35, summary=During peak business hours with only a 30-minute arrival window, garage parking ($30-40 for 3 hours) provides guaranteed availability and eliminates time spent searching. Street meters ($12-15 for 3 hours) are cheaper but finding spots in Midtown during business hours is challenging and risky given the tight timeline.
21:34:18.275 [boundedElastic-1] INFO StreamingWithThinkingAndToolingIntegrationTest - Received reasoning: Evening scenario has free street parking but garage closure risk. If arriving at 6pm with 3-hour stay, would exit at 9pm exactly when garages close - cutting it too close. Street parking is free and has no closure risk
21:34:19.337 [HttpClient-4-Worker-0] WARN StreamHelper - Unhandled event type: CONTENT_BLOCK_STOP
21:34:19.359 [boundedElastic-1] INFO StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Evening (after 6pm, 3-hour stay), option=STREET, cost=0, summary=Street meters are free after 6pm, making this the clear choice. Additionally, with garages closing at 9pm and a 3-hour stay, there's significant risk of garage closure before departure. Street parking eliminates both cost and time constraints.

Note that each reasoning block represents a single line, as Embabel aggregates mini-chunks into a single line.

7. Streaming with Tools and Reasoning

Finally, let’s combine tool calling with streaming and thinking. For a blocking call, the Baeldung article LLM Tool Call Reasoning Using Embabel Agentic AI Framework shows how tool calls benefit from LLM reasoning.

The API is very similar to the reasoning use case, except for the tool registration withToolObject(new ParkingTooling()) and a logging inspector for observability:

void whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning() {
    PromptRunner runner = ai.withDefaultLlm()
      .withToolObject(new ParkingTooling())
      .withToolCallInspectors(new ToolCallLoggingInspector(LogLevel.INFO, logger));
    streamParkingRecommendationsWithThinking(runner, TOOL_PARKING_PROMPT);
}

TOOL_PARKING_PROMPT instructs the model to actively use the tools before giving three recommendations, and the streamParkingRecommendationsWithThinking() method streams them.

The pipeline uses the same createObjectStreamWithThinking() and the same event.isObject() / event.isThinking() pattern. The key difference is tool call timing: the model first calls tools, then reasons over the results, then emits the objects, and finally provides a reasoning summary. As Spring AI starts a new stream after all tool calls are complete, the reasoning blocks are emitted only after tool calls.

We can run the unit test whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning() like this:

$ mvn test -pl embabel-streaming -P integration -Dtest=StreamingWithThinkingAndToolingIntegrationTest#whenStreamingWithThinkingAndTooling_thenReceivesRecommendationAndReasoning

12:48:52.342 [HttpClient-4-Worker-0] WARN  StreamHelper - Unhandled event type: CONTENT_BLOCK_STOP
12:48:53.275 [boundedElastic-2] INFO  StreamingWithThinkingAndToolingIntegrationTest - beforeToolCall: tool=reserveGarage, argsLength=28
12:48:53.278 [boundedElastic-2] INFO  Embabel - [suspicious_tharp] calling tool reserveGarage({"arg0":"Midtown Manhattan"})
12:48:53.281 [boundedElastic-2] INFO  Embabel - [suspicious_tharp] tool reserveGarage returned Garage reserved near Midtown Manhattan ($30/hour, guaranteed) in 2ms with payload {"arg0":"Midtown Manhattan"}
12:48:53.284 [boundedElastic-2] INFO  StreamingWithThinkingAndToolingIntegrationTest - afterToolCall: tool=reserveGarage, status=Text, resultLength=61, durationMs=6
12:48:55.299 [boundedElastic-3] INFO  StreamingWithThinkingAndToolingIntegrationTest - beforeToolCall: tool=findStreetParking, argsLength=38
12:48:55.299 [boundedElastic-3] INFO  Embabel - [suspicious_tharp] calling tool findStreetParking({"arg1":30,"arg0":"Midtown Manhattan"})
12:48:55.300 [boundedElastic-3] INFO  Embabel - [suspicious_tharp] tool findStreetParking returned Street parking found near Midtown Manhattan (free) in 1ms with payload {"arg1":30,"arg0":"Midtown Manhattan"}
12:48:55.300 [boundedElastic-3] INFO  StreamingWithThinkingAndToolingIntegrationTest - afterToolCall: tool=findStreetParking, status=Text, resultLength=50, durationMs=1
12:48:56.953 [boundedElastic-4] INFO  StreamingWithThinkingAndToolingIntegrationTest - beforeToolCall: tool=findMeterParking, argsLength=38
12:48:56.954 [boundedElastic-4] INFO  Embabel - [suspicious_tharp] calling tool findMeterParking({"arg1":30,"arg0":"Midtown Manhattan"})
12:48:56.954 [boundedElastic-4] INFO  Embabel - [suspicious_tharp] tool findMeterParking returned No metered parking found within 30 minutes in 0ms with payload {"arg1":30,"arg0":"Midtown Manhattan"}
12:48:56.954 [boundedElastic-4] INFO  StreamingWithThinkingAndToolingIntegrationTest - afterToolCall: tool=findMeterParking, status=Text, resultLength=42, durationMs=1
12:48:58.801 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received reasoning: I'll help you find the best parking options for your client meeting in Midtown Manhattan. Let me check all available parking options for you.Based on the parking options available, here are my three recommendations with the best option first:
12:49:00.945 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Advisor needs guaranteed parking for 3-hour meeting in 30 minutes, option=GARAGE, cost=90, summary=Reserve garage parking at $30/hour for 3 hours ($90 total). This is the BEST option because it's guaranteed and ensures you won't be late for your client meeting. With only 30 minutes until the meeting, reliability is critical.
12:49:03.093 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Advisor needs guaranteed parking for 3-hour meeting in 30 minutes, option=STREET, cost=0, summary=Free street parking is available but highly risky. While it costs nothing, finding a spot is uncertain and time-consuming in Midtown Manhattan. Given that arriving late is not acceptable and you only have 30 minutes, this option could jeopardize your meeting.
12:49:03.706 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received recommendation: scenario=Advisor needs guaranteed parking for 3-hour meeting in 30 minutes, option=METER, cost=0, summary=Metered parking is NOT available within your 30-minute timeframe in Midtown Manhattan. This option is not viable for your situation.
12:49:05.244 [HttpClient-4-Worker-0] WARN  StreamHelper - Unhandled event type: CONTENT_BLOCK_STOP
12:49:05.247 [boundedElastic-1] INFO  StreamingWithThinkingAndToolingIntegrationTest - Received reasoning: **Strong Recommendation: Choose the garage parking.** With only 30 minutes before your client meeting and the absolute requirement not to be late, the $90 guaranteed garage spot is the only responsible choice. The cost is a small price compared to the professional consequences of arriving late to a client meeting.

Similar to whenStreamingWithThinking_thenReceivesReasoningAndRecommendation(), the stream delivers both a recommendation and a reasoning block (after the tools do their tasks).

8. Conclusion

In this article, we showed how to use the Embabel Fluent API to produce a stream of typed Java objects, with thinking and tool results handled transparently.

Streaming raw text is straightforward in both Spring AI and LangChain4j, but streaming typed objects is not something either framework handles out of the box. Embabel addresses this at the framework level, removing the need to manually buffer tokens, parse NDJSON, or filter reasoning blocks. The same pattern scales from a single object to a collection and from a simple prompt to a multi-tool reasoning chain. For applications where both structure and responsiveness matter, this removes a layer of boilerplate that would otherwise fall on every team building agentic AI applications.

The article code is available on GitHub.