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

fix(streaming): end finished streams cleanly on SDK 3 transport errors and map file download errors

Browse source 
openai SDK 3 reads over httpx2 and, from 3.25, wraps transport errors in APIConnectionError or APITimeoutError. The stream wrapper only knew httpx, so a connection drop after finish_reason raised MidStreamFallbackError and a read timeout fell back as a 500. Transport errors are now classified across httpx, httpx2 and the SDK's explicit cause chain, a drop after finish_reason ends the stream on both the sync and async paths, and file_content(stream=True) raises litellm.Timeout or litellm.APIConnectionError instead of raw transport errors
This commit is contained in:
mateo-berri 2026-10-20 19:44:47 -07:00
parent 9dc0e71aed
commit 53312cb5f2
5 changed files with 205 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
litellm/files/streaming.py
View file

@ -5,7 +5,9 @@ from typing import TYPE_CHECKING, Any, Final, Optional, cast
import anyio
from litellm.exceptions import APIConnectionError, Timeout
from litellm.files.types import FileContentProvider
from litellm.llms.custom_httpx.transport_errors import TransportErrorKind, chained_transport_error_kind
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import (
@ -59,8 +61,11 @@ class FileContentStreamingResponse:
self._log_success_sync()
raise
except Exception as e:
self._log_failure_sync(e)
raise
public_error: Final = self._public_error(e)
self._log_failure_sync(public_error)
if public_error is e:
raise
raise public_error from e
def __aiter__(self) -> "FileContentStreamingResponse":
if not hasattr(self.stream_iterator, "__anext__"):
@ -77,8 +82,28 @@ class FileContentStreamingResponse:
await self._log_success_async()
raise
except Exception as e:
await self._log_failure_async(e)
raise
public_error: Final = self._public_error(e)
await self._log_failure_async(public_error)
if public_error is e:
raise
raise public_error from e
def _public_error(self, error: Exception) -> Exception:
match chained_transport_error_kind(error):
case TransportErrorKind.TIMEOUT:
return Timeout(
message=f"timed out reading file content: {error}",
model=self.model,
llm_provider=self.custom_llm_provider,
)
case TransportErrorKind.DROPPED_CONNECTION:
return APIConnectionError(
message=f"connection dropped while reading file content: {error}",
llm_provider=self.custom_llm_provider,
model=self.model,
)
case None:
return error
async def aclose(self) -> None:
if self._close_completed:

3
litellm/litellm_core_utils/exception_mapping_utils.py
View file

@ -18,6 +18,7 @@ from litellm.litellm_core_utils.bug_report import (
)
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.secret_redaction import redact_string as redact_secret_string
from litellm.llms.custom_httpx.transport_errors import TransportErrorKind, transport_error_classes
from litellm.types.utils import LlmProviders
from ..exceptions import (
@ -2551,7 +2552,7 @@ def exception_type(
################################################################################
if (
isinstance(original_exception, httpx.TimeoutException)
isinstance(original_exception, transport_error_classes(TransportErrorKind.TIMEOUT))
or "Request Timeout Error" in error_str
or "Request timed out" in error_str
or "Timed out generating response" in error_str

62
litellm/litellm_core_utils/streaming_handler.py
View file

@ -35,6 +35,7 @@ from litellm.litellm_core_utils.model_response_utils import (
)
from litellm.litellm_core_utils.redact_messages import LiteLLMLoggingObject
from litellm.litellm_core_utils.thread_pool_executor import executor
from litellm.llms.custom_httpx.transport_errors import TransportErrorKind, chained_transport_error_kind
from litellm.llms.ollama.common_utils import OllamaResponseIdNormalizer
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.llms.openai import OpenAIChatCompletionChunk
@ -54,7 +55,7 @@ from litellm.types.utils import (
)
from litellm.types.utils import GenericStreamingChunk as GChunk
from ..exceptions import OpenAIError, Timeout
from ..exceptions import OpenAIError
from .core_helpers import map_finish_reason, process_response_headers
from .exception_mapping_utils import exception_type, with_deployment_api_base
from .llm_response_utils.get_api_base import get_api_base
@ -122,12 +123,6 @@ def _delta_from_upstream(upstream_delta: BaseModel | None) -> Delta | None:
return None
def _is_transport_timeout(error: BaseException) -> bool:
return isinstance(error, httpx.TimeoutException) or (
isinstance(error, Timeout) and isinstance(error.__cause__, httpx.TimeoutException)
)
def _skips_stream_fallback(status_code: int | None) -> bool:
return status_code is not None and 400 <= status_code < 500 and status_code != 429
@ -2000,7 +1995,7 @@ class CustomStreamWrapper:
):
chunk = self.completion_stream
else:
chunk = next(self.completion_stream)
chunk = self._next_sync_chunk()
if chunk is not None and chunk != b"":
print_verbose(
f"PROCESSED CHUNK PRE CHUNK CREATOR: {chunk.decode('utf-8', errors='replace') if isinstance(chunk, bytes) else chunk}; custom_llm_provider: {self.custom_llm_provider}"
@ -2162,6 +2157,17 @@ class CustomStreamWrapper:
self.close()
self._handle_stream_fallback_error(e)
def _next_sync_chunk(self) -> object:
try:
return next(self.completion_stream)
except Exception as e:
if (
self.received_finish_reason is None
or chained_transport_error_kind(e) is not TransportErrorKind.DROPPED_CONNECTION
):
raise
raise StopIteration from e
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
@ -2282,26 +2288,26 @@ class CustomStreamWrapper:
return processed_chunk
except (StopAsyncIteration, StopIteration):
return await self._finalize_completed_stream(cache_hit=cache_hit)
except httpx.TimeoutException as e: # if httpx read timeout error occues
traceback_exception = traceback.format_exc()
## ADD DEBUG INFORMATION - E.G. LITELLM REQUEST TIMEOUT
traceback_exception += f"\nLiteLLM Default Request Timeout - {litellm.request_timeout}"
if self.logging_obj is not None:
self._record_partial_usage_for_failure()
## LOGGING
self._failure_logging = asyncio.create_task(
self.logging_obj.dispatch_failure_handlers(e, traceback_exception, prefer_async_handlers=True)
)
await self.aclose()
self._handle_stream_fallback_error(e)
except (httpx.ReadError, httpx.RemoteProtocolError) as e:
if self.received_finish_reason is None:
await self.aclose()
self._log_stream_failure_and_raise(e)
return await self._finalize_completed_stream(cache_hit=cache_hit)
except Exception as e:
await self.aclose()
self._log_stream_failure_and_raise(e)
match chained_transport_error_kind(e):
case TransportErrorKind.DROPPED_CONNECTION if self.received_finish_reason is not None:
return await self._finalize_completed_stream(cache_hit=cache_hit)
case TransportErrorKind.TIMEOUT:
traceback_exception: Final = (
f"{traceback.format_exc()}\nLiteLLM Default Request Timeout - {litellm.request_timeout}"
)
if self.logging_obj is not None:
self._record_partial_usage_for_failure()
self._failure_logging = asyncio.create_task(
self.logging_obj.dispatch_failure_handlers(
e, traceback_exception, prefer_async_handlers=True
)
)
await self.aclose()
self._handle_stream_fallback_error(e)
case _:
await self.aclose()
self._log_stream_failure_and_raise(e)
async def _finalize_completed_stream(self, cache_hit: bool) -> "ModelResponseStream":
if self.sent_last_chunk is True:
@ -2556,7 +2562,7 @@ class CustomStreamWrapper:
# Raise non-retriable client errors directly (skip fallback).
# Exception: 429 (rate-limit) IS retriable/transient — allow it
# through so the Router can switch to a different model group.
if not _is_transport_timeout(e) and (
if chained_transport_error_kind(e) is not TransportErrorKind.TIMEOUT and (
_skips_stream_fallback(mapped_status_code) or _skips_stream_fallback(original_status_code)
):
raise mapped_exception

54
litellm/llms/custom_httpx/transport_errors.py Normal file
View file

@ -0,0 +1,54 @@
import sys
from collections.abc import Iterator
from enum import Enum
from typing import Final, TypeAlias
import httpx
class TransportErrorKind(Enum):
TIMEOUT = "timeout"
DROPPED_CONNECTION = "dropped_connection"
_ErrorTable: TypeAlias = tuple[tuple[type[BaseException], TransportErrorKind], ...]
_HTTPX_ERRORS: Final[_ErrorTable] = (
(httpx.TimeoutException, TransportErrorKind.TIMEOUT),
(httpx.ReadError, TransportErrorKind.DROPPED_CONNECTION),
(httpx.RemoteProtocolError, TransportErrorKind.DROPPED_CONNECTION),
)
_MAX_CAUSE_DEPTH: Final = 16
def _transport_errors() -> _ErrorTable:
if "httpx2" not in sys.modules:
return _HTTPX_ERRORS
import httpx2
return (
*_HTTPX_ERRORS,
(httpx2.TimeoutException, TransportErrorKind.TIMEOUT),
(httpx2.ReadError, TransportErrorKind.DROPPED_CONNECTION),
(httpx2.RemoteProtocolError, TransportErrorKind.DROPPED_CONNECTION),
)
def transport_error_kind(error: object) -> TransportErrorKind | None:
return next((kind for error_class, kind in _transport_errors() if isinstance(error, error_class)), None)
def transport_error_classes(kind: TransportErrorKind) -> tuple[type[BaseException], ...]:
return tuple(error_class for error_class, error_kind in _transport_errors() if error_kind is kind)
def _explicit_causes(error: BaseException | None, depth: int) -> Iterator[BaseException]:
if error is None or depth == 0:
return
yield error
yield from _explicit_causes(error.__cause__, depth - 1)
def chained_transport_error_kind(error: BaseException) -> TransportErrorKind | None:
kinds: Final = (transport_error_kind(cause) for cause in _explicit_causes(error, _MAX_CAUSE_DEPTH))
return next((kind for kind in kinds if kind is not None), None)

86
tests/unit/llms/custom_httpx/test_transport_errors.py Normal file
View file

@ -0,0 +1,86 @@
from typing import Final
import httpx
import openai
import pytest
from litellm.llms.custom_httpx.transport_errors import (
TransportErrorKind,
chained_transport_error_kind,
transport_error_classes,
transport_error_kind,
)
REQUEST: Final = httpx.Request("POST", "https://api.openai.com/v1/chat/completions")
def _raised_from(error: Exception, cause: Exception) -> Exception:
try:
raise error from cause
except Exception as raised:
return raised
def _raised_while_handling(error: Exception, handled: Exception) -> Exception:
try:
try:
raise handled
except Exception:
raise error
except Exception as raised:
return raised
@pytest.mark.parametrize("module_name", ("httpx", "httpx2"))
@pytest.mark.parametrize(
("class_name", "expected"),
(
("ReadTimeout", TransportErrorKind.TIMEOUT),
("ConnectTimeout", TransportErrorKind.TIMEOUT),
("ReadError", TransportErrorKind.DROPPED_CONNECTION),
("RemoteProtocolError", TransportErrorKind.DROPPED_CONNECTION),
("ConnectError", None),
("WriteError", None),
),
)
def test_transport_error_kind_matches_httpx_and_httpx2_alike(
module_name: str, class_name: str, expected: TransportErrorKind | None
):
module: Final = pytest.importorskip(module_name)
error: Final = getattr(module, class_name)("transport failure")
assert transport_error_kind(error) is expected
assert chained_transport_error_kind(error) is expected
assert all(isinstance(error, transport_error_classes(kind)) is (kind is expected) for kind in TransportErrorKind)
def test_transport_error_kind_ignores_values_that_are_not_transport_errors():
assert transport_error_kind(ValueError("decoder failure")) is None
assert transport_error_kind({"error": "not an exception"}) is None
def test_chained_kind_reads_the_transport_error_an_sdk_error_was_raised_from():
httpx2: Final = pytest.importorskip("httpx2")
dropped: Final = _raised_from(openai.APIConnectionError(request=REQUEST), httpx2.ReadError("peer closed"))
timed_out: Final = _raised_from(openai.APITimeoutError(request=REQUEST), httpx2.ReadTimeout("timed out"))
twice_wrapped: Final = _raised_from(ValueError("outer"), timed_out)
assert transport_error_kind(dropped) is None
assert chained_transport_error_kind(dropped) is TransportErrorKind.DROPPED_CONNECTION
assert chained_transport_error_kind(timed_out) is TransportErrorKind.TIMEOUT
assert chained_transport_error_kind(twice_wrapped) is TransportErrorKind.TIMEOUT
assert chained_transport_error_kind(openai.APIConnectionError(request=REQUEST)) is None
def test_chained_kind_ignores_an_error_raised_while_handling_a_transport_error():
contextual: Final = _raised_while_handling(ValueError("bug in a handler"), httpx.ReadError("connection reset"))
assert contextual.__context__ is not None
assert chained_transport_error_kind(contextual) is None
def test_chained_kind_stops_on_a_cause_cycle():
cyclic: Final = ValueError("caused by itself")
cyclic.__cause__ = cyclic
assert chained_transport_error_kind(cyclic) is None