feat: stream thinking with text responses - #1819
Closed
jstar0 wants to merge 1 commit into
Closed
Conversation
jstar0
force-pushed
the
feat/thinking-content-metadata
branch
from
July 23, 2026 15:37
5fd413a to
f699d41
Compare
Collaborator
What implementations are we talking about here? We do not have any default construction for existing method |
jorander
reviewed
Jul 24, 2026
Contributor
==> Behavior already in place to accumulate all thinking blocks cross tool loop interation.... could you please substantiate the claim? thanks |
igordayen
reviewed
Jul 25, 2026
Collaborator
My understanding is that this refers to having added native LLM thinking in the existing loop. |
Author
|
@jorander On compatibility: I narrowed the PR body. The defaults added here cover existing @igordayen On accumulation: tagged thinking blocks were already accumulated across tool-loop iterations. The PR body now states the narrower behavior added here: provider-supplied thinking metadata is carried into that accumulation and preserved when output conversion fails. The existing tagged-thinking path is unchanged. |
Contributor
|
@jstar0, @jorander, @azanux, @alexheifetz - after one more round of PR review, I would suggest splitting the PR into 2:
|
Author
|
Yes, I agree with the split. The current commits are not a clean boundary because a few streaming implementation files contain both concerns, so I would narrow this PR by rebuilding it around the first scope rather than mechanically separating the commits. For #1819 I would keep only:
I would move the provider-native work to a follow-up PR: the structured sender/streamer response contracts, Spring AI metadata normalization for OpenAI/Ollama/Anthropic, propagation through tool loops and object streams, and the I will change this PR to |
Contributor
|
Thank you very much, @jstar0! |
Signed-off-by: King Star <mcxin.y@gmail.com>
jstar0
force-pushed
the
feat/thinking-content-metadata
branch
from
July 27, 2026 19:32
08a2242 to
376cbc1
Compare
jstar0
changed the title
igordayen
reviewed
Jul 27, 2026
| * Extracts tagged thinking from text while preserving response order across | ||
| * arbitrary chunk boundaries. Parser state is scoped to each subscription. | ||
| */ | ||
| internal fun Flux<String>.toTaggedThinkingEvents(): Flux<StreamingEvent<String>> = |
Contributor
There was a problem hiding this comment.
What prompted this very significant change?
The current algorithm to collect thinking blocks is lightweight for the following reason:
Every thinking block fits on a single line, so it could be
<think> abc...... NL
xyz.........NL
mnc </think> NL
<think>..... </think> NL
<think>..... </think> ....<reason>.....</reason>
Reason for this - it's the streaming at the very end, so as not to accumulate too much.
Looping in @arnabnandy7 as his PR overlaps with this one.
Also - hard to follow uncommented logic.
Thanks for understanding
Contributor
|
Thanks for narrowing this PR to tagged-text thinking extraction. I reviewed it specifically against the provider-neutral streaming tool loop in #1826. The approaches are complementary. After integration, I found one issue that I think should be addressed. The parser currently buffers an entire tagged-thinking block until its closing delimiter arrives. This delays thinking events and can accumulate a large reasoning block in memory. Emitting completed lines or safe chunks while retaining only a possible closing-delimiter suffix would preserve chunk-boundary correctness while keeping the operation genuinely streaming. There will also be a mechanical rebase adjustment: #1826 changes @igordayen I shared my review |
Contributor
@arnabnandy7 - thank you, right-on-money, that is exactly what I mentioned above as well! |
Contributor
|
@jstar0 - could you please confirm your understanding of the motivation for handling thinking blocks in streaming? |
azanux
requested changes
Jul 29, 2026
azanux
left a comment
azanux
left a comment
Collaborator
There was a problem hiding this comment.
@jstar0 thanks for the contribution.
The points raised by @igordayen look justified to me: a few implementation questions remain and deserve answers before this lands.
Comment on lines
+63
to
+71
| if (tag != null) { | ||
| val endIndex = buffer.indexOf(tag.end) | ||
| if (endIndex < 0) break | ||
| val thinking = buffer.substring(0, endIndex).trim() | ||
| if (thinking.isNotEmpty()) events += StreamingEvent.Thinking(thinking) | ||
| buffer.delete(0, endIndex + tag.end.length) | ||
| activeTag = null | ||
| atLineStart = false | ||
| continue |
Collaborator
There was a problem hiding this comment.
#1716 is about streaming the reasoning, but ThinkingStreamSupport.kt:63-71 emits
nothing until the closing delimiter arrives - the whole block lands at the end. That is
blocking behaviour behind a reactive signature.
It is also O(N²): buffer.indexOf(tag.end) restarts from index 0 on every chunk and the
buffer is never drained. A 50k-char block in 50-char chunks scans ~25M characters.
the solution and the mechanism is already in your code: longestPartialStartSuffix(), the mirror case is simpler - you already know which tag is open, so retaining tag.end.length - 1 chars is enough
Comment on lines
+64
to
+68
| val endIndex = buffer.indexOf(tag.end) | ||
| if (endIndex < 0) break | ||
| val thinking = buffer.substring(0, endIndex).trim() | ||
| if (thinking.isNotEmpty()) events += StreamingEvent.Thinking(thinking) | ||
| buffer.delete(0, endIndex + tag.end.length) |
Collaborator
There was a problem hiding this comment.
One asymmetry worth confirming: the streaming path removes tagged content from the text channel, the blocking path keeps it.
Same model output, different text from generate() and generateStream(). Intended ? Why ?
| val legacy: Boolean, | ||
| ) | ||
|
|
||
| private val tags = ThinkingTags.TAG_DEFINITIONS |
Collaborator
There was a problem hiding this comment.
extractAllThinkingBlocks also runs dynamicTagsDiscoveryAndExtraction, which matches
any <tag>…</tag>, attributes included - and discovers tags that are not in
TAG_DEFINITIONS at all.
The stream parser only does exact indexOf on the 7 tag entries (ThinkingStreamSupport.kt:47-50, :128-129)
So the reasoning ends up rendered as the answer in streamed calls. Intentional scope
reduction for now?
| * | ||
| * The default wraps [doTransformStream] for source compatibility. | ||
| */ | ||
| fun doTransformStreamWithThinking( |
Collaborator
There was a problem hiding this comment.
The default in StreamingLlmOperations.kt:165-173 does no extraction, and both implementations override it with the identical .toTaggedThinkingEvents() line (StreamingLlmOperationsImpl.kt:148-155, StreamingChatClientOperations.kt:202-209).
So the default only ever applies to third-party implementors - who then silently get no thinking at all.
may be change .map { StreamingEvent.Object(it) by toTaggedThinkingEvents() at line 173 and remove implementation in both file
Contributor
|
@jstar0 @jorander @arnabnandy7 @azanux I debugged. Execution flows into: Note - .toolCallbacks(toolCallbacks) - deprecated in Spring 2.x No buffering at all - forgot about this:) But a more appropriate method name IMO might be "Thinking" in this context is given (putting aside the large subject of native thinking) as no object gets created, only LLM responses as strings |
Contributor
Appreciating for taking time for debugging this. I agree that the existing My concern is specifically about the new I don’t think routing through object creation with |
Contributor
Team, could you please retest #1826 at your convenience. thanks |
Contributor
|
@jstar0 - how are you? What is your tentative timeline for this PR? Thank you |
Contributor
|
@jstar0 @arnabnandy7 - as the stream tool loop is already in place: generateStream (line 74) returns raw token-by-token chunks directly from streamWithToolLoop. rawChunksToLines (line 338) already does exactly what's needed — buffers chunks and emits on The simplest fix is to apply it in doTransformStream before returning: return streamWithToolLoop(messagesWithContributions, tools, interaction, agentProcess, action) That reuses rawChunksToLines as-is. The only caveat: it only emits when it hits \n, so if the LLM produces one long paragraph with no newlines, everything buffers until doOnComplete Also, perhaps we can use the existing API and trigger thinking through LLM Options. Could you please evaluate the approach? Thanks. |
Contributor
|
Will be working on a simple yet practical solution post 1.5.0 with full code reusability from the blocking thinking path. Also in output is not triggered by native thinking. Appreciate if you please keep dev on hold, thanks. |
Contributor
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
6 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
StreamingPromptRunnerand the internal streaming operation boundary.Changes
generateStreamWithThinking()now returns orderedStreamingEvent.ThinkingandStreamingEvent.Objectvalues. The built-in streaming implementations parse the existingThinkingTagsformats from their current text streams, while production delegates route the new operation end to end.Parser state is scoped per subscription. Incomplete tagged blocks are preserved as text, and
//THINKING:is recognized only at a line boundary. The tests cover every split point for every configured tag, the legacy format, repeated subscriptions, delegation, and Java API exposure.Provider-native metadata transport, Spring AI provider normalization, tool-loop propagation, object-stream changes, blocking calls, and native enum semantics are intentionally deferred to a separate follow-up.
Verification
mvn -pl embabel-agent-api -Dkotlin.compiler.daemon=false -Dtest=StreamingThinkingJavaApiTest,StreamingLlmOperationsThinkingTest,DelegatingStreamingTest,StreamingImplTest,StreamingChatClientOperationsTest test mvn -pl embabel-agent-api -Dkotlin.compiler.daemon=false -DskipTests verify git diff --checkThe focused run passes 45 tests. The full
embabel-agent-apirun reaches 3,954 tests with no assertion failures; its 37 Spring context errors reproduce on the unchanged base under Java 26 and originate from Spring scanning an existing Kotlin local test class with aBad method descriptor.Part of #1716