Stream LLM token deltas over the conversation SSE endpoint - #1427
Open
yuyixg wants to merge 5 commits into
Open
Conversation
SendMessageSse only subscribed to OnIndicationReceived, so the token deltas the completion providers already raise as OnReceiveLlmStreamMessage never reached the response. They went only to the SignalR hub via ChatHubObserver, which is why BotSharp-UI shows a typewriter effect while an SSE client sees a single complete frame. Subscribing alone was not enough: ConversationObserver invoked listeners only inside its OnIndicationReceived branch, so a listener registered for any other event was silently dropped. Hoisting that dispatch out of the branches lets every registered listener receive its own event. The only other registration in the tree is Twilio's, also on OnIndicationReceived, so nothing dormant is activated. Frames are now written in one atomic write under a lock, because deltas are raised from the provider's execution context and can reach OnChunkReceived while the message callback is midway through a frame, and HttpResponse.Body is not thread-safe. The 10ms delay that sat between the two former writes is replaced by an explicit flush: at one delay per frame it would have added seconds of dead time across a token stream, and nothing was flushing before. Not verified end to end yet -- reproducing a token stream needs an LLM API key, which is not in the repo. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
The previous commit justified a lock by claiming the delta listener and the message callback can write a frame at the same time. That is wrong. MessageHub wraps its subject in Subject.Synchronize, so no two OnNext calls propagate concurrently, and Push runs synchronously on the caller's thread with no scheduler. ConversationObserver then invokes the listener with GetAwaiter().GetResult(), so the provider's streaming loop blocks until the frame is written. The message callback is reached later in the same await chain, via HandleAssistantMessage inside conv.SendMessage, once that loop has already finished. Routing and conversation carry no Task.WhenAll, Task.Run or Parallel work, so the whole path is one logical thread. What the previous commit changed for real reasons stays: one write per frame instead of two, no 10ms delay between them, and an explicit flush rather than relying on Kestrel's buffering to release a frame. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Kestrel's response body is a PipeWriter, and writing to it makes the bytes available to the transport, so the explicit flush after it did nothing. Verified by removing it and streaming a reply against a local OneBrain: 67 frames still arrived over 1.75s across 35 distinct timestamps, the same shape as with the flush in place. Worth recording what this method still costs, because the flush was never the expensive part. A delta frame serializes the whole ChatResponseModel for a few characters of text: measured over a 1008-frame reply, 5,419 bytes of text went out as 607,122 bytes, and 323 of the 602 bytes per frame were an empty Sender object repeated every token. Trimming what a delta carries is where the gain is, not the flushing. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
A delta reused ChatResponseModel, so five characters of text went out as 602 bytes; 323 of them were an empty Sender object, repeated once per token. Over a 1008-frame reply that turned 5,419 bytes of text into 607,122 bytes. Deltas now carry only what a consumer needs to place them: conversation id, message id, the streaming marker and the text. Measured on the same prompt afterwards: 155 bytes per frame, amplification down from 112x to 29.3x. The message id has to stay -- consumers reject a non-indicating frame without one and would drop every delta. Half of what is left is the two ids repeated per frame. Dropping the conversation id would save another 30%, at the cost of a delta no longer looking like the other frame kinds. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Both registered observers ran for every token. ConversationObserver reached the listener that writes the frame; ChatHubObserver answered the same event by building a DTO, serializing it, and awaiting a push to the SignalR group -- for a caller that is reading the response body and has no SignalR connection at all. With a Redis backplane configured, Clients.Group publishes whether or not the group has local members, so that was one Redis round trip per token, awaited synchronously on the completion loop's thread. The endpoint now names the observer it needs. SendMessage is untouched, and that is the endpoint BotSharp-UI posts to, so clients that do want events over SignalR are unaffected. The trade is that a conversation driven through /sse no longer feeds the SignalR hub, so watching the same conversation live in the UI while a caller drives it over SSE will show nothing. Picking one transport per conversation was already the assumption. Verified after the change: 82 delta frames still stream over 2.05s, 155 bytes each, and their concatenation still matches the closing frame's text. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
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
2 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.
SendMessageSse only subscribed to OnIndicationReceived, so the token deltas the completion providers already raise as OnReceiveLlmStreamMessage never reached the response. They went only to the SignalR hub via ChatHubObserver, which is why BotSharp-UI shows a typewriter effect while an SSE client sees a single complete frame.
Subscribing alone was not enough: ConversationObserver invoked listeners only inside its OnIndicationReceived branch, so a listener registered for any other event was silently dropped. Hoisting that dispatch out of the branches lets every registered listener receive its own event. The only other registration in the tree is Twilio's, also on OnIndicationReceived, so nothing dormant is activated.
Frames are now written in one atomic write under a lock, because deltas are raised from the provider's execution context and can reach OnChunkReceived while the message callback is midway through a frame, and HttpResponse.Body is not thread-safe. The 10ms delay that sat between the two former writes is replaced by an explicit flush: at one delay per frame it would have added seconds of dead time across a token stream, and nothing was flushing before.
Not verified end to end yet -- reproducing a token stream needs an LLM API key, which is not in the repo.