scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-07 08:26:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

fix(bedrock): surface modeled HTTP status for mid-stream error events so 5xx is retryable (#24608) (#30946)

Browse source 
* fix(bedrock): surface modeled HTTP status for mid-stream error events (#24608)

* test(bedrock): mid-stream server errors trigger streaming fallback (#24608)

* style(bedrock): black-format stream-error helper (#24608)
This commit is contained in:
Kent 2026-06-22 20:17:38 +08:00 • committed by GitHub
parent 488f7874c6
commit 67c0183fcf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 177 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
litellm/llms/bedrock/chat/invoke_handler.py
View file

@ -70,6 +70,7 @@ from ..base_aws_llm import BaseAWSLLM
from ..common_utils import (
BedrockError,
ModelResponseIterator,
build_bedrock_stream_error,
get_bedrock_response_stream_shape,
get_bedrock_tool_name,
)
@ -1841,23 +1842,7 @@ class AWSEventStreamDecoder:
parsed_response = self.parser.parse(response_dict, response_stream_shape)
if response_dict["status_code"] != 200:
decoded_body = response_dict["body"].decode()
if isinstance(decoded_body, dict):
error_message = decoded_body.get("message")
elif isinstance(decoded_body, str):
error_message = decoded_body
else:
error_message = ""
exception_status = response_dict["headers"].get(":exception-type")
error_message = exception_status + " " + error_message
raise BedrockError(
status_code=response_dict["status_code"],
message=(
json.dumps(error_message)
if isinstance(error_message, dict)
else error_message
),
)
raise build_bedrock_stream_error(response_dict, response_stream_shape)
if "chunk" in parsed_response:
chunk = parsed_response.get("chunk")
if not chunk:

65
litellm/llms/bedrock/common_utils.py
View file

@ -7,9 +7,21 @@ Common utilities used across bedrock chat/embedding/image generation
import functools
import json
import os
from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union
from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
Literal,
Mapping,
Optional,
TypedDict,
Union,
)
if TYPE_CHECKING:
from botocore.model import Shape
from litellm.types.llms.bedrock import BedrockCreateBatchRequest
import httpx
@ -1132,6 +1144,39 @@ def get_bedrock_response_stream_shape():
return _load_bedrock_response_stream_shape()
class BedrockEventStreamResponseDict(TypedDict):
status_code: int
headers: Mapping[str, str]
body: bytes
def build_bedrock_stream_error(
response_dict: BedrockEventStreamResponseDict,
response_stream_shape: Shape | None,
) -> BedrockError:
"""Build a BedrockError for a non-200 event-stream error event.
botocore hard-codes HTTP 400 on every mid-stream error event, so the modeled
ResponseStream member's httpStatusCode is the real status. Resolve it from the
shape and fall back to the raw status when the type is not modeled.
"""
exception_type = response_dict["headers"].get(":exception-type")
decoded_body = response_dict["body"].decode()
message = f"{exception_type} {decoded_body}" if exception_type else decoded_body
status_code = response_dict["status_code"]
if exception_type is not None and response_stream_shape is not None:
member = response_stream_shape.members.get(exception_type)
if member is not None:
modeled_status = (
(member.metadata or {}).get("error", {}).get("httpStatusCode")
)
if modeled_status is not None:
status_code = int(modeled_status)
return BedrockError(status_code=status_code, message=message)
class BedrockEventStreamDecoderBase:
"""
Base class for event stream decoding for Bedrock
@ -1156,23 +1201,7 @@ class BedrockEventStreamDecoderBase:
parsed_response = self.parser.parse(response_dict, response_stream_shape)
if response_dict["status_code"] != 200:
decoded_body = response_dict["body"].decode()
if isinstance(decoded_body, dict):
error_message = decoded_body.get("message")
elif isinstance(decoded_body, str):
error_message = decoded_body
else:
error_message = ""
exception_status = response_dict["headers"].get(":exception-type")
error_message = exception_status + " " + error_message
raise BedrockError(
status_code=response_dict["status_code"],
message=(
json.dumps(error_message)
if isinstance(error_message, dict)
else error_message
),
)
raise build_bedrock_stream_error(response_dict, response_stream_shape)
if "chunk" in parsed_response:
chunk = parsed_response.get("chunk")
if not chunk:

26
tests/llm_translation/test_bedrock_completion.py
View file

@ -2502,19 +2502,34 @@ async def test_bedrock_image_url_sync_client():
mock_post.assert_called_once()
def test_bedrock_error_handling_streaming():
@pytest.mark.parametrize(
"exception_type, expected_status_code",
[
("internalServerException", 500),
("serviceUnavailableException", 503),
("modelTimeoutException", 408),
("modelStreamErrorException", 424),
("validationException", 400),
],
)
def test_bedrock_error_handling_streaming(exception_type, expected_status_code):
"""Bedrock event-stream error events arrive with botocore's hard-coded
status_code=400; the decoder must surface the modeled HTTP status instead
(e.g. internalServerException -> 500). For 5xx this is what makes the error
retryable downstream; for all types it replaces the misleading 400 with the
true code. Regression for #24608."""
from litellm.llms.bedrock.chat.invoke_handler import (
AWSEventStreamDecoder,
BedrockError,
)
from unittest.mock import patch, Mock
from unittest.mock import Mock
event = Mock()
event.to_response_dict = Mock(
return_value={
"status_code": 400,
"headers": {
":exception-type": "serviceUnavailableException",
":exception-type": exception_type,
":content-type": "application/json",
":message-type": "exception",
},
@ -2525,11 +2540,10 @@ def test_bedrock_error_handling_streaming():
decoder = AWSEventStreamDecoder(
model="bedrock/anthropic.claude-3-sonnet-20240229-v1:0"
)
with pytest.raises(Exception) as e:
with pytest.raises(BedrockError) as e:
decoder._parse_message_from_event(event)
assert isinstance(e.value, BedrockError)
assert "Bedrock is unable to process your request." in e.value.message
assert e.value.status_code == 400
assert e.value.status_code == expected_status_code
@pytest.mark.parametrize(

108
tests/test_litellm/litellm_core_utils/test_streaming_handler.py
View file

@ -878,6 +878,114 @@ def test_sync_streaming_bad_request_not_midstream(logging_obj: Logging):
assert "invalid maxOutputTokens" in str(excinfo.value)
def _bedrock_error_event(exception_type: str):
"""A mocked botocore event-stream error event: status_code is botocore's
hard-coded 400, with the real type in the :exception-type header."""
event = Mock()
event.to_response_dict = Mock(
return_value={
"status_code": 400,
"headers": {
":exception-type": exception_type,
":content-type": "application/json",
":message-type": "exception",
},
"body": b'{"message":"Bedrock had an internal error."}',
}
)
return event
@pytest.mark.asyncio
async def test_bedrock_midstream_internal_server_error_wraps_for_fallback(
logging_obj: Logging,
):
"""End-to-end regression for https://github.com/BerriAI/litellm/issues/24608:
a Bedrock mid-stream internalServerException event (botocore stamps it 400)
must flow through the real decoder, gain its modeled 500 status, and wrap
into MidStreamFallbackError so the Router can run streaming fallback.
Calls the real AWSEventStreamDecoder, so reverting the decoder status fix
makes the decoder raise BedrockError(400) and the gate raises BadRequestError
directly -> this test fails without the fix."""
from litellm.exceptions import MidStreamFallbackError
from litellm.llms.bedrock.chat.invoke_handler import AWSEventStreamDecoder
decoder = AWSEventStreamDecoder(model="anthropic.claude-3-sonnet-20240229-v1:0")
async def _bedrock_stream():
decoder._parse_message_from_event(
_bedrock_error_event("internalServerException")
)
yield # unreachable; the line above raises
async def _make_call(**kwargs):
return _bedrock_stream()
response = CustomStreamWrapper(
completion_stream=None,
model="anthropic.claude-3-sonnet-20240229-v1:0",
logging_obj=logging_obj,
custom_llm_provider="bedrock",
make_call=_make_call,
)
with pytest.raises(MidStreamFallbackError):
await response.__anext__()
@pytest.mark.asyncio
async def test_bedrock_5xx_wraps_for_midstream_fallback(logging_obj: Logging):
"""Gate contract: a Bedrock 5xx (here 503 serviceUnavailableException) wraps
into MidStreamFallbackError so the Router can run streaming fallback."""
from litellm.exceptions import MidStreamFallbackError
from litellm.llms.bedrock.chat.invoke_handler import BedrockError
async def _raise_503(**kwargs):
raise BedrockError(
status_code=503,
message="serviceUnavailableException Bedrock is unavailable.",
)
response = CustomStreamWrapper(
completion_stream=None,
model="anthropic.claude-3-sonnet-20240229-v1:0",
logging_obj=logging_obj,
custom_llm_provider="bedrock",
make_call=_raise_503,
)
with pytest.raises(MidStreamFallbackError):
await response.__anext__()
@pytest.mark.asyncio
async def test_bedrock_validation_error_raises_directly(logging_obj: Logging):
"""Gate contract: a Bedrock validationException (400) is a client error and
must surface directly, never wrapped into MidStreamFallbackError."""
from litellm.exceptions import MidStreamFallbackError
from litellm.llms.bedrock.chat.invoke_handler import BedrockError
async def _raise_400(**kwargs):
raise BedrockError(
status_code=400,
message="validationException malformed input.",
)
response = CustomStreamWrapper(
completion_stream=None,
model="anthropic.claude-3-sonnet-20240229-v1:0",
logging_obj=logging_obj,
custom_llm_provider="bedrock",
make_call=_raise_400,
)
with pytest.raises(Exception) as excinfo:
await response.__anext__()
assert not isinstance(excinfo.value, MidStreamFallbackError)
assert getattr(excinfo.value, "status_code", None) == 400
@pytest.mark.asyncio
async def test_async_streaming_read_timeout_triggers_midstream_fallback(
logging_obj: Logging,