diff --git a/litellm/router.py b/litellm/router.py index f0bb6c3a7a9..ef39f5bd983 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -14439,7 +14439,6 @@ class Router: Filter out model in model group, if: - model context window < message length. For azure openai models, requires 'base_model' is set. - https://docs.litellm.ai/docs/proxy/cost_tracking#spend-tracking-for-azure-openai-models - - filter models above rpm limits - if region given, filter out models not in that region / unknown region - [TODO] function call and model doesn't support function calling """ @@ -14454,8 +14453,6 @@ class Router: _context_window_error = False _potential_error_str = "" - _rate_limit_error = False - parent_otel_span: Final = get_parent_otel_span_from_kwargs(request_kwargs) has_countable_input: Final = (messages is not None or input is not None) and not compaction_pending( request_kwargs @@ -14479,13 +14476,6 @@ class Router: else frozenset() ) - ## get model group RPM ## - dt: Final = get_utc_datetime() - current_minute: Final = dt.strftime("%H-%M") - rpm_key: Final = f"{model}:rpm:{current_minute}" - model_group_cache: Final = ( - self.cache.get_cache(key=rpm_key, local_only=True, parent_otel_span=parent_otel_span) or {} - ) # check the in-memory cache used by lowest_latency and usage-based routing. Only check the local cache. for idx, deployment in enumerate(_returned_deployments): # Cache nested dict access to avoid repeated temporary dict allocations _litellm_params = deployment.get("litellm_params", {}) @@ -14508,24 +14498,6 @@ class Router: ) continue - model_id = _model_info.get("id", "") - ## RPM CHECK ## - ### get local router cache ### - current_request_cache_local = ( - self.cache.get_cache(key=model_id, local_only=True, parent_otel_span=parent_otel_span) or 0 - ) - ### get usage based cache ### - if isinstance(model_group_cache, dict) and self.routing_strategy != "usage-based-routing-v2": - model_group_cache[model_id] = model_group_cache.get(model_id, 0) - - current_request = max(current_request_cache_local, model_group_cache[model_id]) - - if isinstance(_litellm_params, dict) and _litellm_params.get("rpm", None) is not None: - if isinstance(_litellm_params["rpm"], int) and _litellm_params["rpm"] <= current_request: - invalid_model_indices.add(idx) - _rate_limit_error = True - continue - ## REGION CHECK ## if request_kwargs is not None and request_kwargs.get("allowed_model_region") is not None: allowed_model_region = request_kwargs.get("allowed_model_region") @@ -14562,12 +14534,6 @@ class Router: ) if len(excluded_indices) == len(_returned_deployments): - """ - - no healthy deployments available b/c context window checks or rate limit error - - - First check for rate limit errors (if this is true, it means the model passed the context window check but failed the rate limit check) - """ - group_deployments: Final = ( tuple(request_deployments) if request_deployments is not None @@ -14589,12 +14555,7 @@ class Router: ), model=model, ) - if _rate_limit_error is True: # allow generic fallback logic to take place - raise RouterRateLimitErrorBasic( - model=model, - ) - - elif _context_window_error is True: + if _context_window_error is True: raise litellm.ContextWindowExceededError( message=f"litellm._pre_call_checks: Context Window exceeded for given call. No models have context window large enough for this call.\n{_potential_error_str}", model=model, diff --git a/tests/integration/routing/test_end_user_region_routing.py b/tests/integration/routing/test_end_user_region_routing.py index ce90280f173..884b61228cc 100644 --- a/tests/integration/routing/test_end_user_region_routing.py +++ b/tests/integration/routing/test_end_user_region_routing.py @@ -19,6 +19,9 @@ UNCOUNTABLE_MODEL: Final = "regional-uncountable-model" PINNED_REGION: Final = "eu" WEIGHT_BY_REGION: Final = {"eu": 1, "us": 99} SILENT_WAV: Final = "UklGRiQAAABXQVZFZm10IBAAAAABAAEAQB8AAIA+AAACABAAZGF0YQAAAAA=" +RPM_CAPPED_MODEL: Final = "regional-rpm-capped-model" +RPM_CAPPED_DEPLOYMENTS: Final = frozenset(f"{RPM_CAPPED_MODEL}-{region}" for region in UPSTREAM_BY_REGION) +RPM_CAPPED_CALLS: Final = 3 @pytest.fixture(scope="module") @@ -53,6 +56,19 @@ def candidate(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Gateway]: } for region, upstream in UPSTREAM_BY_REGION.items() ), + *( + { + "model_name": RPM_CAPPED_MODEL, + "litellm_params": { + "model": f"openai/{upstream}", + "api_base": f"{base.upstream_url}/v1", + "api_key": "synthetic-region-key", + "rpm": 1, + }, + "model_info": {"id": f"{RPM_CAPPED_MODEL}-{region}"}, + } + for region, upstream in UPSTREAM_BY_REGION.items() + ), ] config["router_settings"] = {**config["router_settings"], "enable_pre_call_checks": True} path: Final = directory / "region-routing.yaml" @@ -143,3 +159,16 @@ def test_an_uncountable_prompt_still_pins_an_end_user_to_their_region( observed: Final = upstream.get("/__observations").json()["requests"] pinned_upstream: Final = UPSTREAM_BY_REGION[PINNED_REGION] assert [request["body"]["model"] for request in observed] == [pinned_upstream] * recorded_chat_calls + + +def test_pre_call_checks_keep_serving_deployments_past_their_rpm(candidate: Gateway) -> None: + with candidate.scenario() as scenario: + key: Final = scenario.key(models=[RPM_CAPPED_MODEL]) + body: Final = {"model": RPM_CAPPED_MODEL, "messages": [{"role": "user", "content": "integration control"}]} + responses: Final = tuple( + candidate.request("POST", "/v1/chat/completions", body, key=key) for _ in range(RPM_CAPPED_CALLS) + ) + assert [response.status_code for response in responses] == [200] * RPM_CAPPED_CALLS, [ + response.text for response in responses + ] + assert {response.headers.get("x-litellm-model-id") for response in responses} <= RPM_CAPPED_DEPLOYMENTS diff --git a/tests/unit/test_router/test_router.py b/tests/unit/test_router/test_router.py index c6d02dddae0..4cb994744f2 100644 --- a/tests/unit/test_router/test_router.py +++ b/tests/unit/test_router/test_router.py @@ -27633,38 +27633,89 @@ def test_router_get_model_group_info() -> None: assert model_group_info.output_cost_per_token > 0 -def test_router_rpm_pre_call_check(): - """ - - for a given model not in model cost map - - with rpm set - - check if rpm check is run - """ - try: - model_list = [ +def test_pre_call_checks_keep_deployment_whose_rpm_is_reached(): + model_list: Final = [ + { + "model_name": "fake-openai-endpoint", + "litellm_params": {"model": "openai/my-fake-model", "api_key": "my-fake-key", "rpm": 0}, + "model_info": {"id": "rpm-zero"}, + }, + { + "model_name": "fake-openai-endpoint", + "litellm_params": {"model": "openai/my-fake-model", "api_key": "my-fake-key"}, + "model_info": {"id": "uncapped"}, + }, + ] + router: Final = Router(model_list=model_list, enable_pre_call_checks=True, num_retries=0) + + kept: Final = router._pre_call_checks( + model="fake-openai-endpoint", + healthy_deployments=model_list, + messages=[{"role": "user", "content": "Hey, how's it going?"}], + ) + + assert [deployment["model_info"]["id"] for deployment in kept] == ["rpm-zero", "uncapped"] + + +@pytest.mark.asyncio +async def test_enable_pre_call_checks_alone_serves_past_deployment_rpm(): + router: Final = Router( + model_list=[ { - "model_name": "fake-openai-endpoint", # openai model name - "litellm_params": { # params for litellm completion/embedding call - "model": "openai/my-fake-model", - "api_key": "my-fake-key", - "api_base": "https://openai-function-calling-workers.tasslexyz.workers.dev/", - "rpm": 0, - }, - }, - ] + "model_name": "fake-openai-endpoint", + "litellm_params": {"model": "openai/my-fake-model", "api_key": "my-fake-key", "rpm": 1}, + "model_info": {"id": "rpm-one"}, + } + ], + enable_pre_call_checks=True, + num_retries=0, + ) - router = Router(model_list=model_list, set_verbose=True, enable_pre_call_checks=True, num_retries=0) # type: ignore + served: Final = [ + await router.acompletion( + model="fake-openai-endpoint", + messages=[{"role": "user", "content": "Hey, how's it going?"}], + mock_response="served", + ) + for _ in range(3) + ] - try: - router._pre_call_checks( + assert [response.choices[0].message.content for response in served] == ["served"] * 3 + + +@pytest.mark.asyncio +async def test_enforce_model_rate_limits_returns_429_at_deployment_rpm(): + frozen_minute: Final = datetime(2026, 1, 1, 12, 30, 15, tzinfo=timezone.utc) + router: Final = Router( + model_list=[ + { + "model_name": "fake-openai-endpoint", + "litellm_params": {"model": "openai/my-fake-model", "api_key": "my-fake-key", "rpm": 1}, + "model_info": {"id": "rpm-one"}, + } + ], + enable_pre_call_checks=True, + optional_pre_call_checks=["enforce_model_rate_limits"], + num_retries=0, + ) + + with patch( + "litellm.router_utils.pre_call_checks.model_rate_limit_check.get_utc_datetime", return_value=frozen_minute + ): + first: Final = await router.acompletion( + model="fake-openai-endpoint", + messages=[{"role": "user", "content": "Hey, how's it going?"}], + mock_response="served", + ) + with pytest.raises(litellm.RateLimitError) as rate_limited: + await router.acompletion( model="fake-openai-endpoint", - healthy_deployments=model_list, messages=[{"role": "user", "content": "Hey, how's it going?"}], + mock_response="served", ) - pytest.fail("Expected this to fail") - except Exception: - pass - except Exception as e: - pytest.fail(f"Got unexpected exception on router! - {str(e)}") + + assert first.choices[0].message.content == "served" + assert rate_limited.value.status_code == 429 @pytest.mark.parametrize( @@ -29494,26 +29545,6 @@ def test_pre_call_checks_filters_deployment_when_token_count_exceeds_limit(): assert [d["model_info"]["id"] for d in filtered] == ["large"] -def test_pre_call_checks_rpm_check_still_enforced_when_token_counting_fails(): - from litellm.types.router import RouterRateLimitErrorBasic - - model_list: Final = [ - { - "model_name": "fake-openai-endpoint", - "litellm_params": {"model": "openai/my-fake-model", "api_key": "my-fake-key", "rpm": 0}, - "model_info": {"max_input_tokens": 100}, - }, - ] - router: Final = Router(model_list=model_list, enable_pre_call_checks=True, num_retries=0) - - with pytest.raises(RouterRateLimitErrorBasic): - router._pre_call_checks( - model="fake-openai-endpoint", - healthy_deployments=model_list, - messages=_PRE_CALL_CHECK_UNPARSEABLE_MESSAGES, - ) - - def test_pre_call_checks_context_window_check_is_skipped_but_not_fatal_when_token_counting_fails(): model_list: Final = [ {