From 6ed4eeafb300196c37e56bad227a41d5680810a0 Mon Sep 17 00:00:00 2001 From: lkapadiya-DO Date: Tue, 14 Jul 2026 11:48:16 -0700 Subject: [PATCH] fix(anthropic-adapter): valid Anthropic SSE for reasoning_content and tool-call streams Peek the first upstream chunk before content_block_start so thinking vs text blocks match reasoning_content models, skip empty reasoning_content, and only re-queue non-empty trigger deltas on block transitions. Adds regression tests for Claude Code SSE failures that caused duplicate non-streaming retries. Co-authored-by: Cursor --- .../adapters/streaming_iterator.py | 161 ++++-- .../adapters/transformation.py | 19 +- ...al_pass_through_adapters_transformation.py | 135 +++++ ...st_streaming_iterator_reasoning_content.py | 483 ++++++++++++++++++ .../test_streaming_iterator_tool_args.py | 7 +- .../messages/test_parallel_tool_calls.py | 91 ++-- 6 files changed, 780 insertions(+), 116 deletions(-) create mode 100644 tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_reasoning_content.py 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 44c367ee805..be09041f4f1 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -374,14 +374,65 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if self.sent_content_block_start is False: self.sent_content_block_start = True - self.sent_content_block_finish = False - self.chunk_queue.append( - { - "type": "content_block_start", - "index": self.current_content_block_index, - "content_block": {"type": "text", "text": ""}, - } - ) + # Peek at the first chunk to determine the correct initial + # content block type. Models that use reasoning_content + # (e.g. GLM-5) start with a thinking block, not text. + first_chunk = None + for chunk in self.completion_stream: + if chunk == "None" or chunk is None: + continue + first_chunk = chunk + break + if first_chunk is not None: + ( + block_type, + content_block_start, + ) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic_content_block( + choices=first_chunk.choices + ) + self.current_content_block_type = block_type + self.current_content_block_start = content_block_start + if block_type == "thinking": + initial_block: dict = {"type": "thinking", "thinking": ""} + elif block_type == "tool_use": + initial_block = dict(content_block_start) + else: + initial_block = {"type": "text", "text": ""} + self.chunk_queue.append( + { + "type": "content_block_start", + "index": self.current_content_block_index, + "content_block": initial_block, + } + ) + processed_first = LiteLLMAnthropicMessagesAdapter().translate_streaming_openai_response_to_anthropic( + response=first_chunk, + current_content_block_index=self.current_content_block_index, + ) + # Empty / stop-only first chunk: close the block before the + # terminal message_delta so the sequence stays spec-compliant. + if ( + isinstance(processed_first, dict) + and processed_first.get("type") == "message_delta" + ): + self.chunk_queue.append( + { + "type": "content_block_stop", + "index": self.current_content_block_index, + } + ) + self.sent_content_block_finish = True + self.chunk_queue.append(processed_first) + elif self._trigger_delta_has_content(processed_first): + self.chunk_queue.append(processed_first) + else: + self.chunk_queue.append( + { + "type": "content_block_start", + "index": self.current_content_block_index, + "content_block": {"type": "text", "text": ""}, + } + ) return self.chunk_queue.popleft() for chunk in self.completion_stream: @@ -427,19 +478,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): continue if should_start_new_block and not self.sent_content_block_finish: - # Queue the sequence: content_block_stop -> content_block_start - # -> (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( { "type": "content_block_stop", @@ -447,7 +485,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): } ) - # 2. Start new content block self.chunk_queue.append( { "type": "content_block_start", @@ -593,14 +630,60 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if self.sent_content_block_start is False: self.sent_content_block_start = True - self.sent_content_block_finish = False - self.chunk_queue.append( - { - "type": "content_block_start", - "index": self.current_content_block_index, - "content_block": {"type": "text", "text": ""}, - } - ) + first_chunk = None + async for chunk in self.completion_stream: + if chunk == "None" or chunk is None: + continue + first_chunk = chunk + break + if first_chunk is not None: + ( + block_type, + content_block_start, + ) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic_content_block( + choices=first_chunk.choices + ) + self.current_content_block_type = block_type + self.current_content_block_start = content_block_start + if block_type == "thinking": + initial_block = {"type": "thinking", "thinking": ""} + elif block_type == "tool_use": + initial_block = dict(content_block_start) + else: + initial_block = {"type": "text", "text": ""} + self.chunk_queue.append( + { + "type": "content_block_start", + "index": self.current_content_block_index, + "content_block": initial_block, + } + ) + processed_first = LiteLLMAnthropicMessagesAdapter().translate_streaming_openai_response_to_anthropic( + response=first_chunk, + current_content_block_index=self.current_content_block_index, + ) + if ( + isinstance(processed_first, dict) + and processed_first.get("type") == "message_delta" + ): + self.chunk_queue.append( + { + "type": "content_block_stop", + "index": self.current_content_block_index, + } + ) + self.sent_content_block_finish = True + self.chunk_queue.append(processed_first) + elif self._trigger_delta_has_content(processed_first): + self.chunk_queue.append(processed_first) + else: + self.chunk_queue.append( + { + "type": "content_block_start", + "index": self.current_content_block_index, + "content_block": {"type": "text", "text": ""}, + } + ) return self.chunk_queue.popleft() async for chunk in self.completion_stream: @@ -641,19 +724,6 @@ 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 - # -> (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( { "type": "content_block_stop", @@ -667,13 +737,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): "content_block": self.current_content_block_start, } ) - - # 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 self.sent_content_block_finish = False return self.chunk_queue.popleft() diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py b/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py index 4c981dd36b3..1a52e03cd1c 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py @@ -1410,13 +1410,13 @@ class LiteLLMAnthropicMessagesAdapter: return "thinking", ChatCompletionThinkingBlock( type="thinking", thinking=thinking, signature=signature ) - # OpenAI-compatible reasoning backends (e.g. vLLM/SGLang reasoning - # parsers) populate ``reasoning_content`` without ``thinking_blocks``. - # ``Delta`` deletes the ``thinking_blocks`` attribute when unset, so the - # branch above is skipped entirely; open a ``thinking`` block here so the - # matching ``thinking_delta`` stream is not emitted into a text block. - elif isinstance(choice, StreamingChoices) and getattr(choice.delta, "reasoning_content", None): - return "thinking", ChatCompletionThinkingBlock(type="thinking", thinking="", signature="") + elif isinstance(choice, StreamingChoices) and hasattr( + choice.delta, "reasoning_content" + ): + if choice.delta.reasoning_content: + return "thinking", ChatCompletionThinkingBlock( + type="thinking", thinking="", signature="" + ) return "text", TextBlock(type="text", text="") @@ -1438,8 +1438,7 @@ class LiteLLMAnthropicMessagesAdapter: for choice in choices: if choice.delta.content is not None and len(choice.delta.content) > 0: text += choice.delta.content - if choice.delta.tool_calls: - partial_json = "" + if choice.delta.tool_calls is not None and len(choice.delta.tool_calls) > 0: for tool in choice.delta.tool_calls: if tool.function is not None and tool.function.arguments is not None: partial_json = (partial_json or "") + tool.function.arguments @@ -1459,7 +1458,7 @@ class LiteLLMAnthropicMessagesAdapter: # Handle reasoning_content when thinking_blocks is not present # This handles providers like OpenRouter that return reasoning_content elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "reasoning_content"): - if choice.delta.reasoning_content is not None: + if choice.delta.reasoning_content: reasoning_content += choice.delta.reasoning_content if reasoning_content and reasoning_signature: diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py index f9c55db72b5..72294716cc6 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py @@ -402,6 +402,82 @@ def test_translate_anthropic_messages_to_openai_tool_message_placement(): ), "Tool message should be placed before user message" +def test_translate_streaming_openai_text_delta_no_tool_calls(): + """When tool_calls is None, a text delta with content should be text_delta.""" + choices = [ + StreamingChoices( + finish_reason=None, + index=0, + delta=Delta( + provider_specific_fields=None, + content="Hello", + role="assistant", + function_call=None, + tool_calls=None, + audio=None, + ), + logprobs=None, + ) + ] + + ( + type_of_content, + content_block_delta, + ) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic( + choices=choices # type: ignore[arg-type] + ) + + assert type_of_content == "text_delta" + assert content_block_delta["type"] == "text_delta" + assert content_block_delta["text"] == "Hello" # type: ignore[typeddict-unknown-key] + + +@pytest.mark.parametrize( + "content", + [ + "Hello from DeepSeek", + "Let me think about that.", + ], +) +def test_translate_streaming_openai_text_delta_empty_tool_calls(content: str): + """When tool_calls is an empty list (len==0), content should still be text_delta. + + This reproduces the bug where DeepSeek/vLLM sends tool_calls: [] + instead of tool_calls: None. Before the fix, the `is not None` guard + let this into the tool_calls branch, and a stray partial_json="" caused + an empty input_json_delta to be emitted instead of the text content. + """ + choices = [ + StreamingChoices( + finish_reason=None, + index=0, + delta=Delta( + provider_specific_fields=None, + content=content, + role="assistant", + function_call=None, + tool_calls=[], # empty list — the exact pattern from vLLM/DeepSeek + audio=None, + ), + logprobs=None, + ) + ] + + ( + type_of_content, + content_block_delta, + ) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic( + choices=choices # type: ignore[arg-type] + ) + + assert type_of_content == "text_delta", ( + f"Expected text_delta but got {type_of_content} — " + "empty tool_calls list should not trigger input_json_delta" + ) + assert content_block_delta["type"] == "text_delta" + assert content_block_delta["text"] == content # type: ignore[typeddict-unknown-key] + + def test_translate_openai_content_to_anthropic_empty_function_arguments(): """Test that empty function arguments are handled safely and don't cause JSON parsing errors.""" @@ -1839,6 +1915,65 @@ def test_translate_streaming_openai_chunk_to_anthropic_reasoning_content_without assert content_block_delta["thinking"] == "I need to analyze this carefully..." +def test_translate_streaming_openai_chunk_to_anthropic_content_block_empty_reasoning_content(): + """Empty reasoning_content must not open a thinking block without thinking_delta.""" + choices = [ + StreamingChoices( + finish_reason=None, + index=0, + delta=Delta( + reasoning_content="", + content="", + role="assistant", + function_call=None, + tool_calls=None, + audio=None, + ), + logprobs=None, + ) + ] + + ( + block_type, + content_block_start, + ) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic_content_block( + choices=choices + ) + + assert block_type == "text" + assert content_block_start == {"type": "text", "text": ""} + + +def test_translate_streaming_openai_chunk_to_anthropic_empty_reasoning_content_no_thinking_delta(): + """Empty reasoning_content must not emit thinking_delta on a text block.""" + choices = [ + StreamingChoices( + finish_reason=None, + index=0, + delta=Delta( + reasoning_content="", + content="", + role="assistant", + function_call=None, + tool_calls=None, + audio=None, + ), + logprobs=None, + ) + ] + + ( + type_of_content, + content_block_delta, + ) = LiteLLMAnthropicMessagesAdapter()._translate_streaming_openai_chunk_to_anthropic( + choices=choices + ) + + assert type_of_content == "text_delta" + assert content_block_delta["type"] == "text_delta" + assert content_block_delta["text"] == "" + + def test_translate_openai_response_to_anthropic_with_reasoning_content_only(): """ Test the full response translation when only reasoning_content is present diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_reasoning_content.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_reasoning_content.py new file mode 100644 index 00000000000..9ef38df0877 --- /dev/null +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_reasoning_content.py @@ -0,0 +1,483 @@ +""" +Tests for streaming_iterator.py fixes: + +Fix 2 – Peek at first chunk to determine correct initial content_block_start type. + Models that return reasoning_content (e.g. GLM-5 via Vertex AI) start + the stream with a thinking block, not a text block. + +Fix 3 – Queue the processed_chunk (first delta of a new block) when a content + block transition occurs, so the first token is not silently dropped. +""" + +import os +import sys + +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, +) + + +# --------------------------------------------------------------------------- +# Mock streams +# --------------------------------------------------------------------------- + + +class MockSyncStream: + """Synchronous mock completion stream yielding a fixed list of chunks.""" + + def __init__(self, chunks: list[ModelResponseStream]): + self._chunks = iter(chunks) + + def __iter__(self): + return self + + def __next__(self): + return next(self._chunks) + + +class MockAsyncStream: + """Asynchronous mock completion stream yielding a fixed list of chunks.""" + + def __init__(self, chunks: list[ModelResponseStream]): + self._chunks = iter(chunks) + + def __aiter__(self): + return self + + async def __anext__(self): + try: + return next(self._chunks) + except StopIteration: + raise StopAsyncIteration + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_thinking_chunk(text: str) -> ModelResponseStream: + """Create a streaming chunk with reasoning_content (no thinking_blocks).""" + return ModelResponseStream( + choices=[ + StreamingChoices( + delta=Delta( + reasoning_content=text, + content="", + role="assistant", + ), + index=0, + finish_reason=None, + ) + ], + ) + + +def _make_text_chunk(text: str) -> ModelResponseStream: + """Create a streaming chunk with text content.""" + return ModelResponseStream( + choices=[ + StreamingChoices( + delta=Delta( + content=text, + role="assistant", + ), + index=0, + finish_reason=None, + ) + ], + ) + + +def _make_stop_chunk() -> ModelResponseStream: + """Create a streaming chunk signalling end of generation.""" + return ModelResponseStream( + choices=[ + StreamingChoices( + delta=Delta(content=""), + index=0, + finish_reason="stop", + ) + ], + ) + + +def _collect_all_events(wrapper) -> list[dict]: + """Collect all events from a sync AnthropicStreamWrapper.""" + events = [] + for raw in wrapper: + events.append(raw) + return events + + +async def _collect_all_events_async(wrapper) -> list[dict]: + """Collect all events from an async AnthropicStreamWrapper.""" + events = [] + async for raw in wrapper: + events.append(raw) + return events + + +def _assert_monotonic_indices(events: list[dict]) -> None: + """Indices on content_block events must be non-decreasing with matched pairs.""" + open_indices: list[int] = [] + for event in events: + event_type = event.get("type") + if event_type not in ( + "content_block_start", + "content_block_delta", + "content_block_stop", + ): + continue + idx = event.get("index") + assert isinstance(idx, int) + if event_type == "content_block_start": + if open_indices: + assert idx >= open_indices[-1] + open_indices.append(idx) + elif event_type == "content_block_stop": + assert open_indices, f"unexpected content_block_stop at index {idx}" + assert open_indices.pop() == idx + + +# --------------------------------------------------------------------------- +# Fix 2 – Initial content_block_start reflects first chunk type +# --------------------------------------------------------------------------- + + +class TestInitialBlockTypePeek: + """ + When the first chunk from the upstream model contains reasoning_content, + the initial content_block_start must have type "thinking", not "text". + """ + + def test_sync_thinking_first_chunk(self): + chunks = [ + _make_thinking_chunk("Let me think..."), + _make_thinking_chunk(" about this."), + _make_text_chunk("The answer is 42."), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockSyncStream(chunks), model="glm-5" + ) + events = _collect_all_events(wrapper) + + assert events[0]["type"] == "message_start" + + content_block_start = events[1] + assert content_block_start["type"] == "content_block_start" + assert content_block_start["content_block"]["type"] == "thinking" + + first_delta = events[2] + assert first_delta["type"] == "content_block_delta" + assert first_delta["delta"]["type"] == "thinking_delta" + + def test_sync_text_first_chunk(self): + """Text-first streams should still emit type 'text' (no regression).""" + chunks = [ + _make_text_chunk("Hello"), + _make_text_chunk(" world"), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockSyncStream(chunks), model="gpt-4o" + ) + events = _collect_all_events(wrapper) + + content_block_start = events[1] + assert content_block_start["type"] == "content_block_start" + assert content_block_start["content_block"]["type"] == "text" + + def test_sync_empty_reasoning_content_first_chunk_opens_text_block(self): + """Empty reasoning_content placeholder must not open a thinking block.""" + chunks = [ + _make_thinking_chunk(""), + _make_text_chunk("Hello"), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockSyncStream(chunks), model="glm-5" + ) + events = _collect_all_events(wrapper) + + content_block_start = events[1] + assert content_block_start["type"] == "content_block_start" + assert content_block_start["content_block"]["type"] == "text" + + first_delta = events[2] + assert first_delta["type"] == "content_block_delta" + assert first_delta["delta"]["type"] == "text_delta" + + @pytest.mark.asyncio + async def test_async_thinking_first_chunk(self): + chunks = [ + _make_thinking_chunk("Let me think..."), + _make_thinking_chunk(" about this."), + _make_text_chunk("The answer is 42."), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockAsyncStream(chunks), model="glm-5" + ) + events = await _collect_all_events_async(wrapper) + + assert events[0]["type"] == "message_start" + + content_block_start = events[1] + assert content_block_start["type"] == "content_block_start" + assert content_block_start["content_block"]["type"] == "thinking" + + first_delta = events[2] + assert first_delta["type"] == "content_block_delta" + assert first_delta["delta"]["type"] == "thinking_delta" + + @pytest.mark.asyncio + async def test_async_text_first_chunk(self): + chunks = [ + _make_text_chunk("Hello"), + _make_text_chunk(" world"), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockAsyncStream(chunks), model="gpt-4o" + ) + events = await _collect_all_events_async(wrapper) + + content_block_start = events[1] + assert content_block_start["type"] == "content_block_start" + assert content_block_start["content_block"]["type"] == "text" + + +# --------------------------------------------------------------------------- +# Fix 3 – Block transition queues the trigger chunk +# --------------------------------------------------------------------------- + + +class TestBlockTransitionIncludesFirstDelta: + """ + When the stream transitions from one block type to another (e.g. thinking + → text), the processed chunk that triggered the transition must be queued + and eventually yielded. Without Fix 3 the first token of the new block + would be silently dropped. + """ + + def test_sync_thinking_to_text_no_token_drop(self): + chunks = [ + _make_thinking_chunk("Reasoning step."), + _make_text_chunk("Answer text."), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockSyncStream(chunks), model="glm-5" + ) + events = _collect_all_events(wrapper) + + text_deltas = [ + e + for e in events + if e.get("type") == "content_block_delta" + and e.get("delta", {}).get("type") == "text_delta" + ] + assert len(text_deltas) >= 1, ( + "The first text delta after a thinking→text transition must not be " + "dropped. Got text_delta events: " + repr(text_deltas) + ) + assert text_deltas[0]["delta"]["text"] == "Answer text." + + @pytest.mark.asyncio + async def test_async_thinking_to_text_no_token_drop(self): + chunks = [ + _make_thinking_chunk("Reasoning step."), + _make_text_chunk("Answer text."), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockAsyncStream(chunks), model="glm-5" + ) + events = await _collect_all_events_async(wrapper) + + text_deltas = [ + e + for e in events + if e.get("type") == "content_block_delta" + and e.get("delta", {}).get("type") == "text_delta" + ] + assert len(text_deltas) >= 1, ( + "The first text delta after a thinking→text transition must not be " + "dropped. Got text_delta events: " + repr(text_deltas) + ) + assert text_deltas[0]["delta"]["text"] == "Answer text." + + def test_sync_event_sequence_is_valid(self): + """ + The full event sequence for a thinking→text stream should follow the + Anthropic SSE spec: + message_start → + content_block_start (thinking) → + content_block_delta (thinking_delta) → + content_block_stop → + content_block_start (text) → + content_block_delta (text_delta) → + content_block_stop → + message_delta → + message_stop + """ + chunks = [ + _make_thinking_chunk("Think."), + _make_text_chunk("Answer."), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockSyncStream(chunks), model="glm-5" + ) + events = _collect_all_events(wrapper) + + types = [e["type"] for e in events] + + assert types[0] == "message_start" + assert types[1] == "content_block_start" + assert events[1]["content_block"]["type"] == "thinking" + assert "content_block_delta" in types + idx_first_stop = types.index("content_block_stop") + assert idx_first_stop > 1 + idx_second_start = types.index("content_block_start", idx_first_stop) + assert events[idx_second_start]["content_block"]["type"] == "text" + text_delta_idx = types.index("content_block_delta", idx_second_start) + assert events[text_delta_idx]["delta"]["type"] == "text_delta" + + +# --------------------------------------------------------------------------- +# Extra coverage – text-only full sequence, no duplicates, stop-only guard +# --------------------------------------------------------------------------- + + +class TestTextOnlyFullSequence: + def test_sync_text_only_sequence(self): + chunks = [ + _make_text_chunk("Hello"), + _make_text_chunk(" world"), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockSyncStream(chunks), model="gpt-4o" + ) + events = _collect_all_events(wrapper) + types = [e["type"] for e in events] + + assert types == [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ] + assert events[1]["content_block"]["type"] == "text" + assert events[2]["delta"]["text"] == "Hello" + assert events[3]["delta"]["text"] == " world" + + @pytest.mark.asyncio + async def test_async_text_only_sequence(self): + chunks = [ + _make_text_chunk("Hello"), + _make_text_chunk(" world"), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockAsyncStream(chunks), model="gpt-4o" + ) + events = await _collect_all_events_async(wrapper) + types = [e["type"] for e in events] + + assert types == [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ] + + +class TestNoDuplicateFirstDelta: + def test_sync_peeked_first_delta_not_duplicated(self): + chunks = [ + _make_text_chunk("Hello"), + _make_text_chunk(" world"), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockSyncStream(chunks), model="gpt-4o" + ) + events = _collect_all_events(wrapper) + + text_deltas = [ + e["delta"]["text"] + for e in events + if e.get("type") == "content_block_delta" + and e.get("delta", {}).get("type") == "text_delta" + ] + assert text_deltas == ["Hello", " world"] + _assert_monotonic_indices(events) + + @pytest.mark.asyncio + async def test_async_peeked_first_delta_not_duplicated(self): + chunks = [ + _make_thinking_chunk("Think."), + _make_text_chunk("Answer."), + _make_stop_chunk(), + ] + wrapper = AnthropicStreamWrapper( + completion_stream=MockAsyncStream(chunks), model="glm-5" + ) + events = await _collect_all_events_async(wrapper) + _assert_monotonic_indices(events) + + +class TestStopOnlyFirstChunk: + """Empty/stop-only first chunk must still emit content_block_stop.""" + + def test_sync_stop_only_first_chunk(self): + chunks = [_make_stop_chunk()] + wrapper = AnthropicStreamWrapper( + completion_stream=MockSyncStream(chunks), model="gpt-4o" + ) + events = _collect_all_events(wrapper) + types = [e["type"] for e in events] + + assert types == [ + "message_start", + "content_block_start", + "content_block_stop", + "message_delta", + "message_stop", + ] + + @pytest.mark.asyncio + async def test_async_stop_only_first_chunk(self): + chunks = [_make_stop_chunk()] + wrapper = AnthropicStreamWrapper( + completion_stream=MockAsyncStream(chunks), model="gpt-4o" + ) + events = await _collect_all_events_async(wrapper) + types = [e["type"] for e in events] + + assert types == [ + "message_start", + "content_block_start", + "content_block_stop", + "message_delta", + "message_stop", + ] diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_tool_args.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_tool_args.py index bd39e420607..9c473924ebf 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_tool_args.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_tool_args.py @@ -152,7 +152,7 @@ async def test_async_stream_no_extra_delta_when_tool_args_empty(): """ When a provider sends tool name/id WITHOUT arguments in the first chunk (OpenAI-style), the wrapper should NOT emit an extra input_json_delta - after content_block_start. This verifies backward compatibility. + from the trigger chunk — only the follow-up chunk with real args. """ # Chunk 1: text text_chunk = _make_chunk(Delta(content="Hi", role="assistant", tool_calls=None)) @@ -221,9 +221,8 @@ async def test_async_stream_no_extra_delta_when_tool_args_empty(): assert tool_start_idx is not None - # Count how many input_json_delta events appear after the tool_use block start. - # With empty args in the trigger chunk, only the subsequent tool_args_chunk - # should produce one — not the trigger chunk itself. + # Count input_json_delta events after the tool_use block start. Only the + # follow-up chunk with real arguments should produce a delta. input_json_deltas = [ e for e in events[tool_start_idx + 1 :] 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 e44413cf837..edd89f9ed99 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 @@ -131,22 +131,17 @@ def test_anthropic_stream_wrapper_single_tool_call(): chunks.append(chunk) chunk_types.append(chunk.get("type")) - # Verify the expected sequence of chunk types + # Verify the expected sequence of chunk types. + # Peek opens tool_use directly (no empty text block). Empty args on the + # first tool-name chunk are suppressed by _trigger_delta_has_content. expected_types = [ - "message_start", # Initial message start - # TODO: for future contributors: if the initial content_block_start - # respects the upstream's starting chunk, the initial empty text block - # should be removed (and this test should be updated accordingly) - # --------------------------------------------------------------------- - "content_block_start", # Initial empty text block start - "content_block_stop", # End of empty text block - # --------------------------------------------------------------------- - "content_block_start", # Start of first tool_use content block + "message_start", + "content_block_start", # tool_use (from peek) "content_block_delta", # {"city": "content_block_delta", # "NY"} - "content_block_stop", # End of first tool_use content block - "message_delta", # Stop reason with merged usage - "message_stop", # Final message stop + "content_block_stop", + "message_delta", + "message_stop", ] assert expected_types == chunk_types @@ -193,26 +188,20 @@ def test_anthropic_stream_wrapper_back_to_back_tool_calls(): chunks.append(chunk) chunk_types.append(chunk.get("type")) - # Verify the expected sequence of chunk types + # Verify the expected sequence of chunk types. + # Empty tool-name trigger deltas are suppressed; only real args stream. expected_types = [ - "message_start", # Initial message start - # TODO: for future contributors: if the initial content_block_start - # respects the upstream's starting chunk, the initial empty text block - # should be removed (and this test should be updated accordingly) - # --------------------------------------------------------------------- - "content_block_start", # Initial empty text block start - "content_block_stop", # End of empty text block - # --------------------------------------------------------------------- - "content_block_start", # Start of first tool_use content block + "message_start", + "content_block_start", # tool_use (from peek) "content_block_delta", # {"city": "content_block_delta", # "NY"} - "content_block_stop", # End of first tool_use content block - "content_block_start", # Start of second tool_use content block + "content_block_stop", + "content_block_start", # second tool_use "content_block_delta", # {"city": - "content_block_delta", # " SF"} - "content_block_stop", # End of second tool_use content block - "message_delta", # Stop reason with merged usage - "message_stop", # Final message stop + "content_block_delta", # "SF"} + "content_block_stop", + "message_delta", + "message_stop", ] assert expected_types == chunk_types @@ -264,36 +253,30 @@ def test_anthropic_stream_wrapper_interleaved_tool_calls_and_text(): chunks.append(chunk) chunk_types.append(chunk.get("type")) - # Verify the expected sequence of chunk types + # Verify the expected sequence of chunk types. + # Empty tool-name trigger deltas are suppressed; bundled text deltas are kept. expected_types = [ - "message_start", # Initial message start - # TODO: for future contributors: if the initial content_block_start - # respects the upstream's starting chunk, the initial empty text block - # should be removed (and this test should be updated accordingly) - # --------------------------------------------------------------------- - "content_block_start", # Initial empty text block start - "content_block_stop", # End of empty text block - # --------------------------------------------------------------------- - "content_block_start", # Start of first tool_use content block + "message_start", + "content_block_start", # tool_use (from peek) "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" 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": - "content_block_delta", # " SF"} - "content_block_stop", # End of second tool_use content block - "content_block_start", # Start of third tool_use content block - "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" text block - "content_block_delta", # "The weather is not so nice today." text_delta + "content_block_start", # text + "content_block_delta", # "The weather is nice today." "content_block_stop", - "message_delta", # Stop reason with merged usage - "message_stop", # Final message stop + "content_block_start", # second tool_use + "content_block_delta", # {"city": + "content_block_delta", # "SF"} + "content_block_stop", + "content_block_start", # third tool_use + "content_block_delta", # {"city": + "content_block_delta", # "CHI"} + "content_block_stop", + "content_block_start", # text + "content_block_delta", # "The weather is not so nice today." + "content_block_stop", + "message_delta", + "message_stop", ] assert expected_types == chunk_types