# `ReqLLM.Provider.ChunkAccumulator` [šŸ”—](https://github.com/agentjido/req_llm/blob/main/lib/req_llm/provider/chunk_accumulator.ex#L1) Shared streaming-chunk reducer used by both `ReqLLM.StreamServer` (the hot path, one chunk at a time) and `ReqLLM.Provider.Defaults.ResponseBuilder` (batch, full chunk list at end-of-stream). Maintains running iodata buffers for text/thinking, ordered content events, complete content parts, a running tool-call list, and per-index argument-fragment buffers. Reasoning details, logprobs, and annotations are also collected from `:meta` chunks. ## Finalizers Two different finalizers exist because StreamServer and ResponseBuilder consume the accumulator differently: * `finalize_tool_calls_for_response/1` ā€” preserves the historical `ResponseBuilder` contract: returns raw maps with `:id`, `:name`, `:arguments` (decoded JSON when fragments are present, else the raw args from the tool_call chunk). Used to feed `ResponseBuilder.normalize_tool_calls/1`. * `finalize_message/1` ā€” preserves the historical `StreamServer` contract: returns either `nil` (empty acc) or an assistant `%ReqLLM.Message{}` ready to attach to OTel content-capture metadata. Adjacent text chunks become a single `:text` `ContentPart`; complete content parts are retained; tool calls become `%ReqLLM.ToolCall{}` structs (with builtin flag preserved). Reasoning text is intentionally not surfaced through `finalize_message/1` ā€” OTel content capture redacts it anyway and the canonical response message is built separately by `ResponseBuilder` with full reasoning details. ## Performance notes The accumulator is on the streaming hot path. To keep `push/2` O(1) per chunk we prepend list entries (tool calls, reasoning details, logprobs, annotations) and reverse them at finalize time. Text and thinking buffers are iodata ā€” also O(1) per chunk. Argument fragments are iodata buffers keyed by tool-call index, joined only at finalize time. A stream with N chunks costs O(N) total work, not O(NĀ²). ## Text boundaries A `:content` chunk whose metadata carries both `:phase` and `:output_index` belongs to one labelled output item (OpenAI Responses assistant phases). Adjacent text chunks merge only when they share that boundary, so two consecutive items stay separate parts and each part keeps `%{phase: phase}` in its metadata. Chunks without a boundary merge with each other exactly as plain text always does. `ordered_content?/1` reports when the accumulated content needs `finalize_ordered_content/2` to keep that structure. # `content_event` ```elixir @type content_event() :: {:text, iodata(), text_boundary()} | {:thinking, iodata()} | {:content_part, ReqLLM.Message.ContentPart.t()} ``` # `t` ```elixir @type t() :: %ReqLLM.Provider.ChunkAccumulator{ annotations: [term()], arg_fragments: %{optional(non_neg_integer()) => iodata()}, content_events: [content_event()], content_parts: [ReqLLM.Message.ContentPart.t()], finish_reason: atom() | String.t() | nil, logprobs: [term()], reasoning_details: [term()], segmented_text?: boolean(), text_content: iodata(), thinking_content: iodata(), tool_call_metadata: %{optional(non_neg_integer()) => map()}, tool_calls: [tool_call_record()], usage: map() | nil } ``` # `text_boundary` ```elixir @type text_boundary() :: {String.t(), non_neg_integer()} | nil ``` # `tool_call_record` ```elixir @type tool_call_record() :: %{ :id => String.t(), :name => String.t(), :arguments => term(), :index => non_neg_integer(), optional(:builtin?) => true, optional(:expects_arg_fragments) => true, optional(:metadata) => map() } ``` # `finalize_annotations` ```elixir @spec finalize_annotations(t()) :: [term()] ``` Returns annotations in arrival order, with exact duplicates dropped. Deduplication matters because a provider may re-send an annotation it already streamed (or emit both an incremental event and a final list). # `finalize_content_parts` ```elixir @spec finalize_content_parts(t()) :: [ReqLLM.Message.ContentPart.t()] ``` Returns complete content parts in arrival order. # `finalize_finish_reason` ```elixir @spec finalize_finish_reason(t()) :: atom() | String.t() | nil ``` Returns the most recently observed `finish_reason` from meta chunks, or `nil` if no meta chunk surfaced one. The value is returned raw (atom or string) ā€” callers normalize. # `finalize_logprobs` ```elixir @spec finalize_logprobs(t()) :: [term()] ``` Returns logprob tokens in arrival order. # `finalize_message` ```elixir @spec finalize_message(t()) :: ReqLLM.Message.t() | nil ``` Returns a partial assistant `%ReqLLM.Message{}` for OTel content capture, or `nil` when the accumulator has no text and no tool calls. Reasoning is intentionally `nil` ā€” OTel content capture redacts it. # `finalize_ordered_content` ```elixir @spec finalize_ordered_content(t(), keyword()) :: [ReqLLM.Message.ContentPart.t()] ``` Returns text, thinking, and complete content parts in arrival order. Adjacent text or thinking chunks become one content part. Adjacent text chunks with different boundaries stay separate, and a bounded text part carries `%{phase: phase}` in its metadata. Set `:include_thinking?` to `false` to omit thinking content. # `finalize_reasoning_details` ```elixir @spec finalize_reasoning_details(t()) :: [term()] ``` Returns reasoning details in arrival order. # `finalize_text` ```elixir @spec finalize_text(t()) :: String.t() ``` Returns the concatenated text content as a binary. # `finalize_thinking` ```elixir @spec finalize_thinking(t()) :: String.t() ``` Returns the concatenated thinking content as a binary. # `finalize_tool_calls_for_response` ```elixir @spec finalize_tool_calls_for_response(t()) :: [map()] ``` Returns tool calls in the format `ResponseBuilder.normalize_tool_calls/1` expects: maps with `:id`, `:name`, `:arguments`, and optionally a `:builtin?` flag. If argument fragments were observed and decode successfully, arguments are the decoded JSON; otherwise they fall back to the raw arguments captured from the tool_call chunk. # `finalize_usage` ```elixir @spec finalize_usage(t()) :: map() | nil ``` Returns the merged usage map (or `nil` if no meta chunk surfaced usage). # `new` ```elixir @spec new() :: t() ``` Returns an empty accumulator. # `ordered_content?` ```elixir @spec ordered_content?(t()) :: boolean() ``` Returns `true` when the accumulated content has structure that a single aggregated text part would lose: a complete content part, or text chunks that carry an output-item boundary. Callers use `finalize_ordered_content/2` in that case. # `push` ```elixir @spec push(t(), ReqLLM.StreamChunk.t()) :: t() ``` Folds a single chunk into the accumulator. Hot path ā€” O(1) per chunk. # `reduce` ```elixir @spec reduce(t(), [ReqLLM.StreamChunk.t()]) :: t() ``` Folds a list of chunks through `push/2`. Convenience wrapper for the batch path (`ResponseBuilder`). --- *Consult [api-reference.md](api-reference.md) for complete listing*