diff --git a/litellm/llms/anthropic/messages/adapters/streaming_iterator.py b/litellm/llms/anthropic/messages/adapters/streaming_iterator.py index c41deb00f1d..0e0f95bbaba 100644 --- a/litellm/llms/anthropic/messages/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/messages/adapters/streaming_iterator.py @@ -674,6 +674,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): self._paused_turn: bool = False self._claude_refusal_stop_details: AnthropicStopDetails | None = None self.sent_compaction_block: bool = False + self._current_chunk_is_finish: bool = False # Per-phase flags so the compaction block's start/delta/stop events # are emitted (and the public state machine is advanced) in # lock-step with the caller actually consuming each event. Pre- @@ -935,7 +936,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): continue should_start_new_block = self._should_start_new_content_block(chunk) - is_opening_first_block = self.sent_content_block_start is False + is_opening_first_block = self.sent_content_block_start is False and not self._current_chunk_is_finish if is_opening_first_block and self._is_blank_delta(chunk): continue if is_opening_first_block: @@ -1035,19 +1036,22 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if processed_chunk["type"] == "message_delta" and self.sent_content_block_finish is False: # Queue both the content_block_stop and the message_delta - self.chunk_queue.append( - { - "type": "content_block_stop", - "index": self.current_content_block_index, - } - ) - self.sent_content_block_finish = True + if self.sent_content_block_start: + self.chunk_queue.append( + { + "type": "content_block_stop", + "index": self.current_content_block_index, + } + ) + self.sent_content_block_finish = True if processed_chunk.get("delta", {}).get("stop_reason") is not None: self.holding_stop_reason_chunk = processed_chunk else: processed_chunk = self._augment_message_delta_usage(processed_chunk) self.chunk_queue.append(processed_chunk) - return self.chunk_queue.popleft() + if self.chunk_queue: + return self.chunk_queue.popleft() + continue elif self.holding_chunk is not None: self.chunk_queue.append(self.holding_chunk) if processed_chunk.get("type") == "message_delta": @@ -1079,7 +1083,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): # valid Anthropic order (... -> content_block_stop -> # message_delta). Emit ``content_block_stop`` here if # the active content block was not already closed. - if not self.sent_content_block_finish: + if self.sent_content_block_start and not self.sent_content_block_finish: self.chunk_queue.append( { "type": "content_block_stop", @@ -1108,14 +1112,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): # Anthropic SSE ordering is preserved (content_block_stop -> # message_delta). if self.holding_stop_reason_chunk is not None: - if not self.sent_content_block_finish: - self.sent_content_block_finish = True - self.chunk_queue.append(self._augment_message_delta_usage(self.holding_stop_reason_chunk)) - self.holding_stop_reason_chunk = None - return { - "type": "content_block_stop", - "index": self.current_content_block_index, - } held: Final = self._augment_message_delta_usage(self.holding_stop_reason_chunk) self.holding_stop_reason_chunk = None return held @@ -1170,7 +1166,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): continue should_start_new_block = self._should_start_new_content_block(chunk) - is_opening_first_block = self.sent_content_block_start is False + is_opening_first_block = self.sent_content_block_start is False and not self._current_chunk_is_finish if is_opening_first_block and self._is_blank_delta(chunk): continue if is_opening_first_block: @@ -1262,19 +1258,22 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if processed_chunk["type"] == "message_delta" and self.sent_content_block_finish is False: # Queue both the content_block_stop and the holding chunk - self.chunk_queue.append( - { - "type": "content_block_stop", - "index": self.current_content_block_index, - } - ) - self.sent_content_block_finish = True + if self.sent_content_block_start: + self.chunk_queue.append( + { + "type": "content_block_stop", + "index": self.current_content_block_index, + } + ) + self.sent_content_block_finish = True if processed_chunk.get("delta", {}).get("stop_reason") is not None: self.holding_stop_reason_chunk = processed_chunk else: processed_chunk = self._augment_message_delta_usage(processed_chunk) self.chunk_queue.append(processed_chunk) - return self.chunk_queue.popleft() + if self.chunk_queue: + return self.chunk_queue.popleft() + continue elif self.holding_chunk is not None: # Queue both chunks self.chunk_queue.append(self.holding_chunk) @@ -1307,7 +1306,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): # valid Anthropic order (... -> content_block_stop -> # message_delta). Emit ``content_block_stop`` here if # the active content block was not already closed. - if not self.sent_content_block_finish: + if self.sent_content_block_start and not self.sent_content_block_finish: self.chunk_queue.append( { "type": "content_block_stop", @@ -1341,14 +1340,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): # Anthropic SSE ordering is preserved (content_block_stop -> # message_delta). if self.holding_stop_reason_chunk is not None: - if not self.sent_content_block_finish: - self.sent_content_block_finish = True - self.chunk_queue.append(self._augment_message_delta_usage(self.holding_stop_reason_chunk)) - self.holding_stop_reason_chunk = None - return { - "type": "content_block_stop", - "index": self.current_content_block_index, - } held: Final = self._augment_message_delta_usage(self.holding_stop_reason_chunk) self.holding_stop_reason_chunk = None return held @@ -1544,7 +1535,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): from .transformation import LiteLLMAnthropicMessagesAdapter self._record_anthropic_stop_fields(chunk.choices[0].delta) - if chunk.choices[0].finish_reason is not None: + self._current_chunk_is_finish = chunk.choices[0].finish_reason is not None + if self._current_chunk_is_finish: return False refusal_text: Final = refusal_reply_text(chunk.choices[0].delta) diff --git a/tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_first_delta.py b/tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_first_delta.py index 79cde87b4e0..2df129df5fc 100644 --- a/tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_first_delta.py +++ b/tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_first_delta.py @@ -817,9 +817,9 @@ async def test_role_only_lead_chunk_does_not_open_text_block_before_reasoning_co def test_finish_first_chunk_is_not_deferred_sync(): """A stream whose first upstream chunk is already the finish event must not be skipped by the blank-delta deferral. ``_is_blank_delta`` returns False - for a finish chunk so the message_delta still flows (with an empty text - block opened and closed first); without that guard the deferral would drop - the terminal event entirely. + for a finish chunk so the message_delta still flows directly without + opening/closing a spurious empty text block (Issue #47576); without that + guard the deferral would drop the terminal event entirely. """ chunks = [_make_chunk(Delta(content=None), finish_reason="stop")] wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x") @@ -827,8 +827,6 @@ def test_finish_first_chunk_is_not_deferred_sync(): assert [e["type"] for e in events] == [ "message_start", - "content_block_start", - "content_block_stop", "message_delta", "message_stop", ] @@ -1534,3 +1532,82 @@ async def test_signed_thinking_blocks_keep_their_own_text_and_signature(is_async assert _last_delta_types_per_thinking_block(events) == ("signature_delta", "signature_delta") assert _text_deltas(events) == ["Done"] _assert_deltas_match_their_block_type(events) + + +@pytest.mark.parametrize("is_async", [False, True]) +@pytest.mark.asyncio +async def test_finish_only_first_chunk_emits_no_synthetic_content_block(is_async: bool): + """A stream whose first choice chunk is finish-only (0-output stream) must + not synthesize a fake empty text block {"type": "text", "text": ""}. + The emitted sequence must strictly be message_start -> message_delta -> message_stop. + Regression test for issue #47576.""" + chunks = [ + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + if is_async: + wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="claude-x") + events = await _drain_async(wrapper) + else: + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x") + events = _drain_sync(wrapper) + + event_types = [e.get("type") for e in events] + assert event_types == ["message_start", "message_delta", "message_stop"] + assert not any(e.get("type") == "content_block_start" for e in events) + assert not any(e.get("type") == "content_block_stop" for e in events) + message_delta = next(e for e in events if e.get("type") == "message_delta") + assert message_delta["delta"]["stop_reason"] == "end_turn" + + +@pytest.mark.parametrize("is_async", [False, True]) +@pytest.mark.asyncio +async def test_finish_only_chunk_with_trailing_usage_emits_no_synthetic_content_block(is_async: bool): + """A 0-output stream with a trailing usage chunk must not emit synthetic + content blocks and must correctly merge usage into the terminal message_delta. + Regression test for issue #47576.""" + usage_chunk = MagicMock() + usage_chunk.choices = [] + usage_chunk.usage = Usage(prompt_tokens=15, completion_tokens=0, total_tokens=15) + usage_chunk._hidden_params = {} + + chunks = [ + _make_chunk(Delta(content=None), finish_reason="stop"), + usage_chunk, + ] + if is_async: + wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="claude-x") + events = await _drain_async(wrapper) + else: + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x") + events = _drain_sync(wrapper) + + event_types = [e.get("type") for e in events] + assert event_types == ["message_start", "message_delta", "message_stop"] + assert not any(e.get("type") == "content_block_start" for e in events) + assert not any(e.get("type") == "content_block_stop" for e in events) + message_delta = next(e for e in events if e.get("type") == "message_delta") + assert message_delta["delta"]["stop_reason"] == "end_turn" + assert message_delta["usage"]["input_tokens"] == 15 + assert message_delta["usage"]["output_tokens"] == 0 + + +@pytest.mark.parametrize("is_async", [False, True]) +@pytest.mark.asyncio +async def test_blank_deltas_before_finish_chunk_emit_no_synthetic_content_block(is_async: bool): + """Blank deltas before the terminal chunk must be skipped, and the finish + chunk must not open a synthetic content block. + Regression test for issue #47576.""" + chunks = [ + _make_chunk(Delta(content=None)), + _make_chunk(Delta(content=None), finish_reason="stop"), + ] + if is_async: + wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="claude-x") + events = await _drain_async(wrapper) + else: + wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="claude-x") + events = _drain_sync(wrapper) + + event_types = [e.get("type") for e in events] + assert event_types == ["message_start", "message_delta", "message_stop"] + assert not any(e.get("type") == "content_block_start" for e in events)