LlmComposer.Agent.StreamCollector (llm_composer v0.20.0)

Copy Markdown View Source

Accumulates the LlmComposer.StreamChunks of a single agent turn and reassembles them into a synthetic LlmComposer.LlmResponse.

Streaming providers emit tool calls incrementally and in provider-specific shapes:

  • OpenAI / OpenRouter send :tool_call_delta chunks whose tool_calls are raw maps keyed by "index", with the function.arguments JSON split across several chunks. They must be grouped by index and concatenated.
  • OpenAI Responses sends a "function_call_started" map (carries "call_id" and "name") followed by one or more "function_call_arguments_delta" maps (carry "call_id" and "arguments_delta"). Fragments are grouped by "call_id" and concatenated in arrival order.
  • Google sends already-complete LlmComposer.FunctionCall structs (one per chunk).
  • Bedrock sends two event types per tool call: a start event with "toolUseId" and "name", followed by one or more delta events that carry only "inputJson" fragments. Fragments are grouped by "toolUseId" and concatenated in arrival order.
  • Ollama (native provider) does not emit tool-call deltas in its streaming format; text streaming works. For tool-call streaming with Ollama, use the :open_ai provider pointed at Ollama's OpenAI-compatible endpoint.

This collector hides those differences. Feed every chunk of a turn through add/2, then call tool_turn?/1 to know whether the model requested tools, and to_llm_response/1 to obtain a response shaped exactly like a non-streaming one (so the rest of LlmComposer.Agent can reuse the synchronous loop unchanged).

Summary

Types

t()

Functions

add(collector, chunk)

Accumulates a single LlmComposer.StreamChunk into the collector.

aggregate_cost_infos(cost_infos)

Sums a list of per-turn LlmComposer.CostInfo into a single aggregate, or nil when empty.

new(provider)

Creates a collector for the given provider.

reasoning(stream_collector)

Returns the accumulated reasoning text, or nil when none was streamed.

to_function_calls(stream_collector)

Finalizes the accumulated tool-call deltas into complete LlmComposer.FunctionCall structs.

to_llm_response(collector)

Builds a synthetic LlmComposer.LlmResponse equivalent to a non-streaming response for the turn.

tool_turn?(stream_collector)

Returns true once any tool-call delta has been accumulated.

Types

t()

@type t() :: %LlmComposer.Agent.StreamCollector{
  cost_info: LlmComposer.CostInfo.t() | nil,
  current_tool_id: String.t() | nil,
  function_calls: [LlmComposer.FunctionCall.t()],
  provider: atom(),
  reasoning: iodata(),
  saw_tool_call: boolean(),
  text: iodata(),
  tool_fragments: %{optional(non_neg_integer() | String.t()) => map()},
  tool_id_sequence: [String.t()],
  usage: LlmComposer.StreamChunk.usage() | nil
}

Functions

add(collector, chunk)

@spec add(t(), LlmComposer.StreamChunk.t()) :: t()

Accumulates a single LlmComposer.StreamChunk into the collector.

aggregate_cost_infos(cost_infos)

@spec aggregate_cost_infos([LlmComposer.CostInfo.t()]) ::
  LlmComposer.CostInfo.t() | nil

Sums a list of per-turn LlmComposer.CostInfo into a single aggregate, or nil when empty.

Token counts are added and Decimal cost fields are summed. provider_name, provider_model and currency are taken from the first entry (a run may span models).

new(provider)

@spec new(atom()) :: t()

Creates a collector for the given provider.

Raises ArgumentError for providers that are not yet supported by the streaming agent.

reasoning(stream_collector)

@spec reasoning(t()) :: String.t() | nil

Returns the accumulated reasoning text, or nil when none was streamed.

to_function_calls(stream_collector)

@spec to_function_calls(t()) :: [LlmComposer.FunctionCall.t()]

Finalizes the accumulated tool-call deltas into complete LlmComposer.FunctionCall structs.

to_llm_response(collector)

@spec to_llm_response(t()) :: LlmComposer.LlmResponse.t()

Builds a synthetic LlmComposer.LlmResponse equivalent to a non-streaming response for the turn.

The returned response has an :assistant main_response carrying the accumulated text, reasoning and reassembled function calls, plus token usage and cost info captured from the stream.

tool_turn?(stream_collector)

@spec tool_turn?(t()) :: boolean()

Returns true once any tool-call delta has been accumulated.