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

Merge pull request #40988 from BerriAI/litellm_responses_stream_error_exception_mapping

Browse source 
fix(responses): route mid-stream error events through exception_type so content_policy_fallbacks fire
This commit is contained in:
Mateo Wang 2026-09-14 18:36:20 -07:00 • committed by GitHub
commit d2859e18d7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 298 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
litellm/responses/streaming_iterator.py
View file

@ -221,6 +221,13 @@ def _status_code_for_error_fields(error_type: str | None, error_code: str | None
)
def _mid_stream_fallback_eligible(mapped_exception: Exception) -> bool:
if isinstance(mapped_exception, litellm.ContentPolicyViolationError):
return True
status_code: Final = getattr(mapped_exception, "status_code", None)
return not isinstance(status_code, int) or status_code >= 500 or status_code == 429
class BaseResponsesAPIStreamingIterator:
"""
Base class for streaming iterators that process responses from the Responses API.
@ -521,15 +528,8 @@ class BaseResponsesAPIStreamingIterator:
getattr(self.completed_response, "response", None) if self.completed_response else None
)
error_info: Final = getattr(response_obj, "error", None) if response_obj else None
error_message, error_type, error_code = _error_event_fields(error_info)
self._record_failed_response_usage(response_obj)
exception: Final = litellm.APIError(
status_code=_status_code_for_error_fields(error_type, error_code),
message=error_message,
llm_provider=self.custom_llm_provider or "",
model=self.model or "",
)
self._handle_failure(exception)
self._handle_failure(self._map_error_event_exception(error_info))
def _record_failed_response_usage(self, response_obj: ResponsesAPIResponse | None) -> None:
if response_obj is None or self.logging_obj is None:
@ -551,6 +551,28 @@ class BaseResponsesAPIStreamingIterator:
self.logging_obj._response_cost_calculator(result=response_obj) or 0.0
)
def _map_error_event_exception(self, error_obj: object) -> Exception:
from litellm.llms.base_llm.chat.transformation import BaseLLMException
error_message, error_type, error_code = _error_event_fields(error_obj)
status_code: Final = _status_code_for_error_fields(error_type, error_code)
error_body: Final = {"message": error_message, "type": error_type, "code": error_code}
provider_exception: Final = BaseLLMException(
status_code=status_code,
message=f"Error code: {status_code} - {{'error': {error_body}}}",
body=error_body,
)
try:
return litellm.exception_type(
model=self.model or "",
custom_llm_provider=self.custom_llm_provider or "",
original_exception=provider_exception,
completion_kwargs={},
extra_kwargs={},
)
except Exception as mapped_exception:
return mapped_exception
def _maybe_raise_for_error_event(self, result: object) -> None:
chunk_type: Final = getattr(result, "type", None)
if chunk_type not in ("error", "response.failed"):
@ -562,15 +584,8 @@ class BaseResponsesAPIStreamingIterator:
else getattr(result, "error", None)
)
error_message, error_type, error_code = _error_event_fields(error_obj)
status_code: Final = _status_code_for_error_fields(error_type, error_code)
mapped_exception: Final = litellm.APIError(
status_code=status_code,
message=error_message,
llm_provider=self.custom_llm_provider or "",
model=self.model or "",
)
if 400 <= status_code < 500 and status_code != 429:
mapped_exception: Final = self._map_error_event_exception(error_obj)
if not _mid_stream_fallback_eligible(mapped_exception):
raise mapped_exception
raise MidStreamFallbackError(
message=str(mapped_exception),

9
litellm/router.py
View file

@ -3268,8 +3268,15 @@ class Router:
kwargs=initial_kwargs,
metadata_variable_name="litellm_metadata",
)
# The content-policy dispatch branch matches on the trigger's own type, so a refusal's
# MidStreamFallbackError envelope is unwrapped here or the wrong fallback list is consulted.
fallback_trigger: Final[Exception] = (
e.original_exception
if isinstance(e.original_exception, litellm.ContentPolicyViolationError)
else e
)
fallback_response = await self.async_function_with_fallbacks_common_utils(
e=e,
e=fallback_trigger,
disable_fallbacks=False,
fallbacks=fallbacks,
context_window_fallbacks=context_window_fallbacks,

9
tests/llm_responses_api_testing/test_openai_responses_api.py
View file

@ -1627,9 +1627,10 @@ async def test_openai_responses_api_token_limit_error():
Parsing the in-stream ErrorEvent must not raise
"pydantic_core._pydantic_core.ValidationError: 3 validation errors for ErrorEvent".
The iterator now surfaces the event as litellm.APIError with status 400
(invalid_request_error is a non-retriable client error, so no
MidStreamFallbackError wrapping) carrying the provider's message.
The iterator routes the event through litellm.exception_type, so it surfaces as
the typed 400 client error the non-streaming path raises (litellm.BadRequestError)
carrying the provider's message. invalid_request_error is a non-retriable client
error, so there is no MidStreamFallbackError wrapping.
"""
litellm._turn_on_debug()
@ -1644,7 +1645,7 @@ async def test_openai_responses_api_token_limit_error():
async for event in response:
print(event)
with pytest.raises(litellm.APIError) as exc_info:
with pytest.raises(litellm.BadRequestError) as exc_info:
await _drain()
assert exc_info.value.status_code == 400

