fix(router): map stream opening exceptions and prevent bytes concat (#49933)

This commit is contained in:
Charlie Freeman 2026-11-04 20:58:57 +08:00
parent fd4cef22ea
commit e8eccd8864
5 changed files with 121 additions and 12 deletions

View file

@ -2503,16 +2503,24 @@ class CustomStreamWrapper:
def fetch_sync_stream(self):
if self.completion_stream is None and self.make_call is not None:
# Call make_call to get the completion stream
self.completion_stream = self.make_call(client=litellm.module_level_client)
self._stream_iter = self.completion_stream.__iter__()
try:
self.completion_stream = self.make_call(client=litellm.module_level_client)
self._stream_iter = self.completion_stream.__iter__()
except Exception as e:
self.close()
raise self._map_stream_exception(e) from e
return self.completion_stream
async def fetch_stream(self):
if self.completion_stream is None and self.make_call is not None:
# Call make_call to get the completion stream
self.completion_stream = await self.make_call(client=litellm.module_level_aclient)
self._stream_iter = self.completion_stream.__aiter__()
try:
self.completion_stream = await self.make_call(client=litellm.module_level_aclient)
self._stream_iter = self.completion_stream.__aiter__()
except Exception as e:
await self.aclose()
raise self._map_stream_exception(e) from e
return self.completion_stream
@ -2833,12 +2841,15 @@ class CustomStreamWrapper:
self._emitted_disqualifying_content = True
def _map_stream_exception(self, original_exception: Exception) -> Exception:
return exception_type(
model=self.model,
custom_llm_provider=self.custom_llm_provider,
original_exception=original_exception,
extra_kwargs=with_deployment_api_base(self.logging_obj.model_call_details.get("litellm_params")),
)
try:
return exception_type(
model=self.model,
custom_llm_provider=self.custom_llm_provider,
original_exception=original_exception,
extra_kwargs=with_deployment_api_base(self.logging_obj.model_call_details.get("litellm_params")),
)
except Exception as e:
return e
def _handle_stream_fallback_error(self, e: Exception) -> "NoReturn":
"""

View file

@ -685,6 +685,10 @@ class MaskedHTTPStatusError(httpx.HTTPStatusError):
request=masked_request,
),
)
if isinstance(message, bytes):
message = message.decode("utf-8", errors="replace")
if isinstance(text, bytes):
text = text.decode("utf-8", errors="replace")
self.message = message
self.text = text
self.status_code = original_error.response.status_code

View file

@ -3367,8 +3367,13 @@ def make_sync_call(
timeout=timeout,
)
except httpx.HTTPStatusError as e:
exception_string: Final = str(e.response.read())
known_cached_contents.forget_rejected(request_body.get("cachedContent"), e.response.status_code)
raise
raise VertexAIError(
status_code=e.response.status_code,
message=VertexGeminiConfig().translate_exception_str(exception_string),
headers=e.response.headers,
)
if response.status_code != 200 and response.status_code != 201:
raise VertexAIError(

View file

@ -9421,12 +9421,16 @@ class PythonCore(DeprecatedRouterMembers):
and not no_fallback_group_explained
and not nested_fallback_hop
):
original_exception.message += format_fallback_outcome_message(
outcome_msg: Final = format_fallback_outcome_message(
model_group,
attempted_fallback_group,
fallback_failure_exception_str,
skipped=fallback_walk.skipped,
)
if isinstance(original_exception.message, str):
original_exception.message += outcome_msg
elif isinstance(original_exception.message, bytes):
original_exception.message += outcome_msg.encode("utf-8", errors="replace")
raise original_exception

View file

@ -0,0 +1,85 @@
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
import pytest
import litellm
class _Gemini400Server(BaseHTTPRequestHandler):
def do_POST(self):
self.rfile.read(int(self.headers.get("content-length", 0)))
body = b'{"error": {"code": 400, "message": "bad request", "status": "INVALID_ARGUMENT"}}'
self.send_response(400)
self.send_header("content-type", "application/json")
self.send_header("content-length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
@pytest.fixture(scope="module")
def gemini_400_url():
server = HTTPServer(("127.0.0.1", 0), _Gemini400Server)
port = server.server_port
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
yield f"http://127.0.0.1:{port}"
server.shutdown()
def test_router_sync_streaming_gemini_error_mapped(gemini_400_url):
"""
Issue #49933: Router.completion(stream=True) with a Gemini deployment raising
HTTP 400 should map to litellm.BadRequestError instead of TypeError (can't concat str to bytes)
or raw MaskedHTTPStatusError.
"""
params = {
"model": "gemini/gemini-2.5-flash",
"api_base": gemini_400_url,
"api_key": "fake-key",
}
router = litellm.Router(
model_list=[{"model_name": "test-gemini", "litellm_params": params}],
num_retries=0,
)
with pytest.raises(litellm.BadRequestError) as exc_info:
for _ in router.completion(
model="test-gemini",
messages=[{"role": "user", "content": "hi"}],
stream=True,
):
pass
assert isinstance(exc_info.value, litellm.BadRequestError)
@pytest.mark.asyncio
async def test_router_async_streaming_gemini_error_mapped(gemini_400_url):
"""
Issue #49933: Router.acompletion(stream=True) with a Gemini deployment raising
HTTP 400 should map to litellm.BadRequestError instead of raw VertexAIError.
"""
params = {
"model": "gemini/gemini-2.5-flash",
"api_base": gemini_400_url,
"api_key": "fake-key",
}
router = litellm.Router(
model_list=[{"model_name": "test-gemini", "litellm_params": params}],
num_retries=0,
)
with pytest.raises(litellm.BadRequestError) as exc_info:
stream = await router.acompletion(
model="test-gemini",
messages=[{"role": "user", "content": "hi"}],
stream=True,
)
async for _ in stream:
pass
assert isinstance(exc_info.value, litellm.BadRequestError)