From fa152013a8610eba674e54bb874e940bbfd8b00b Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Thu, 29 Oct 2026 04:32:37 -0700 Subject: [PATCH] fix(proxy): name the deployment in stream error alerts on every provider and route The llm_exceptions alert for a cut stream on Anthropic chat, /v1/completions, native /v1/responses and native /v1/messages had no API Base, model_group or deployment line. The shared exception mapper now attaches its debug info at its single exit, mid-stream fallback errors inherit it from the error they wrap, and the native relays add it to the same error object they re-raise. --- litellm/exceptions.py | 5 +- .../exception_mapping_utils.py | 37 ++++++- .../messages/native/streaming_iterator.py | 26 +++-- litellm/main.py | 15 +-- .../streaming_handler.py | 2 + litellm/responses/streaming_iterator.py | 33 +++--- litellm/utils.py | 5 +- .../test_stream_error_alert_api_base.py | 100 +++++++++++++++++- .../test_exception_mapping_utils.py | 36 +++++++ .../native/test_streaming_iterator.py | 61 +++++++++++ .../unit/responses/test_streaming_iterator.py | 28 +++++ .../test_streaming_iterator.py | 29 +++++ 12 files changed, 340 insertions(+), 37 deletions(-) diff --git a/litellm/exceptions.py b/litellm/exceptions.py index cd5139f5f79..707b23b2caf 100644 --- a/litellm/exceptions.py +++ b/litellm/exceptions.py @@ -1236,7 +1236,10 @@ class MidStreamFallbackError(ServiceUnavailableError): self.model = model self.llm_provider = llm_provider self.original_exception = original_exception - self.litellm_debug_info = litellm_debug_info + inherited_debug_info: Final[object] = getattr(original_exception, "litellm_debug_info", None) + self.litellm_debug_info = litellm_debug_info or ( + inherited_debug_info if isinstance(inherited_debug_info, str) else None + ) self.max_retries = max_retries self.num_retries = num_retries self.generated_content = generated_content diff --git a/litellm/litellm_core_utils/exception_mapping_utils.py b/litellm/litellm_core_utils/exception_mapping_utils.py index 47ec8331c83..ef24dda3857 100644 --- a/litellm/litellm_core_utils/exception_mapping_utils.py +++ b/litellm/litellm_core_utils/exception_mapping_utils.py @@ -2443,6 +2443,39 @@ def with_deployment_api_base(litellm_params: object) -> Mapping[str, object]: return MappingProxyType({**params, "api_base": deployment_api_base}) +_DEBUG_INFO_ATTRIBUTE: Final = "litellm_debug_info" + + +def _attach_debug_info(exception: Exception, debug_info: str) -> None: + if debug_info and getattr(exception, _DEBUG_INFO_ATTRIBUTE, None) is None: + setattr(exception, _DEBUG_INFO_ATTRIBUTE, debug_info) + + +def mapped_stream_exception( + original_exception: Exception, model: str, custom_llm_provider: str, litellm_params: object +) -> Exception: + try: + return exception_type( + model=model, + custom_llm_provider=custom_llm_provider, + original_exception=original_exception, + completion_kwargs={}, + extra_kwargs=with_deployment_api_base(litellm_params), + ) + except Exception as mapped_exception: + return mapped_exception + + +def with_deployment_debug_info( + error: Exception, model: str, custom_llm_provider: str, litellm_params: object +) -> Exception: + debug_info: Final[object] = getattr( + mapped_stream_exception(error, model, custom_llm_provider, litellm_params), _DEBUG_INFO_ATTRIBUTE, None + ) + _attach_debug_info(error, debug_info if isinstance(debug_info, str) else "") + return error + + def _api_base_without_credentials(api_base: str) -> str: scheme, separator, _ = api_base.partition("://") scheme_prefix: Final = scheme + separator if separator and _URL_SCHEME.fullmatch(scheme) else "" @@ -2492,11 +2525,11 @@ def exception_type( print() # noqa: T201 litellm_response_headers: Final = _get_response_headers(original_exception=original_exception) + extra_information = "" try: error_str = ( redact_secret_string(str(original_exception)) if _ENABLE_SECRET_REDACTION else str(original_exception) ) - extra_information = "" if model or custom_llm_provider: if hasattr(original_exception, "message"): error_str = ( @@ -2837,11 +2870,13 @@ def exception_type( # don't let an error with mapping interrupt the user from receiving an error from the llm api calls if exception_mapping_worked: setattr(e, "litellm_response_headers", litellm_response_headers) + _attach_debug_info(e, extra_information) raise e else: for error_type in litellm.LITELLM_EXCEPTION_TYPES: if isinstance(e, error_type): setattr(e, "litellm_response_headers", litellm_response_headers) + _attach_debug_info(e, extra_information) raise e # it's already mapped raised_exc: Final = APIConnectionError( message=f"{original_exception}\n{redact_string(traceback.format_exc())}", diff --git a/litellm/llms/anthropic/messages/native/streaming_iterator.py b/litellm/llms/anthropic/messages/native/streaming_iterator.py index 52b3f1042f8..f3f22f4ed2a 100644 --- a/litellm/llms/anthropic/messages/native/streaming_iterator.py +++ b/litellm/llms/anthropic/messages/native/streaming_iterator.py @@ -15,6 +15,7 @@ from litellm.constants import ( ANTHROPIC_MESSAGES_STREAM_RELAY_QUEUE_MAXSIZE, ) from litellm.litellm_core_utils.core_helpers import process_response_headers +from litellm.litellm_core_utils.exception_mapping_utils import with_deployment_debug_info from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.llms.anthropic.common_utils import ANTHROPIC_ERROR_STATUS_CODE_MAP @@ -562,6 +563,7 @@ class BaseAnthropicMessagesStreamingIterator: start_time=self.start_time, passthrough_success_handler_obj=GLOBAL_PASS_THROUGH_SUCCESS_HANDLER_OBJ, url_route="/v1/messages", + on_upstream_error=self._with_deployment_debug_info, ) def _convert_chunk_to_sse_format(self, chunk: dict | Any) -> bytes: @@ -609,8 +611,9 @@ class BaseAnthropicMessagesStreamingIterator: An upstream failure (Bedrock read / decode / chunk-conversion error) that happens while the client is still connected is forwarded through the queue and re-raised here, so the original provider exception (and - its status) reaches the proxy's failure handling unchanged rather than - being masked by a generic incomplete-stream event. + its status) reaches the proxy's failure handling with the deployment's + debug info attached, rather than being masked by a generic + incomplete-stream event. This method provides the common logic for both Anthropic and Bedrock implementations. """ @@ -797,18 +800,29 @@ class BaseAnthropicMessagesStreamingIterator: """ from litellm.proxy.pass_through_endpoints.streaming_handler import PassThroughStreamingHandler + described: Final = self._with_deployment_debug_info(exc) await PassThroughStreamingHandler.schedule_stream_failure_logging( litellm_logging_obj=self.litellm_logging_obj, endpoint_type=EndpointType.ANTHROPIC, request_body=self.request_body, raw_bytes=collected_chunks, - exception=exc, + exception=described, ) - if not client_detached.is_set() and await self._enqueue_for_client(queue, client_detached, exc): + if not client_detached.is_set() and await self._enqueue_for_client(queue, client_detached, described): await client_detached.wait() - if not _exception_left_unconsumed(queue, exc): + if not _exception_left_unconsumed(queue, described): return - await self._fire_detached_failure_hook(exc) + await self._fire_detached_failure_hook(described) + + def _with_deployment_debug_info(self, exc: Exception) -> Exception: + details: Final = self.litellm_logging_obj.model_call_details + provider: Final = details.get("custom_llm_provider") + return with_deployment_debug_info( + exc, + self.litellm_logging_obj.model, + provider if isinstance(provider, str) and provider else "anthropic", + details.get("litellm_params"), + ) async def _fire_detached_failure_hook(self, exc: Exception) -> None: from litellm._logging import verbose_proxy_logger diff --git a/litellm/main.py b/litellm/main.py index 41c4302499e..0f1bddbd5ee 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -79,6 +79,7 @@ from litellm.litellm_core_utils.chat_completion_agentic_loop import ( ) from litellm.litellm_core_utils.completion_timeout import CompletionTimeout, StreamKind from litellm.litellm_core_utils.dd_tracing import tracer +from litellm.litellm_core_utils.exception_mapping_utils import mapped_stream_exception from litellm.litellm_core_utils.get_litellm_params import ( AWS_CREDENTIAL_KWARGS_KEYS, OAUTH_TOKEN_EXCHANGE_KWARGS_KEYS, @@ -817,7 +818,7 @@ def _text_completion_response_from_cached_dict(cached_response_dict: Mapping[str return TextCompletionResponse(stream=bool(cached_response_dict.get("stream", False)), **fields) -async def _async_streaming(response, model, custom_llm_provider, args): +async def _async_streaming(response, model, custom_llm_provider, litellm_params: Mapping[str, object]): try: print_verbose(f"received response in _async_streaming: {response}") if asyncio.iscoroutine(response): @@ -826,11 +827,11 @@ async def _async_streaming(response, model, custom_llm_provider, args): print_verbose(f"line in async streaming: {line}") yield line except Exception as e: - custom_llm_provider = custom_llm_provider or "openai" - raise exception_type( - model=model, - custom_llm_provider=custom_llm_provider, - original_exception=e, + raise mapped_stream_exception( + e, + model if isinstance(model, str) else "", + custom_llm_provider if isinstance(custom_llm_provider, str) and custom_llm_provider else "openai", + litellm_params, ) @@ -7804,7 +7805,7 @@ async def atext_completion(*args, **kwargs) -> TextCompletionResponse | TextComp response=response, model=model, custom_llm_provider=custom_llm_provider, - args=args, + litellm_params=_REQUEST_KWARGS_ADAPTER.validate_python(kwargs), ), model=model, custom_llm_provider=custom_llm_provider, diff --git a/litellm/proxy/pass_through_endpoints/streaming_handler.py b/litellm/proxy/pass_through_endpoints/streaming_handler.py index c94dc16a8f5..55a9cff732e 100644 --- a/litellm/proxy/pass_through_endpoints/streaming_handler.py +++ b/litellm/proxy/pass_through_endpoints/streaming_handler.py @@ -166,6 +166,7 @@ class PassThroughStreamingHandler: passthrough_success_handler_obj: PassThroughEndpointLogging, url_route: str, route_streaming_logging: RouteStreamingLogging | None = None, + on_upstream_error: Callable[[Exception], object] = lambda _: None, ): resolved_route_streaming_logging: Final[RouteStreamingLogging] = ( route_streaming_logging or PassThroughStreamingHandler.route_streaming_logging_to_handler @@ -265,6 +266,7 @@ class PassThroughStreamingHandler: litellm_logging_obj._deferred_stream_complete_args = (_build_logging_coroutine(),) except Exception as e: verbose_proxy_logger.error("Error in chunk_processor: %s", e) + on_upstream_error(e) if response.status_code < 400: logging_scheduled = True await PassThroughStreamingHandler.schedule_stream_failure_logging( diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 6c57bfd2a3f..d8fd1386d8e 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -44,7 +44,7 @@ from litellm.litellm_core_utils.core_helpers import ( provider_reported_usage_cost, set_response_cost_in_hidden_params, ) -from litellm.litellm_core_utils.exception_mapping_utils import with_deployment_api_base +from litellm.litellm_core_utils.exception_mapping_utils import mapped_stream_exception, with_deployment_debug_info from litellm.litellm_core_utils.internal_call_metadata import ( is_background_response, requests_background, @@ -318,27 +318,21 @@ def map_stream_error_to_exception( return _mapped_exception(provider_exception, model, custom_llm_provider, logging_obj) +def _logged_litellm_params(logging_obj: LiteLLMLoggingObj | None) -> object: + return ( + None + if logging_obj is None + else _typed_gets_litellm_params(logging_obj.model_call_details.get)("litellm_params", {}) + ) + + def _mapped_exception( original_exception: Exception, model: str, custom_llm_provider: str, logging_obj: LiteLLMLoggingObj | None, ) -> Exception: - litellm_params: Final = ( - None - if logging_obj is None - else _typed_gets_litellm_params(logging_obj.model_call_details.get)("litellm_params", {}) - ) - try: - return litellm.exception_type( - model=model, - custom_llm_provider=custom_llm_provider, - original_exception=original_exception, - completion_kwargs={}, - extra_kwargs=with_deployment_api_base(litellm_params), - ) - except Exception as mapped_exception: - return mapped_exception + return mapped_stream_exception(original_exception, model, custom_llm_provider, _logged_litellm_params(logging_obj)) def _event_field(event: object, name: str) -> object: @@ -1175,9 +1169,12 @@ class BaseResponsesAPIStreamingIterator: def _raise_for_transport_error(self, error: httpx.ReadError | httpx.RemoteProtocolError) -> NoReturn: self._handle_failure(error) + described: Final = with_deployment_debug_info( + error, self.model or "", self.custom_llm_provider or "", _logged_litellm_params(self.logging_obj) + ) if self._output_started: - raise error - raise self._fallback_error(error) from error + raise described + raise self._fallback_error(described) from described async def call_post_streaming_hooks_for_testing( diff --git a/litellm/utils.py b/litellm/utils.py index 72c37c3760e..4a5b2b4c8a6 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -6008,9 +6008,8 @@ def get_model_region(litellm_params: LiteLLM_Params, mode: str | None) -> str | def get_first_chars_messages(kwargs: dict) -> str: try: - _messages = kwargs.get("messages") - _messages = str(_messages)[:100] - return _messages + messages: Final = kwargs.get("messages") + return "" if messages is None else str(messages)[:100] except Exception: return "" diff --git a/tests/integration/observability/test_stream_error_alert_api_base.py b/tests/integration/observability/test_stream_error_alert_api_base.py index d7f325da8c7..b91cf7e3025 100644 --- a/tests/integration/observability/test_stream_error_alert_api_base.py +++ b/tests/integration/observability/test_stream_error_alert_api_base.py @@ -13,7 +13,15 @@ import openai import pytest import yaml from anthropic.types import RawContentBlockDeltaEvent, TextDelta -from integration._support.anthropic_sse import DROP_PAUSE, dropping_reply, stream_reply +from integration._support.anthropic_sse import ( + CONTENT_BLOCK_START, + DROP_PAUSE, + dropping_reply, + error_body, + message_start, + stream_reply, + text_delta, +) from integration._support.client import ( JSON_OBJECT, Gateway, @@ -30,6 +38,7 @@ from integration._support.responses_ws_peer import PROVIDER_MODEL, Script, event from integration._support.tls import write_self_signed_cert from integration._support.vertex import service_account_json from integration._support.wire import Reply, Request, Wire, wire_server +from openai.types import Completion from openai.types.chat import ChatCompletionChunk from openai.types.responses import ResponseFailedEvent, ResponseTextDeltaEvent from pydantic import JsonValue @@ -39,6 +48,7 @@ pytestmark: Final = [pytest.mark.timeout(2 * graceful_stop_seconds() + 120), pyt AZURE_KEY: Final = "synthetic-azure-key" API_VERSION: Final = "2025-04-01-preview" AZURE_CHAT_BACKEND: Final = "azure/gpt-4o-mini" +ANTHROPIC_BACKEND: Final = "anthropic/claude-opus-5-5" PARTIAL_TEXT: Final = "partial answer" ALERT_SECONDS: Final = 40 NO_RETRY_NO_CACHE: Final[Mapping[str, JsonValue]] = MappingProxyType({"num_retries": 0, "cache": {"no-cache": True}}) @@ -234,6 +244,10 @@ def responses_text(item: object) -> str: return item.delta if isinstance(item, ResponseTextDeltaEvent) else "" +def completion_text(item: object) -> str: + return "".join(choice.text for choice in item.choices) if isinstance(item, Completion) else "" + + def failure_of(item: object) -> str: if isinstance(item, Exception): return str(item) @@ -296,6 +310,19 @@ def responses_outcome(gateway: Gateway, key: str, model: str) -> Outcome: return outcome(items, responses_text) +def completions_outcome(gateway: Gateway, key: str, model: str) -> Outcome: + with openai_client(gateway, key) as client: + items: Final = tuple( + items_or_error( + lambda: client.completions.create( + model=model, prompt="stream an answer", stream=True, extra_body=dict(NO_RETRY_NO_CACHE) + ), + openai.APIError, + ) + ) + return outcome(items, completion_text) + + ENTRYPOINTS: Final[Mapping[str, Callable[[Gateway, str, str], Outcome]]] = MappingProxyType( {"chat": chat_outcome, "messages": messages_outcome, "responses-bridge": responses_outcome} ) @@ -480,3 +507,74 @@ def test_fallback_mid_stream_failure_alert_names_the_fallback_deployment(alertin alert: Final = alert_naming(alerting.alerts, marker) assert_names_deployment(alert, backup_wire.url, backup, AZURE_CHAT_BACKEND, caller) assert primary_wire.url not in alert, alert + + +PING_NEVER_SENT: Final = b"event: ping\ndata: {}\n\n" + + +def anthropic_cut_after_text(marker: str) -> Reply: + return dropping_reply( + ( + message_start(f"msg_{marker}", ANTHROPIC_BACKEND), + CONTENT_BLOCK_START, + text_delta(PARTIAL_TEXT), + PING_NEVER_SENT, + ), + abort_after=3, + ) + + +def responses_cut_after_text(marker: str) -> Reply: + identity: Final = f"resp_{marker}" + return dropping_reply( + (frame(created(identity)), frame(delta(identity, PARTIAL_TEXT)), b": never sent\n\n"), abort_after=2 + ) + + +@dataclass(frozen=True, slots=True) +class CutPath: + backend: str + reply: Callable[[str], Reply] + call: Callable[[Gateway, str, str], Outcome] + + +CUT_PATHS: Final[Mapping[str, CutPath]] = MappingProxyType( + { + "anthropic-chat": CutPath(ANTHROPIC_BACKEND, anthropic_cut_after_text, chat_outcome), + "completions": CutPath(AZURE_CHAT_BACKEND, cut_after_text, completions_outcome), + "native-responses": CutPath(AZURE_CHAT_BACKEND, responses_cut_after_text, responses_outcome), + "native-messages": CutPath(ANTHROPIC_BACKEND, anthropic_cut_after_text, messages_outcome), + } +) + + +def scripted_model(scenario: Scenario, backend: str, api_base: str) -> str: + if backend == AZURE_CHAT_BACKEND: + return azure_chat(scenario, api_base) + return scenario.model(model=backend, api_base=api_base, api_key="synthetic-anthropic-key", num_retries=0) + + +@pytest.mark.parametrize("path", list(CUT_PATHS)) +def test_stream_cut_alert_names_the_deployment(alerting: AlertingProxy, path: str) -> None: + marker: Final = uuid.uuid4().hex + cut: Final = CUT_PATHS[path] + with wire_server(chat_upstream(cut.reply, marker)) as wire, alerting.gateway.scenario() as scenario: + model: Final = scripted_model(scenario, cut.backend, wire.url) + caller: Final = keyed_caller(scenario) + seen: Final = cut.call(alerting.gateway, caller.key, model) + assert seen.text == PARTIAL_TEXT and seen.failure, seen + assert_names_deployment(alert_naming(alerting.alerts, model), wire.url, model, cut.backend, caller) + + +def test_anthropic_rejection_before_stream_alert_names_the_deployment(alerting: AlertingProxy) -> None: + marker: Final = uuid.uuid4().hex + + def rejected(_: str) -> Reply: + return Reply(status=500, body=error_body(500, provider_message(marker))) + + with wire_server(chat_upstream(rejected, marker)) as wire, alerting.gateway.scenario() as scenario: + model: Final = scripted_model(scenario, ANTHROPIC_BACKEND, wire.url) + caller: Final = keyed_caller(scenario) + seen: Final = chat_outcome(alerting.gateway, caller.key, model) + assert provider_message(marker) in seen.failure and not seen.text, seen + assert_names_deployment(alert_naming(alerting.alerts, marker), wire.url, model, ANTHROPIC_BACKEND, caller) diff --git a/tests/unit/litellm_core_utils/test_exception_mapping_utils.py b/tests/unit/litellm_core_utils/test_exception_mapping_utils.py index 55e7c47e2c9..5d9f3af3fda 100644 --- a/tests/unit/litellm_core_utils/test_exception_mapping_utils.py +++ b/tests/unit/litellm_core_utils/test_exception_mapping_utils.py @@ -20,6 +20,7 @@ from litellm.litellm_core_utils.exception_mapping_utils import ( exception_type, extract_and_raise_litellm_exception, ) +from litellm.llms.anthropic.common_utils import AnthropicError from litellm.llms.base_llm.chat.transformation import BaseLLMException from litellm.llms.bedrock.common_utils import BedrockError from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler @@ -156,6 +157,41 @@ def test_api_base_line_keeps_an_at_sign_in_the_path(api_base: str) -> None: assert f"API Base: `{api_base}`" in _debug_info_for_api_base(api_base) +@pytest.mark.parametrize( + ("provider_error", "mapped_class"), + [ + ( + httpx.ReadError("peer closed connection", request=httpx.Request("POST", "https://relay.example/v1/messages")), + litellm.APIConnectionError, + ), + (RuntimeError("stream decoder failed"), litellm.APIConnectionError), + (AnthropicError(status_code=500, message="[Errno 104] Connection reset by peer"), litellm.InternalServerError), + (AnthropicError(status_code=429, message="rate_limit_error"), litellm.RateLimitError), + ], + ids=["unmapped_with_request", "unmapped_without_request", "anthropic_500", "anthropic_429"], +) +def test_mapped_provider_error_carries_the_deployment_debug_info( + provider_error: Exception, mapped_class: type[Exception] +) -> None: + with pytest.raises(mapped_class) as caught: + exception_type( + model="claude-opus-5-5", + original_exception=provider_error, + custom_llm_provider="anthropic", + completion_kwargs={}, + extra_kwargs={ + "api_base": "https://relay.example", + "metadata": {"model_group": "claude-alias", "deployment": "anthropic/claude-opus-5-5"}, + }, + ) + debug_info: Final = caught.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example`" in debug_info + assert "model_group: `claude-alias`" in debug_info + assert "deployment: `anthropic/claude-opus-5-5`" in debug_info + assert "Messages:" not in debug_info + + # Test cases for is_error_str_context_window_exceeded # Tuple format: (error_message, expected_result) context_window_test_cases = [ diff --git a/tests/unit/llms/anthropic/messages/native/test_streaming_iterator.py b/tests/unit/llms/anthropic/messages/native/test_streaming_iterator.py index dd50f60250a..39f7a4a05cb 100644 --- a/tests/unit/llms/anthropic/messages/native/test_streaming_iterator.py +++ b/tests/unit/llms/anthropic/messages/native/test_streaming_iterator.py @@ -645,6 +645,67 @@ async def test_async_sse_wrapper_reraises_upstream_error_to_connected_client(): assert detached_hook.exceptions == [], "the relay re-raised the error, so the proxy failure hook already ran" +_DEPLOYMENT_PARAMS: Final = { + "metadata": { + "api_base": "https://relay.example/anthropic", + "model_group": "claude-alias", + "deployment": "anthropic/claude-opus-5-5", + } +} + + +def _assert_names_the_deployment(error: Exception) -> None: + debug_info: Final = getattr(error, "litellm_debug_info", None) + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/anthropic`" in debug_info + assert "model_group: `claude-alias`" in debug_info + assert "deployment: `anthropic/claude-opus-5-5`" in debug_info + + +@pytest.mark.asyncio +async def test_relayed_upstream_error_reaches_the_client_naming_the_deployment(): + async def _cut_stream(): + yield {"type": "message_start", "message": {"id": "msg_1", "usage": {"input_tokens": 52, "output_tokens": 1}}} + raise _ProviderStreamError("Response payload is not completed", status_code=529) + + iterator = _RecordingLoggingIterator( + litellm_logging_obj=_make_logging_obj("test_upstream_error_names_the_deployment"), + request_body={}, + ) + iterator.litellm_logging_obj.model_call_details["litellm_params"] = _DEPLOYMENT_PARAMS + + with pytest.raises(_ProviderStreamError) as excinfo: + async for _ in iterator.async_sse_wrapper(_cut_stream()): + pass + + assert excinfo.value.status_code == 529 + _assert_names_the_deployment(excinfo.value) + + +@pytest.mark.asyncio +async def test_native_anthropic_stream_cut_reaches_the_client_naming_the_deployment() -> None: + logging_obj: Final = _make_logging_obj("native_anthropic_stream_cut") + logging_obj.model_call_details["litellm_params"] = _DEPLOYMENT_PARAMS + cut: Final = httpx.ReadError("Response payload is not completed") + + async def _cut_upstream() -> AsyncIterator[bytes]: + yield b'event: message_start\ndata: {"type": "message_start"}\n\n' + raise cut + + stream: Final = AnthropicMessagesConfig().get_async_streaming_response_iterator( + model=logging_obj.model, + httpx_response=httpx.Response(200, content=_cut_upstream()), + request_body={"model": logging_obj.model}, + litellm_logging_obj=logging_obj, + ) + + with pytest.raises(httpx.ReadError) as excinfo: + _ = [chunk async for chunk in stream] + + assert excinfo.value is cut + _assert_names_the_deployment(cut) + + @pytest.mark.asyncio async def test_async_sse_wrapper_logs_failure_on_upstream_error_after_disconnect(): """ diff --git a/tests/unit/responses/test_streaming_iterator.py b/tests/unit/responses/test_streaming_iterator.py index b9454ae6c1a..a3bfb87465d 100644 --- a/tests/unit/responses/test_streaming_iterator.py +++ b/tests/unit/responses/test_streaming_iterator.py @@ -365,6 +365,34 @@ def test_sync_transport_read_timeout_raises_litellm_timeout(prefix): assert "Request timed out. Error_str: ReadTimeout" in str(exc_info.value) +@pytest.mark.asyncio +@pytest.mark.parametrize("prefix", _READ_TIMEOUT_PREFIXES) +@pytest.mark.parametrize("error_class", [httpx.ReadError, httpx.RemoteProtocolError]) +async def test_transport_cut_error_names_the_deployment(prefix, error_class): + logging_obj: Final = _logging_obj_stub() + logging_obj.model_call_details = { + "litellm_params": { + "metadata": { + "api_base": "https://relay.example/openai/v1", + "model_group": "responses-alias", + "deployment": "openai/gpt-4o-mini", + } + } + } + trailing_error: Final = error_class("peer closed connection without sending complete message body") + iterator: Final = _make_iterator(sse_events=prefix, logging_obj=logging_obj, trailing_error=trailing_error) + + with pytest.raises((error_class, MidStreamFallbackError)) as exc_info: + async for _ in iterator: + pass + + debug_info: Final = exc_info.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/openai/v1`" in debug_info + assert "model_group: `responses-alias`" in debug_info + assert "deployment: `openai/gpt-4o-mini`" in debug_info + + @pytest.mark.asyncio @pytest.mark.parametrize("trailer", [[], [_DONE_MARKER]], ids=["eof", "done-marker"]) async def test_stream_ending_after_partial_output_without_terminal_event_raises(trailer): diff --git a/tests/unit/text_completions/test_streaming_iterator.py b/tests/unit/text_completions/test_streaming_iterator.py index b7edffc0753..be82131d56b 100644 --- a/tests/unit/text_completions/test_streaming_iterator.py +++ b/tests/unit/text_completions/test_streaming_iterator.py @@ -1,9 +1,11 @@ from collections.abc import AsyncIterator, Iterator from typing import Final +import httpx import pytest from litellm.exceptions import APIConnectionError +from litellm.main import _async_streaming from litellm.text_completions.streaming_iterator import TextCompletionStreamWrapper from litellm.types.utils import Choices, Delta, ModelResponseStream, StreamingChoices, Usage @@ -142,3 +144,30 @@ def test_sync_text_completion_stream_maps_upstream_failure_after_content() -> No next(wrapper) assert (error.value.model, error.value.llm_provider) == ("test-model", "openai") + + +_DEPLOYMENT_PARAMS: Final = { + "api_base": "https://relay.example/v1", + "metadata": {"model_group": "instruct-alias", "deployment": "text-completion-openai/gpt-instruct"}, +} + + +@pytest.mark.asyncio +async def test_async_text_completion_upstream_cut_names_the_deployment() -> None: + async def cut_stream() -> AsyncIterator[ModelResponseStream]: + raise httpx.ReadError("Response payload is not completed") + yield + + wrapper: Final = TextCompletionStreamWrapper( + completion_stream=_async_streaming(cut_stream(), "test-model", None, _DEPLOYMENT_PARAMS), + model="test-model", + ) + + with pytest.raises(APIConnectionError, match="OpenAIException") as error: + await anext(wrapper) + + debug_info: Final = error.value.litellm_debug_info + assert isinstance(debug_info, str) + assert "API Base: `https://relay.example/v1`" in debug_info + assert "model_group: `instruct-alias`" in debug_info + assert "deployment: `text-completion-openai/gpt-instruct`" in debug_info