diff --git a/litellm/files/streaming.py b/litellm/files/streaming.py index 460050872f4..6956d3baf32 100644 --- a/litellm/files/streaming.py +++ b/litellm/files/streaming.py @@ -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: diff --git a/litellm/litellm_core_utils/exception_mapping_utils.py b/litellm/litellm_core_utils/exception_mapping_utils.py index 1f49cc2c4bb..e7f2f5dc359 100644 --- a/litellm/litellm_core_utils/exception_mapping_utils.py +++ b/litellm/litellm_core_utils/exception_mapping_utils.py @@ -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 diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 6ec426d6da5..045943fdd88 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -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 diff --git a/litellm/llms/custom_httpx/transport_errors.py b/litellm/llms/custom_httpx/transport_errors.py new file mode 100644 index 00000000000..f3a2b04a4ea --- /dev/null +++ b/litellm/llms/custom_httpx/transport_errors.py @@ -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) diff --git a/tests/unit/llms/custom_httpx/test_transport_errors.py b/tests/unit/llms/custom_httpx/test_transport_errors.py new file mode 100644 index 00000000000..92aad0318c8 --- /dev/null +++ b/tests/unit/llms/custom_httpx/test_transport_errors.py @@ -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