Normandy.Components.StreamProcessor (normandy v2.2.0)

View Source

Utilities for processing streaming LLM responses.

Provides functions to accumulate streaming events into complete messages, handle text deltas, and build final responses from event streams.

Summary

Functions

Accumulates text deltas from a stream of events.

Builds a complete message from all stream events.

Returns the failure a collected stream carries, or nil for a complete stream.

Functions

accumulate_text(event_stream)

@spec accumulate_text(Enumerable.t()) :: Enumerable.t()

Accumulates text deltas from a stream of events.

Returns a stream that emits accumulated text chunks.

Example

events
|> StreamProcessor.accumulate_text()
|> Enum.each(&IO.write/1)

build_final_message(events)

@spec build_final_message([map()]) :: map()

Builds a complete message from all stream events.

Accumulates all events and constructs a final response structure similar to non-streaming responses.

Example

events = Enum.to_list(stream)
final_message = StreamProcessor.build_final_message(events)

stream_error(events)

@spec stream_error([term()]) :: Normandy.LLMError.t() | nil

Returns the failure a collected stream carries, or nil for a complete stream.

A stream fails when it contains an error event (e.g. a mid-stream overloaded_error), an {:error, reason} item (a transport failure after the stream started), or ends without a message_stop event. The result is a Normandy.LLMError with type: :stream, retryable?: false (callbacks have already received partial output).