scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-26 01:12:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 98

fix(router): honor disable_fallbacks on mid-stream fallback (#43111)

Browse source 
* fix(router): honor disable_fallbacks on mid-stream fallback

The mid-stream fallback hop on chat, Responses, and Messages hardcoded
disable_fallbacks=False, so a request or key that opted out of fallbacks
still got a fallback deployment's answer when the primary's stream died
before its first chunk. Each hop now reads the opt-out from the request
the way the pre-stream path does, and the sync chat stream re-raises the
primary's error instead of re-entering the fallback chain

* test(router): prove disable_fallbacks reaches the mid-stream hop through the public entrypoints

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-24 20:06:53 -07:00 • committed by GitHub
parent d0d3b6a67e
commit f3cf1cdfef
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 195 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
litellm/router.py
View file

@ -2940,7 +2940,7 @@ class Router:
self._update_kwargs_before_fallbacks(model=model_group, kwargs=initial_kwargs)
fallback_response = await self.async_function_with_fallbacks_common_utils(
e=e,
disable_fallbacks=False,
disable_fallbacks=fallbacks_disabled_for_request(initial_kwargs),
fallbacks=fallbacks,
context_window_fallbacks=context_window_fallbacks,
content_policy_fallbacks=content_policy_fallbacks,
@ -3384,7 +3384,7 @@ class Router:
)
fallback_response = await self.async_function_with_fallbacks_common_utils(
e=fallback_trigger,
disable_fallbacks=False,
disable_fallbacks=fallbacks_disabled_for_request(initial_kwargs),
fallbacks=fallbacks,
context_window_fallbacks=context_window_fallbacks,
content_policy_fallbacks=content_policy_fallbacks,
@ -3475,8 +3475,9 @@ class Router:
for item in model_response:
yield item
except MidStreamFallbackError as e:
if not e.is_pre_first_chunk and (
e.generated_content or _stream_chunks_have_generated_content(model_response.chunks)
if fallbacks_disabled_for_request(initial_kwargs) or (
not e.is_pre_first_chunk
and (e.generated_content or _stream_chunks_have_generated_content(model_response.chunks))
):
if e.original_exception is not None:
raise e.original_exception from e
@ -5611,7 +5612,7 @@ class Router:
)
fallback_response = await self.async_function_with_fallbacks_common_utils( # rebind-ok: set on success
e=fallback_trigger,
disable_fallbacks=False,
disable_fallbacks=fallbacks_disabled_for_request(initial_kwargs),
fallbacks=fallbacks,
context_window_fallbacks=context_window_fallbacks,
content_policy_fallbacks=content_policy_fallbacks,

189
tests/test_litellm/test_router.py
View file

@ -25,6 +25,7 @@ from litellm import Router
from litellm.caching.caching import DualCache
from litellm.caching.redis_cache import _redis_circuit_breaker_guard
from litellm.exceptions import MidStreamFallbackError
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.integrations.custom_guardrail import CustomGuardrail
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
@ -49,6 +50,7 @@ from litellm.router import (
from litellm.router_strategy import simple_shuffle
from litellm.router_utils.client_initalization_utils import MaxParallelRequestsLimit
from litellm.router_utils.cooldown_handlers import _async_get_cooldown_deployments
from litellm.router_utils.fallback_event_handlers import DISABLE_FALLBACKS_METADATA_KEY
from litellm.router_utils.router_callbacks.track_deployment_metrics import get_deployment_successes_for_current_minute
from litellm.types.llms.openai import ChatCompletionRequest
from litellm.types.router import Deployment, DeploymentTypedDict, LiteLLM_Params, ModelInfo, PreRoutingHookResponse, RetryPolicy
@ -14392,6 +14394,193 @@ async def test_anthropic_messages_fallback_also_catches_raised_midstream_error()
assert mock_fallback.await_args.kwargs["e"] is raised_error
_MID_STREAM_OPT_OUT_SHAPES: Final = (
pytest.param({"disable_fallbacks": True}, id="raw-kwarg"),
pytest.param({"metadata": {DISABLE_FALLBACKS_METADATA_KEY: True}}, id="metadata-stamp"),
pytest.param({"litellm_metadata": {DISABLE_FALLBACKS_METADATA_KEY: True}}, id="litellm_metadata-stamp"),
)
def _mid_stream_opt_out_router() -> Router:
return Router(
model_list=[
{"model_name": "primary", "litellm_params": {"model": "openai/gpt-5.4", "api_key": "k1"}},
{"model_name": "fallback", "litellm_params": {"model": "openai/gpt-5.4-mini", "api_key": "k2"}},
],
fallbacks=[{"primary": ["fallback"]}],
)
def _mid_stream_opt_out_primary_error() -> litellm.InternalServerError:
return litellm.InternalServerError(message="primary failed at stream start", llm_provider="openai", model="primary")
def _mid_stream_opt_out_trigger(primary_error: Exception) -> MidStreamFallbackError:
return MidStreamFallbackError(
message=str(primary_error),
model="primary",
llm_provider="openai",
original_exception=primary_error,
is_pre_first_chunk=True,
)
class _MidStreamOptOutChatStream(CustomStreamWrapper):
"""A chat deployment stream, as the router sees one, that dies before its first chunk."""
def __init__(self, error: Exception, model: str = "primary") -> None:
super().__init__(completion_stream=object(), model=model, custom_llm_provider="openai", logging_obj=MagicMock())
self._error: Final = error
def __aiter__(self):
return self
async def __anext__(self) -> object:
raise self._error
def __iter__(self):
return self
def __next__(self) -> object:
raise self._error
@pytest.mark.asyncio
@pytest.mark.parametrize("opt_out", _MID_STREAM_OPT_OUT_SHAPES)
async def test_acompletion_streaming_iterator_honors_disable_fallbacks(opt_out):
"""A chat stream that fails before its first chunk on a request that opted out of fallbacks
surfaces the primary's own error and never tries the fallback deployment."""
router = _mid_stream_opt_out_router()
primary_error = _mid_stream_opt_out_primary_error()
source = _MidStreamOptOutChatStream(_mid_stream_opt_out_trigger(primary_error))
with patch.object(router, "_acompletion", new=AsyncMock(return_value=_AsyncList([]))) as fallback_attempt:
wrapped = await router._acompletion_streaming_iterator(
model_response=source,
messages=[{"role": "user", "content": "Hi"}],
initial_kwargs={"model": "primary", "stream": True, **copy.deepcopy(opt_out)},
)
with pytest.raises(litellm.InternalServerError) as raised:
[chunk async for chunk in wrapped]
assert raised.value is primary_error
fallback_attempt.assert_not_awaited()
@pytest.mark.parametrize("opt_out", _MID_STREAM_OPT_OUT_SHAPES)
def test_completion_streaming_iterator_honors_disable_fallbacks(opt_out):
"""Sync counterpart of test_acompletion_streaming_iterator_honors_disable_fallbacks."""
router = _mid_stream_opt_out_router()
primary_error = _mid_stream_opt_out_primary_error()
source = _MidStreamOptOutChatStream(_mid_stream_opt_out_trigger(primary_error))
with patch.object(router, "_completion", new=MagicMock(return_value=iter([]))) as fallback_attempt:
wrapped = router._completion_streaming_iterator(
model_response=source,
messages=[{"role": "user", "content": "Hi"}],
initial_kwargs={"model": "primary", "stream": True, **copy.deepcopy(opt_out)},
)
with pytest.raises(litellm.InternalServerError) as raised:
list(wrapped)
assert raised.value is primary_error
fallback_attempt.assert_not_called()
@pytest.mark.asyncio
@pytest.mark.parametrize("opt_out", _MID_STREAM_OPT_OUT_SHAPES)
async def test_aresponses_streaming_iterator_honors_disable_fallbacks(opt_out):
"""Same opt-out contract on the Responses API mid-stream fallback path."""
router = _mid_stream_opt_out_router()
primary_error = _mid_stream_opt_out_primary_error()
source = _make_responses_iterator(error=_mid_stream_opt_out_trigger(primary_error), model="primary")
with patch.object(
router,
"_ageneric_api_call_with_fallbacks_responses_attempt",
new=AsyncMock(return_value=_AsyncList([])),
) as fallback_attempt:
wrapped = await router._aresponses_streaming_iterator(
response=source,
initial_kwargs={
"model": "primary",
"stream": True,
"input": "Hi",
"original_generic_function": litellm.aresponses,
**copy.deepcopy(opt_out),
},
)
with pytest.raises(litellm.InternalServerError) as raised:
[chunk async for chunk in wrapped]
assert raised.value is primary_error
fallback_attempt.assert_not_awaited()
@pytest.mark.asyncio
@pytest.mark.parametrize("opt_out", _MID_STREAM_OPT_OUT_SHAPES)
async def test_anthropic_messages_streaming_iterator_honors_disable_fallbacks(opt_out):
"""Same opt-out contract on the Anthropic Messages mid-stream fallback path."""
router = _mid_stream_opt_out_router()
primary_error = _mid_stream_opt_out_primary_error()
source = _AnthropicMessagesRaisingByteStream([], _mid_stream_opt_out_trigger(primary_error))
with patch.object(
router,
"_ageneric_api_call_with_fallbacks_anthropic_messages_attempt",
new=AsyncMock(return_value=_AnthropicMessagesFallbackByteStream([])),
) as fallback_attempt:
wrapped = await router._aanthropic_messages_streaming_iterator(
response=source,
initial_kwargs={"model": "primary", "stream": True, **copy.deepcopy(opt_out)},
)
with pytest.raises(litellm.InternalServerError) as raised:
[chunk async for chunk in wrapped]
assert raised.value is primary_error
fallback_attempt.assert_not_awaited()
@pytest.mark.asyncio
async def test_acompletion_disable_fallbacks_reaches_the_mid_stream_hop():
"""`disable_fallbacks=True` sent to the public entrypoint survives the fallback wrapper's handoff
into the stream: the primary's own error surfaces and no fallback deployment is ever called."""
router = _mid_stream_opt_out_router()
primary_error = _mid_stream_opt_out_primary_error()
async def primary_stream(**kwargs):
return _MidStreamOptOutChatStream(_mid_stream_opt_out_trigger(primary_error), model=kwargs["model"])
with patch("litellm.acompletion", side_effect=primary_stream) as provider_calls:
response = await router.acompletion(
model="primary", messages=[{"role": "user", "content": "Hi"}], stream=True, disable_fallbacks=True
)
with pytest.raises(litellm.InternalServerError) as raised:
[chunk async for chunk in response]
assert raised.value is primary_error
assert [call.kwargs["metadata"]["model_group"] for call in provider_calls.call_args_list] == ["primary"]
def test_completion_disable_fallbacks_reaches_the_mid_stream_hop():
"""Sync counterpart of test_acompletion_disable_fallbacks_reaches_the_mid_stream_hop."""
router = _mid_stream_opt_out_router()
primary_error = _mid_stream_opt_out_primary_error()
def primary_stream(**kwargs):
return _MidStreamOptOutChatStream(_mid_stream_opt_out_trigger(primary_error), model=kwargs["model"])
with patch("litellm.completion", side_effect=primary_stream) as provider_calls:
response = router.completion(
model="primary", messages=[{"role": "user", "content": "Hi"}], stream=True, disable_fallbacks=True
)
with pytest.raises(litellm.InternalServerError) as raised:
list(response)
assert raised.value is primary_error
assert [call.kwargs["metadata"]["model_group"] for call in provider_calls.call_args_list] == ["primary"]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"raised_error",