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)