mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-27 01:22:18 +00:00
fix(responses): fall back on pre-output stream drops, fail truncated streams, honor request_timeout (#43133)
* fix(responses): fall back on pre-output stream drops, fail truncated streams, honor request_timeout A native /v1/responses stream that drops before any output item now raises the router's fallback-eligible MidStreamFallbackError, so configured fallbacks retry the original input. A stream that ends with a clean EOF or a [DONE] marker but no response.completed, response.incomplete or response.failed event now raises litellm.APIConnectionError instead of ending as if it had completed: fallback-eligible before any output, an explicit error after partial output. The sync iterator mirrors every branch. resolve_llm_passthrough_timeout now consults an explicitly set litellm_settings.request_timeout right after the router timeout and before general_settings.pass_through_request_timeout, so the router's native responses path honors it. * test(responses): give the normal-completion stream tests a terminal event --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot]
GitHub
parent
8327cd6d47
commit
1fd04abb92
6 changed files with 420 additions and 28 deletions
|
|
@ -5,6 +5,8 @@ from typing import Final
|
|||
|
||||
from pydantic import TypeAdapter
|
||||
|
||||
from litellm.litellm_core_utils.request_timeout_resolver import get_configured_request_timeout
|
||||
|
||||
DEFAULT_PASS_THROUGH_REQUEST_TIMEOUT_SECONDS: Final = 600.0
|
||||
|
||||
_SECONDS: Final = TypeAdapter(float)
|
||||
|
|
@ -48,8 +50,8 @@ def resolve_llm_passthrough_timeout(
|
|||
Anthropic /v1/messages).
|
||||
|
||||
Non-streaming precedence: kwargs timeout/request_timeout -> litellm_params
|
||||
timeout/request_timeout -> router_timeout -> general_settings.pass_through_request_timeout
|
||||
-> 600s default.
|
||||
timeout/request_timeout -> router_timeout -> litellm.request_timeout (litellm_settings.request_timeout,
|
||||
when explicitly set) -> general_settings.pass_through_request_timeout -> 600s default.
|
||||
|
||||
Streaming (``kwargs["stream"]`` truthy) resolves ``stream_timeout`` at every level before
|
||||
any generic timeout, matching ``Router._get_stream_timeout`` on the completion route:
|
||||
|
|
@ -73,6 +75,7 @@ def resolve_llm_passthrough_timeout(
|
|||
deployment.get("timeout"),
|
||||
deployment.get("request_timeout"),
|
||||
router_timeout,
|
||||
get_configured_request_timeout(),
|
||||
)
|
||||
winner: Final = next((val for val in candidates if val is not None), None)
|
||||
return resolve_pass_through_request_timeout() if winner is None else _SECONDS.validate_python(winner)
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
|
|||
from datetime import datetime
|
||||
from functools import lru_cache
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, overload, runtime_checkable
|
||||
from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, Protocol, overload, runtime_checkable
|
||||
|
||||
import httpx
|
||||
from openai._streaming import SSEDecoder
|
||||
|
|
@ -265,6 +265,9 @@ def _mid_stream_fallback_eligible(mapped_exception: Exception) -> bool:
|
|||
return not isinstance(status_code, int) or status_code >= 500 or status_code == 429
|
||||
|
||||
|
||||
_PRE_OUTPUT_LIFECYCLE_EVENT_TYPES: Final = frozenset({"response.created", "response.in_progress", "response.queued"})
|
||||
|
||||
|
||||
class BaseResponsesAPIStreamingIterator:
|
||||
"""
|
||||
Base class for streaming iterators that process responses from the Responses API.
|
||||
|
|
@ -292,6 +295,7 @@ class BaseResponsesAPIStreamingIterator:
|
|||
self.start_time = getattr(logging_obj, "start_time", datetime.now())
|
||||
self._failure_handled = False # Track if failure handler has been called
|
||||
self._yielded_first_chunk = False
|
||||
self._output_started = False
|
||||
self._generated_content = ""
|
||||
self._generated_tool_arguments = ""
|
||||
self._completed_response_cached = False
|
||||
|
|
@ -879,6 +883,46 @@ class BaseResponsesAPIStreamingIterator:
|
|||
except Exception:
|
||||
pass
|
||||
|
||||
def _note_yielded_event(self, event: ResponsesAPIStreamingResponse) -> None:
|
||||
self._yielded_first_chunk = True
|
||||
if event.type not in _PRE_OUTPUT_LIFECYCLE_EVENT_TYPES:
|
||||
self._output_started = True
|
||||
|
||||
def _fallback_error(self, original: Exception) -> MidStreamFallbackError:
|
||||
return MidStreamFallbackError(
|
||||
message=str(original),
|
||||
model=self.model or "",
|
||||
llm_provider=self.custom_llm_provider or "",
|
||||
original_exception=original,
|
||||
generated_content="",
|
||||
is_pre_first_chunk=not self._yielded_first_chunk,
|
||||
)
|
||||
|
||||
def _stream_ended_early_error(self) -> litellm.APIConnectionError:
|
||||
return litellm.APIConnectionError(
|
||||
message=(
|
||||
f"{self.custom_llm_provider or 'provider'} closed the responses stream before any terminal event "
|
||||
"(response.completed, response.incomplete or response.failed)"
|
||||
),
|
||||
llm_provider=self.custom_llm_provider or "",
|
||||
model=self.model or "",
|
||||
)
|
||||
|
||||
def _raise_if_ended_without_terminal_event(self) -> None:
|
||||
if self.completed_response is not None:
|
||||
return
|
||||
error: Final = self._stream_ended_early_error()
|
||||
self._handle_failure(error)
|
||||
if self._output_started:
|
||||
raise error
|
||||
raise self._fallback_error(error) from error
|
||||
|
||||
def _raise_for_transport_error(self, error: httpx.ReadError | httpx.RemoteProtocolError) -> NoReturn:
|
||||
self._handle_failure(error)
|
||||
if self._output_started:
|
||||
raise error
|
||||
raise self._fallback_error(error) from error
|
||||
|
||||
|
||||
async def call_post_streaming_hooks_for_testing(
|
||||
iterator: object, chunk: ResponsesAPIStreamingResponse
|
||||
|
|
@ -934,12 +978,14 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
sse = await self.stream_iterator.__anext__()
|
||||
except StopAsyncIteration:
|
||||
self.finished = True
|
||||
self._raise_if_ended_without_terminal_event()
|
||||
raise StopAsyncIteration
|
||||
|
||||
self._check_max_streaming_duration()
|
||||
result = self._process_chunk(sse.data)
|
||||
|
||||
if self.finished:
|
||||
self._raise_if_ended_without_terminal_event()
|
||||
raise StopAsyncIteration
|
||||
elif result is not None:
|
||||
self._maybe_raise_for_error_event(result)
|
||||
|
|
@ -948,7 +994,7 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
result = await self._call_post_streaming_deployment_hook(
|
||||
chunk=result,
|
||||
)
|
||||
self._yielded_first_chunk = True
|
||||
self._note_yielded_event(result)
|
||||
return result
|
||||
# If result is None, continue the loop to get the next chunk
|
||||
|
||||
|
|
@ -957,10 +1003,9 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
raise
|
||||
except (httpx.ReadError, httpx.RemoteProtocolError) as e:
|
||||
self.finished = True
|
||||
if self.completed_response is None:
|
||||
self._handle_failure(e)
|
||||
raise
|
||||
raise StopAsyncIteration from e
|
||||
if self.completed_response is not None:
|
||||
raise StopAsyncIteration from e
|
||||
self._raise_for_transport_error(e)
|
||||
except httpx.HTTPError as e:
|
||||
# Handle HTTP errors
|
||||
self.finished = True
|
||||
|
|
@ -1016,12 +1061,14 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
sse = next(self.stream_iterator)
|
||||
except StopIteration:
|
||||
self.finished = True
|
||||
self._raise_if_ended_without_terminal_event()
|
||||
raise StopIteration
|
||||
|
||||
self._check_max_streaming_duration()
|
||||
result = self._process_chunk(sse.data)
|
||||
|
||||
if self.finished:
|
||||
self._raise_if_ended_without_terminal_event()
|
||||
raise StopIteration
|
||||
elif result is not None:
|
||||
self._maybe_raise_for_error_event(result)
|
||||
|
|
@ -1030,7 +1077,7 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
async_function=self._call_post_streaming_deployment_hook,
|
||||
chunk=result,
|
||||
)
|
||||
self._yielded_first_chunk = True
|
||||
self._note_yielded_event(result)
|
||||
return result
|
||||
# If result is None, continue the loop to get the next chunk
|
||||
|
||||
|
|
@ -1039,10 +1086,9 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
raise
|
||||
except (httpx.ReadError, httpx.RemoteProtocolError) as e:
|
||||
self.finished = True
|
||||
if self.completed_response is None:
|
||||
self._handle_failure(e)
|
||||
raise
|
||||
raise StopIteration from e
|
||||
if self.completed_response is not None:
|
||||
raise StopIteration from e
|
||||
self._raise_for_transport_error(e)
|
||||
except httpx.HTTPError as e:
|
||||
# Handle HTTP errors
|
||||
self.finished = True
|
||||
|
|
|
|||
|
|
@ -381,6 +381,36 @@ class TestBaseResponsesAPIStreamingIterator:
|
|||
)
|
||||
raise
|
||||
|
||||
@staticmethod
|
||||
def _config_completing_after_one_delta() -> Mock:
|
||||
mock_config = Mock(spec=BaseResponsesAPIConfig)
|
||||
completed_response = ResponsesAPIResponse(
|
||||
id="resp_123",
|
||||
created_at=0,
|
||||
status="completed",
|
||||
model="gpt-5.5",
|
||||
object="response",
|
||||
output=[],
|
||||
usage=ResponseAPIUsage(input_tokens=1, output_tokens=1, total_tokens=2),
|
||||
)
|
||||
|
||||
def _transform(model, parsed_chunk, logging_obj):
|
||||
if parsed_chunk.get("type") == "response.completed":
|
||||
return ResponseCompletedEvent(
|
||||
type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
|
||||
response=completed_response,
|
||||
)
|
||||
return OutputTextDeltaEvent(
|
||||
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
|
||||
item_id="msg_123",
|
||||
output_index=0,
|
||||
content_index=0,
|
||||
delta=parsed_chunk["delta"],
|
||||
)
|
||||
|
||||
mock_config.transform_streaming_response.side_effect = _transform
|
||||
return mock_config
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_async_iteration_not_logged_as_failure(self):
|
||||
"""
|
||||
|
|
@ -399,6 +429,7 @@ class TestBaseResponsesAPIStreamingIterator:
|
|||
|
||||
async def mock_aiter_bytes():
|
||||
yield b'data: {"type": "response.output_text.delta", "delta": "test"}\n\n'
|
||||
yield b'data: {"type": "response.completed", "response": {"id": "resp_123"}}\n\n'
|
||||
|
||||
mock_response.aiter_bytes = mock_aiter_bytes
|
||||
|
||||
|
|
@ -408,11 +439,7 @@ class TestBaseResponsesAPIStreamingIterator:
|
|||
mock_logging_obj.async_failure_handler = Mock()
|
||||
mock_logging_obj.failure_handler = Mock()
|
||||
|
||||
mock_config = Mock(spec=BaseResponsesAPIConfig)
|
||||
mock_delta_event = Mock()
|
||||
mock_delta_event.type = ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA
|
||||
mock_delta_event.delta = "test"
|
||||
mock_config.transform_streaming_response.return_value = mock_delta_event
|
||||
mock_config = self._config_completing_after_one_delta()
|
||||
|
||||
# Create the iterator instance
|
||||
iterator = ResponsesAPIStreamingIterator(
|
||||
|
|
@ -432,8 +459,9 @@ class TestBaseResponsesAPIStreamingIterator:
|
|||
except StopAsyncIteration:
|
||||
pass # This is expected
|
||||
|
||||
# Verify we got the chunk
|
||||
assert len(chunks_received) == 1
|
||||
# Verify we got the delta and the terminal event
|
||||
assert len(chunks_received) == 2
|
||||
assert iterator.completed_response is not None
|
||||
|
||||
# CRITICAL: Verify that failure handlers were NOT called
|
||||
# StopAsyncIteration is a normal end of stream, not a failure
|
||||
|
|
@ -460,6 +488,7 @@ class TestBaseResponsesAPIStreamingIterator:
|
|||
|
||||
def mock_iter_bytes():
|
||||
yield b'data: {"type": "response.output_text.delta", "delta": "test"}\n\n'
|
||||
yield b'data: {"type": "response.completed", "response": {"id": "resp_123"}}\n\n'
|
||||
|
||||
mock_response.iter_bytes = mock_iter_bytes
|
||||
|
||||
|
|
@ -469,11 +498,7 @@ class TestBaseResponsesAPIStreamingIterator:
|
|||
mock_logging_obj.async_failure_handler = Mock()
|
||||
mock_logging_obj.failure_handler = Mock()
|
||||
|
||||
mock_config = Mock(spec=BaseResponsesAPIConfig)
|
||||
mock_delta_event = Mock()
|
||||
mock_delta_event.type = ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA
|
||||
mock_delta_event.delta = "test"
|
||||
mock_config.transform_streaming_response.return_value = mock_delta_event
|
||||
mock_config = self._config_completing_after_one_delta()
|
||||
|
||||
# Create the iterator instance
|
||||
iterator = SyncResponsesAPIStreamingIterator(
|
||||
|
|
@ -493,8 +518,9 @@ class TestBaseResponsesAPIStreamingIterator:
|
|||
except StopIteration:
|
||||
pass # This is expected
|
||||
|
||||
# Verify we got the chunk
|
||||
assert len(chunks_received) == 1
|
||||
# Verify we got the delta and the terminal event
|
||||
assert len(chunks_received) == 2
|
||||
assert iterator.completed_response is not None
|
||||
|
||||
# CRITICAL: Verify that failure handlers were NOT called
|
||||
# StopIteration is a normal end of stream, not a failure
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ from starlette.datastructures import UploadFile as StarletteUploadFile
|
|||
|
||||
import litellm
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm.constants import DEFAULT_REQUEST_TIMEOUT_SECONDS
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.proxy._types import ProxyException, UserAPIKeyAuth
|
||||
|
|
@ -1165,6 +1166,29 @@ def test_resolve_llm_passthrough_timeout_precedence():
|
|||
assert resolve_llm_passthrough_timeout() == 6.0
|
||||
|
||||
|
||||
def test_resolve_llm_passthrough_timeout_honors_explicit_global_request_timeout(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr("litellm.request_timeout", 44.0, raising=False)
|
||||
monkeypatch.setattr("litellm.request_timeout_explicitly_set", True, raising=False)
|
||||
|
||||
with patch("litellm.proxy.proxy_server.general_settings", {"pass_through_request_timeout": 6}):
|
||||
assert resolve_llm_passthrough_timeout() == 44.0
|
||||
assert resolve_llm_passthrough_timeout(kwargs={"stream": True}) == 44.0
|
||||
assert resolve_llm_passthrough_timeout(router_timeout=120) == 120.0
|
||||
assert resolve_llm_passthrough_timeout(kwargs={"stream": True}, router_stream_timeout=900) == 900.0
|
||||
assert resolve_llm_passthrough_timeout(litellm_params={"timeout": 90}) == 90.0
|
||||
assert resolve_llm_passthrough_timeout(kwargs={"timeout": 45}) == 45.0
|
||||
|
||||
|
||||
def test_resolve_llm_passthrough_timeout_skips_unset_global_request_timeout(monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setattr("litellm.request_timeout", float(DEFAULT_REQUEST_TIMEOUT_SECONDS), raising=False)
|
||||
monkeypatch.setattr("litellm.request_timeout_explicitly_set", False, raising=False)
|
||||
|
||||
with patch("litellm.proxy.proxy_server.general_settings", {"pass_through_request_timeout": 6}):
|
||||
assert resolve_llm_passthrough_timeout() == 6.0
|
||||
with patch("litellm.proxy.proxy_server.general_settings", {}):
|
||||
assert resolve_llm_passthrough_timeout() == DEFAULT_PASS_THROUGH_REQUEST_TIMEOUT_SECONDS
|
||||
|
||||
|
||||
def test_resolve_llm_passthrough_timeout_stream_timeout_precedence():
|
||||
assert (
|
||||
resolve_llm_passthrough_timeout(
|
||||
|
|
|
|||
|
|
@ -6,13 +6,14 @@ completion_start_time = end_time."""
|
|||
import json
|
||||
from datetime import datetime
|
||||
from typing import Final, Optional
|
||||
from unittest.mock import Mock, patch
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from pydantic_core import PydanticSerializationError
|
||||
|
||||
import litellm
|
||||
from litellm.exceptions import MidStreamFallbackError
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
|
||||
from litellm.responses.streaming_iterator import (
|
||||
|
|
@ -251,6 +252,171 @@ def test_sync_transport_error_before_completed_event_raises():
|
|||
pass
|
||||
|
||||
|
||||
_DONE_MARKER: Final = b"data: [DONE]\n\n"
|
||||
_CREATED_EVENT: Final = _sse_event({"type": "response.created"})
|
||||
_IN_PROGRESS_EVENT: Final = _sse_event({"type": "response.in_progress"})
|
||||
_PARTIAL_OUTPUT_EVENTS: Final = _COMPLETE_STREAM_EVENTS[:-1]
|
||||
_PRE_OUTPUT_PREFIXES: Final = [
|
||||
pytest.param([], True, id="nothing-yielded"),
|
||||
pytest.param([_CREATED_EVENT], False, id="created"),
|
||||
pytest.param([_CREATED_EVENT, _IN_PROGRESS_EVENT], False, id="created-and-in-progress"),
|
||||
]
|
||||
|
||||
|
||||
def _failure_tracking_logging_obj() -> Mock:
|
||||
logging_obj: Final = _logging_obj_stub()
|
||||
logging_obj.async_failure_handler = AsyncMock()
|
||||
return logging_obj
|
||||
|
||||
|
||||
def _assert_failure_logged_once(logging_obj: Mock, exception: Exception) -> None:
|
||||
assert logging_obj.async_failure_handler.await_count == 1
|
||||
assert logging_obj.async_failure_handler.await_args.kwargs["exception"] is exception
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("prefix, pre_first_chunk", _PRE_OUTPUT_PREFIXES)
|
||||
@pytest.mark.parametrize("trailing_error", _TRAILING_ERRORS, ids=type)
|
||||
async def test_transport_error_before_any_output_raises_fallback_error(prefix, pre_first_chunk, trailing_error):
|
||||
"""A connection lost while only lifecycle events (response.created / response.in_progress)
|
||||
have streamed is fallback-eligible, so it must surface as the MidStreamFallbackError the
|
||||
router re-routes, carrying the raw transport error and no generated content."""
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
iterator: Final = _make_iterator(sse_events=prefix, logging_obj=logging_obj, trailing_error=trailing_error)
|
||||
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
async for _ in iterator:
|
||||
pass
|
||||
|
||||
assert exc_info.value.original_exception is trailing_error
|
||||
assert exc_info.value.is_pre_first_chunk is pre_first_chunk
|
||||
assert exc_info.value.generated_content == ""
|
||||
_assert_failure_logged_once(logging_obj, trailing_error)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transport_error_after_output_started_is_not_fallback_eligible():
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
trailing_error: Final = httpx.ReadError("Response payload is not completed")
|
||||
iterator: Final = _make_iterator(
|
||||
sse_events=_PARTIAL_OUTPUT_EVENTS, logging_obj=logging_obj, trailing_error=trailing_error
|
||||
)
|
||||
|
||||
with pytest.raises(httpx.ReadError) as exc_info:
|
||||
async for _ in iterator:
|
||||
pass
|
||||
|
||||
assert exc_info.value is trailing_error
|
||||
_assert_failure_logged_once(logging_obj, trailing_error)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("trailer", [[], [_DONE_MARKER]], ids=["eof", "done-marker"])
|
||||
async def test_stream_ending_after_partial_output_without_terminal_event_raises(trailer):
|
||||
"""A clean EOF or `[DONE]` after output text but with no response.completed /
|
||||
response.incomplete / response.failed is a truncated answer: the partial events still
|
||||
reach the caller, then an explicit error follows instead of a normal end of stream."""
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
iterator: Final = _make_iterator(sse_events=[*_PARTIAL_OUTPUT_EVENTS, *trailer], logging_obj=logging_obj)
|
||||
|
||||
created: Final = await iterator.__anext__()
|
||||
delta: Final = await iterator.__anext__()
|
||||
with pytest.raises(litellm.APIConnectionError) as exc_info:
|
||||
await iterator.__anext__()
|
||||
|
||||
assert (created.type, delta.type) == ("response.created", "response.output_text.delta")
|
||||
assert not isinstance(exc_info.value, MidStreamFallbackError)
|
||||
assert exc_info.value.llm_provider == "openai"
|
||||
_assert_failure_logged_once(logging_obj, exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("prefix, pre_first_chunk", _PRE_OUTPUT_PREFIXES)
|
||||
@pytest.mark.parametrize("trailer", [[], [_DONE_MARKER]], ids=["eof", "done-marker"])
|
||||
async def test_stream_ending_before_any_output_raises_fallback_error(prefix, pre_first_chunk, trailer):
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
iterator: Final = _make_iterator(sse_events=[*prefix, *trailer], logging_obj=logging_obj)
|
||||
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
async for _ in iterator:
|
||||
pass
|
||||
|
||||
assert isinstance(exc_info.value.original_exception, litellm.APIConnectionError)
|
||||
assert exc_info.value.is_pre_first_chunk is pre_first_chunk
|
||||
assert exc_info.value.generated_content == ""
|
||||
_assert_failure_logged_once(logging_obj, exc_info.value.original_exception)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("trailer", [[], [_DONE_MARKER]], ids=["eof", "done-marker"])
|
||||
async def test_complete_stream_still_ends_normally(trailer):
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
iterator: Final = _make_iterator(sse_events=[*_COMPLETE_STREAM_EVENTS, *trailer], logging_obj=logging_obj)
|
||||
|
||||
seen: Final = [event.type async for event in iterator]
|
||||
|
||||
assert seen[-1] == ResponsesAPIStreamEvents.RESPONSE_COMPLETED
|
||||
assert logging_obj.async_failure_handler.await_count == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("trailing_error", _TRAILING_ERRORS, ids=type)
|
||||
def test_sync_transport_error_before_any_output_raises_fallback_error(trailing_error):
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
iterator: Final = _make_sync_iterator(
|
||||
sse_events=[_CREATED_EVENT, _IN_PROGRESS_EVENT],
|
||||
logging_obj=logging_obj,
|
||||
trailing_error=trailing_error,
|
||||
)
|
||||
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
for _ in iterator:
|
||||
pass
|
||||
|
||||
assert exc_info.value.original_exception is trailing_error
|
||||
assert exc_info.value.is_pre_first_chunk is False
|
||||
assert exc_info.value.generated_content == ""
|
||||
_assert_failure_logged_once(logging_obj, trailing_error)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("trailer", [[], [_DONE_MARKER]], ids=["eof", "done-marker"])
|
||||
def test_sync_stream_ending_after_partial_output_without_terminal_event_raises(trailer):
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
iterator: Final = _make_sync_iterator(sse_events=[*_PARTIAL_OUTPUT_EVENTS, *trailer], logging_obj=logging_obj)
|
||||
|
||||
created: Final = next(iterator)
|
||||
delta: Final = next(iterator)
|
||||
with pytest.raises(litellm.APIConnectionError) as exc_info:
|
||||
next(iterator)
|
||||
|
||||
assert (created.type, delta.type) == ("response.created", "response.output_text.delta")
|
||||
assert not isinstance(exc_info.value, MidStreamFallbackError)
|
||||
_assert_failure_logged_once(logging_obj, exc_info.value)
|
||||
|
||||
|
||||
def test_sync_stream_ending_before_any_output_raises_fallback_error():
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
iterator: Final = _make_sync_iterator(sse_events=[_CREATED_EVENT], logging_obj=logging_obj)
|
||||
|
||||
with pytest.raises(MidStreamFallbackError) as exc_info:
|
||||
for _ in iterator:
|
||||
pass
|
||||
|
||||
assert isinstance(exc_info.value.original_exception, litellm.APIConnectionError)
|
||||
assert exc_info.value.is_pre_first_chunk is False
|
||||
_assert_failure_logged_once(logging_obj, exc_info.value.original_exception)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("trailer", [[], [_DONE_MARKER]], ids=["eof", "done-marker"])
|
||||
def test_sync_complete_stream_still_ends_normally(trailer):
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
iterator: Final = _make_sync_iterator(sse_events=[*_COMPLETE_STREAM_EVENTS, *trailer], logging_obj=logging_obj)
|
||||
|
||||
seen: Final = [event.type for event in iterator]
|
||||
|
||||
assert seen[-1] == ResponsesAPIStreamEvents.RESPONSE_COMPLETED
|
||||
assert logging_obj.async_failure_handler.await_count == 0
|
||||
|
||||
|
||||
def test_stream_cache_write_completes_when_asyncio_run_closes_the_loop(monkeypatch):
|
||||
"""
|
||||
Regression test for LIT-6184 on the /v1/responses streaming surface: the
|
||||
|
|
|
|||
|
|
@ -4486,6 +4486,107 @@ async def test_aresponses_streaming_iterator_pre_first_chunk_skips_continuation(
|
|||
assert fbk["input"] == "Hello" # original input, no continuation messages
|
||||
|
||||
|
||||
def _make_native_responses_iterator(*, sse_payloads: tuple[dict[str, str], ...], trailing_error: Exception | None):
|
||||
"""A real ResponsesAPIStreamingIterator over canned SSE bytes, so the router test covers the
|
||||
iterator's own transport-error classification instead of a hand-built MidStreamFallbackError."""
|
||||
from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig
|
||||
from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator
|
||||
|
||||
async def aiter_bytes():
|
||||
for payload in sse_payloads:
|
||||
yield f"data: {json.dumps(payload)}\n\n".encode()
|
||||
if trailing_error is not None:
|
||||
raise trailing_error
|
||||
|
||||
def transform(model, parsed_chunk, logging_obj):
|
||||
return MagicMock(type=parsed_chunk["type"])
|
||||
|
||||
response: Final = MagicMock()
|
||||
response.headers = {}
|
||||
response.aiter_bytes = aiter_bytes
|
||||
config: Final = MagicMock(spec=BaseResponsesAPIConfig)
|
||||
config.transform_streaming_response.side_effect = transform
|
||||
logging_obj: Final = MagicMock(spec=LiteLLMLogging)
|
||||
logging_obj.completion_start_time = None
|
||||
logging_obj.model_call_details = {"litellm_params": {}}
|
||||
return ResponsesAPIStreamingIterator(
|
||||
response=response,
|
||||
model="gpt-4",
|
||||
responses_api_provider_config=config,
|
||||
logging_obj=logging_obj,
|
||||
litellm_metadata={},
|
||||
custom_llm_provider="openai",
|
||||
)
|
||||
|
||||
|
||||
_RESPONSES_LIFECYCLE_PAYLOADS: Final = ({"type": "response.created"}, {"type": "response.in_progress"})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aresponses_streaming_iterator_falls_back_on_transport_drop_before_output():
|
||||
"""A connection lost after response.created but before any output item is re-routed to the
|
||||
fallback with the original input, the same as a provider error event would be."""
|
||||
router: Final = _make_router_with_fallback()
|
||||
src: Final = _make_native_responses_iterator(
|
||||
sse_payloads=_RESPONSES_LIFECYCLE_PAYLOADS,
|
||||
trailing_error=httpx.ReadError("Response payload is not completed"),
|
||||
)
|
||||
|
||||
with patch.object(
|
||||
router,
|
||||
"async_function_with_fallbacks_common_utils",
|
||||
return_value=_AsyncList([MagicMock(type="response.completed")]),
|
||||
) as mock_fallback_utils:
|
||||
wrapped: Final = await router._aresponses_streaming_iterator(
|
||||
response=src,
|
||||
initial_kwargs={
|
||||
"model": "gpt-4",
|
||||
"stream": True,
|
||||
"input": "Hello",
|
||||
"original_generic_function": litellm.aresponses,
|
||||
},
|
||||
)
|
||||
seen: Final = [chunk.type async for chunk in wrapped]
|
||||
|
||||
assert seen == ["response.created", "response.in_progress", "response.completed"]
|
||||
assert isinstance(mock_fallback_utils.call_args.kwargs["e"], MidStreamFallbackError)
|
||||
assert mock_fallback_utils.call_args.kwargs["kwargs"]["input"] == "Hello"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aresponses_streaming_iterator_surfaces_transport_drop_when_no_fallback_lands():
|
||||
transport_error: Final = httpx.ReadError("Response payload is not completed")
|
||||
router: Final = _make_router_with_fallback()
|
||||
src: Final = _make_native_responses_iterator(
|
||||
sse_payloads=_RESPONSES_LIFECYCLE_PAYLOADS, trailing_error=transport_error
|
||||
)
|
||||
|
||||
async def reraise_trigger(**kwargs):
|
||||
raise kwargs["e"]
|
||||
|
||||
with patch.object(
|
||||
router, "async_function_with_fallbacks_common_utils", new=AsyncMock(side_effect=reraise_trigger)
|
||||
) as mock_fallback_utils:
|
||||
wrapped: Final = await router._aresponses_streaming_iterator(
|
||||
response=src,
|
||||
initial_kwargs={
|
||||
"model": "gpt-4",
|
||||
"stream": True,
|
||||
"input": "Hello",
|
||||
"original_generic_function": litellm.aresponses,
|
||||
},
|
||||
)
|
||||
with pytest.raises(httpx.ReadError) as exc_info:
|
||||
async for _ in wrapped:
|
||||
pass
|
||||
|
||||
assert exc_info.value is transport_error
|
||||
assert mock_fallback_utils.await_count == 1
|
||||
trigger: Final = mock_fallback_utils.await_args.kwargs["e"]
|
||||
assert isinstance(trigger, MidStreamFallbackError)
|
||||
assert trigger.original_exception is transport_error
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aresponses_streaming_iterator_partial_content_injects_continuation():
|
||||
"""Mid-stream error: input is rewritten to include user prompt +
|
||||
|
|
@ -6090,6 +6191,32 @@ def test_update_kwargs_with_deployment_passthrough_router_stream_timeout_sources
|
|||
assert _passthrough_timeout(default_router, default_router.model_list[0], stream=False) == 120.0
|
||||
|
||||
|
||||
def test_update_kwargs_with_deployment_passthrough_honors_global_request_timeout(monkeypatch: pytest.MonkeyPatch):
|
||||
"""litellm_settings.request_timeout must bound the native responses route when neither the
|
||||
deployment nor the router carries a timeout, while a deployment timeout keeps winning."""
|
||||
monkeypatch.setattr("litellm.request_timeout", 44.0, raising=False)
|
||||
monkeypatch.setattr("litellm.request_timeout_explicitly_set", True, raising=False)
|
||||
router: Final = litellm.Router(
|
||||
model_list=[
|
||||
{
|
||||
"model_name": "responses-global-timeout",
|
||||
"litellm_params": {"model": "openai/gpt-5.4-mini", "api_key": "fake-key"},
|
||||
},
|
||||
{
|
||||
"model_name": "responses-deployment-timeout",
|
||||
"litellm_params": {"model": "openai/gpt-5.4-mini", "api_key": "fake-key", "timeout": 3},
|
||||
},
|
||||
],
|
||||
)
|
||||
global_only, per_deployment = router.model_list
|
||||
|
||||
with patch("litellm.proxy.proxy_server.general_settings", {"pass_through_request_timeout": 6}):
|
||||
assert _passthrough_timeout(router, global_only, stream=True) == 44.0
|
||||
assert _passthrough_timeout(router, global_only, stream=False) == 44.0
|
||||
assert _passthrough_timeout(router, per_deployment, stream=True) == 3.0
|
||||
assert _passthrough_timeout(router, per_deployment, stream=False) == 3.0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_router_acompletion_with_unknown_model_and_default_fallback():
|
||||
"""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue