diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index eeb728721f..8087a983d5 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -89,6 +89,9 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self._next_tool_output_index: int = 1 # output_index=0 reserved for the message item self._final_tool_events_queued: bool = False self._sequence_number: int = 0 + self._cached_reasoning_item_id: Optional[str] = None + # -- GENERIC RESPONSE-EVENTS PENDING QUEUE as required by fix -- + self._pending_response_events: List[BaseLiteLLMOpenAIResponseObject] = [] def _get_or_assign_tool_output_index(self, call_id: str) -> int: existing = self._tool_output_index_by_call_id.get(call_id) @@ -509,9 +512,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): elif self.sent_response_in_progress_event is False: self.sent_response_in_progress_event = True return self.create_response_in_progress_event() - elif self.sent_content_part_added_event is False: - self.sent_content_part_added_event = True - return self.create_content_part_added_event() return None def is_stream_finished(self) -> bool: @@ -558,6 +558,58 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): else: raise StopAsyncIteration + def _ensure_output_item_for_chunk(self, chunk: ModelResponseStream) -> None: + # Change: Never return a value, just enqueue output item events + if self.sent_output_item_added_event: + return + print("first chunk=", chunk) + delta = chunk.choices[0].delta + + self._sequence_number += 1 + self.sent_output_item_added_event = True + + # Reasoning-first + if hasattr(delta, "reasoning_content") and delta.reasoning_content: + if self._cached_reasoning_item_id is None: + self._cached_reasoning_item_id = f"rs_{uuid.uuid4()}" + + event = OutputItemAddedEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + output_index=0, + item=BaseLiteLLMOpenAIResponseObject( + id=self._cached_reasoning_item_id, + type="reasoning", + status="in_progress", + summary=None, + ), + sequence_number=self._sequence_number, + ) + self._pending_response_events.append(event) + return + + # Tool-first + if hasattr(delta, "tool_calls") and delta.tool_calls: + # Tool calls already handled via _queue_tool_call_delta_events + # DO NOT create message item + return + + # Default: message + self._cached_item_id = self._cached_item_id or f"msg_{uuid.uuid4()}" + event = OutputItemAddedEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + output_index=0, + item=BaseLiteLLMOpenAIResponseObject( + id=self._cached_item_id, + type="message", + role="assistant", + status="in_progress", + content=[], + ), + sequence_number=self._sequence_number, + ) + self._pending_response_events.append(event) + return + async def __anext__( self, ) -> Union[ @@ -573,11 +625,19 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): result = self.return_default_initial_events() if result: return result - # Get the next chunk from the stream + # Emit any pending output_item or other response events before reading a new chunk + if self._pending_response_events: + return self._pending_response_events.pop(0) + # Emit any pending tool events before reading a new chunk + if self._pending_tool_events: + return self._pending_tool_events.pop(0) + try: chunk = await self.litellm_custom_stream_wrapper.__anext__() if chunk is not None: chunk = cast(ModelResponseStream, chunk) + self._ensure_output_item_for_chunk(chunk) + # Proceed to transformation self.collected_chat_completion_chunks.append(chunk) response_api_chunk = ( self._transform_chat_completion_chunk_to_response_api_chunk( @@ -585,7 +645,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): ) ) if response_api_chunk: - return response_api_chunk + self._pending_response_events.append(response_api_chunk) + + if self._pending_response_events: + return self._pending_response_events.pop(0) + except StopAsyncIteration: return self.common_done_event_logic(sync_mode=False) @@ -608,13 +672,21 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): while True: if self.finished is True: raise StopIteration - # Get the next chunk from the stream - result = self.return_default_initial_events() if result: return result + # Emit any pending output_item or other response events before reading a new chunk + if self._pending_response_events: + return self._pending_response_events.pop(0) + # Emit any pending tool events before reading a new chunk + if self._pending_tool_events: + return self._pending_tool_events.pop(0) try: chunk = self.litellm_custom_stream_wrapper.__next__() + self._ensure_output_item_for_chunk(chunk) + # Emit any just-queued output_item event + if self._pending_response_events: + return self._pending_response_events.pop(0) self.collected_chat_completion_chunks.append(chunk) response_api_chunk = ( self._transform_chat_completion_chunk_to_response_api_chunk( @@ -623,6 +695,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): ) if response_api_chunk: return response_api_chunk + # Otherwise, loop to next chunk except StopIteration: return self.common_done_event_logic(sync_mode=True) except Exception as e: