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
@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)
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)
@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).