2
tests/router_unit_tests/test_router_aresponses_streaming_fallback.py
View file

@ -372,7 +372,7 @@ async def test_aresponses_fallback_on_in_stream_error_event():
raised = mock_fallback.await_args.kwargs["e"]
assert isinstance(raised, MidStreamFallbackError)
assert raised.status_code == 429
assert isinstance(raised.original_exception, litellm.APIError)
assert isinstance(raised.original_exception, litellm.RateLimitError)
assert raised.original_exception.status_code == 429
assert mock_fallback.await_args.kwargs["kwargs"]["input"] == "original question"

167
tests/test_litellm/responses/test_streaming_iterator_error_events.py
View file

@ -1,9 +1,14 @@
"""
Regression: in-stream error events (type="error", type="response.failed") must
raise instead of being returned as benign chunks, mirroring chat streaming
semantics (_handle_stream_fallback_error): non-retriable 4xx (except 429)
raise litellm.APIError directly; 429 and 5xx are wrapped in
MidStreamFallbackError so the Router's mid-stream fallback machinery fires.
semantics (_handle_stream_fallback_error). The event's code, type and status go
through litellm.exception_type, so each event raises the same typed exception
the non-streaming path raises for that provider error: non-retriable 4xx
(except 429) raise that typed exception directly, so a context-length event
surfaces as ContextWindowExceededError(400) with no MidStreamFallbackError
wrapping, while 429, 5xx and ContentPolicyViolationError are wrapped in
MidStreamFallbackError so the Router's mid-stream fallback machinery fires and
its content_policy_fallbacks dispatch sees the trigger it matches on.
Status mapping must consider both the OpenAI error `type` (e.g.
"invalid_request_error") and `code` (e.g. "invalid_prompt",
@ -66,12 +71,12 @@ def test_maybe_raise_for_error_event_wraps_unknown_error_in_mid_stream_fallback(
with pytest.raises(MidStreamFallbackError) as exc_info:
iterator._maybe_raise_for_error_event(chunk)
assert exc_info.value.status_code == 500
assert isinstance(exc_info.value.original_exception, litellm.APIError)
assert isinstance(exc_info.value.original_exception, litellm.InternalServerError)
assert exc_info.value.original_exception.status_code == 500
def test_maybe_raise_for_error_event_maps_rate_limit_code_to_429_mid_stream_fallback():
"""429 is retriable: it must be wrapped so the Router can fall back, carrying the mapped APIError."""
"""429 is retriable: it must be wrapped so the Router can fall back, carrying the mapped RateLimitError."""
iterator = _make_iterator()
chunk = _make_error_chunk("tokens", "rate_limit_exceeded", "Too many requests")
with pytest.raises(MidStreamFallbackError) as exc_info:
@ -79,15 +84,15 @@ def test_maybe_raise_for_error_event_maps_rate_limit_code_to_429_mid_stream_fall
assert exc_info.value.status_code == 429
assert exc_info.value.generated_content == ""
assert exc_info.value.is_pre_first_chunk is True
assert isinstance(exc_info.value.original_exception, litellm.APIError)
assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
assert exc_info.value.original_exception.status_code == 429
def test_maybe_raise_for_error_event_maps_invalid_request_type_to_400():
"""Client errors classified via the `type` field must raise APIError directly (no fallback)."""
"""Client errors classified via the `type` field must raise BadRequestError directly (no fallback)."""
iterator = _make_iterator()
chunk = _make_error_chunk("invalid_request_error", "invalid_prompt", "bad request")
with pytest.raises(litellm.APIError) as exc_info:
with pytest.raises(litellm.BadRequestError) as exc_info:
iterator._maybe_raise_for_error_event(chunk)
assert exc_info.value.status_code == 400
assert not isinstance(exc_info.value, MidStreamFallbackError)
@ -99,12 +104,86 @@ def test_maybe_raise_for_error_event_maps_context_length_code_to_400():
chunk = Mock()
chunk.type = "error"
chunk.error = {"code": "context_length_exceeded", "message": "too long"}
with pytest.raises(litellm.APIError) as exc_info:
with pytest.raises(litellm.BadRequestError) as exc_info:
iterator._maybe_raise_for_error_event(chunk)
assert exc_info.value.status_code == 400
assert not isinstance(exc_info.value, MidStreamFallbackError)
def test_maybe_raise_for_error_event_raises_context_window_exceeded_directly():
"""A context-length error event maps to ContextWindowExceededError exactly like the non-streaming
path and, being a non-retriable client error, is raised directly rather than wrapped for mid-stream
fallback, preserving the direct-SDK 400 contract from issue #15785."""
iterator = _make_iterator()
chunk = _make_error_chunk(
"invalid_request_error",
"context_length_exceeded",
"This model's maximum context length is 128000 tokens. However, your messages resulted in 130000 tokens.",
)
with pytest.raises(litellm.ContextWindowExceededError) as exc_info:
iterator._maybe_raise_for_error_event(chunk)
assert exc_info.value.status_code == 400
assert not isinstance(exc_info.value, MidStreamFallbackError)
assert "maximum context length" in str(exc_info.value)
CONTENT_POLICY_MESSAGE = "This content was flagged for possible cybersecurity risk. The response was halted mid-stream."
@pytest.mark.parametrize("custom_llm_provider", ["openai", "azure"])
def test_maybe_raise_for_error_event_wraps_content_policy_violation_for_content_policy_fallbacks(
custom_llm_provider: str,
):
"""Regression: a content_policy_violation error event used to raise a bare APIError, so the Router's
content_policy_fallbacks never fired. It must map to ContentPolicyViolationError (the same exception the
non-streaming path raises) and be wrapped so the Router's mid-stream fallback catches it."""
iterator = _make_iterator()
iterator.custom_llm_provider = custom_llm_provider
chunk = _make_error_chunk("invalid_request_error", "content_policy_violation", CONTENT_POLICY_MESSAGE)
with pytest.raises(MidStreamFallbackError) as exc_info:
iterator._maybe_raise_for_error_event(chunk)
assert isinstance(exc_info.value.original_exception, litellm.ContentPolicyViolationError)
assert exc_info.value.original_exception.status_code == 400
assert exc_info.value.status_code == 400
assert exc_info.value.is_pre_first_chunk is True
assert CONTENT_POLICY_MESSAGE in str(exc_info.value.original_exception)
def test_maybe_raise_for_response_failed_event_wraps_content_policy_violation():
iterator = _make_iterator()
chunk = _make_failed_chunk(
{"type": "invalid_request_error", "code": "content_policy_violation", "message": CONTENT_POLICY_MESSAGE}
)
with pytest.raises(MidStreamFallbackError) as exc_info:
iterator._maybe_raise_for_error_event(chunk)
assert isinstance(exc_info.value.original_exception, litellm.ContentPolicyViolationError)
@pytest.mark.parametrize(
"error_type,error_code,expected_exception",
[
("invalid_request_error", "content_policy_violation", litellm.ContentPolicyViolationError),
("tokens", "rate_limit_exceeded", litellm.RateLimitError),
("invalid_request_error", "insufficient_quota", litellm.RateLimitError),
("server_error", "internal_error", litellm.InternalServerError),
("invalid_request_error", "invalid_prompt", litellm.BadRequestError),
("invalid_request_error", "model_not_found", litellm.NotFoundError),
("server_error", "vector_store_timeout", litellm.Timeout),
],
)
def test_error_event_raises_the_same_typed_exception_as_the_non_streaming_path(
error_type: str, error_code: str, expected_exception: type[Exception]
):
iterator = _make_iterator()
chunk = _make_error_chunk(error_type, error_code, "provider message")
with pytest.raises((MidStreamFallbackError, expected_exception)) as exc_info:
iterator._maybe_raise_for_error_event(chunk)
raised = exc_info.value
typed_exception = raised.original_exception if isinstance(raised, MidStreamFallbackError) else raised
assert type(typed_exception) is expected_exception
assert "provider message" in str(typed_exception)
def test_maybe_raise_for_error_event_maps_insufficient_quota_to_429():
"""OpenAI returns HTTP 429 for insufficient_quota; it must not map to 400 even though its type
is invalid_request_error-adjacent, and it must be wrapped for fallback."""
@ -113,6 +192,7 @@ def test_maybe_raise_for_error_event_maps_insufficient_quota_to_429():
with pytest.raises(MidStreamFallbackError) as exc_info:
iterator._maybe_raise_for_error_event(chunk)
assert exc_info.value.status_code == 429
assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
def test_maybe_raise_for_error_event_passes_through_normal_chunk():
@ -186,10 +266,40 @@ async def test_async_iterator_raises_mid_stream_fallback_on_rate_limit_error_eve
assert exc_info.value.status_code == 429
assert exc_info.value.is_pre_first_chunk is True
assert exc_info.value.generated_content == ""
assert isinstance(exc_info.value.original_exception, litellm.APIError)
assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
assert exc_info.value.original_exception.status_code == 429
@pytest.mark.asyncio
async def test_async_iterator_content_policy_violation_after_first_chunk_carries_generated_content():
"""The customer's case: text streams, then the provider halts the stream with a
content_policy_violation error event. The iterator must surface ContentPolicyViolationError
inside MidStreamFallbackError, together with the text already streamed."""
iterator = _make_async_iterator_with_events(
[
{"type": "response.output_text.delta", "delta": "partial "},
{
"type": "error",
"error": {
"type": "invalid_request_error",
"code": "content_policy_violation",
"message": CONTENT_POLICY_MESSAGE,
},
},
]
)
stream = aiter(iterator)
first_chunk = await anext(stream)
assert first_chunk is not None
with pytest.raises(MidStreamFallbackError) as exc_info:
await anext(stream)
assert isinstance(exc_info.value.original_exception, litellm.ContentPolicyViolationError)
assert exc_info.value.is_pre_first_chunk is False
assert exc_info.value.generated_content == "partial "
@pytest.mark.asyncio
async def test_async_iterator_error_after_first_chunk_carries_generated_content():
"""An error after streamed output must expose the accumulated text so the router's
@ -205,14 +315,13 @@ async def test_async_iterator_error_after_first_chunk_carries_generated_content(
]
)
chunks = []
async def _drain():
async for chunk in iterator:
chunks.append(chunk)
stream = aiter(iterator)
first_chunk = await anext(stream)
second_chunk = await anext(stream)
assert first_chunk is not None and second_chunk is not None
with pytest.raises(MidStreamFallbackError) as exc_info:
await _drain()
assert len(chunks) == 2
await anext(stream)
assert exc_info.value.status_code == 500
assert exc_info.value.is_pre_first_chunk is False
assert exc_info.value.generated_content == "hello world"
@ -265,7 +374,7 @@ def test_handle_logging_failed_response_maps_rate_limit_to_429():
):
iterator._handle_logging_failed_response()
logged_exception = mock_run_async.call_args.kwargs["exception"]
assert isinstance(logged_exception, litellm.APIError)
assert isinstance(logged_exception, litellm.RateLimitError)
assert logged_exception.status_code == 429
assert "throttled" in str(logged_exception)
@ -282,10 +391,28 @@ def test_handle_logging_failed_response_maps_type_field_to_400():
):
iterator._handle_logging_failed_response()
logged_exception = mock_run_async.call_args.kwargs["exception"]
assert isinstance(logged_exception, litellm.APIError)
assert isinstance(logged_exception, litellm.BadRequestError)
assert logged_exception.status_code == 400
def test_handle_logging_failed_response_logs_content_policy_violation():
"""Failure logging must record the same typed exception the stream raises, so logging
integrations see a content policy violation instead of a generic APIError."""
iterator = _make_iterator()
iterator.completed_response = _make_failed_chunk(
{"type": "invalid_request_error", "code": "content_policy_violation", "message": CONTENT_POLICY_MESSAGE}
)
with (
patch.object(import_module("litellm.responses.streaming_iterator"), "run_async_function") as mock_run_async,
patch.object(import_module("litellm.responses.streaming_iterator"), "executor"),
):
iterator._handle_logging_failed_response()
logged_exception = mock_run_async.call_args.kwargs["exception"]
assert isinstance(logged_exception, litellm.ContentPolicyViolationError)
assert logged_exception.status_code == 400
assert CONTENT_POLICY_MESSAGE in str(logged_exception)
def test_handle_logging_failed_response_records_usage_and_cost():
"""Usage on a response.failed event must reach failure spend accounting via combined_usage_object."""
iterator = _make_iterator()
@ -357,7 +484,7 @@ def test_sync_iterator_raises_mid_stream_fallback_on_rate_limit_error_event():
for _ in iterator:
pass
assert exc_info.value.status_code == 429
assert isinstance(exc_info.value.original_exception, litellm.APIError)
assert isinstance(exc_info.value.original_exception, litellm.RateLimitError)
def test_every_openai_sdk_response_error_code_has_explicit_status_mapping():
@ -413,7 +540,7 @@ def test_maybe_raise_for_response_failed_event_maps_image_code_to_400():
chunk = Mock()
chunk.type = "response.failed"
chunk.response = mock_response_obj
with pytest.raises(litellm.APIError) as exc_info:
with pytest.raises(litellm.BadRequestError) as exc_info:
iterator._maybe_raise_for_error_event(chunk)
assert exc_info.value.status_code == 400
assert not isinstance(exc_info.value, MidStreamFallbackError)

105
tests/test_litellm/test_router.py
View file

@ -3694,6 +3694,111 @@ async def test_aresponses_streaming_iterator_fallback():
assert call_kwargs["disable_fallbacks"] is False
@pytest.mark.asyncio
async def test_aresponses_streaming_content_policy_error_event_routes_to_content_policy_fallback():
"""Regression: a mid-stream content_policy_violation error event never reached
content_policy_fallbacks. The iterator raised a bare APIError the wrapper does not
catch, and even once wrapped, the MidStreamFallbackError envelope was handed to the
fallback dispatch, whose isinstance branch on ContentPolicyViolationError never matched.
The stream below is the customer's shape: a raw OpenAI error event with code
content_policy_violation, transformed by the real OpenAI config, and the router must
call the content_policy_fallbacks target, not the general fallbacks one."""
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator
router = litellm.Router(
model_list=[
{"model_name": "primary", "litellm_params": {"model": "openai/gpt-5.4", "api_key": "k1"}},
{
"model_name": "content-fallback",
"litellm_params": {"model": "gemini/gemini-2.5-flash", "api_key": "k2"},
},
{"model_name": "general-fallback", "litellm_params": {"model": "openai/gpt-5-mini", "api_key": "k3"}},
],
fallbacks=[{"primary": ["general-fallback"]}],
content_policy_fallbacks=[{"primary": ["content-fallback"]}],
)
error_event = {
"type": "error",
"sequence_number": 2,
"error": {
"type": "invalid_request_error",
"code": "content_policy_violation",
"message": "This content was flagged for possible cybersecurity risk. The response was halted mid-stream.",
"param": None,
},
}
async def aiter_bytes():
yield f"data: {json.dumps(error_event)}\n\n".encode()
raw_response = MagicMock()
raw_response.headers = {}
raw_response.aiter_bytes = aiter_bytes
logging_obj = MagicMock(spec=LiteLLMLogging)
logging_obj.model_call_details = {"litellm_params": {}}
logging_obj.completion_start_time = None
source = ResponsesAPIStreamingIterator(
response=raw_response,
model="gpt-5.4",
responses_api_provider_config=OpenAIResponsesAPIConfig(),
logging_obj=logging_obj,
custom_llm_provider="openai",
)
fallback_chunks = [MagicMock(type="response.output_text.delta"), MagicMock(type="response.completed")]
fallback_call = AsyncMock(return_value=_AsyncList(fallback_chunks))
wrapped = await router._aresponses_streaming_iterator(
response=source,
initial_kwargs={
"model": "primary",
"stream": True,
"input": "Hi",
"original_generic_function": fallback_call,
},
)
collected = [chunk async for chunk in wrapped]
assert collected == fallback_chunks
fallback_call.assert_awaited_once()
assert fallback_call.await_args.kwargs["model"] == "gemini/gemini-2.5-flash"
@pytest.mark.asyncio
async def test_aresponses_streaming_iterator_unwraps_content_policy_trigger_for_fallback_dispatch():
"""The fallback dispatch matches on the trigger's own type, so the wrapper must hand it the
ContentPolicyViolationError carried inside MidStreamFallbackError, not the envelope."""
router = _make_router_with_fallback("openai/gpt-5.4", "openai/gpt-5-mini")
content_policy_error = litellm.ContentPolicyViolationError(
message="flagged mid-stream", llm_provider="openai", model="openai/gpt-5.4"
)
src = _make_responses_iterator(
chunks=[MagicMock(type="response.created")],
error=MidStreamFallbackError(
message=str(content_policy_error),
model="openai/gpt-5.4",
llm_provider="openai",
original_exception=content_policy_error,
is_pre_first_chunk=True,
),
model="openai/gpt-5.4",
)
with patch.object(
router,
"async_function_with_fallbacks_common_utils",
new=AsyncMock(return_value=_AsyncList([MagicMock(type="response.completed")])),
) as mock_fallback_utils:
wrapped = await router._aresponses_streaming_iterator(
response=src,
initial_kwargs={"model": "openai/gpt-5.4", "stream": True, "input": "Hi"},
)
[chunk async for chunk in wrapped]
mock_fallback_utils.assert_awaited_once()
assert mock_fallback_utils.await_args.kwargs["e"] is content_policy_error
@pytest.mark.asyncio
@pytest.mark.parametrize(
"fallback_headers",