From 46807759117a45836fbede0ea32dcf1411192be6 Mon Sep 17 00:00:00 2001 From: inchang-ing <197932532+inchang-ing@users.noreply.github.com> Date: Thu, 29 Oct 2026 17:29:52 +0800 Subject: [PATCH] fix(streaming_handler): coerce list delta content to text Rebase onto current main; keep the delta-content-list accumulation fix and null-text handling alongside main's dict-usage streaming tests. --- .../streaming_chunk_builder_utils.py | 25 ++++++- .../litellm_core_utils/streaming_handler.py | 25 ++++++- .../test_streaming_chunk_builder_utils.py | 65 +++++++++++++++++++ .../test_streaming_handler.py | 63 ++++++++++++++++++ 4 files changed, 173 insertions(+), 5 deletions(-) diff --git a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py index 719f75546b9..d99ee23bfae 100644 --- a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py +++ b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py @@ -708,7 +708,7 @@ class ChunkProcessor: content = delta.get(delta_key, "") if content is None: continue # openai v1.0.0 sets content = None for chunks - content_list.append(content) + content_list.append(delta_content_as_str(content)) # Combine the "content" strings into a single string || combine the 'function' strings into a single string combined_content: Final = "".join(content_list) @@ -1478,12 +1478,33 @@ def _delta_carries_more_than_text(delta: Mapping[str, object]) -> bool: return any(delta.get(field) is not None for field in _NON_TEXT_DELTA_FIELDS) +def delta_content_as_str(content: object) -> str: + """Coerce a streamed delta content value to plain text. + + OpenAI-compatible providers are supposed to stream delta.content as a + string, but some stream a list of content parts instead. Join the text + parts into one string and skip every part that carries no usable text, so + the running answer text neither loses the content nor raises when a part + holds a null value or is not a mapping at all. + """ + if isinstance(content, str): + return content + if not isinstance(content, list): + return "" + texts: Final[list[str]] = [] # mutable-ok: accumulator of streamed delta content text parts + for part in content: # pyright: ignore[reportUnknownVariableType] # delta content parts are dynamically typed + text = part.get("text") if isinstance(part, Mapping) else getattr(part, "text", None) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType, reportUnknownArgumentType] # dynamic delta parts + if isinstance(text, str): # pyright: ignore[reportUnknownArgumentType] # dynamic text value + texts.append(text) # pyright: ignore[reportUnknownArgumentType] # dynamic text value + return "".join(texts) + + def _simple_text_part(choices: Sequence[object]) -> str | None: deltas: Final = tuple(_stream_choice_delta(choice) for choice in choices) if any(_delta_carries_more_than_text(delta) for delta in deltas): return None content: Final = deltas[0].get("content") - return content if isinstance(content, str) else "" + return delta_content_as_str(content) _NO_TRACE_FIELDS: Final[Mapping[str, object]] = MappingProxyType({}) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 5542249d2c7..68f1b230378 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -980,9 +980,17 @@ class CustomStreamWrapper: model_response: ModelResponseStream, response_obj: Mapping[str, object], ) -> bool: + from litellm.litellm_core_utils.streaming_chunk_builder_utils import ( + delta_content_as_str, + ) + + # Providers may stream delta.content as a list of content parts; fold it to + # text here too, otherwise every chunk after the first is judged empty and + # dropped (only the first survives, via the role clause below). + content_str: Final = delta_content_as_str(completion_obj.get("content")) if ( "content" in completion_obj - and (isinstance(completion_obj["content"], str) and len(completion_obj["content"]) > 0) + and len(content_str) > 0 or ( "tool_calls" in completion_obj and completion_obj["tool_calls"] is not None @@ -2559,9 +2567,20 @@ class CustomStreamWrapper: def _accumulate_streamed_delta(self, delta: object) -> None: # Shared by the sync, async, and non-aiohttp iteration sites so answer # text and the disqualifying-content latch stay in step across all three. + from litellm.litellm_core_utils.streaming_chunk_builder_utils import ( + delta_content_as_str, + ) + get: Final = getattr(delta, "get", None) - content: Final = get("content") if callable(get) else None - self.response_uptil_now += content if isinstance(content, str) else "" + content: Final = get("content") if callable(get) else getattr(delta, "content", None) + if isinstance(content, str): + self.response_uptil_now += content + else: + # Providers that stream delta.content as a list of content parts must + # still accumulate as plain text. delta_content_as_str folds those + # parts and yields "" for anything that is not a content list, which is + # exactly what a non-text delta contributed before. + self.response_uptil_now += delta_content_as_str(content) if not self._emitted_disqualifying_content and self._delta_disqualifies_continuation(delta): self._emitted_disqualifying_content = True diff --git a/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py b/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py index dd1f495b0d0..12da6a66b8a 100644 --- a/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py +++ b/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py @@ -2286,6 +2286,71 @@ def test_stream_chunk_builder_rebuilds_each_text_completion_choice_by_index() -> assert (response.usage.prompt_tokens, response.usage.completion_tokens) == (1, 4) +def test_stream_chunk_builder_joins_content_part_list_delta(): + """A provider streaming delta.content as a content-part list must aggregate as text.""" + chunks = [ + ModelResponseStream( + id="chatcmpl-123", + created=1, + model="gpt-4.1-mini", + object="chat.completion.chunk", + choices=[ + StreamingChoices( + finish_reason=None, + index=0, + delta=Delta(content=[{"type": "text", "text": "Hello "}], role="assistant"), + ) + ], + ), + ModelResponseStream( + id="chatcmpl-123", + created=2, + model="gpt-4.1-mini", + object="chat.completion.chunk", + choices=[ + StreamingChoices( + finish_reason="stop", + index=0, + delta=Delta(content=[{"type": "text", "text": "world"}], role=None), + ) + ], + ), + ] + + response = stream_chunk_builder(chunks=chunks) + assert response is not None + assert response.choices[0].message.content == "Hello world" + + +def test_delta_content_as_str_skips_parts_without_usable_text(): + """Parts holding a null value or not being mappings must not abort the join.""" + from litellm.litellm_core_utils.streaming_chunk_builder_utils import delta_content_as_str + + assert delta_content_as_str([{"type": "text", "text": "a"}, {"type": "text", "text": "b"}]) == "ab" + assert delta_content_as_str([{"type": "text", "text": None}]) == "" + assert delta_content_as_str(["bare string", 42, None]) == "" + assert ( + delta_content_as_str( + [ + {"type": "text", "text": "a"}, + {"type": "text", "text": None}, + {"type": "text", "text": "b"}, + ] + ) + == "ab" + ) + + +def test_delta_content_as_str_reads_text_from_non_mapping_parts(): + """A part exposing only a ``text`` attribute still contributes its text.""" + from litellm.litellm_core_utils.streaming_chunk_builder_utils import delta_content_as_str + + class _TextPart: + text = "from attribute" + + assert delta_content_as_str([_TextPart()]) == "from attribute" + + def test_stream_chunk_builder_keeps_null_text_completion_logprobs_fields_null() -> None: response: Final = litellm.stream_chunk_builder( chunks=[ diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index f4502937fce..13c48c48e8f 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -8248,3 +8248,66 @@ def test_chunk_creator_usage_only_chunk_with_dict_usage_keeps_details( assert usage.prompt_tokens_details.cached_tokens == 80 assert usage.completion_tokens_details is not None assert usage.completion_tokens_details.reasoning_tokens == 5 + + +def _chunk_with_list_content(text: str, chunk_id: str, created: int) -> ModelResponseStream: + return ModelResponseStream( + id=chunk_id, + object="chat.completion.chunk", + created=created, + model="gpt-5.6", + choices=[ + StreamingChoices( + index=0, + delta=Delta(role="assistant", content=[{"type": "text", "text": text}]), + finish_reason=None, + ) + ], + ) + +def test_streaming_delta_content_list_accumulates_as_str(): + """A provider streaming delta.content as a content-part list must still accumulate as text.""" + chunks: Final = [ + _chunk_with_list_content("hello ", "chunk1", 1741037890), + _chunk_with_list_content("world", "chunk2", 1741037891), + ] + + response: Final = CustomStreamWrapper( + completion_stream=iter(chunks), + model="gpt-5.6", + logging_obj=MagicMock(), + custom_llm_provider="openai", + ) + + for _ in response: + pass + + assert response.response_uptil_now == "hello world" + +def test_streaming_accumulates_content_part_with_null_text(): + """A content part carrying a null value must not abort the accumulated stream.""" + chunk: Final = ModelResponseStream( + id="chunk-null", + object="chat.completion.chunk", + created=1741037892, + model="gpt-5.6", + choices=[ + StreamingChoices( + index=0, + delta=Delta(role="assistant", content=[{"type": "text", "text": None}]), + finish_reason=None, + ) + ], + ) + + response: Final = CustomStreamWrapper( + completion_stream=iter([chunk]), + model="gpt-5.6", + logging_obj=MagicMock(), + custom_llm_provider="openai", + ) + + for _ in response: + pass + + assert response.response_uptil_now == ""