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

Merge pull request #41564 from BerriAI/litellm_responses_bridge_message_item_lit4622

Browse source 
fix(responses): announce message item before text events in the chat completions bridge
This commit is contained in:
Mateo Wang 2026-09-18 22:36:28 -07:00 • committed by GitHub
commit b7f07469bc
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 237 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -22,7 +22,6 @@ from litellm.types.llms.openai import (
ContentPartAddedEvent,
ContentPartDoneEvent,
ContentPartDonePartOutputText,
ContentPartDonePartReasoningText,
FunctionCallArgumentsDeltaEvent,
FunctionCallArgumentsDoneEvent,
OutputItemAddedEvent,
@ -102,6 +101,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
@ -111,6 +111,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self.completed_response = None
self.final_text: str = ""
self._cached_item_id: str | None = None
self._message_output_index: int = 0
self._cached_response_id: str | None = None
self._buffered_chunk: ModelResponseStream | None = None
self._upstream_exhausted: bool = False
@ -563,7 +564,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self._sequence_number += 1
event: Final = OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=0,
output_index=self._message_output_index,
item=BaseLiteLLMOpenAIResponseObject(
**{
"id": self._cached_item_id,
@ -585,13 +586,41 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
event: Final = ContentPartAddedEvent(
type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED,
item_id=self._cached_item_id,
output_index=0,
output_index=self._message_output_index,
content_index=0,
part=BaseLiteLLMOpenAIResponseObject(**{"type": "output_text", "text": "", "annotations": []}),
)
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
if self._cached_reasoning_item_id is not None:
self._message_output_index = self._next_tool_output_index
self._next_tool_output_index += 1
else:
self._message_output_index = 0
self._sequence_number += 1
event: Final = OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=self._message_output_index,
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.
@ -711,7 +740,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
return OutputTextDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
item_id=self._cached_item_id,
output_index=0,
output_index=self._message_output_index,
content_index=0,
text=getattr(litellm_complete_object.choices[0].message, "content", "") or "",
)
@ -721,33 +750,24 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self._cached_item_id = f"msg_{uuid.uuid4()}"
text: Final = getattr(litellm_complete_object.choices[0].message, "content", "") or ""
reasoning_content = getattr(litellm_complete_object.choices[0].message, "reasoning_content", "") or ""
annotations: Final = getattr(litellm_complete_object.choices[0].message, "annotations", None)
part: PART_UNION_TYPES | None = None
if reasoning_content:
part = ContentPartDonePartReasoningText(
type="reasoning_text",
reasoning=reasoning_content,
)
else:
response_annotations: Final = (
LiteLLMCompletionResponsesConfig._transform_chat_completion_annotations_to_response_output_annotations(
annotations=annotations
)
)
part = ContentPartDonePartOutputText(
type="output_text",
text=text,
annotations=response_annotations,
logprobs=None,
response_annotations: Final = (
LiteLLMCompletionResponsesConfig._transform_chat_completion_annotations_to_response_output_annotations(
annotations=annotations
)
)
part: Final[PART_UNION_TYPES] = ContentPartDonePartOutputText(
type="output_text",
text=text,
annotations=response_annotations,
logprobs=None,
)
return ContentPartDoneEvent(
type=ResponsesAPIStreamEvents.CONTENT_PART_DONE,
item_id=self._cached_item_id,
output_index=0,
output_index=self._message_output_index,
content_index=0,
part=part,
)
@ -766,7 +786,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
)
return OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=0,
output_index=self._message_output_index,
sequence_number=1,
item=BaseLiteLLMOpenAIResponseObject(
**{
@ -832,6 +852,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)
@ -936,31 +965,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__(
@ -1115,12 +1120,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self.collected_chat_completion_chunks.append(
self._snapshot_chunk_for_stream_chunk_builder(cast(ModelResponseStream, chunk))
)
# Emit any just-queued output_item event
if self._pending_response_events:
return self._pending_response_events.pop(0)
response_api_chunk = self._transform_chat_completion_chunk_to_response_api_chunk(chunk)
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)
# Otherwise, loop to next chunk
except StopIteration:
return self.common_done_event_logic(sync_mode=True)
@ -1162,7 +1166,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
event = OutputTextAnnotationAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_ANNOTATION_ADDED,
item_id=item_id,
output_index=0,
output_index=self._message_output_index,
content_index=0,
annotation_index=idx,
annotation=annotation_dict,
@ -1189,11 +1193,13 @@ 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,
item_id=item_id,
output_index=0,
output_index=self._message_output_index,
content_index=0,
delta=delta_content,
)

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

@ -20,7 +20,10 @@ from litellm.responses.litellm_completion_transformation.streaming_iterator impo
LiteLLMCompletionStreamingIterator,
)
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.types.llms.openai import ResponsesAPIStreamEvents
from litellm.types.llms.openai import (
BaseLiteLLMOpenAIResponseObject,
ResponsesAPIStreamEvents,
)
from litellm.types.responses.main import build_web_search_call
from litellm.types.utils import (
Delta,
@ -957,3 +960,174 @@ 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[BaseLiteLLMOpenAIResponseObject]:
if sync_mode:
return list(iterator)
return [event async for event in iterator]
def _is_message_item(event: BaseLiteLLMOpenAIResponseObject) -> 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: bool):
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: bool):
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()
announced_indexes_by_item_type: dict[str, int] = {}
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:
announced_indexes_by_item_type[event.item.type] = event.output_index
if _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
assert "".join(
event.delta for event in events if getattr(event, "type", None) == ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA
) == "Hello!"
assert announced_indexes_by_item_type["message"] != announced_indexes_by_item_type["reasoning"]
@pytest.mark.asyncio
async def test_reasoning_item_closes_before_message_item_opens():
iterator: Final = _build_iterator(
[
_reasoning_chunk("let me think"),
_chunk("Hello"),
_chunk("!", finish_reason="stop"),
]
)
events: Final = await _collect_events(iterator, sync_mode=False)
item_lifecycle: Final = [
(event.type, event.item.type)
for event in events
if getattr(event, "type", None)
in (ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE)
]
assert item_lifecycle == [
(ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, "reasoning"),
(ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, "reasoning"),
(ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, "message"),
(ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, "message"),
]
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_tool_then_reasoning_then_text_gives_message_its_own_output_index(sync_mode: bool):
iterator: Final = _build_iterator(
[
_tool_call_chunk(),
_reasoning_chunk("thinking"),
_chunk("Hello"),
_chunk("!", finish_reason="stop"),
]
)
events: Final = await _collect_events(iterator, sync_mode)
output_item_added_events: Final = [
event for event in events if getattr(event, "type", None) == ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED
]
message_item_adds: Final = [event for event in output_item_added_events if _is_message_item(event)]
function_call_adds: Final = [
event for event in output_item_added_events if getattr(event.item, "type", None) == "function_call"
]
assert len(message_item_adds) == 1
assert all(message_item_adds[0].output_index != event.output_index for event in function_call_adds)
output_indexes_by_item_id: Final = {event.item.id: event.output_index for event in output_item_added_events}
assert len(output_indexes_by_item_id) == len(set(output_indexes_by_item_id.values()))
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_plain_text_stream_announces_exactly_one_message_item(sync_mode: bool):
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