mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
fix(router): drop the deployment rpm filter from enable_pre_call_checks (#48873)
enable_pre_call_checks filtered deployments whose in-memory minute count
reached litellm_params.rpm, undocumented since 0d1cca9aa0. Deployment rpm
and tpm stay routing weights unless optional_pre_call_checks includes
enforce_model_rate_limits, which is now the only deployment rate-limit path.
Resolves LIT-10233
Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu
GitHub
parent
43ce3e22aa
commit
ca4a2ba006
3 changed files with 107 additions and 86 deletions
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 = [
|
||||
{
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue