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 6bddad09f21..ad2156f1042 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -14,6 +14,31 @@ if TYPE_CHECKING: from litellm.types.utils import ModelResponseStream +def _trigger_delta_has_content(processed_chunk: Optional[Dict[str, Any]]) -> bool: + """Return True when a ``content_block_delta`` carries non-empty content. + + Used on content-block transitions to decide whether the trigger chunk's delta + should be re-emitted alongside the synthetic ``content_block_stop`` / + ``content_block_start`` pair. Re-emitting is required for text and thinking + transitions (where the trigger chunk holds the first characters of the new + block) and is intentionally skipped for tool_use openers, whose delta is an + empty ``input_json_delta`` because the tool name is already carried by + ``content_block_start``. + """ + if not processed_chunk or processed_chunk.get("type") != "content_block_delta": + return False + delta = processed_chunk.get("delta") or {} + if not isinstance(delta, dict): + return False + # Anthropic delta variants all carry their content under exactly one of + # these fields; any non-empty value means the chunk is worth emitting. + for key in ("text", "thinking", "partial_json", "signature"): + value = delta.get(key) + if value: + return True + return False + + class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): """ - first chunk return 'message_start' @@ -129,8 +154,21 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if should_start_new_block and not self.sent_content_block_finish: # Queue the sequence: content_block_stop -> content_block_start - # The trigger chunk itself is not emitted as a delta since the - # content_block_start already carries the relevant information. + # (-> content_block_delta, when the trigger chunk carries + # content). + # + # NOTE: a non-empty trigger delta MUST be re-emitted here. The + # `content_block_start` event carries only the block type and an + # empty body (see + # `_translate_streaming_openai_chunk_to_anthropic_content_block` + # which returns `TextBlock(text="")` for text transitions), so + # dropping a non-empty `processed_chunk` silently loses the first + # characters of every new content block for providers that stream + # reasoning then text (e.g. Bedrock Converse MiniMax / Kimi / + # Claude extended thinking). For tool_use openers the trigger + # chunk carries only the tool name + empty arguments, so its + # delta is empty and is intentionally skipped - the tool name is + # already part of `content_block_start`. self.chunk_queue.append( { "type": "content_block_stop", @@ -144,6 +182,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): "content_block": self.current_content_block_start, } ) + if _trigger_delta_has_content(processed_chunk): + self.chunk_queue.append(processed_chunk) self.sent_content_block_finish = False return self.chunk_queue.popleft() @@ -305,8 +345,13 @@ 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 - # The trigger chunk itself is not emitted as a delta since the - # content_block_start already carries the relevant information. + # (-> content_block_delta, when the trigger chunk carries + # content). See the sync `__next__` path for a full + # explanation - the short version is that + # `content_block_start` only carries the block type and an + # empty body, so a non-empty trigger delta must be emitted + # separately or the first characters of every new content + # block are silently dropped. # 1. Stop current content block self.chunk_queue.append( @@ -325,6 +370,13 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): } ) + # 3. Emit the trigger chunk's delta when it has content + # (skip empty deltas such as tool_use openers whose + # arguments are empty and whose name is already carried by + # `content_block_start`). + if _trigger_delta_has_content(processed_chunk): + self.chunk_queue.append(processed_chunk) + # Reset state for new block self.sent_content_block_finish = False diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_first_chunk_on_block_transition.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_first_chunk_on_block_transition.py new file mode 100644 index 00000000000..890f72b15a2 --- /dev/null +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_first_chunk_on_block_transition.py @@ -0,0 +1,221 @@ +""" +Regression tests for AnthropicStreamWrapper dropping the first chunk of content +when a new content block is started. + +Context: +Before the fix, `__next__` / `__anext__` emitted the sequence +`content_block_stop` -> `content_block_start` on a detected block transition +(e.g. text -> thinking, thinking -> text, text -> tool_use), but the `processed_chunk` +that actually *triggered* the transition was silently discarded. Because +`_translate_streaming_openai_chunk_to_anthropic_content_block()` returns a block +with an empty body for text transitions (`TextBlock(text="")`), dropping the +trigger chunk meant the first character(s) of every new block were lost on the +wire. For Bedrock Converse reasoning providers (MiniMax, Kimi, Claude extended +thinking), this typically manifested as responses starting mid-sentence or, if +the model emitted the text as a single chunk, as an empty text block with zero +`content_block_delta` events. + +The fix enqueues the trigger chunk's `content_block_delta` after the +`content_block_start` so that no content is lost. +""" + +import os +import sys +from typing import List + +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 Delta, ModelResponseStream, StreamingChoices, Usage + + +def _text_chunk(text: str) -> ModelResponseStream: + return ModelResponseStream( + choices=[ + StreamingChoices( + delta=Delta(content=text, role="assistant"), + index=0, + finish_reason=None, + ) + ], + ) + + +def _thinking_chunk(thinking: str) -> ModelResponseStream: + return ModelResponseStream( + choices=[ + StreamingChoices( + delta=Delta( + content="", + role="assistant", + reasoning_content=thinking, + thinking_blocks=[ + { + "type": "thinking", + "thinking": thinking, + "signature": None, + } + ], + provider_specific_fields={ + "thinking_blocks": [ + { + "type": "thinking", + "thinking": thinking, + "signature": None, + } + ] + }, + ), + index=0, + finish_reason=None, + ) + ], + ) + + +def _stop_chunk() -> ModelResponseStream: + return ModelResponseStream( + choices=[ + StreamingChoices( + delta=Delta(content="", role="assistant"), + index=0, + finish_reason="stop", + ) + ], + usage=Usage(prompt_tokens=10, completion_tokens=10, total_tokens=20), + ) + + +class _MockStreamThinkingThenText: + """Simulates a Bedrock Converse reasoning model stream. + + The first chunk is a thinking block (which forces a text->thinking transition + on the wrapper's initial state) and the first text chunk forces a + thinking->text transition. Both transition chunks carry real content that + must not be dropped. + """ + + def __init__(self) -> None: + self.responses: List[ModelResponseStream] = [ + _thinking_chunk("The user says hi. I should"), + _thinking_chunk(" greet them back."), + _text_chunk("안녕하세요, 저는 MiniMax입니다."), + _text_chunk(" 무엇을 도와드릴까요?"), + _stop_chunk(), + ] + self.index = 0 + + def __iter__(self): + return self + + def __next__(self): + if self.index >= len(self.responses): + raise StopIteration + response = self.responses[self.index] + self.index += 1 + return response + + def __aiter__(self): + return self + + async def __anext__(self): + if self.index >= len(self.responses): + raise StopAsyncIteration + response = self.responses[self.index] + self.index += 1 + return response + + +def _collect_deltas(chunks: List[dict]) -> dict: + """Aggregate text_delta / thinking_delta content from a stream of wrapper chunks.""" + text = "" + thinking = "" + for chunk in chunks: + if chunk.get("type") != "content_block_delta": + continue + delta = chunk.get("delta", {}) or {} + if delta.get("type") == "text_delta": + text += delta.get("text", "") or "" + elif delta.get("type") == "thinking_delta": + thinking += delta.get("thinking", "") or "" + return {"text": text, "thinking": thinking} + + +EXPECTED_THINKING = "The user says hi. I should greet them back." +EXPECTED_TEXT = "안녕하세요, 저는 MiniMax입니다. 무엇을 도와드릴까요?" + + +def test_sync_first_chunk_preserved_on_block_transitions(): + """Sync path: the first chunk of each new content block must not be dropped.""" + wrapper = AnthropicStreamWrapper( + completion_stream=_MockStreamThinkingThenText(), + model="bedrock/converse/minimax.minimax-m2.5", + ) + + chunks = list(wrapper) + aggregated = _collect_deltas(chunks) + + assert aggregated["thinking"] == EXPECTED_THINKING, ( + "Thinking content was dropped on the text->thinking transition. " + f"Expected {EXPECTED_THINKING!r}, got {aggregated['thinking']!r}" + ) + assert aggregated["text"] == EXPECTED_TEXT, ( + "Text content was dropped on the thinking->text transition. " + f"Expected {EXPECTED_TEXT!r}, got {aggregated['text']!r}" + ) + + +@pytest.mark.asyncio +async def test_async_first_chunk_preserved_on_block_transitions(): + """Async path: same guarantee as the sync path.""" + wrapper = AnthropicStreamWrapper( + completion_stream=_MockStreamThinkingThenText(), + model="bedrock/converse/minimax.minimax-m2.5", + ) + + chunks: List[dict] = [] + async for chunk in wrapper: + chunks.append(chunk) + + aggregated = _collect_deltas(chunks) + + assert aggregated["thinking"] == EXPECTED_THINKING + assert aggregated["text"] == EXPECTED_TEXT + + +def test_sync_block_transition_event_ordering(): + """On a block transition the wrapper must emit + content_block_stop -> content_block_start -> content_block_delta in that order, + with the delta carrying the trigger chunk's content.""" + wrapper = AnthropicStreamWrapper( + completion_stream=_MockStreamThinkingThenText(), + model="bedrock/converse/minimax.minimax-m2.5", + ) + + chunks = list(wrapper) + types = [c.get("type") for c in chunks] + + # Find the first text->thinking transition: content_block_stop then start then delta. + # The wrapper emits an initial content_block_start(index=0,text) before the first + # real chunk, so the first stop we see is the index=0 text block being closed. + first_stop = types.index("content_block_stop") + assert types[first_stop + 1] == "content_block_start" + assert ( + types[first_stop + 2] == "content_block_delta" + ), "First trigger chunk's delta must immediately follow content_block_start." + + # And the delta must carry the *first* trigger chunk's content verbatim - + # i.e. the text that would otherwise have been dropped. This is the + # specific regression this test guards against. + delta_chunk = chunks[first_stop + 2] + inner_delta = delta_chunk.get("delta", {}) or {} + assert inner_delta.get("type") == "thinking_delta" + assert inner_delta.get("thinking") == "The user says hi. I should", ( + "Expected the first thinking chunk's content to be emitted as the first " + "delta after content_block_start, but got " + f"{inner_delta.get('thinking')!r}" + ) 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..707c0c2885e 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 @@ -279,6 +279,7 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text(): "content_block_delta", # "NY"} "content_block_stop", # End of first tool_use content block "content_block_start", # "The weather is nice today" + "content_block_delta", # "The weather is nice today." (trigger chunk) "content_block_stop", "content_block_start", # Start of second tool_use content block "content_block_delta", # {"city": @@ -289,6 +290,7 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text(): "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_delta", # "The weather is not so nice today." (trigger chunk) "content_block_stop", "message_delta", # Stop reason with merged usage "message_stop", # Final message stop @@ -296,6 +298,21 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text(): assert expected_types == chunk_types + # Verify the content of the text content blocks is preserved across the + # transition (previously the trigger chunk was dropped, losing the first + # characters of every new content block - see + # test_first_chunk_on_block_transition.py for a dedicated regression test). + text_deltas = [ + (chunk.get("delta") or {}).get("text", "") + for chunk in chunks + if chunk.get("type") == "content_block_delta" + and (chunk.get("delta") or {}).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: