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..5bfe1f56a67 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py @@ -105,13 +105,50 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if self.sent_content_block_start is False: self.sent_content_block_start = True - 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, + ) + 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: @@ -128,9 +165,6 @@ 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. self.chunk_queue.append( { "type": "content_block_stop", @@ -144,6 +178,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): "content_block": self.current_content_block_start, } ) + self.chunk_queue.append(processed_chunk) self.sent_content_block_finish = False return self.chunk_queue.popleft() @@ -226,13 +261,47 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): if self.sent_content_block_start is False: self.sent_content_block_start = True - 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, + ) + 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: @@ -304,19 +373,12 @@ 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. - - # 1. Stop current content block self.chunk_queue.append( { "type": "content_block_stop", "index": max(self.current_content_block_index - 1, 0), } ) - - # 2. Start new content block self.chunk_queue.append( { "type": "content_block_start", @@ -324,11 +386,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): "content_block": self.current_content_block_start, } ) - - # Reset state for new block + self.chunk_queue.append(processed_chunk) self.sent_content_block_finish = False - - # Return the first queued item return self.chunk_queue.popleft() if ( diff --git a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py b/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py index ed49943b7fe..26ff7a4f9bb 100644 --- a/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py +++ b/litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py @@ -1397,6 +1397,13 @@ class LiteLLMAnthropicMessagesAdapter: return "thinking", ChatCompletionThinkingBlock( type="thinking", thinking=thinking, signature=signature ) + elif isinstance(choice, StreamingChoices) and hasattr( + choice.delta, "reasoning_content" + ): + if choice.delta.reasoning_content is not None: + return "thinking", ChatCompletionThinkingBlock( + type="thinking", thinking="", signature="" + ) return "text", TextBlock(type="text", text="") 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..10e20ba283a --- /dev/null +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_reasoning_content.py @@ -0,0 +1,313 @@ +""" +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 + + +# --------------------------------------------------------------------------- +# 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" + + @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" 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..7dc081f2144 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,19 @@ 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. + # The initial content_block_start now peeks at the first upstream chunk + # to determine the correct block type, so we get tool_use directly + # instead of a spurious empty text block. 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", # first tool chunk (empty args) "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 +190,21 @@ 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. 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", # first tool chunk (empty args) "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", # first chunk of second tool "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_stop", + "message_delta", + "message_stop", ] assert expected_types == chunk_types @@ -264,34 +256,32 @@ 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. 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", # first tool chunk (empty args) "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_stop", - "content_block_start", # Start of second tool_use content block + "content_block_start", # text + "content_block_delta", # "The weather is nice today." + "content_block_stop", + "content_block_start", # second tool_use + "content_block_delta", # first chunk of second tool "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_stop", + "content_block_start", # third tool_use + "content_block_delta", # first chunk of third tool "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_stop", - "message_delta", # Stop reason with merged usage - "message_stop", # Final message 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