Multiline Streaming Aggregator

[igordayen]

In streaming mode, getting results as soon as they arrive is paramount.
However, there is also a practical need to massage data before responding to the user.
Example:
line: <think> abc...
line: ```xyz

Ideally, the user would be interested in getting an aggregated cross-line response, such as "abc...xyz".

An aggregator can be used as an additional streaming callback in parallel to getting low-level chunks of data.

@jorander @arnabnandy7 - FYI

[arnabnandy7]

A few things food for thought:

• what happens if a delimiter never closes (truncated/error stream) - do we flush on end or cap the buffer?
• also tag boundaries can split mid-chunk depending on provider SSE chunking, not just at line breaks, so the parser needs to handle partial matches across calls.
• since this is a second parallel callback, worth being explicit about ordering guarantees vs the raw chunk stream so consumers don't get confused subscribing to both.

@igordayen @jorander

[jorander]

line: <think> abc...
line: xyz

These two lines will probably arrive as a list of response (content) SSE-events from the LLM. E.g [<,th,ink,>, abc,\n,x,yz].

Optimally we would like to process this so that each event recieved (except for the <think>-tag) would generate a StreamEvent, in this case with the abc, \n, x and yz as thinking content. Four StreamEvents in total.

To do that, we probably need to introduce some buffering since we cannot decide from the first token (<) wether it is the start of a <think>-tag or just a single "less than" in the text stream or the start of another tag that does not mark the start of a thinking block (e.g. if the whole response i XML or HTML). I think the groundrules for the implementation should be:

1. Convert every recieved event to its proper StreamEvent and send it immediately, unless
2. the content of the event potentially could be the start of a thinking delimiter. If so, buffer until the new (content/thinking) state could be determined, and then
3. convert buffered events to its proper StreamEvent and send them immediately. Then goto 1.

@igordayen @arnabnandy7

[arnabnandy7]

Connecting #1873

[igordayen]

@jorander @arnabnandy7 Thanks for continuing the discussion.

Aggregator abstraction to include rules:

1. regexp-start regexp-end rule
2. capacity control rule

Once regexp-start is detected:
Aggregator continues to accumulate tokens till one of the events occurs: regexp-end is detected, OR buffer size crosses the limits.

As such, the user may receive a callback on reasoning-start-tag with the following text, without a reasoning-end-tag.

Streaming indeed sends mini-tokens, so <think> comes char by char.

That is one of the reasons for having an out-of-the-box standardized NL-delimiter buffering mechanism.

It does not cover all scenarios, but practically most.

Aggregator is an add-on to the framework; the framework continues "as-is", and aggregator adds parallel, extensible callbacks.