diff --git a/litellm/router.py b/litellm/router.py index da04e59f18a..7a6e02aa2d9 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -6324,6 +6324,7 @@ class Router: self, response: AsyncIterator[bytes], initial_kwargs: dict[str, object], # mutable-ok: mutated in-place before re-entering the fallback chain + allow_fallback: bool = True, ) -> AsyncIterator[bytes]: """ Wrap an anthropic_messages (/v1/messages) streaming response so a @@ -6479,6 +6480,10 @@ class Router: for buffered_chunk in buffered_lifecycle_chunks: yield buffered_chunk except Exception as stream_error: # noqa: BLE001 # any raised provider error must reach the fallback gate + if not allow_fallback: + if isinstance(stream_error, MidStreamFallbackError) and stream_error.original_exception is not None: + raise stream_error.original_exception from stream_error + raise async for item in self._aanthropic_messages_recover_stream_error( stream_error, has_generated_content, @@ -6830,7 +6835,16 @@ class Router: include_fallback_errors=initial_kwargs.get("include_fallback_errors", False) is True, ) ) - async for fallback_item in self._aanthropic_messages_yield_recovered(fallback_response, wrapper): + recovered: Final = ( + await self._aanthropic_messages_streaming_iterator( + response=cast("AsyncIterator[bytes]", fallback_response), # cast-ok: __aiter__ checked below + initial_kwargs=initial_kwargs, + allow_fallback=False, + ) + if hasattr(fallback_response, "__aiter__") + else fallback_response + ) + async for fallback_item in self._aanthropic_messages_yield_recovered(recovered, wrapper): yield fallback_item except Exception as fallback_error: verbose_router_logger.error("Anthropic messages streaming fallback also failed: %s", fallback_error) diff --git a/tests/unit/test_router/test_router.py b/tests/unit/test_router/test_router.py index 9cc1bf6ce79..480294ee194 100644 --- a/tests/unit/test_router/test_router.py +++ b/tests/unit/test_router/test_router.py @@ -27855,6 +27855,95 @@ class CompletionCustomHandler( self.errors.append(traceback.format_exc()) +@pytest.mark.asyncio +@pytest.mark.parametrize("failure_kind", ["raised", "wrapped" ]) +async def test_anthropic_messages_double_failure_holds_back_fallback_lifecycle_frames(failure_kind): + router = Router( + model_list=[ + {"model_name": name, "litellm_params": {"model": "anthropic/" + name, "api_key": "sk-test"}} + for name in ["primary", "fallback"] + ], + num_retries=0, + fallbacks=[{"primary": ["fallback"]}], + ) + calls = [] + fallback_error = litellm.APIError( + status_code=503, message="fallback also overloaded", llm_provider="anthropic", model="fallback" + ) + + provider_error = ( + MidStreamFallbackError( + message="fallback also overloaded", + model="fallback", + llm_provider="anthropic", + original_exception=fallback_error, + is_pre_first_chunk=True, + ) + if failure_kind == "wrapped" + else fallback_error + ) + + async def provider(**kwargs): + calls.append(kwargs["model"]) + if kwargs["model"] == "anthropic/primary": + return _AnthropicMessagesFakeByteStream([_anthropic_messages_overloaded_error_chunk()]) + return _AnthropicMessagesRaisingByteStream([_anthropic_messages_message_start_chunk()], provider_error) + + stream = await router._aanthropic_messages_with_streaming_fallbacks( + original_function=provider, + model="primary", + stream=True, + messages=[{"role": "user", "content": "hi"}], + max_tokens=10, + ) + chunks = [] + + async def collect_chunks(): + async for chunk in stream: + chunks.append(chunk) + + with pytest.raises(litellm.APIError, match="fallback also overloaded") as caught: + await collect_chunks() + + assert caught.value is fallback_error + + assert chunks == [] + assert calls == ["anthropic/primary", "anthropic/fallback"] + + +@pytest.mark.asyncio +async def test_anthropic_messages_final_fallback_unwraps_provider_error_and_closes_stream(): + router = _anthropic_messages_make_router() + provider_error = litellm.APIError( + status_code=503, message="final fallback overloaded", llm_provider="anthropic", model="fallback" + ) + source = _AnthropicMessagesRaisingByteStream( + [_anthropic_messages_message_start_chunk()], + MidStreamFallbackError( + message="final fallback overloaded", + model="fallback", + llm_provider="anthropic", + original_exception=provider_error, + is_pre_first_chunk=True, + ), + ) + stream = await router._aanthropic_messages_streaming_iterator( + response=source, initial_kwargs={"model": "primary"}, allow_fallback=False + ) + chunks = [] + + async def collect_chunks(): + async for chunk in stream: + chunks.append(chunk) + + with pytest.raises(litellm.APIError) as caught: + await collect_chunks() + + assert caught.value is provider_error + assert chunks == [] + assert source.closed is True + + _FALLBACK_WIRE_PRIMARY: Final = "http://primary.wire.test/v1" _FALLBACK_WIRE_BACKUP: Final = "http://backup.wire.test/v1" _FALLBACK_WIRE_BACKUP_REPLY: Final = {