scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-22 00:31:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 51

fix(responses): announce message item before text events in the chat completions bridge

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Devin AI 2026-09-17 04:46:26 +00:00
parent 5ef40a630b
commit 441021fc96
2 changed files with 160 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

67
litellm/responses/litellm_completion_transformation/streaming_iterator.py
View file

@ -102,6 +102,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self.sent_response_created_event: bool = False
self.sent_response_in_progress_event: bool = False
self.sent_output_item_added_event: bool = False
self.sent_message_item_added_event: bool = False
self.sent_content_part_added_event: bool = False
self.sent_output_text_done_event: bool = False
self.sent_output_content_part_done_event: bool = False
@ -592,6 +593,29 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
event.__dict__["sequence_number"] = self._sequence_number
return event
def _queue_message_item_added_events(self) -> None:
if self._cached_item_id is None:
self._cached_item_id = f"msg_{uuid.uuid4()}"
self.sent_message_item_added_event = True
self.sent_content_part_added_event = True
self._sequence_number += 1
event: Final = OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=0,
item=BaseLiteLLMOpenAIResponseObject(
**{
"id": self._cached_item_id,
"type": "message",
"role": "assistant",
"status": "in_progress",
"content": [],
}
),
)
event.__dict__["sequence_number"] = self._sequence_number
self._pending_response_events.append(event)
self._pending_response_events.append(self.create_content_part_added_event())
def _merge_provider_specific_fields(self, src: dict) -> None:
"""Merge provider_specific_fields using last-value-wins for lists.
@ -832,6 +856,15 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
def return_default_done_events(
self, litellm_complete_object: ModelResponse
) -> BaseLiteLLMOpenAIResponseObject | None:
if self.sent_message_item_added_event is False:
final_content: Final = litellm_complete_object.choices[0].message.content or ""
if not final_content:
self.sent_output_text_done_event = True
self.sent_output_content_part_done_event = True
self.sent_output_item_done_event = True
return None
self._queue_message_item_added_events()
return self._pending_response_events.pop(0)
if self.sent_output_text_done_event is False:
self.sent_output_text_done_event = True
return self.create_output_text_done_event(litellm_complete_object)
@ -898,6 +931,12 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
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:
if (
not self.sent_message_item_added_event
and chunk.choices
and self._get_delta_string_from_streaming_choices(chunk.choices)
):
self._queue_message_item_added_events()
return
if not chunk.choices:
return
@ -936,31 +975,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
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": [],
}
),
)
event.__dict__["sequence_number"] = self._sequence_number
self._pending_response_events.append(event)
# Emit content_part.added immediately after output_item.added for message
# items. The OpenAI Responses spec requires this event before any
# output_text.delta events so downstream parsers can initialize the
# text part structure.
if not self.sent_content_part_added_event:
self.sent_content_part_added_event = True
content_part_event: Final = self.create_content_part_added_event()
self._pending_response_events.append(content_part_event)
self._queue_message_item_added_events()
return
async def __anext__(
@ -1189,6 +1204,8 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
# Priority 2: Handle text deltas
delta_content: Final = self._get_delta_string_from_streaming_choices(chunk.choices)
if delta_content:
if not self.sent_message_item_added_event:
self._queue_message_item_added_events()
self._sequence_number += 1
text_delta_event: Final = OutputTextDeltaEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,

118
tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py
View file

@ -957,3 +957,121 @@ def test_streamed_unrecognized_tool_choice_is_echoed_as_auto() -> None:
]
assert [event.response.tool_choice for event in response_events] == ["auto", "auto", "auto"]
def _reasoning_chunk(reasoning: str, finish_reason: str | None = None) -> ModelResponseStream:
return ModelResponseStream(
id=CHAT_COMPLETION_ID,
created=1748575031,
model="claude-haiku-4-5",
object="chat.completion.chunk",
choices=[
StreamingChoices(
index=0,
delta=Delta(role="assistant", reasoning_content=reasoning),
finish_reason=finish_reason,
)
],
)
async def _collect_events(iterator: LiteLLMCompletionStreamingIterator, sync_mode: bool) -> list:
if sync_mode:
return list(iterator)
return [event async for event in iterator]
def _is_message_item(event) -> bool:
return getattr(getattr(event, "item", None), "type", None) == "message"
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_tool_only_stream_emits_no_message_item_events(sync_mode):
"""
A turn that only calls tools must not announce or close a message output item:
Vercel AI SDK clients reject text/item events that reference a message id they
never saw in response.output_item.added.
"""
iterator: Final = _build_iterator([_tool_call_chunk(), _chunk("", finish_reason="tool_calls")])
events: Final = await _collect_events(iterator, sync_mode)
message_item_events = [
event
for event in events
if getattr(event, "type", None)
in (ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE)
and _is_message_item(event)
]
assert message_item_events == []
assert [
event
for event in events
if str(getattr(event, "type", "")).startswith("response.output_text")
or getattr(event, "type", None)
in (ResponsesAPIStreamEvents.CONTENT_PART_ADDED, ResponsesAPIStreamEvents.CONTENT_PART_DONE)
] == []
assert any(getattr(event, "type", None) == ResponsesAPIStreamEvents.RESPONSE_COMPLETED for event in events)
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_reasoning_then_text_announces_message_item_before_text_events(sync_mode):
"""
When reasoning is announced first, a later text delta still has to be preceded by
the message output_item.added/content_part.added, and every text-scoped event must
reference that announced message item id.
"""
iterator: Final = _build_iterator(
[
_reasoning_chunk("let me think"),
_chunk("Hello"),
_chunk("!", finish_reason="stop"),
]
)
events: Final = await _collect_events(iterator, sync_mode)
announced_message_ids: set[str] = set()
content_part_added_seen = False
saw_text_delta = False
for event in events:
event_type = getattr(event, "type", None)
if event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED and _is_message_item(event):
announced_message_ids.add(event.item.id)
elif event_type == ResponsesAPIStreamEvents.CONTENT_PART_ADDED:
content_part_added_seen = True
elif event_type in (
ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
ResponsesAPIStreamEvents.CONTENT_PART_DONE,
):
assert event.item_id in announced_message_ids
if event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA:
assert content_part_added_seen
saw_text_delta = True
elif event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE and _is_message_item(event):
assert event.item.id in announced_message_ids
assert saw_text_delta
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_plain_text_stream_announces_exactly_one_message_item(sync_mode):
iterator: Final = _build_iterator([_chunk("Hel"), _chunk("lo", finish_reason="stop")])
events: Final = await _collect_events(iterator, sync_mode)
message_item_adds = [
event
for event in events
if getattr(event, "type", None) == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED and _is_message_item(event)
]
assert len(message_item_adds) == 1
for event in events:
if getattr(event, "type", None) in (
ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
):
assert event.item_id == message_item_adds[0].item.id