Generate Stream with Thinking

[igordayen]

#OVERVIEW

• Add Streaming Line Classifier
• Add Generate Stream with Thinking API

Related PR: #1819 , #1864

[igordayen]

@jstar0 / @arnabnandy7 - please integrate with your PR and test e2e

In your PR, please use StreamingLlmOperationsImp from this PR

@arimu1 @jorander - FYI

Thank you

[igordayen]

@jorander forwarded your request on Claude to Alex H.
Please contact me on Discord. Thanks

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
100.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[arnabnandy7]

@igordayen #1864 is a replacement for #1819, not sure why we are treating it separately.

@arnabnandy7 Got it, then please integrate #1864. I assume #1819 should be closed then.

Any inquiries - let me know. Thanks.

[arnabnandy7]

Thanks, @igordayen . I reviewed the StreamingLlmOperationsImpl changes in #1873 specifically, as requested.

The placement is compatible with #1864: generateStreamWithThinking() runs above doTransformStream(), so it receives the final text stream from streamWithToolLoop(). I also agree that moving reusable classification logic into embabel-agent-ai is a good direction.

However, I don’t think we can use the implementation unchanged because its current classification contract comes from object streaming:

• ordinary non-JSON text is emitted as Thinking(CONTINUATION)
• JSON lines are dropped;
• no StreamingEvent.Object is emitted for the final text answer.

For example:

<think>reasoning</think>
Final answer

the implementation in #1873 produces two thinking events. The contract in #1864 requires:

Thinking("reasoning")
Object("Final answer")

There are also a few streaming differences to reconcile:

• rawChunksToLines() waits for a newline and therefore buffers long reasoning or answer paragraphs
• it trims lines and removes blank lines, changing the generated text
• its mutable buffer is not scoped with Flux.defer, so repeated or concurrent subscriptions can share parser state
• multi-line closing tags may remain in the emitted content
• dynamic tags with attributes are not handled equivalently

I suggest we reuse the architectural direction from #1873, but first evolve StreamingLineClassifier into a subscription-scoped text-stream classifier that emits both Thinking and Object, preserves formatting, and handles arbitrary chunk boundaries. At that point, #1864 can delegate to it instead of maintaining separate parsing logic.

I’m happy to integrate that version and add end-to-end coverage across the tool loop. Also, #1819 can be closed because #1864 replaces it.

[jorander]

One request on deferred line-accumulation.

[jorander]

What if we have pure streaming text content, with no thinking tags included? We would still buffer that stream until we find a newline character. I don't think that is a good behavior. Would it be possible to hold of buffering until we identify a chunk that could be the start of a thinking tag?

The use-case I'm thinking of is where we use this method to get StreamingEvent but the thinking we are looking for is native thinking triggered by setting a thinking budget. (I know, not yet implemented or designed, but given the name of the methods I think it is reasonable to assume they should pick up both types of thinking.) ==> that complies with the current behavior for object creation, when Thinking by definition is having tagType=as {XML-tag, PREFIX, NO-PREFIX}. in object creation - everything that is not a JSON is modelled as thinking, see PROMPT definition, for blocking and streaming events.

In summary: replicate the same logic as for object creation and drop any object creation.
Thanks

[igordayen]

@jorander The intent here is to model every line as a ThinkingEvent - as the method states "withThinking".
Fully aligned with object creation. Same behavior.
Native thinking is a very challenging area; eager to start after release 2.0.0, main focus this week.
Intentionally made this simple.
Headups, I'm reviewing discussion forums; new items coming. One of them is related to providing the user with both:

• streaming event without buffering + additional interceptor (in parallel for buffering). So the user can define StreamingEventsAggregatorInterceptor, but it will not block the user from getting low-level streaming events.

Thinking type is having tagType=as {XML-tag, PREFIX, NO-PREFIX}. In object creation, everything that is not JSON is modelled as thinking; see PROMPT definition for blocking and streaming events.

User can opt to use just createObject if needed; mix of thinking + String (final response)

[igordayen]

• no StreamingEvent.Object is emitted for the final text answer.

@arnabnandy7 - yes, that is the intent. Just buffering per line + emitting every line as a thinking event.
That is done intentionally, not to mix concepts.
Use createObject for stream if an object is required.

[arnabnandy7]

• no StreamingEvent.Object is emitted for the final text answer.

@arnabnandy7 - yes, that is the intent. Just buffering per line + emitting every line as a thinking event. That is done intentionally, not to mix concepts. Use createObject for stream if an object is required.

Thanks, @igordayen. Understood that emitting every non-JSON line as Thinking is intentional in #1873.

The distinction I want to clarify is that StreamingEvent.Object<String> in #1864 does not imply JSON conversion or createObjectStream(). It represents the final text channel in the mixed stream.

For example:

<think>reasoning</think>
Final answer

needs to expose both channels to the caller:

Thinking("reasoning")
Object("Final answer")

If every line is emitted as Thinking, the caller cannot distinguish the final answer from reasoning. createObjectStream() does not cover this use case because the requested output is ordinary streamed text, not a structured object.

I also saw @jorander's concern about buffering pure text until a newline. That is another reason I don’t think #1864 can adopt StreamingLlmOperationsImpl from #1873 unchanged.

It looks like we currently have two different intended APIs:

• a thinking-only, newline-classified stream in Generate Stream with Thinking #1873
• a mixed thinking-and-final-text stream in Streaming review #1864

Before integrating them, I think we need agreement on which contract generateStreamWithThinking() should expose. I’ll keep #1864 development on hold as requested until that contract is clarified.

[igordayen]

I suggest we reuse the architectural direction from #1873, but first evolve StreamingLineClassifier into a subscription-scoped text-stream classifier that emits both Thinking and Object

==> intent was not to mix concepts. Object creation is the responsibility of the "objectCreate" API. Thanks

[igordayen]

Thinking("reasoning")
Object("Final answer")

==> Do you see a practical need there? Perhaps consider createObject instead?

[arnabnandy7]

@igordayen it's done, open for review.

cc @jorander

[jorander]

@igordayen I add a link between the discussion in #1881 and this PR since I think they are very much connected.