ReqLLM.Response.Stream (ReqLLM v1.13.0)

Stream processing utilities for ReqLLM responses.

This module contains helper functions for working with streaming responses, particularly for joining stream chunks into complete responses.

Summary

Types

summary()

Summary of accumulated stream data.

Functions

join(stream, response)

Join a stream of chunks into a complete response.

summarize(chunks)

Summarize a stream of chunks into accumulated data.

Types

summary()

@type summary() :: %{
  text: String.t(),
  thinking: String.t(),
  tool_calls: [map()],
  finish_reason: atom() | nil,
  usage: map() | nil
}

Summary of accumulated stream data.

Contains all the extracted content from a stream of chunks, suitable for building responses or classifying stream results.

Functions

join(stream, response)

@spec join(Enumerable.t(), ReqLLM.Response.t()) ::
  {:ok, ReqLLM.Response.t()} | {:error, term()}

Join a stream of chunks into a complete response.

This function consumes the entire stream, builds the complete message from content chunks, and returns a new response with the stream consumed and message populated.

Implementation Notes

The joining process involves several steps:

Collect all stream chunks by consuming the enumerable
Filter and concatenate content chunks to build the response text
Extract final usage statistics from meta chunks, merging with existing usage
Build a complete assistant message with the concatenated text content
Return an updated response with materialized data and stream cleared

Parameters

stream - The stream enumerable containing stream chunks
response - The original response to update with materialized data

Returns

{:ok, updated_response} on success
{:error, %ReqLLM.Error.API.Stream{}} on stream processing failure

summarize(chunks)

@spec summarize(Enumerable.t()) :: summary()

Summarize a stream of chunks into accumulated data.

Processes all chunks and returns a map with:

text - Accumulated text content
thinking - Accumulated thinking/reasoning content
tool_calls - List of reconstructed tool calls with merged argument fragments
finish_reason - The finish reason from metadata chunks (normalized to atom)
usage - Token usage statistics from metadata chunks

This function is the shared core for both join/2 and ReqLLM.Stream.ToolCalls. It delegates accumulation to ReqLLM.Provider.ChunkAccumulator so the streaming chunk reducer stays a single source of truth.

Examples

chunks = Enum.to_list(stream_response.stream)
summary = ReqLLM.Response.Stream.summarize(chunks)
summary.text        #=> "Hello, world!"
summary.tool_calls  #=> [%{id: "call_123", name: "get_weather", arguments: %{...}}]