diff --git a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py index 0d02b4fa969..8ddf0b50e33 100644 --- a/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py @@ -3,7 +3,7 @@ import json import traceback from collections import deque -from typing import Any, AsyncIterator, Dict +from typing import Any, AsyncIterator, Dict, Optional from litellm import verbose_logger from litellm._uuid import uuid @@ -38,6 +38,7 @@ class AnthropicResponsesStreamWrapper: self._pending_tool_ids: Dict[str, str] = {} # item_id -> call_id / name accumulator self._sent_message_start = False self._sent_message_stop = False + self._open_block_type: Optional[str] = None self._chunk_queue: deque = deque() def _make_message_start(self) -> Dict[str, Any]: @@ -64,6 +65,38 @@ class AnthropicResponsesStreamWrapper: self._current_block_index += 1 return self._current_block_index + def _close_open_block(self) -> None: + if self._open_block_type is not None: + self._chunk_queue.append( + { + "type": "content_block_stop", + "index": self._current_block_index, + } + ) + self._open_block_type = None + + def _open_content_block(self, content_block: Dict[str, Any], item_id: Optional[str] = None) -> int: + block_idx = self._next_block_index() + if item_id: + self._item_id_to_block_index[item_id] = block_idx + self._open_block_type = content_block["type"] + self._chunk_queue.append( + { + "type": "content_block_start", + "index": block_idx, + "content_block": content_block, + } + ) + return block_idx + + def _ensure_block_of_type( + self, block_type: str, content_block: Dict[str, Any], item_id: Optional[str] = None + ) -> int: + if self._open_block_type != block_type: + self._close_open_block() + return self._open_content_block(content_block, item_id) + return self._current_block_index + def _process_event(self, event: Any) -> None: """Convert one Responses API event into zero or more Anthropic chunks queued for emission.""" event_type = getattr(event, "type", None) @@ -75,6 +108,8 @@ class AnthropicResponsesStreamWrapper: # ---- message_start ---- if event_type == "response.created": + if self._sent_message_start: + return self._sent_message_start = True self._chunk_queue.append(self._make_message_start()) return @@ -88,69 +123,29 @@ class AnthropicResponsesStreamWrapper: item_id = getattr(item, "id", None) or (item.get("id") if isinstance(item, dict) else None) if item_type == "message": - block_idx = self._next_block_index() - if item_id: - self._item_id_to_block_index[item_id] = block_idx - self._chunk_queue.append( - { - "type": "content_block_start", - "index": block_idx, - "content_block": {"type": "text", "text": ""}, - } - ) - elif item_type == "function_call": + return + if item_type == "function_call": call_id = ( getattr(item, "call_id", None) or (item.get("call_id") if isinstance(item, dict) else None) or "" ) name = getattr(item, "name", None) or (item.get("name") if isinstance(item, dict) else None) or "" - block_idx = self._next_block_index() if item_id: - self._item_id_to_block_index[item_id] = block_idx self._pending_tool_ids[item_id] = call_id - self._chunk_queue.append( - { - "type": "content_block_start", - "index": block_idx, - "content_block": { - "type": "tool_use", - "id": call_id, - "name": name, - "input": {}, - }, - } + self._close_open_block() + self._open_content_block( + {"type": "tool_use", "id": call_id, "name": name, "input": {}}, + item_id, ) elif item_type == "reasoning": - block_idx = self._next_block_index() - if item_id: - self._item_id_to_block_index[item_id] = block_idx - self._chunk_queue.append( - { - "type": "content_block_start", - "index": block_idx, - "content_block": {"type": "thinking", "thinking": ""}, - } - ) + self._close_open_block() + self._open_content_block({"type": "thinking", "thinking": ""}, item_id) return # ---- text delta ---- if event_type == "response.output_text.delta": item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None) delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "") - block_idx = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index - if block_idx < 0: - # Some providers (e.g. LMStudio) skip response.output_item.added, - # so no text block is open yet; synthesize content_block_start - # instead of emitting a delta with index -1 - block_idx = self._next_block_index() - if item_id: - self._item_id_to_block_index[item_id] = block_idx - self._chunk_queue.append( - { - "type": "content_block_start", - "index": block_idx, - "content_block": {"type": "text", "text": ""}, - } - ) + block_idx = self._ensure_block_of_type("text", {"type": "text", "text": ""}, item_id) self._chunk_queue.append( { "type": "content_block_delta", @@ -164,11 +159,7 @@ class AnthropicResponsesStreamWrapper: if event_type == "response.reasoning_summary_text.delta": item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None) delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "") - block_idx = ( - self._item_id_to_block_index.get(item_id, self._current_block_index) - if item_id - else self._current_block_index - ) + block_idx = self._ensure_block_of_type("thinking", {"type": "thinking", "thinking": ""}, item_id) self._chunk_queue.append( { "type": "content_block_delta", @@ -198,21 +189,7 @@ class AnthropicResponsesStreamWrapper: # ---- output item done -> content_block_stop ---- if event_type == "response.output_item.done": - item = getattr(event, "item", None) or (event.get("item") if isinstance(event, dict) else None) - item_id = ( - getattr(item, "id", None) or (item.get("id") if isinstance(item, dict) else None) if item else None - ) - block_idx = ( - self._item_id_to_block_index.get(item_id, self._current_block_index) - if item_id - else self._current_block_index - ) - self._chunk_queue.append( - { - "type": "content_block_stop", - "index": block_idx, - } - ) + self._close_open_block() return # ---- response completed -> message_delta + message_stop ---- @@ -264,6 +241,8 @@ class AnthropicResponsesStreamWrapper: if cache_read_tokens: usage_delta["cache_read_input_tokens"] = cache_read_tokens + self._close_open_block() + self._chunk_queue.append( { "type": "message_delta", diff --git a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py index 450f69fb87c..3307c700232 100644 --- a/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py +++ b/tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py @@ -6,6 +6,8 @@ Tests for AnthropicResponsesStreamWrapper import os import sys +import pytest + sys.path.insert( 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")) ) @@ -22,6 +24,27 @@ def _process_all(events: list) -> list: return list(wrapper._chunk_queue) +class _FakeResponsesStream: + def __init__(self, events: list) -> None: + self._it = iter(events) + + def __aiter__(self): + return self + + async def __anext__(self): + try: + return next(self._it) + except StopIteration: + raise StopAsyncIteration + + +async def _drain(events: list) -> list: + wrapper = AnthropicResponsesStreamWrapper( + responses_stream=_FakeResponsesStream(events), model="m" + ) + return [chunk async for chunk in wrapper] + + class TestProcessEventTextDeltaWithoutOutputItemAdded: """Streams that skip response.output_item.added (e.g. LMStudio) must still open a text block before any delta and never emit index -1.""" @@ -59,9 +82,15 @@ class TestProcessEventTextDeltaWithoutOutputItemAdded: {"type": "response.output_text.delta", "item_id": "m1", "delta": "Hi"}, ] ) - assert chunks[1]["type"] == "content_block_start" - assert chunks[1]["content_block"] == {"type": "text", "text": ""} - assert [c["index"] for c in chunks[1:]] == [1, 1] + assert [c["type"] for c in chunks] == [ + "content_block_start", + "content_block_stop", + "content_block_start", + "content_block_delta", + ] + assert chunks[0]["content_block"] == {"type": "thinking", "thinking": ""} + assert chunks[2]["content_block"] == {"type": "text", "text": ""} + assert [c["index"] for c in chunks] == [0, 0, 1, 1] def test_process_event_registered_item_id_does_not_synthesize_start(self): chunks = _process_all( @@ -77,3 +106,84 @@ class TestProcessEventTextDeltaWithoutOutputItemAdded: ("content_block_start", 0), ("content_block_delta", 0), ] + + +def _reasoning_first_bridge_events() -> list: + """The event sequence a chat-completions -> Responses API bridge emits for + an OpenAI-compatible reasoning backend: a message output item is announced + first (from the role-only chunk), reasoning arrives as + reasoning_summary_text.delta events with unrelated item_ids, then the text + answer arrives on the message item.""" + return [ + {"type": "response.created"}, + {"type": "response.in_progress"}, + {"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}}, + {"type": "response.content_part.added", "item_id": "msg_1"}, + {"type": "response.reasoning_summary_text.delta", "item_id": "rs_a", "delta": "I"}, + {"type": "response.reasoning_summary_text.delta", "item_id": "rs_b", "delta": " am"}, + {"type": "response.reasoning_summary_text.delta", "item_id": "rs_c", "delta": " thinking"}, + {"type": "response.output_text.delta", "item_id": "msg_1", "delta": "OK"}, + {"type": "response.output_text.done", "item_id": "msg_1"}, + {"type": "response.content_part.done", "item_id": "msg_1"}, + {"type": "response.output_item.done", "item": {"type": "message", "id": "msg_1"}}, + {"type": "response.completed"}, + ] + + +class TestReasoningContentIsNotStreamedIntoTextBlock: + """Regression tests for https://github.com/BerriAI/litellm/issues/32357""" + + @pytest.mark.asyncio + async def test_message_start_emitted_exactly_once(self): + chunks = await _drain(_reasoning_first_bridge_events()) + assert [c["type"] for c in chunks].count("message_start") == 1 + + @pytest.mark.asyncio + async def test_reasoning_streams_into_its_own_thinking_block(self): + chunks = await _drain(_reasoning_first_bridge_events()) + + assert [c["type"] for c in chunks] == [ + "message_start", + "content_block_start", + "content_block_delta", + "content_block_delta", + "content_block_delta", + "content_block_stop", + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", + "message_stop", + ] + + thinking_block = chunks[1] + assert thinking_block["content_block"] == {"type": "thinking", "thinking": ""} + assert thinking_block["index"] == 0 + assert [c["delta"]["type"] for c in chunks[2:5]] == [ + "thinking_delta", + "thinking_delta", + "thinking_delta", + ] + assert all(c["index"] == 0 for c in chunks[2:5]) + + text_block = chunks[6] + assert text_block["content_block"] == {"type": "text", "text": ""} + assert text_block["index"] == 1 + assert chunks[7]["delta"] == {"type": "text_delta", "text": "OK"} + assert chunks[7]["index"] == 1 + + @pytest.mark.asyncio + async def test_no_thinking_delta_is_ever_emitted_into_a_text_block(self): + chunks = await _drain(_reasoning_first_bridge_events()) + + text_block_indexes = { + c["index"] + for c in chunks + if c["type"] == "content_block_start" and c["content_block"]["type"] == "text" + } + thinking_delta_indexes = { + c["index"] + for c in chunks + if c["type"] == "content_block_delta" and c["delta"]["type"] == "thinking_delta" + } + assert text_block_indexes.isdisjoint(thinking_delta_indexes)