From 8fb2b97d5b2b09d0265122c8120951548f017d1d Mon Sep 17 00:00:00 2001 From: joshua-berri Date: Tue, 10 Nov 2026 12:55:49 -0800 Subject: [PATCH] fix(router): sync responses() applies the Router's num_retries and fallbacks (#51011) * fix(router): sync responses() applies the Router's num_retries and fallbacks Sync Router.responses() made one attempt and raised, while aresponses() retried and fell back on the same Router. The sync plan now walks retries and fallbacks like the async one, stamps the deployment's num_retries on the error the same way, and the SDK no longer nests its own retries under the Router's for sync generic calls Fixes #49938 * fix(router): sync generate_content() retries and falls back through the Router Reading litellm_metadata in the SDK retry check left sync generate_content with no retries at all, since its Router plan made one attempt and the inner completion() call stopped retrying. Sync generate_content and generate_content_stream now walk the Router's retries and fallbacks like their async versions, so retries are neither lost nor nested --------- Co-authored-by: Joshua Valluru <326636767+joshua-berri@users.noreply.github.com> --- litellm/router_rust/invoke/endpoints.py | 22 ++++-- litellm/utils.py | 16 ++--- ...test_endpoints_characterization_generic.py | 2 +- .../core/test_parity_regress5_errors.py | 43 +++++++++++- tests/unit/test_utils.py | 68 +++++++++++++------ 5 files changed, 111 insertions(+), 40 deletions(-) diff --git a/litellm/router_rust/invoke/endpoints.py b/litellm/router_rust/invoke/endpoints.py index d728583cb31..a5307ff8af9 100644 --- a/litellm/router_rust/invoke/endpoints.py +++ b/litellm/router_rust/invoke/endpoints.py @@ -293,7 +293,7 @@ class Plan: return get_router_metadata_variable_name(function_name=self.function_name) -def _generic(endpoint: str, host_op: str | None = None, *, sync: bool = False) -> Plan: +def _generic(endpoint: str, host_op: str | None = None, *, sync: bool = False, single_attempt: bool = True) -> Plan: if sync: return Plan( endpoint=endpoint, @@ -303,9 +303,9 @@ def _generic(endpoint: str, host_op: str | None = None, *, sync: bool = False) - prepare_field="litellm_metadata", function_name="generic_api_call", splits_stream_timeout=_handler_name(host_op or endpoint) in STREAM_TIMEOUT_AWARE_GENERIC_HANDLERS, - stamp="none", + stamp="none" if single_attempt else "deployment", fail_key="group", - single_attempt=True, + single_attempt=single_attempt, ) return Plan( endpoint=endpoint, @@ -483,7 +483,7 @@ PLANS: Final[Mapping[str, Plan]] = MappingProxyType( "aanthropic_messages": _generic("aanthropic_messages", "anthropic_messages"), "anthropic_messages": _generic("anthropic_messages", "anthropic_messages"), "aresponses": _generic("aresponses"), - "responses": _generic("responses", sync=True), + "responses": _generic("responses", sync=True, single_attempt=False), } ) @@ -583,6 +583,8 @@ _SYNC_GENERIC: Final[Mapping[str, str]] = MappingProxyType( } ) +_ROUTER_RETRIED_SYNC_GENERIC: Final = frozenset({"generate_content", "generate_content_stream"}) + _CONTAINER_FILE_ROUTED: Final = frozenset( { "aupload_container_file", @@ -708,7 +710,12 @@ def generic_plan(endpoint: str) -> Plan | None: if endpoint in _VIDEO_DIRECT: return _generic(endpoint, f"videos.{endpoint}") if endpoint in _SYNC_GENERIC: - return _generic(endpoint, _SYNC_GENERIC[endpoint], sync=True) + return _generic( + endpoint, + _SYNC_GENERIC[endpoint], + sync=True, + single_attempt=endpoint not in _ROUTER_RETRIED_SYNC_GENERIC, + ) return None @@ -1658,6 +1665,7 @@ class RustEndpoints: def _sync_factory(self, name: str, plan: Plan) -> Callable[..., object]: is_vector_store: Final = name.startswith("vector_store_") + attempts: Final = _SINGLE_ATTEMPT if plan.single_attempt else None def sync_wrapper( custom_llm_provider: str | None = None, @@ -1669,7 +1677,7 @@ class RustEndpoints: raise TypeError( "_generic_api_call_with_fallbacks() missing 1 required positional argument: 'model'" ) - return self._core.call_sync(plan, kwargs, _SINGLE_ATTEMPT) + return self._core.call_sync(plan, kwargs, attempts) provider_kwargs: dict[ # mutable-ok: the request kwargs the Python router mutates in place # rebind-ok: rebound as the Python router rebinds it str, object ] = ( @@ -1683,7 +1691,7 @@ class RustEndpoints: "_direct_vector_store_embedding_executor": self._embedding_executor(kwargs), } if provider_kwargs.get("model"): - return self._core.call_sync(plan, provider_kwargs, _SINGLE_ATTEMPT) + return self._core.call_sync(plan, provider_kwargs, attempts) if name == "vector_store_search": return self._invoke(plan.host_op)(**{**provider_kwargs, "router": self._core.public}) return self._invoke(plan.host_op)(**provider_kwargs) diff --git a/litellm/utils.py b/litellm/utils.py index 717b1b9009a..a69d3c063de 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -2060,11 +2060,9 @@ def _get_call_type_from_public_name(function_name: str) -> str: return function_name -def _is_litellm_router_call(kwargs: Mapping[str, object], *, is_async: bool) -> bool: - """Router completion uses metadata. Async generic calls retry with litellm_metadata; sync generic calls need SDK retries.""" - metadata_buckets: Final = ( - (kwargs.get("metadata"), kwargs.get("litellm_metadata")) if is_async else (kwargs.get("metadata"),) - ) +def _is_litellm_router_call(kwargs: Mapping[str, object]) -> bool: + """Router completion stamps the group on metadata, generic endpoints on litellm_metadata.""" + metadata_buckets: Final = (kwargs.get("metadata"), kwargs.get("litellm_metadata")) return any(isinstance(bucket, Mapping) and "model_group" in bucket for bucket in metadata_buckets) @@ -2322,7 +2320,7 @@ def client(original_function): litellm.num_retries = None # set retries to None to prevent infinite loops context_window_fallback_dict: Final = kwargs.get("context_window_fallback_dict", {}) - is_completion_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=False) + is_completion_litellm_router_call: Final = _is_litellm_router_call(kwargs) if ( num_retries and not is_completion_litellm_router_call @@ -2360,7 +2358,7 @@ def client(original_function): if not is_resent_failure(e): litellm.num_retries = None # set retries to None to prevent infinite loops - is_responses_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=False) + is_responses_litellm_router_call: Final = _is_litellm_router_call(kwargs) if ( num_retries and not is_responses_litellm_router_call @@ -2664,7 +2662,7 @@ def client(original_function): if retry_call_type == CallTypes.acompletion.value: context_window_fallback_dict: Final = kwargs.get("context_window_fallback_dict", {}) - is_acompletion_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=True) + is_acompletion_litellm_router_call: Final = _is_litellm_router_call(kwargs) if ( num_retries @@ -2700,7 +2698,7 @@ def client(original_function): result = await original_function(*args, **kwargs) return result elif retry_call_type == CallTypes.aresponses.value: - is_aresponses_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=True) + is_aresponses_litellm_router_call: Final = _is_litellm_router_call(kwargs) if ( num_retries diff --git a/tests/unit/test_router/core/test_endpoints_characterization_generic.py b/tests/unit/test_router/core/test_endpoints_characterization_generic.py index b1a356cf8eb..f7669bff80c 100644 --- a/tests/unit/test_router/core/test_endpoints_characterization_generic.py +++ b/tests/unit/test_router/core/test_endpoints_characterization_generic.py @@ -195,7 +195,7 @@ async def test_sync_generic_endpoints_set_no_routing_hidden_params( assert response._hidden_params == {} -@pytest.mark.parametrize("endpoint", ["ocr", "generate_content"]) +@pytest.mark.parametrize("endpoint", ["ocr", "video_generation"]) async def test_sync_generic_endpoints_are_not_retried(make_core: MakeCore, host: EndpointHost, endpoint: str): host.answer("openai/d1", litellm.InternalServerError(message="boom", llm_provider="openai", model="d1")) _, request = ROUTED_SYNC[endpoint] diff --git a/tests/unit/test_router/core/test_parity_regress5_errors.py b/tests/unit/test_router/core/test_parity_regress5_errors.py index 3e8537af878..bfeca17b3a2 100644 --- a/tests/unit/test_router/core/test_parity_regress5_errors.py +++ b/tests/unit/test_router/core/test_parity_regress5_errors.py @@ -114,12 +114,51 @@ async def test_a_providers_bad_request_falls_back_and_notes_the_failed_fallback( def test_a_single_attempt_endpoint_raises_with_no_note(make_core: MakeCore, host: EndpointHost) -> None: - core: Final = build(make_core, fallbacks=[{"a": ["b"]}]) + core: Final = build(make_core, fallbacks=[{"a": ["b"]}], num_retries=1) show_router_debug(host) host.answer("openai/a", server_error()) with pytest.raises(litellm.InternalServerError) as raised: - core.responses(model="a", input="hi") # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType] # an endpoint + core.ocr(model="a", document={}) # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType] # an endpoint assert NOTE not in str(raised.value) + assert host.deployments_called() == ["a"] + + +ROUTER_RETRIED: Final[dict[str, tuple[str, str, Mapping[str, object]]]] = { + "responses": ("responses", "aresponses", {"input": "hi"}), + "generate_content": ("generate_content", "agenerate_content", {"contents": []}), +} + + +async def call_routed(core: RouterCore, endpoint: str, mode: str) -> object: + sync_name, async_name, request = ROUTER_RETRIED[endpoint] + if mode == "sync": + return getattr(core, sync_name)(model="a", **request) + return await getattr(core, async_name)(model="a", **request) + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("endpoint", sorted(ROUTER_RETRIED)) +async def test_a_router_retried_endpoint_retries_a_failed_attempt_with_the_router_num_retries( + make_core: MakeCore, host: EndpointHost, endpoint: str, mode: str +) -> None: + core: Final = build(make_core, num_retries=1) + recovered: Final = completion_response(content="recovered") + host.answer("openai/a", server_error(), recovered) + assert await call_routed(core, endpoint, mode) is recovered + assert host.deployments_called() == ["a", "a"] + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +@pytest.mark.parametrize("endpoint", sorted(ROUTER_RETRIED)) +async def test_a_router_retried_endpoint_falls_back_to_the_next_group_when_its_group_fails( + make_core: MakeCore, host: EndpointHost, endpoint: str, mode: str +) -> None: + core: Final = build(make_core, fallbacks=[{"a": ["b"]}]) + fallback: Final = completion_response(content="from b") + host.answer("openai/a", server_error()) + host.answer("openai/b", fallback) + assert await call_routed(core, endpoint, mode) is fallback + assert host.deployments_called() == ["a", "b"] async def test_an_empty_request_list_says_no_fallbacks_are_configured(make_core: MakeCore, host: EndpointHost) -> None: diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 045d78fc1e7..da8c3c6a451 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -5706,20 +5706,17 @@ async def test_wrapper_async_logs_converted_responses_stream_with_standard_loggi @pytest.mark.parametrize( - "kwargs, is_async, expected", + "kwargs, expected", [ - ({"metadata": {"model_group": "g"}}, False, True), - ({"metadata": {"model_group": "g"}}, True, True), - ({"litellm_metadata": {"model_group": "g"}}, False, False), - ({"litellm_metadata": {"model_group": "g"}}, True, True), - ({}, False, False), - ({}, True, False), - ({"metadata": None}, False, False), - ({"metadata": None}, True, False), + ({"metadata": {"model_group": "g"}}, True), + ({"litellm_metadata": {"model_group": "g"}}, True), + ({"metadata": {"user": "u"}, "litellm_metadata": {"user": "u"}}, False), + ({}, False), + ({"metadata": None}, False), ], ) -def test_is_litellm_router_call_is_async_aware(kwargs: Mapping[str, object], is_async: bool, expected: bool) -> None: - assert _is_litellm_router_call(kwargs, is_async=is_async) is expected +def test_is_litellm_router_call_reads_both_metadata_buckets(kwargs: Mapping[str, object], expected: bool) -> None: + assert _is_litellm_router_call(kwargs) is expected @pytest.mark.asyncio @@ -5758,11 +5755,24 @@ async def test_router_aresponses_does_not_run_sdk_retries(monkeypatch: pytest.Mo litellm.in_memory_llm_clients_cache.flush_cache() -def test_router_responses_keeps_sdk_retries_for_sync_router_call( - monkeypatch: pytest.MonkeyPatch, -) -> None: +SYNC_ROUTER_RETRIED_CALLS: Final[dict[str, tuple[str, Callable[[litellm.Router], object]]]] = { + "responses": ("/v1/responses", lambda router: router.responses(model="responses-retry", input="hi")), + "generate_content": ( + "/v1/chat/completions", + lambda router: router.generate_content( + model="responses-retry", contents=[{"role": "user", "parts": [{"text": "hi"}]}] + ), + ), +} + + +def _count_sync_router_attempts( + monkeypatch: pytest.MonkeyPatch, endpoint: str, router_num_retries: int, deployment_num_retries: int | None +) -> int: + path, call = SYNC_ROUTER_RETRIED_CALLS[endpoint] monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) litellm.in_memory_llm_clients_cache.flush_cache() + retries: Final = {} if deployment_num_retries is None else {"num_retries": deployment_num_retries} model_list: Final = [ { "model_name": "responses-retry", @@ -5770,15 +5780,16 @@ def test_router_responses_keeps_sdk_retries_for_sync_router_call( "model": "openai/gpt-4o-mini", "api_key": "sk-test", "api_base": "https://responses-retry.local/v1", - "num_retries": 2, + **retries, }, } ] - router: Final = litellm.Router(model_list=model_list, num_retries=0, retry_after=0, disable_cooldowns=True) - + router: Final = litellm.Router( + model_list=model_list, num_retries=router_num_retries, retry_after=0, disable_cooldowns=True + ) try: with respx.mock(assert_all_called=True) as respx_mock: - upstream: Final = respx_mock.post("https://responses-retry.local/v1/responses").mock( + upstream: Final = respx_mock.post(f"https://responses-retry.local{path}").mock( return_value=httpx.Response( 503, headers={"retry-after": "0"}, @@ -5786,14 +5797,29 @@ def test_router_responses_keeps_sdk_retries_for_sync_router_call( ) ) with pytest.raises(litellm.ServiceUnavailableError): - router.responses(model="responses-retry", input="hi") - - assert upstream.call_count == 3 + call(router) + return upstream.call_count finally: router.discard() litellm.in_memory_llm_clients_cache.flush_cache() +@pytest.mark.parametrize("router_num_retries", [0, 1]) +@pytest.mark.parametrize("endpoint", sorted(SYNC_ROUTER_RETRIED_CALLS)) +def test_sync_router_call_retries_as_often_as_the_deployment_num_retries( + monkeypatch: pytest.MonkeyPatch, router_num_retries: int, endpoint: str +) -> None: + assert _count_sync_router_attempts(monkeypatch, endpoint, router_num_retries, deployment_num_retries=2) == 3 + + +@pytest.mark.parametrize("endpoint", sorted(SYNC_ROUTER_RETRIED_CALLS)) +def test_sync_router_call_does_not_nest_the_global_num_retries_under_router_retries( + monkeypatch: pytest.MonkeyPatch, endpoint: str +) -> None: + monkeypatch.setattr(litellm, "num_retries", 2) + assert _count_sync_router_attempts(monkeypatch, endpoint, router_num_retries=1, deployment_num_retries=None) == 2 + + @pytest.mark.asyncio async def test_aresponses_uses_sdk_retries_without_router(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)