mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-02 02:11:58 +00:00
Merge 7639e47d32 into 1ceeefbf84
This commit is contained in:
Rain.
GitHub
commit
d49721ffaa
5 changed files with 336 additions and 0 deletions
|
|
@ -87,6 +87,14 @@ class ResponsesToCompletionBridgeHandler:
|
|||
response: Final = self._coerce_response_object(response_obj, hidden_params)
|
||||
if not isinstance(response, ResponsesAPIResponse):
|
||||
raise ValueError("Stream completed response is invalid")
|
||||
# Some providers emit a terminal response.completed event with an empty output
|
||||
# array even though output_item.done events streamed the full answer. Rebuild
|
||||
# the output from those streamed items before failing on empty choices.
|
||||
get_streamed_items: Final = getattr(stream_iter, "get_streamed_output_items", None)
|
||||
if len(response.output) == 0 and callable(get_streamed_items):
|
||||
streamed_items: Final = get_streamed_items()
|
||||
if streamed_items:
|
||||
response.output = streamed_items # pyright: ignore[reportAttributeAccessIssue] # assigning typed SSE items back to ResponsesAPIResponse.output
|
||||
return response
|
||||
|
||||
async def _collect_response_from_stream_async(self, stream_iter: AsyncIterable[object]) -> "ResponsesAPIResponse":
|
||||
|
|
@ -102,6 +110,11 @@ class ResponsesToCompletionBridgeHandler:
|
|||
response: Final = self._coerce_response_object(response_obj, hidden_params)
|
||||
if not isinstance(response, ResponsesAPIResponse):
|
||||
raise ValueError("Stream completed response is invalid")
|
||||
get_streamed_items: Final = getattr(stream_iter, "get_streamed_output_items", None)
|
||||
if len(response.output) == 0 and callable(get_streamed_items):
|
||||
streamed_items: Final = get_streamed_items()
|
||||
if streamed_items:
|
||||
response.output = streamed_items # pyright: ignore[reportAttributeAccessIssue] # assigning typed SSE items back to ResponsesAPIResponse.output
|
||||
return response
|
||||
|
||||
def validate_input_kwargs(self, kwargs: dict) -> ResponsesToCompletionBridgeHandlerInputKwargs:
|
||||
|
|
|
|||
|
|
@ -324,6 +324,10 @@ class BaseResponsesAPIStreamingIterator:
|
|||
self._completed_response_logged = False
|
||||
self._completed_response_cache_hit: bool | None = None
|
||||
self._persist_completed_response_before_logging = True
|
||||
# output_item.done events seen during the stream, keyed by output_index.
|
||||
# Some providers emit a terminal response.completed event with an empty
|
||||
# output array even though items were streamed; these are used to rebuild it.
|
||||
self._streamed_output_items: dict[int, object] = {} # mutable-ok: streaming accumulator keyed by output_index
|
||||
self._stream_created_time: float = time.time()
|
||||
|
||||
# track request context for hooks
|
||||
|
|
@ -431,6 +435,22 @@ class BaseResponsesAPIStreamingIterator:
|
|||
custom_llm_provider=self.custom_llm_provider,
|
||||
model_id=_stream_model_id,
|
||||
)
|
||||
if _event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE:
|
||||
_done_item: Final[object] = getattr(openai_responses_api_chunk, "item", None)
|
||||
if _done_item is not None:
|
||||
_output_index: Final = getattr(openai_responses_api_chunk, "output_index", None)
|
||||
if (
|
||||
isinstance(_output_index, int)
|
||||
and not isinstance(_output_index, bool)
|
||||
and _output_index not in self._streamed_output_items
|
||||
):
|
||||
self._streamed_output_items[_output_index] = _done_item
|
||||
else:
|
||||
# missing/invalid/duplicate index: append after the highest known
|
||||
# index so the fallback can never collide with a real index and
|
||||
# silently discard a streamed item
|
||||
_fallback_index: Final = max(self._streamed_output_items.keys(), default=-1) + 1
|
||||
self._streamed_output_items[_fallback_index] = _done_item
|
||||
elif _event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_ANNOTATION_ADDED:
|
||||
_annotation: Final[object] = getattr(openai_responses_api_chunk, "annotation", None)
|
||||
if _annotation is not None:
|
||||
|
|
@ -533,6 +553,18 @@ class BaseResponsesAPIStreamingIterator:
|
|||
self._handle_failure(e)
|
||||
raise
|
||||
|
||||
def get_streamed_output_items(self) -> list[object]: # mutable-ok: response.output takes a real list
|
||||
"""
|
||||
Output items received via response.output_item.done events, ordered by output_index.
|
||||
|
||||
Used to rebuild a terminal response.completed payload whose output array is empty.
|
||||
"""
|
||||
if not self._streamed_output_items:
|
||||
return [] # mutable-ok: response.output takes a real list
|
||||
return [ # mutable-ok: response.output takes a real list
|
||||
item for _, item in sorted(self._streamed_output_items.items())
|
||||
]
|
||||
|
||||
def _log_completed_response(self, *, is_async: bool) -> None:
|
||||
if self._completed_response_logged:
|
||||
return
|
||||
|
|
|
|||
|
|
@ -335,3 +335,131 @@ async def test_acompletion_keeps_provider_native_model_id_through_responses(
|
|||
|
||||
handed_model = responses_call.call_args.kwargs["model"]
|
||||
assert _upstream_model_for(handed_model, custom_llm_provider) == expected_upstream_model
|
||||
# ------------------------------------------------------------------------
|
||||
# #41009/#41014: rebuild empty streamed output from output_item.done events
|
||||
# ------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _FakeCompletedResponse:
|
||||
response = {
|
||||
"id": "resp_41014",
|
||||
"object": "response",
|
||||
"created_at": 0,
|
||||
"status": "completed",
|
||||
"model": "gpt-5.4",
|
||||
"output": [],
|
||||
}
|
||||
|
||||
|
||||
class _FakeStreamIter:
|
||||
def __init__(self, streamed_items):
|
||||
self.completed_response = _FakeCompletedResponse()
|
||||
self._hidden_params = None
|
||||
self._items = streamed_items
|
||||
|
||||
def __iter__(self):
|
||||
return iter([])
|
||||
|
||||
def get_streamed_output_items(self):
|
||||
return self._items
|
||||
|
||||
|
||||
class _FakeAsyncStreamIter(_FakeStreamIter):
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def __anext__(self):
|
||||
raise StopAsyncIteration
|
||||
|
||||
|
||||
def test_collect_response_from_stream_rebuilds_empty_output_from_streamed_items():
|
||||
from litellm.completion_extras.litellm_responses_transformation.handler import (
|
||||
ResponsesToCompletionBridgeHandler,
|
||||
)
|
||||
|
||||
bridge = ResponsesToCompletionBridgeHandler()
|
||||
streamed_items = [
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_1",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "Hello", "annotations": []}],
|
||||
}
|
||||
]
|
||||
response = bridge._collect_response_from_stream(_FakeStreamIter(streamed_items))
|
||||
assert len(response.output) == 1
|
||||
assert response.output[0]["id"] == "msg_1"
|
||||
|
||||
|
||||
def test_collect_response_from_stream_keeps_nonempty_output_untouched():
|
||||
from litellm.completion_extras.litellm_responses_transformation.handler import (
|
||||
ResponsesToCompletionBridgeHandler,
|
||||
)
|
||||
|
||||
bridge = ResponsesToCompletionBridgeHandler()
|
||||
|
||||
class _NonEmptyCompleted:
|
||||
response = {
|
||||
"id": "resp_nonempty",
|
||||
"object": "response",
|
||||
"created_at": 0,
|
||||
"status": "completed",
|
||||
"model": "gpt-5.4",
|
||||
"output": [{"type": "message", "id": "msg_0"}],
|
||||
}
|
||||
|
||||
class _StreamNoItems:
|
||||
completed_response = _NonEmptyCompleted()
|
||||
_hidden_params = None
|
||||
|
||||
def __iter__(self):
|
||||
return iter([])
|
||||
|
||||
response = bridge._collect_response_from_stream(_StreamNoItems())
|
||||
assert len(response.output) == 1
|
||||
assert response.output[0]["id"] == "msg_0"
|
||||
|
||||
|
||||
def test_collect_response_from_stream_without_streamed_items_keeps_empty_output():
|
||||
from litellm.completion_extras.litellm_responses_transformation.handler import (
|
||||
ResponsesToCompletionBridgeHandler,
|
||||
)
|
||||
|
||||
bridge = ResponsesToCompletionBridgeHandler()
|
||||
|
||||
class _NoStreamedItems:
|
||||
completed_response = _FakeCompletedResponse()
|
||||
_hidden_params = None
|
||||
|
||||
def __iter__(self):
|
||||
return iter([])
|
||||
|
||||
response = bridge._collect_response_from_stream(_NoStreamedItems())
|
||||
assert len(response.output) == 0
|
||||
|
||||
|
||||
def test_collect_response_from_stream_async_rebuilds_empty_output():
|
||||
from litellm.completion_extras.litellm_responses_transformation.handler import (
|
||||
ResponsesToCompletionBridgeHandler,
|
||||
)
|
||||
from litellm.types.utils import ResponsesAPIResponse
|
||||
import asyncio
|
||||
|
||||
async def run():
|
||||
bridge = ResponsesToCompletionBridgeHandler()
|
||||
streamed_items = [
|
||||
{
|
||||
"type": "message",
|
||||
"id": "msg_async_1",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "Hi", "annotations": []}],
|
||||
}
|
||||
]
|
||||
response = await bridge._collect_response_from_stream_async(_FakeAsyncStreamIter(streamed_items))
|
||||
assert isinstance(response, ResponsesAPIResponse)
|
||||
assert len(response.output) == 1
|
||||
assert response.output[0]["id"] == "msg_async_1"
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1280,3 +1280,89 @@ def test_persist_completed_response_to_cache_survives_an_unserializable_response
|
|||
iterator._persist_completed_response_to_cache(is_async=False)
|
||||
|
||||
cache.add_cache.assert_not_called()
|
||||
|
||||
|
||||
def test_process_chunk_collects_output_item_done_in_order():
|
||||
"""output_item.done events are keyed by index and returned in order."""
|
||||
logging_obj = _logging_obj_for_collector()
|
||||
iterator = _make_collector_iterator(logging_obj=logging_obj)
|
||||
|
||||
for index, item in enumerate([{"id": "msg_1"}, {"id": "msg_2"}]):
|
||||
iterator._process_chunk(
|
||||
json.dumps({"type": "response.output_item.done", "output_index": index, "item": item})
|
||||
)
|
||||
|
||||
assert iterator.get_streamed_output_items() == [{"id": "msg_1"}, {"id": "msg_2"}]
|
||||
|
||||
|
||||
def test_process_chunk_collector_fallback_index_avoids_collision():
|
||||
"""A missing/invalid index appends after the max key, never overwriting an entry."""
|
||||
logging_obj = _logging_obj_for_collector()
|
||||
iterator = _make_collector_iterator(logging_obj=logging_obj)
|
||||
|
||||
iterator._process_chunk(
|
||||
json.dumps(
|
||||
{"type": "response.output_item.done", "output_index": 5, "item": {"id": "msg_5"}}
|
||||
)
|
||||
)
|
||||
iterator._process_chunk(
|
||||
json.dumps({"type": "response.output_item.done", "item": {"id": "msg_fallback"}})
|
||||
)
|
||||
items = iterator.get_streamed_output_items()
|
||||
assert items == [{"id": "msg_5"}, {"id": "msg_fallback"}]
|
||||
|
||||
|
||||
def test_process_chunk_collector_duplicate_index_appends():
|
||||
"""A repeated index is never dropped: the second item is appended after the max key."""
|
||||
logging_obj = _logging_obj_for_collector()
|
||||
iterator = _make_collector_iterator(logging_obj=logging_obj)
|
||||
|
||||
iterator._process_chunk(
|
||||
json.dumps(
|
||||
{"type": "response.output_item.done", "output_index": 1, "item": {"id": "msg_a"}}
|
||||
)
|
||||
)
|
||||
iterator._process_chunk(
|
||||
json.dumps(
|
||||
{"type": "response.output_item.done", "output_index": 1, "item": {"id": "msg_b"}}
|
||||
)
|
||||
)
|
||||
assert iterator.get_streamed_output_items() == [{"id": "msg_a"}, {"id": "msg_b"}]
|
||||
|
||||
|
||||
def test_get_streamed_output_items_empty_when_no_done_events():
|
||||
logging_obj = _logging_obj_for_collector()
|
||||
iterator = _make_collector_iterator(logging_obj=logging_obj)
|
||||
assert iterator.get_streamed_output_items() == []
|
||||
|
||||
|
||||
def _streamed_item_config() -> Mock:
|
||||
"""transform_streaming_response returns output_item.done events with item/index preserved."""
|
||||
mock_config = Mock()
|
||||
mock_config.transform_streaming_response.side_effect = (
|
||||
lambda model, parsed_chunk, logging_obj: Mock(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
|
||||
item=parsed_chunk.get("item"),
|
||||
output_index=parsed_chunk.get("output_index"),
|
||||
)
|
||||
)
|
||||
return mock_config
|
||||
|
||||
|
||||
def _make_collector_iterator(*, logging_obj: Mock) -> SyncResponsesAPIStreamingIterator:
|
||||
return SyncResponsesAPIStreamingIterator(
|
||||
response=Mock(headers={}),
|
||||
model="gpt-4o-mini",
|
||||
responses_api_provider_config=_streamed_item_config(),
|
||||
logging_obj=logging_obj,
|
||||
litellm_metadata={},
|
||||
custom_llm_provider="openai",
|
||||
call_type="responses",
|
||||
)
|
||||
|
||||
|
||||
def _logging_obj_for_collector() -> Mock:
|
||||
logging_obj = Mock(spec=LiteLLMLoggingObj)
|
||||
logging_obj.completion_start_time = None
|
||||
logging_obj.model_call_details = {"litellm_params": {}}
|
||||
return logging_obj
|
||||
|
|
|
|||
|
|
@ -66,6 +66,83 @@ def test_should_collect_response_from_stream():
|
|||
assert collected._hidden_params.get("headers") == {"x-test": "1"}
|
||||
|
||||
|
||||
class _RecoverableResponsesStream(_FakeResponsesStream):
|
||||
"""Terminal response.completed carries an empty output array even though
|
||||
output_item.done events streamed the full answer (issue #41009)."""
|
||||
|
||||
def __init__(self, response, streamed_items):
|
||||
super().__init__(response)
|
||||
self._streamed_items = streamed_items
|
||||
|
||||
def get_streamed_output_items(self):
|
||||
return self._streamed_items
|
||||
|
||||
|
||||
class _RecoverableAsyncResponsesStream(_RecoverableResponsesStream):
|
||||
def __aiter__(self):
|
||||
return self
|
||||
|
||||
async def __anext__(self):
|
||||
if not self._emitted:
|
||||
self._emitted = True
|
||||
self.completed_response = _CompletedEvent(self._response)
|
||||
return {"type": "response.completed"}
|
||||
raise StopAsyncIteration
|
||||
|
||||
|
||||
def test_collect_response_from_stream_rebuilds_empty_output_from_streamed_items():
|
||||
handler = ResponsesToCompletionBridgeHandler()
|
||||
response = ResponsesAPIResponse.model_construct(
|
||||
id="resp-recover",
|
||||
created_at=0,
|
||||
output=[],
|
||||
object="response",
|
||||
model="gpt-5.6-sol",
|
||||
)
|
||||
streamed_items = [
|
||||
{
|
||||
"id": "msg_1",
|
||||
"type": "message",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "Hi! How can I help?", "annotations": []}],
|
||||
}
|
||||
]
|
||||
stream = _RecoverableResponsesStream(response, streamed_items)
|
||||
|
||||
collected = handler._collect_response_from_stream(stream)
|
||||
|
||||
assert len(collected.output) == 1
|
||||
assert "Hi! How can I help?" in str(collected.output[0])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_collect_response_from_stream_async_rebuilds_empty_output_from_streamed_items():
|
||||
handler = ResponsesToCompletionBridgeHandler()
|
||||
response = ResponsesAPIResponse.model_construct(
|
||||
id="resp-recover-async",
|
||||
created_at=0,
|
||||
output=[],
|
||||
object="response",
|
||||
model="gpt-5.6-sol",
|
||||
)
|
||||
streamed_items = [
|
||||
{
|
||||
"id": "msg_1",
|
||||
"type": "message",
|
||||
"status": "completed",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "Hi! How can I help?", "annotations": []}],
|
||||
}
|
||||
]
|
||||
stream = _RecoverableAsyncResponsesStream(response, streamed_items)
|
||||
|
||||
collected = await handler._collect_response_from_stream_async(stream)
|
||||
|
||||
assert len(collected.output) == 1
|
||||
assert "Hi! How can I help?" in str(collected.output[0])
|
||||
|
||||
|
||||
def create_mock_completion_response(
|
||||
model: str = "gpt-4",
|
||||
prompt_tokens: int = 10,
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue