Implementation:BerriAI Litellm Streaming Chunk Builder

Attribute	Value
Sources	litellm/litellm_core_utils/streaming_chunk_builder_utils.py
Domains	Streaming, Response Assembly, Token Counting, Tool Calls
last_updated	2026-02-15 16:00 GMT

Overview

The Streaming Chunk Builder provides the ChunkProcessor class that reassembles streaming response chunks into a complete ModelResponse object, including content concatenation, tool call merging, thinking block assembly, audio content handling, and usage calculation.

Description

When LiteLLM processes streaming responses from LLM providers, it receives a sequence of ModelResponseStream chunks, each containing incremental delta content. The ChunkProcessor class is responsible for combining these chunks into a single, complete ModelResponse.

Key responsibilities:

• Chunk sorting -- Sorts chunks by their created_at hidden parameter if available, ensuring correct ordering.
• Base response construction -- build_base_response() creates the skeleton ModelResponse from the first chunk's metadata (id, model, created, system_fingerprint) while scanning all chunks for the actual model name (important for Azure Model Router where the first chunk may have a generic model name).
• Content assembly:
  • get_combined_content() -- Concatenates text content from all delta chunks.
  • get_combined_reasoning_content() -- Concatenates reasoning content from reasoning_content delta key.
  • get_combined_thinking_content() -- Assembles Anthropic thinking blocks, handling both thinking and redacted_thinking block types with signature-based flush logic.
  • get_combined_audio_content() -- Concatenates base64-encoded audio data and transcripts from audio deltas into a ChatCompletionAudioResponse.
• Tool call merging -- get_combined_tool_content() collects tool call fragments across chunks, grouping by index, concatenating function arguments, and preserving provider_specific_fields (e.g., Gemini thought signatures). Handles both dict and object tool call formats.
• Function call merging -- get_combined_function_call_content() handles legacy function_call format.
• Usage calculation -- calculate_usage() aggregates token counts from chunk usage data, falling back to token_counter() when provider-reported counts are unavailable. Handles:
  • Prompt and completion tokens
  • Anthropic prompt caching tokens (cache_creation_input_tokens, cache_read_input_tokens)
  • Completion token details (reasoning tokens, etc.)
  • Prompt token details (web search requests)
  • Server tool use tracking

The module also provides concatenate_base64_list() as a standalone utility for merging base64-encoded audio chunks.

Usage

Used internally by the streaming chunk builder when finalizing a streaming response:

from litellm.litellm_core_utils.streaming_chunk_builder_utils import ChunkProcessor

Code Reference

Source Location

/litellm/litellm_core_utils/streaming_chunk_builder_utils.py (686 lines)

Class: ChunkProcessor

Method	Signature	Purpose
__init__	def __init__(self, chunks: List, messages: Optional[list] = None)	Sorts chunks and stores reference data
build_base_response	def build_base_response(self, chunks: List[Dict]) -> ModelResponse	Creates skeleton response from chunk metadata
get_combined_content	def get_combined_content(self, chunks, delta_key="content") -> str	Concatenates text content from deltas
get_combined_tool_content	def get_combined_tool_content(self, tool_call_chunks) -> List[ChatCompletionMessageToolCall]	Merges tool call fragments by index
get_combined_thinking_content	def get_combined_thinking_content(self, chunks) -> Optional[List[...]]	Assembles thinking and redacted thinking blocks
get_combined_audio_content	def get_combined_audio_content(self, chunks) -> ChatCompletionAudioResponse	Merges base64 audio data and transcripts
get_combined_reasoning_content	def get_combined_reasoning_content(self, chunks) -> str	Concatenates reasoning content deltas
get_combined_function_call_content	def get_combined_function_call_content(self, chunks) -> FunctionCall	Merges legacy function call arguments
calculate_usage	def calculate_usage(self, chunks, model, completion_output, messages=None, reasoning_tokens=None) -> Usage	Computes complete usage statistics
count_reasoning_tokens	def count_reasoning_tokens(self, response: ModelResponse) -> int	Counts tokens in reasoning content

Standalone Function

Function	Signature	Purpose
concatenate_base64_list	def concatenate_base64_list(base64_strings: List[str]) -> str	Decodes, concatenates, and re-encodes base64 strings

Import

from litellm.litellm_core_utils.streaming_chunk_builder_utils import (
    ChunkProcessor,
    concatenate_base64_list,
)

I/O Contract

Inputs (ChunkProcessor.__init__)

Parameter	Type	Description
chunks	List[ModelResponseStream]	Ordered list of streaming response chunks
messages	Optional[list]	Original input messages (for token counting fallback)

Inputs (calculate_usage)

Parameter	Type	Description
chunks	List[Union[Dict, ModelResponse]]	Response chunks with usage data
model	str	Model name for token counting
completion_output	str	Combined completion text for token counting
messages	Optional[List]	Input messages for prompt token counting
reasoning_tokens	Optional[int]	Pre-counted reasoning tokens

Outputs

Return	Type	Description
Base response	ModelResponse	Skeleton response with metadata
Combined content	str	Concatenated text content
Tool calls	List[ChatCompletionMessageToolCall]	Merged tool call objects
Usage	Usage	Complete token usage statistics

Usage Examples

from litellm.litellm_core_utils.streaming_chunk_builder_utils import ChunkProcessor

# Process accumulated streaming chunks
processor = ChunkProcessor(chunks=collected_chunks, messages=original_messages)

# Build the complete response
chunks_as_dicts = [chunk.model_dump() for chunk in collected_chunks]
response = processor.build_base_response(chunks_as_dicts)

# Assemble content
content = processor.get_combined_content(chunks_as_dicts)
response.choices[0].message.content = content

# Merge tool calls if present
tool_calls = processor.get_combined_tool_content(chunks_as_dicts)
if tool_calls:
    response.choices[0].message.tool_calls = tool_calls

# Calculate usage
usage = processor.calculate_usage(
    chunks=chunks_as_dicts,
    model="gpt-4",
    completion_output=content,
    messages=original_messages,
)
response.usage = usage

Related Pages

• BerriAI_Litellm_Core_Helpers - provides map_finish_reason used during response building
• BerriAI_Litellm_Constants - defines token counting constants
• BerriAI_Litellm_Logging_Worker - logging callbacks triggered after response assembly