Package org.springframework.ai.chat.messages.part

Class StreamingPartIndexer

java.lang.Object
org.springframework.ai.chat.messages.part.StreamingPartIndexer
public final class StreamingPartIndexer extends Object
Synthesizes the StreamingParts index for providers that stream message content without a content block index, so that MessageAggregator can rebuild the complete parts from the chunks.

The aggregator merges streamed parts by index: a partial part is appended to what was received earlier at the same index, and a complete part replaces it. Providers that stream content blocks, such as Anthropic, send that index with every delta. Others, such as Chat Completions, Gemini or Ollama, send reasoning deltas, text deltas and complete tool calls without an index, possibly several of them in one chunk. This class derives the index from the order in which the parts arrive:

  • A ReasoningPart or TextPart is a delta. It continues the current index when the previous delta was of the same kind, and starts the next index otherwise. So a run of reasoning deltas becomes one ReasoningPart, and a run of text deltas one TextPart.
  • Any other part, such as a tool call, is complete when it arrives and gets the next index for itself. It also ends the current run, so a text delta that follows it starts a new text part.
  • An empty TextPart without a payload, such as the placeholder of a chunk that only carries usage, carries nothing. It is returned unchanged and does not end the current run, so that such a chunk in the middle of the reasoning does not split it in two.
For example, the deltas reasoning("Think "), reasoning("more."), text("Hel") and text("lo"), followed by one tool call, are stamped as partial 0, partial 0, partial 1, partial 1 and complete 2. They aggregate to a reasoning part "Think more.", a text part "Hello" and the tool call, in that order.

A caller that must not index a part, for instance to keep streamed media out of the aggregated message, keeps that part on the chunk unstamped rather than passing it to stamp(MessagePart).

The aggregator groups indexed parts by response id, so one instance must serve all the parts that share a response id, in the order they appear in the stream. It is stateful and not thread-safe; the chunks of a stream are processed one at a time.

Since:
2.1.0
Author:
Christian Tzolov, Dimitar Proynov
See Also:

  • Constructor Details

    • StreamingPartIndexer

      public StreamingPartIndexer()

  • Method Details

    • stamp

      public MessagePart stamp(MessagePart part)
      Returns a copy of the part with its stream index set, and advances this indexer. Call it once per part, in the order the parts appear in the stream.
      Parameters:
      part - the part of a streamed chunk, without an index
      Returns:
      an empty TextPart without a payload unchanged; any other TextPart or a ReasoningPart stamped as partial, with the index of the current run of deltas of its kind, or the next index when it starts a new run; any other part stamped as complete, with the next index