diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py index 8c20f4c430e..f049abcf47f 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -469,12 +469,16 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if should_start_new_block and not self.sent_content_block_finish: # Queue the sequence: content_block_stop -> content_block_start - # For text blocks the trigger chunk is not emitted as a separate - # delta because content_block_start carries the information. - # For tool_use blocks we must also emit the trigger chunk's delta - # when it carries input_json_delta data, because some providers - # (e.g. xAI, Gemini) include tool arguments in the same streaming - # chunk as the function name/id. + # -> (optionally) the trigger chunk's delta. + # + # The synthesized content_block_start always carries an + # empty body, so the chunk that *triggered* the transition + # also carries the new block's first delta. It must be + # re-emitted or the first token of the new block is lost. + # This applies to text_delta and thinking_delta (the first + # non-empty text/thinking token) as well as input_json_delta + # (providers like xAI/Gemini bundle tool arguments with the + # function name/id in a single chunk). # 1. Stop current content block self.chunk_queue.append( @@ -493,14 +497,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): } ) - # 3. If the trigger chunk carries tool argument data, queue it - # so the input_json_delta is not silently dropped. - if ( - processed_chunk.get("type") == "content_block_delta" - and isinstance(processed_chunk.get("delta"), dict) - and processed_chunk["delta"].get("type") == "input_json_delta" - and processed_chunk["delta"].get("partial_json") - ): + # 3. If the trigger chunk carries delta content, queue it + # so the first delta of the new block is not silently dropped. + if self._trigger_delta_has_content(processed_chunk): self.chunk_queue.append(processed_chunk) self.sent_content_block_finish = False @@ -711,12 +710,16 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if not self.queued_usage_chunk: if should_start_new_block and not self.sent_content_block_finish: # Queue the sequence: content_block_stop -> content_block_start - # For text blocks the trigger chunk is not emitted as a separate - # delta because content_block_start carries the information. - # For tool_use blocks we must also emit the trigger chunk's delta - # when it carries input_json_delta data, because some providers - # (e.g. xAI, Gemini) include tool arguments in the same streaming - # chunk as the function name/id. + # -> (optionally) the trigger chunk's delta. + # + # The synthesized content_block_start always carries an + # empty body, so the chunk that *triggered* the transition + # also carries the new block's first delta. It must be + # re-emitted or the first token of the new block is lost. + # This applies to text_delta and thinking_delta (the + # first non-empty text/thinking token) as well as + # input_json_delta (providers like xAI/Gemini bundle tool + # arguments with the function name/id in a single chunk). # 1. Stop current content block self.chunk_queue.append( @@ -733,15 +736,9 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): } ) - # 3. If the trigger chunk carries tool argument data, queue it - # so the input_json_delta is not silently dropped. - if ( - processed_chunk.get("type") == "content_block_delta" - and isinstance(processed_chunk.get("delta"), dict) - and processed_chunk["delta"].get("type") - == "input_json_delta" - and processed_chunk["delta"].get("partial_json") - ): + # 3. If the trigger chunk carries delta content, queue it + # so the first delta of the new block is not silently dropped. + if self._trigger_delta_has_content(processed_chunk): self.chunk_queue.append(processed_chunk) # Reset state for new block @@ -898,6 +895,38 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): def _increment_content_block_index(self): self.current_content_block_index += 1 + @staticmethod + def _trigger_delta_has_content(processed_chunk: Dict[str, Any]) -> bool: + """Return True if a translated trigger chunk carries a non-empty + ``content_block_delta`` payload that must be re-emitted after a + block transition. + + When an upstream chunk both *triggers* a new content block (its type + differs from the active block) and *carries* delta content, that + content belongs to the new block. The synthesized + ``content_block_start`` only ever carries an empty body — see + ``_translate_streaming_openai_chunk_to_anthropic_content_block``, + which returns an empty ``TextBlock``/``ToolUseBlock``/thinking block — + so the trigger chunk's delta must be re-queued or the first token of + the new block (the first non-empty text/thinking delta, or bundled + tool arguments) is silently dropped. + """ + if processed_chunk.get("type") != "content_block_delta": + return False + delta = processed_chunk.get("delta") + if not isinstance(delta, dict): + return False + delta_type = delta.get("type") + if delta_type == "text_delta": + return bool(delta.get("text")) + if delta_type == "input_json_delta": + return bool(delta.get("partial_json")) + if delta_type == "thinking_delta": + return bool(delta.get("thinking")) + if delta_type == "signature_delta": + return bool(delta.get("signature")) + return False + def _should_start_new_content_block(self, chunk: "ModelResponseStream") -> bool: """ Determine if we should start a new content block based on the processed chunk. diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py new file mode 100644 index 00000000000..8bc39a6d85e --- /dev/null +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py @@ -0,0 +1,312 @@ +""" +Regression tests for issue #30014. + +When LiteLLM proxies ``client -> /v1/messages -> /v1/chat/completions`` and a +streaming chunk both *triggers* a new Anthropic content block (its type differs +from the active block) and *carries* the first delta of that new block, the +trigger chunk's delta must be re-emitted as a ``content_block_delta``. + +The synthesized ``content_block_start`` always carries an empty body, so before +the fix the first non-empty ``text_delta`` of every transitioned block was +silently dropped — e.g. text resuming after a tool call started from the second +token ("The weather is nice." was lost, "Hi" rendered as ""). Bundled +``input_json_delta`` tool arguments were already preserved and must stay +preserved, and empty trigger deltas must not produce spurious events. +""" + +import os +import sys +from typing import List, Optional +from unittest.mock import MagicMock + +import pytest + +sys.path.insert(0, os.path.abspath("../../../../..")) + +from litellm.llms.anthropic.experimental_pass_through.adapters.streaming_iterator import ( + AnthropicStreamWrapper, +) +from litellm.types.utils import ( + ChatCompletionDeltaToolCall, + Delta, + Function, + StreamingChoices, +) + + +def _make_chunk(delta: Delta, finish_reason: Optional[str] = None) -> MagicMock: + chunk = MagicMock() + chunk.choices = [ + StreamingChoices( + finish_reason=finish_reason, + index=0, + delta=delta, + logprobs=None, + ) + ] + chunk.usage = None + chunk._hidden_params = {} + return chunk + + +def _tool_chunk( + call_id: str, name: Optional[str], arguments: Optional[str] +) -> MagicMock: + return _make_chunk( + Delta( + content=None, + tool_calls=[ + ChatCompletionDeltaToolCall( + id=call_id, + function=Function(name=name, arguments=arguments), + type="function", + index=0, + ) + ], + ) + ) + + +class _AsyncStream: + def __init__(self, items: List[MagicMock]): + self._it = iter(items) + + def __aiter__(self): + return self + + async def __anext__(self): + try: + return next(self._it) + except StopIteration: + raise StopAsyncIteration + + +def _drain_sync(wrapper: AnthropicStreamWrapper) -> List[dict]: + return list(wrapper) + + +async def _drain_async(wrapper: AnthropicStreamWrapper) -> List[dict]: + return [event async for event in wrapper] + + +def _text_deltas(events: List[dict]) -> List[str]: + return [ + e["delta"]["text"] + for e in events + if e.get("type") == "content_block_delta" + and e["delta"].get("type") == "text_delta" + ] + + +def _input_json_deltas(events: List[dict]) -> List[str]: + return [ + e["delta"]["partial_json"] + for e in events + if e.get("type") == "content_block_delta" + and e["delta"].get("type") == "input_json_delta" + ] + + +def test_first_text_delta_after_tool_use_is_not_dropped_sync(): + """A tool_use -> text transition (text resuming after a tool call) carries + the resumed text's first token in the trigger chunk. Without the fix it was + dropped, so "The weather is nice." vanished and the answer began at " Bye.". + """ + chunks = [ + _make_chunk(Delta(content="Let me check.")), + _tool_chunk("call_1", "get_weather", '{"city":'), + _tool_chunk("call_1", None, ' "NY"}'), + _make_chunk(Delta(content="The weather is nice.")), + _make_chunk(Delta(content=" Bye.")), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x") + events = _drain_sync(wrapper) + + assert _input_json_deltas(events) == ['{"city":', ' "NY"}'] + assert _text_deltas(events) == [ + "Let me check.", + "The weather is nice.", + " Bye.", + ] + + +@pytest.mark.asyncio +async def test_first_text_delta_after_tool_use_is_not_dropped_async(): + """Async path mirrors the sync regression — the proxy serves the async + iterator, so it must preserve the first resumed text delta too. + """ + chunks = [ + _make_chunk(Delta(content="Let me check.")), + _tool_chunk("call_1", "get_weather", '{"city": "NY"}'), + _make_chunk(Delta(content="The weather is nice.")), + _make_chunk(Delta(content=" Bye.")), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=_AsyncStream(chunks), model="claude-x" + ) + events = await _drain_async(wrapper) + + assert _input_json_deltas(events) == ['{"city": "NY"}'] + assert _text_deltas(events) == [ + "Let me check.", + "The weather is nice.", + " Bye.", + ] + + +def test_single_first_text_token_after_tool_use_preserved_sync(): + """Minimal reproduction of the issue's example: a single short text token + ("Hi") resuming after a tool call. Without the fix the whole answer is + dropped because its only delta sits in the transition trigger chunk. + """ + chunks = [ + _tool_chunk("call_1", "get_weather", '{"city": "NY"}'), + _make_chunk(Delta(content="Hi")), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x") + events = _drain_sync(wrapper) + + assert _text_deltas(events) == ["Hi"] + + +def test_multiple_text_deltas_after_tool_use_preserved_sync(): + """Multiple-delta edge case: only the *first* text delta sits in the + transition trigger chunk; the rest stream normally. All of them — leading + one included — must reach the client in order. + """ + chunks = [ + _tool_chunk("call_1", "get_weather", '{"city": "NY"}'), + _make_chunk(Delta(content="Hi")), + _make_chunk(Delta(content=", how ")), + _make_chunk(Delta(content="can I help ")), + _make_chunk(Delta(content="you?")), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x") + events = _drain_sync(wrapper) + + assert _text_deltas(events) == ["Hi", ", how ", "can I help ", "you?"] + assert "".join(_text_deltas(events)) == "Hi, how can I help you?" + + +def test_empty_trigger_delta_is_not_re_emitted_sync(): + """A transition whose trigger chunk carries no content (empty text) must + NOT produce a spurious empty ``content_block_delta`` — only the synthesized + ``content_block_start`` is emitted for the new block. Here a ``tool_use -> + text`` transition is triggered by an empty-content chunk; the re-emit guard + must reject it so the new text block opens without a leading empty delta. + """ + chunks = [ + _tool_chunk("call_1", "get_weather", '{"city": "NY"}'), + # tool_use -> text transition triggered by an empty content chunk; the + # real text arrives in the following chunk. + _make_chunk(Delta(content="")), + _make_chunk(Delta(content="real text")), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x") + events = _drain_sync(wrapper) + + # No empty-string text_delta should be present. + assert "" not in _text_deltas(events) + assert "".join(_text_deltas(events)) == "real text" + + +def test_bundled_tool_args_on_transition_still_preserved_sync(): + """Existing behavior guard: when the trigger chunk that opens a tool_use + block also carries arguments (xAI/Gemini style), the ``input_json_delta`` + must still be emitted after ``content_block_start``. + """ + chunks = [ + _make_chunk(Delta(content="Calling a tool.")), + _tool_chunk("call_1", "get_weather", '{"city": "NY"}'), + _make_chunk(Delta(content=None), finish_reason="tool_calls"), + ] + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x") + events = _drain_sync(wrapper) + + assert _text_deltas(events) == ["Calling a tool."] + assert _input_json_deltas(events) == ['{"city": "NY"}'] + + +@pytest.mark.parametrize( + "processed_chunk, expected", + [ + # Non-empty deltas of every type must be re-emitted. + ( + { + "type": "content_block_delta", + "delta": {"type": "text_delta", "text": "x"}, + }, + True, + ), + ( + { + "type": "content_block_delta", + "delta": {"type": "input_json_delta", "partial_json": "{}"}, + }, + True, + ), + ( + { + "type": "content_block_delta", + "delta": {"type": "thinking_delta", "thinking": "t"}, + }, + True, + ), + ( + { + "type": "content_block_delta", + "delta": {"type": "signature_delta", "signature": "s"}, + }, + True, + ), + # Empty deltas must NOT be re-emitted (no spurious events). + ( + { + "type": "content_block_delta", + "delta": {"type": "text_delta", "text": ""}, + }, + False, + ), + ( + { + "type": "content_block_delta", + "delta": {"type": "input_json_delta", "partial_json": ""}, + }, + False, + ), + ( + { + "type": "content_block_delta", + "delta": {"type": "thinking_delta", "thinking": ""}, + }, + False, + ), + ( + { + "type": "content_block_delta", + "delta": {"type": "signature_delta", "signature": ""}, + }, + False, + ), + # Unknown delta type / non-content_block_delta / malformed delta. + ( + {"type": "content_block_delta", "delta": {"type": "other_delta"}}, + False, + ), + ({"type": "message_delta", "delta": {"stop_reason": "stop"}}, False), + ({"type": "content_block_delta", "delta": None}, False), + ], +) +def test_trigger_delta_has_content_branches(processed_chunk, expected): + """Directly exercise the re-emit predicate across all delta types and the + empty/malformed guards, so the helper's behavior is pinned independently of + upstream chunk-translation details. + """ + assert ( + AnthropicStreamWrapper._trigger_delta_has_content(processed_chunk) is expected + ) diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py index 1d25d719384..e44413cf837 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_parallel_tool_calls.py @@ -278,7 +278,8 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text(): "content_block_delta", # {"city": "content_block_delta", # "NY"} "content_block_stop", # End of first tool_use content block - "content_block_start", # "The weather is nice today" + "content_block_start", # "The weather is nice today" text block + "content_block_delta", # "The weather is nice today." text_delta "content_block_stop", "content_block_start", # Start of second tool_use content block "content_block_delta", # {"city": @@ -288,7 +289,8 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text(): "content_block_delta", # {"city": "content_block_delta", # " CHI"} "content_block_stop", # End of third tool_use content block - "content_block_start", # "The weather is not so nice today" + "content_block_start", # "The weather is not so nice today" text block + "content_block_delta", # "The weather is not so nice today." text_delta "content_block_stop", "message_delta", # Stop reason with merged usage "message_stop", # Final message stop @@ -296,6 +298,20 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text(): assert expected_types == chunk_types + # Regression: the first (and only) text delta of each text block sits in + # the chunk that *triggered* the tool_use -> text transition. It must be + # re-emitted as a content_block_delta instead of being silently dropped. + text_deltas = [ + chunk["delta"]["text"] + for chunk in chunks + if chunk.get("type") == "content_block_delta" + and chunk["delta"].get("type") == "text_delta" + ] + assert text_deltas == [ + "The weather is nice today.", + "The weather is not so nice today.", + ] + get_weather_calls = 0 for chunk in chunks: