feat(router): add deployment rpd and tpd daily limits to enforce_model_rate_limits (#48875)

* feat(router): add deployment rpd and tpd daily limits to enforce_model_rate_limits

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(router): route rate limit cache increments through typed helpers

* chore(ui): regenerate schema.d.ts for deployment rpd and tpd fields

* test(types): add rpd and tpd to the owned litellm_params inventory

* fix(router): run daily rpd/tpd checks after the minute checks and increment rpm and rpd in one cache trip

* fix(router): flatten the async rpm reject branch for the ruff strict gate

* fix(router): hand rpm and rpd back on any rate limit reject

A request the rpm, rpd or tpd check rejects no longer keeps its request counts, so retries against a deployment whose rpd is spent cannot fill its rpm and block the in-group fallover with the minute 429.

* test(router): a minute-rejected request no longer keeps its rpm count

---------

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu 2026-10-28 15:52:44 -07:00 • committed by GitHub
parent 752b5c053a
commit 0f786fa1c7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 589 additions and 70 deletions

View file

@ -274,7 +274,9 @@ class InMemoryCache(BaseCache):
) -> list[float] | None:
results: Final = []
for increment in increment_list:
result = await self.async_increment(increment["key"], increment["increment_value"], **kwargs)
result = await self.async_increment(
increment["key"], increment["increment_value"], ttl=increment["ttl"], **kwargs
)
results.append(result)
return results

View file

@ -75,6 +75,8 @@ OPTIONAL_KWARGS_KEYS: Final = (
"rpm",
"itpm",
"otpm",
"tpd",
"rpd",
"use_xai_oauth",
"chatgpt_auth_file",
"github_copilot_auth_type",

View file

@ -5,15 +5,19 @@ When enabled via router_settings.optional_pre_call_checks: ["enforce_model_rate_
- tpm/rpm: combined TPM + optional RPM (legacy)
- itpm/otpm: separate input/output tokens per minute
- tpd/rpd: tokens and requests per UTC calendar day
When a deployment sets both itpm/otpm and tpm/rpm, both are enforced. A warning
is logged the first time such a deployment is seen.
"""
import contextlib
from typing import TYPE_CHECKING, Any, Final
from collections.abc import Callable
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any, Final, Literal
import httpx
from pydantic import BaseModel, ConfigDict
import litellm
from litellm._internal_context import with_service_target
@ -33,6 +37,7 @@ from litellm.router_utils.pre_call_checks.io_token_rate_limit_check import (
io_token_refund_failure,
)
from litellm.router_utils.routing_read_batch import ROUTER_USAGE_TARGET
from litellm.types.caching import RedisPipelineIncrementOperation
from litellm.types.router import RouterErrors
from litellm.types.utils import StandardLoggingPayload
from litellm.utils import get_utc_datetime
@ -47,6 +52,62 @@ else:
class RoutingArgs:
ttl: int = 60 # 1min (RPM/TPM expire key)
daily_ttl: int = 86400
DAY_FORMAT: Final = "%Y-%m-%d"
def seconds_until_utc_midnight(now: datetime) -> int:
next_midnight: Final = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
return max(1, int((next_midnight - now).total_seconds()))
def _first_set(*values: int | None) -> int | None:
return next((value for value in values if value is not None), None)
class _DailyLimitFields(BaseModel):
model_config = ConfigDict(frozen=True, extra="ignore", from_attributes=True)
tpd: int | None = None
rpd: int | None = None
class _DailyLimitParams(_DailyLimitFields):
model: str | None = None
class _DailyLimitModelInfo(_DailyLimitFields):
id: str | None = None
class _DailyLimitDeployment(_DailyLimitFields):
model_name: str = ""
litellm_params: _DailyLimitParams = _DailyLimitParams()
model_info: _DailyLimitModelInfo = _DailyLimitModelInfo()
def limits(self) -> tuple[int | None, int | None]:
return (
_first_set(self.tpd, self.litellm_params.tpd, self.model_info.tpd),
_first_set(self.rpd, self.litellm_params.rpd, self.model_info.rpd),
)
def cache_keys(self, current_day: str) -> tuple[str, str]:
prefix: Final = f"{self.model_info.id}:{self.litellm_params.model}"
return f"{prefix}:tpd:{current_day}", f"{prefix}:rpd:{current_day}"
def limit_error(self, kind: Literal["tpd", "rpd"], limit: int, usage: int, now: datetime) -> litellm.RateLimitError:
return litellm.RateLimitError(
message=f"Model rate limit exceeded. {kind.upper()} limit={limit}, current usage={usage}",
llm_provider="",
model=self.litellm_params.model,
response=httpx.Response(
status_code=429,
content=f"{RouterErrors.user_defined_ratelimit_error.value} {kind} limit={limit}. current usage={usage}. id={self.model_info.id}, model_group={self.model_name}",
headers={"retry-after": str(seconds_until_utc_midnight(now))},
request=httpx.Request(method="model_rate_limit_check", url="https://github.com/BerriAI/litellm"),
),
)
class ModelRateLimitingCheck(CustomLogger):
@ -60,12 +121,60 @@ class ModelRateLimitingCheck(CustomLogger):
this check actively enforces those limits across ALL routing strategies.
"""
def __init__(self, dual_cache: DualCache):
def __init__(self, dual_cache: DualCache, clock: Callable[[], datetime] = get_utc_datetime):
self.dual_cache = dual_cache
self.clock = clock
# model_ids already warned about itpm/otpm + tpm/rpm on the same deployment,
# so the warning is logged once per deployment rather than per request.
self._io_token_conflict_warned_ids: set[str] = set()
def _increment(self, key: str, value: int, ttl: int) -> int:
return self.dual_cache.increment_cache(key=key, value=value, ttl=ttl)
async def _async_increment(
self, key: str, value: int, ttl: int, parent_otel_span: Span | None = None
) -> int | None:
count: Final = await self.dual_cache.async_increment_cache(
key=key, value=value, ttl=ttl, parent_otel_span=parent_otel_span
)
return None if count is None else int(count)
@staticmethod
def _request_count_operations(
rpm_key: str | None, rpd_key: str | None, increment_value: int
) -> tuple[RedisPipelineIncrementOperation, ...]:
return tuple(
RedisPipelineIncrementOperation(key=key, increment_value=increment_value, ttl=ttl)
for key, ttl in ((rpm_key, RoutingArgs.ttl), (rpd_key, RoutingArgs.daily_ttl))
if key is not None
)
async def _async_refund_request_counts(
self, rpm_key: str | None, rpd_key: str | None, parent_otel_span: Span | None
) -> None:
operations: Final = self._request_count_operations(rpm_key, rpd_key, -1)
if operations:
await self.dual_cache.async_increment_cache_pipeline(
increment_list=list(operations), parent_otel_span=parent_otel_span
)
async def _async_increment_request_counts(
self, rpm_key: str | None, rpd_key: str | None, parent_otel_span: Span | None
) -> tuple[int | None, int | None]:
operations: Final = self._request_count_operations(rpm_key, rpd_key, 1)
if not operations:
return None, None
counts: Final = await self.dual_cache.async_increment_cache_pipeline(
increment_list=list(operations), parent_otel_span=parent_otel_span
)
if counts is None or len(counts) != len(operations):
return None, None
count_by_key: Final = {op["key"]: int(count) for op, count in zip(operations, counts)}
return (
None if rpm_key is None else count_by_key.get(rpm_key),
None if rpd_key is None else count_by_key.get(rpd_key),
)
def _warn_io_token_and_tpm_rpm_coexist_once(self, deployment: dict) -> None:
tpm_limit, rpm_limit = self._get_deployment_limits(deployment)
if tpm_limit is None and rpm_limit is None:
@ -129,6 +238,21 @@ class ModelRateLimitingCheck(CustomLogger):
return tpm, rpm
def _check_daily_limits(self, deployment: _DailyLimitDeployment, now: datetime) -> None:
tpd_limit, rpd_limit = deployment.limits()
if tpd_limit is None and rpd_limit is None:
return
tpd_key, rpd_key = deployment.cache_keys(now.strftime(DAY_FORMAT))
if tpd_limit is not None:
current_tpd: Final = self._get_current_tpm(tpd_key, tpd_limit)
if current_tpd is not None and current_tpd >= tpd_limit:
raise deployment.limit_error("tpd", tpd_limit, current_tpd, now)
if rpd_limit is not None:
current_rpd: Final = self._increment(rpd_key, 1, RoutingArgs.daily_ttl)
if current_rpd > rpd_limit:
self._increment(rpd_key, -1, RoutingArgs.daily_ttl)
raise deployment.limit_error("rpd", rpd_limit, current_rpd, now)
def _get_cache_keys(self, deployment: dict, current_minute: str) -> tuple[str, str]:
"""Get the cache keys for TPM and RPM tracking."""
model_id: Final = deployment.get("model_info", {}).get("id")
@ -178,13 +302,10 @@ class ModelRateLimitingCheck(CustomLogger):
)
io_reservation_made = True
dt: Final = self.clock()
daily: Final = _DailyLimitDeployment.model_validate(deployment)
tpm_limit, rpm_limit = self._get_deployment_limits(deployment)
# If no limits are set, allow the request
if tpm_limit is None and rpm_limit is None:
return deployment
dt: Final = get_utc_datetime()
current_minute: Final = dt.strftime("%H-%M")
tpm_key, rpm_key = self._get_cache_keys(deployment, current_minute)
@ -213,8 +334,9 @@ class ModelRateLimitingCheck(CustomLogger):
# Check RPM limit (atomic increment-first to avoid race conditions)
if rpm_limit is not None:
current_rpm: Final = self.dual_cache.increment_cache(key=rpm_key, value=1, ttl=RoutingArgs.ttl)
current_rpm: Final = self._increment(rpm_key, 1, RoutingArgs.ttl)
if current_rpm is not None and current_rpm > rpm_limit:
self._increment(rpm_key, -1, RoutingArgs.ttl)
raise litellm.RateLimitError(
message=f"Model rate limit exceeded. RPM limit={rpm_limit}, current usage={current_rpm}",
llm_provider="",
@ -230,6 +352,12 @@ class ModelRateLimitingCheck(CustomLogger):
),
)
try:
self._check_daily_limits(daily, dt)
except litellm.RateLimitError:
if rpm_limit is not None:
self._increment(rpm_key, -1, RoutingArgs.ttl)
raise
return deployment
except litellm.RateLimitError:
@ -259,15 +387,14 @@ class ModelRateLimitingCheck(CustomLogger):
)
io_reservation_made = True
dt: Final = self.clock()
daily: Final = _DailyLimitDeployment.model_validate(deployment)
tpd_limit, rpd_limit = daily.limits()
tpm_limit, rpm_limit = self._get_deployment_limits(deployment)
# If no limits are set, allow the request
if tpm_limit is None and rpm_limit is None:
return deployment
dt: Final = get_utc_datetime()
current_minute: Final = dt.strftime("%H-%M")
tpm_key, rpm_key = self._get_cache_keys(deployment, current_minute)
tpd_key, rpd_key = daily.cache_keys(dt.strftime(DAY_FORMAT))
model_id: Final = deployment.get("model_info", {}).get("id")
model_name: Final = deployment.get("litellm_params", {}).get("model")
@ -293,30 +420,37 @@ class ModelRateLimitingCheck(CustomLogger):
num_retries=0, # Don't retry - return 429 immediately
)
# Check RPM limit (atomic increment-first to avoid race conditions)
if rpm_limit is not None:
current_rpm: Final = await self.dual_cache.async_increment_cache(
key=rpm_key,
value=1,
ttl=RoutingArgs.ttl,
parent_otel_span=parent_otel_span,
)
if current_rpm is not None and current_rpm > rpm_limit:
raise litellm.RateLimitError(
message=f"Model rate limit exceeded. RPM limit={rpm_limit}, current usage={current_rpm}",
llm_provider="",
model=model_name,
response=httpx.Response(
status_code=429,
content=f"{RouterErrors.user_defined_ratelimit_error.value} rpm limit={rpm_limit}. current usage={current_rpm}. id={model_id}, model_group={model_group}",
headers={"retry-after": str(60)},
request=httpx.Request(
method="model_rate_limit_check",
url="https://github.com/BerriAI/litellm",
),
if tpd_limit is not None:
current_tpd: Final = await self._async_get_current_tpm(tpd_key, tpd_limit, parent_otel_span)
if current_tpd is not None and current_tpd >= tpd_limit:
raise daily.limit_error("tpd", tpd_limit, current_tpd, dt)
charged_rpm_key: Final = rpm_key if rpm_limit is not None else None
charged_rpd_key: Final = rpd_key if rpd_limit is not None else None
current_rpm, current_rpd = await self._async_increment_request_counts(
charged_rpm_key, charged_rpd_key, parent_otel_span
)
if rpm_limit is not None and current_rpm is not None and current_rpm > rpm_limit:
await self._async_refund_request_counts(charged_rpm_key, charged_rpd_key, parent_otel_span)
raise litellm.RateLimitError(
message=f"Model rate limit exceeded. RPM limit={rpm_limit}, current usage={current_rpm}",
llm_provider="",
model=model_name,
response=httpx.Response(
status_code=429,
content=f"{RouterErrors.user_defined_ratelimit_error.value} rpm limit={rpm_limit}. current usage={current_rpm}. id={model_id}, model_group={model_group}",
headers={"retry-after": str(60)},
request=httpx.Request(
method="model_rate_limit_check",
url="https://github.com/BerriAI/litellm",
),
num_retries=0, # Don't retry - return 429 immediately
)
),
num_retries=0, # Don't retry - return 429 immediately
)
if rpd_limit is not None and current_rpd is not None and current_rpd > rpd_limit:
await self._async_refund_request_counts(charged_rpm_key, charged_rpd_key, parent_otel_span)
raise daily.limit_error("rpd", rpd_limit, current_rpd, dt)
return deployment
@ -373,17 +507,16 @@ class ModelRateLimitingCheck(CustomLogger):
if not model or not total_tokens:
return
dt: Final = get_utc_datetime()
dt: Final = self.clock()
current_minute: Final = dt.strftime("%H-%M")
tpm_key: Final = f"{model_id}:{model}:tpm:{current_minute}"
tpd_key: Final = f"{model_id}:{model}:tpd:{dt.strftime(DAY_FORMAT)}"
verbose_router_logger.debug("[TPM TRACKING] Incrementing %s by %s", tpm_key, total_tokens)
await self.dual_cache.async_increment_cache(
key=tpm_key,
value=total_tokens,
ttl=RoutingArgs.ttl,
)
tokens: Final = int(total_tokens)
await self._async_increment(tpm_key, tokens, RoutingArgs.ttl)
await self._async_increment(tpd_key, tokens, RoutingArgs.daily_ttl)
except Exception as e:
verbose_router_logger.debug("Error in ModelRateLimitingCheck.async_log_success_event: %s", e)
@ -434,15 +567,14 @@ class ModelRateLimitingCheck(CustomLogger):
if not model or not total_tokens:
return
dt: Final = get_utc_datetime()
dt: Final = self.clock()
current_minute: Final = dt.strftime("%H-%M")
tpm_key: Final = f"{model_id}:{model}:tpm:{current_minute}"
tpd_key: Final = f"{model_id}:{model}:tpd:{dt.strftime(DAY_FORMAT)}"
self.dual_cache.increment_cache(
key=tpm_key,
value=total_tokens,
ttl=RoutingArgs.ttl,
)
tokens: Final = int(total_tokens)
self._increment(tpm_key, tokens, RoutingArgs.ttl)
self._increment(tpd_key, tokens, RoutingArgs.daily_ttl)
except Exception as e:
verbose_router_logger.debug("Error in ModelRateLimitingCheck.log_success_event: %s", e)

View file

@ -189,6 +189,8 @@ class DeploymentOptions:
tpm: int | None = None
itpm: int | None = None
otpm: int | None = None
rpd: int | None = None
tpd: int | None = None
kubernetes_pod_discovery: bool | None = None
default_api_key_rpm_limit: int | None = None
default_api_key_tpm_limit: int | None = None

View file

@ -533,6 +533,8 @@ class GenericLiteLLMParams(CredentialLiteLLMParams, CustomPricingLiteLLMParams):
rpm: int | None = None
itpm: int | None = None
otpm: int | None = None
tpd: int | None = None
rpd: int | None = None
timeout: float | str | httpx.Timeout | SDKTimeout | None = None # if str, pass in as os.environ/
# read timeout of a live stream, a faked stream takes the longer of this and timeout. If str, pass in as os.environ/
stream_timeout: float | str | None = None
@ -738,6 +740,8 @@ class LiteLLMParamsTypedDict(TypedDict, total=False):
rpm: int | None
itpm: int | None
otpm: int | None
tpd: ReadOnly[int | None]
rpd: ReadOnly[int | None]
order: int | None
weight: int | None
max_parallel_requests: int | None

View file

@ -0,0 +1,96 @@
from __future__ import annotations
import uuid
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Final
import httpx
import pytest
from integration._support.client import gateway_from_environment
from integration._support.openai_wire import chat_reply
from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process
from integration._support.wire import Reply, Request, Wire, wire_server
from integration.providers._cache_control_marks_support import owned_config
from integration.routing.test_payment_required_mapping import OPENAI_MODEL, is_model_info_probe, model_list_reply
from pydantic import JsonValue
pytestmark: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
GROUP: Final = "daily-limited-group"
LIMITED_ID: Final = "free-tier-two-a-day"
UNLIMITED_ID: Final = "paid-tier"
RPD: Final = 2
EXTRA_REQUESTS: Final = 2
@dataclass(frozen=True, slots=True)
class Rig:
owned: OwnedProxy
limited: Wire
unlimited: Wire
def _reply(request: Request) -> Reply:
if is_model_info_probe(request):
return model_list_reply()
return chat_reply("chatcmpl-daily", OPENAI_MODEL, "served", stream=False)
def _deployment(model_id: str, wire: Wire, **params: JsonValue) -> dict[str, JsonValue]:
return {
"model_name": GROUP,
"litellm_params": {
"model": OPENAI_MODEL,
"api_base": wire.url + "/v1",
"api_key": "synthetic-openai-key",
**params,
},
"model_info": {"id": model_id},
}
def _config(directory: Path, limited: Wire, unlimited: Wire) -> Path:
return owned_config(
directory,
[_deployment(LIMITED_ID, limited, rpd=RPD, order=1), _deployment(UNLIMITED_ID, unlimited, order=2)],
litellm_settings={"cache": False},
router_settings={"optional_pre_call_checks": ["enforce_model_rate_limits"], "num_retries": 1},
)
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
directory: Final = tmp_path_factory.mktemp("daily-limits")
with (
gateway_from_environment() as gateway,
wire_server(_reply) as limited,
wire_server(_reply) as unlimited,
owned_proxy_process(gateway, directory, {}, config=_config(directory, limited, unlimited)) as owned,
):
yield Rig(owned, limited, unlimited)
def _chat(rig: Rig) -> httpx.Response:
return rig.owned.gateway.request(
"POST",
"/v1/chat/completions",
{"model": GROUP, "messages": [{"role": "user", "content": f"daily {uuid.uuid4().hex}"}]},
)
def _forwarded(wire: Wire) -> int:
return sum(1 for request in wire.drain() if not is_model_info_probe(request))
def test_requests_past_the_deployment_rpd_are_served_by_the_other_deployment_in_the_group(rig: Rig) -> None:
responses: Final = tuple(_chat(rig) for _ in range(RPD + EXTRA_REQUESTS))
assert [response.status_code for response in responses] == [200] * (RPD + EXTRA_REQUESTS), [
response.text for response in responses
]
assert [response.headers["x-litellm-model-id"] for response in responses] == [LIMITED_ID] * RPD + [
UNLIMITED_ID
] * EXTRA_REQUESTS
assert (_forwarded(rig.limited), _forwarded(rig.unlimited)) == (RPD, EXTRA_REQUESTS)

View file

@ -235,13 +235,9 @@ def test_in_memory_cache_eviction_order():
# Add items with different TTLs
now = time.time()
in_memory_cache.set_cache(
key="early_expire", value="value_1", ttl=100
) # expires in 100 seconds
in_memory_cache.set_cache(key="early_expire", value="value_1", ttl=100) # expires in 100 seconds
time.sleep(0.01)
in_memory_cache.set_cache(
key="late_expire", value="value_2", ttl=200
) # expires in 200 seconds
in_memory_cache.set_cache(key="late_expire", value="value_2", ttl=200) # expires in 200 seconds
# Verify TTL order
early_ttl = in_memory_cache.ttl_dict["early_expire"]
@ -379,3 +375,20 @@ def test_in_memory_cache_injected_clock_controls_expiry_and_eviction() -> None:
assert cache.get_cache("second") is None
assert cache.get_cache("third") == "replacement"
assert cache.get_cache("fourth") == "new"
@pytest.mark.asyncio
async def test_async_increment_pipeline_keeps_each_key_for_its_own_ttl() -> None:
now: Final = {"t": 1_000.0}
cache: Final = InMemoryCache(default_ttl=600, clock=lambda: now["t"])
await cache.async_increment_pipeline(
increment_list=[
{"key": "minute", "increment_value": 1, "ttl": 60},
{"key": "day", "increment_value": 1, "ttl": 86400},
]
)
now["t"] += 601
assert cache.get_cache("minute") is None
assert cache.get_cache("day") == 1

View file

@ -6,7 +6,7 @@ regardless of the routing strategy being used.
"""
import asyncio
from datetime import timedelta
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock
import pytest
@ -222,11 +222,9 @@ class TestModelRateLimitingCheck:
check.log_success_event(kwargs, None, None, None)
# Verify increment_cache was called
mock_cache.increment_cache.assert_called_once()
_, kwarg_params = mock_cache.increment_cache.call_args
assert "test-id:gpt-4:tpm:" in kwarg_params["key"]
assert kwarg_params["value"] == 50
keys = tuple(call.kwargs["key"] for call in mock_cache.increment_cache.call_args_list)
assert any(key.startswith("test-id:gpt-4:tpm:") for key in keys), keys
assert all(call.kwargs["value"] == 50 for call in mock_cache.increment_cache.call_args_list)
class TestModelRateLimitingCheckAsync:
@ -253,7 +251,7 @@ class TestModelRateLimitingCheckAsync:
"""Test that RateLimitError is raised when RPM limit is exceeded (async)."""
mock_cache = MagicMock()
mock_cache.async_get_cache = AsyncMock(return_value=None)
mock_cache.async_increment_cache = AsyncMock(return_value=11) # Over limit
mock_cache.async_increment_cache_pipeline = AsyncMock(return_value=[11.0]) # Over limit
check = ModelRateLimitingCheck(dual_cache=mock_cache)
@ -274,7 +272,7 @@ class TestModelRateLimitingCheckAsync:
"""Test that requests are allowed when under the limit (async)."""
mock_cache = MagicMock()
mock_cache.async_get_cache = AsyncMock(return_value=None)
mock_cache.async_increment_cache = AsyncMock(return_value=6)
mock_cache.async_increment_cache_pipeline = AsyncMock(return_value=[6.0])
check = ModelRateLimitingCheck(dual_cache=mock_cache)
@ -339,7 +337,7 @@ class TestModelRateLimitingCheckAsync:
):
redis_cache = MagicMock()
redis_cache.async_get_cache = AsyncMock(side_effect=RedisCircuitBreakerOpenError())
redis_cache.async_increment = AsyncMock(return_value=2)
redis_cache.async_increment_pipeline = AsyncMock(return_value=[2.0])
check = ModelRateLimitingCheck(dual_cache=_dual_cache_with_local_tpm(5, redis_cache))
with pytest.raises(litellm.RateLimitError) as exc_info:
@ -375,11 +373,9 @@ class TestModelRateLimitingCheckAsync:
await check.async_log_success_event(kwargs, None, None, None)
# Verify async_increment_cache was called
mock_cache.async_increment_cache.assert_called_once()
_, kwarg_params = mock_cache.async_increment_cache.call_args
assert "test-id:gpt-4:tpm:" in kwarg_params["key"]
assert kwarg_params["value"] == 50
keys = tuple(call.kwargs["key"] for call in mock_cache.async_increment_cache.call_args_list)
assert any(key.startswith("test-id:gpt-4:tpm:") for key in keys), keys
assert all(call.kwargs["value"] == 50 for call in mock_cache.async_increment_cache.call_args_list)
class TestRouterWithEnforceModelRateLimits:
@ -466,3 +462,265 @@ class TestModelRateLimitConcurrency:
assert len(successes) == 2, f"Expected 2 successes, got {len(successes)}"
assert len(failures) == 2, f"Expected 2 rate limit errors, got {len(failures)}"
DAY_ONE = datetime(2026, 10, 28, 23, 59, 30, tzinfo=timezone.utc)
RPD_DEPLOYMENT = {
"litellm_params": {"model": "gpt-4.1-mini", "rpd": 2},
"model_info": {"id": "daily-test-id"},
"model_name": "daily-model",
}
RPM_AND_RPD_DEPLOYMENT = {**RPD_DEPLOYMENT, "litellm_params": {"model": "gpt-4.1-mini", "rpm": 1, "rpd": 2}}
class Clock:
def __init__(self, now: datetime):
self.now = now
def __call__(self) -> datetime:
return self.now
def _pre_call(check: ModelRateLimitingCheck, deployment: dict[str, object]) -> object:
return check.pre_call_check(deployment)
async def _async_pre_call(check: ModelRateLimitingCheck, deployment: dict[str, object]) -> object:
return await check.async_pre_call_check(deployment)
async def _async_log_success(check: ModelRateLimitingCheck, kwargs: dict[str, object]) -> None:
await check.async_log_success_event(kwargs, None, None, None)
class TestDailyLimits:
def test_pre_call_check_rejects_the_request_over_rpd(self):
check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=Clock(DAY_ONE))
assert _pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
assert _pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
with pytest.raises(litellm.RateLimitError) as exc_info:
_pre_call(check, RPD_DEPLOYMENT)
assert "RPD limit=2, current usage=3" in str(exc_info.value)
assert "rpd limit=2. current usage=3. id=daily-test-id, model_group=daily-model" in exc_info.value.response.text
@pytest.mark.asyncio
async def test_async_pre_call_check_rejects_the_request_over_rpd_and_lets_the_router_retry(self):
check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=Clock(DAY_ONE))
assert await _async_pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
assert await _async_pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
with pytest.raises(litellm.RateLimitError) as exc_info:
await _async_pre_call(check, RPD_DEPLOYMENT)
assert "RPD limit=2" in str(exc_info.value)
assert exc_info.value.num_retries is None
@pytest.mark.asyncio
async def test_async_pre_call_check_rejects_once_the_day_tokens_reach_tpd(self):
deployment = {**RPD_DEPLOYMENT, "litellm_params": {"model": "gpt-4.1-mini", "tpd": 100}}
check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=Clock(DAY_ONE))
kwargs = {
"standard_logging_object": {
"model_id": "daily-test-id",
"total_tokens": 60,
"hidden_params": {"litellm_model_name": "gpt-4.1-mini"},
}
}
await _async_log_success(check, kwargs)
assert await _async_pre_call(check, deployment) == deployment
await _async_log_success(check, kwargs)
with pytest.raises(litellm.RateLimitError) as exc_info:
await _async_pre_call(check, deployment)
assert "TPD limit=100, current usage=120" in str(exc_info.value)
def test_pre_call_check_minute_rejected_request_does_not_spend_rpd(self):
clock = Clock(DAY_ONE)
check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
assert _pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
with pytest.raises(litellm.RateLimitError, match="RPM limit=1"):
_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
clock.now = DAY_ONE - timedelta(minutes=1)
assert _pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
clock.now = DAY_ONE - timedelta(minutes=2)
with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
@pytest.mark.asyncio
async def test_async_pre_call_check_minute_rejected_request_does_not_spend_rpd(self):
clock = Clock(DAY_ONE)
check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
with pytest.raises(litellm.RateLimitError, match="RPM limit=1"):
await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
clock.now = DAY_ONE - timedelta(minutes=1)
assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
clock.now = DAY_ONE - timedelta(minutes=2)
with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
def test_pre_call_check_rejected_requests_hand_back_rpm_and_rpd(self):
clock = Clock(DAY_ONE)
check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
assert _pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
with pytest.raises(litellm.RateLimitError, match="RPM limit=1, current usage=2"):
_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
with pytest.raises(litellm.RateLimitError, match="RPM limit=1, current usage=2"):
_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
clock.now = DAY_ONE - timedelta(minutes=1)
assert _pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
clock.now = DAY_ONE - timedelta(minutes=2)
with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
@pytest.mark.asyncio
async def test_async_pre_call_check_rejected_requests_hand_back_rpm_and_rpd(self):
clock = Clock(DAY_ONE)
check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
with pytest.raises(litellm.RateLimitError, match="RPM limit=1, current usage=2"):
await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
with pytest.raises(litellm.RateLimitError, match="RPM limit=1, current usage=2"):
await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
clock.now = DAY_ONE - timedelta(minutes=1)
assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
clock.now = DAY_ONE - timedelta(minutes=2)
with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
@pytest.mark.asyncio
async def test_async_pre_call_check_refunds_rpm_and_rpd_in_one_cache_trip(self):
mock_cache = MagicMock()
mock_cache.async_get_cache = AsyncMock(return_value=None)
mock_cache.async_increment_cache_pipeline = AsyncMock(side_effect=[[1.0, 3.0], [0.0, 2.0]])
check = ModelRateLimitingCheck(dual_cache=mock_cache, clock=Clock(DAY_ONE))
with pytest.raises(litellm.RateLimitError, match="RPD limit=2, current usage=3"):
await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT)
assert mock_cache.async_increment_cache_pipeline.await_count == 2
refund = mock_cache.async_increment_cache_pipeline.call_args_list[1].kwargs["increment_list"]
assert [(op["key"], op["increment_value"], op["ttl"]) for op in refund] == [
("daily-test-id:gpt-4.1-mini:rpm:23-59", -1, 60),
("daily-test-id:gpt-4.1-mini:rpd:2026-10-28", -1, 86400),
]
@pytest.mark.asyncio
async def test_async_pre_call_check_increments_rpm_and_rpd_in_one_cache_trip(self):
mock_cache = MagicMock()
mock_cache.async_get_cache = AsyncMock(return_value=None)
mock_cache.async_increment_cache_pipeline = AsyncMock(return_value=[1.0, 1.0])
check = ModelRateLimitingCheck(dual_cache=mock_cache, clock=Clock(DAY_ONE))
assert await _async_pre_call(check, RPM_AND_RPD_DEPLOYMENT) == RPM_AND_RPD_DEPLOYMENT
assert mock_cache.async_increment_cache_pipeline.await_count == 1
operations = mock_cache.async_increment_cache_pipeline.call_args.kwargs["increment_list"]
assert [(op["key"], op["ttl"]) for op in operations] == [
("daily-test-id:gpt-4.1-mini:rpm:23-59", 60),
("daily-test-id:gpt-4.1-mini:rpd:2026-10-28", 86400),
]
assert not mock_cache.async_increment_cache.called
def test_pre_call_check_reads_the_shared_tpd_count_from_redis(self):
redis_cache = MagicMock()
redis_cache.get_cache.return_value = 100
deployment = {**RPD_DEPLOYMENT, "litellm_params": {"model": "gpt-4.1-mini", "tpd": 100}}
check = ModelRateLimitingCheck(dual_cache=DualCache(redis_cache=redis_cache), clock=Clock(DAY_ONE))
with pytest.raises(litellm.RateLimitError) as exc_info:
_pre_call(check, deployment)
assert "TPD limit=100, current usage=100" in str(exc_info.value)
assert redis_cache.get_cache.call_args.kwargs["key"] == "daily-test-id:gpt-4.1-mini:tpd:2026-10-28"
def test_rpd_count_starts_fresh_after_utc_midnight(self):
clock = Clock(DAY_ONE)
check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=clock)
_pre_call(check, RPD_DEPLOYMENT)
_pre_call(check, RPD_DEPLOYMENT)
with pytest.raises(litellm.RateLimitError):
_pre_call(check, RPD_DEPLOYMENT)
clock.now = DAY_ONE + timedelta(seconds=30)
assert _pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
assert _pre_call(check, RPD_DEPLOYMENT) == RPD_DEPLOYMENT
with pytest.raises(litellm.RateLimitError) as exc_info:
_pre_call(check, RPD_DEPLOYMENT)
assert "current usage=3" in str(exc_info.value)
@pytest.mark.parametrize(
("now", "seconds_to_midnight"),
[
(DAY_ONE, 30),
(datetime(2026, 10, 28, 0, 0, 0, tzinfo=timezone.utc), 86400),
(datetime(2026, 10, 28, 12, 30, 15, 500000, tzinfo=timezone.utc), 41384),
],
)
def test_daily_429_retry_after_is_seconds_until_utc_midnight(self, now: datetime, seconds_to_midnight: int):
deployment = {**RPD_DEPLOYMENT, "litellm_params": {"model": "gpt-4.1-mini", "rpd": 0}}
check = ModelRateLimitingCheck(dual_cache=DualCache(), clock=Clock(now))
with pytest.raises(litellm.RateLimitError) as exc_info:
_pre_call(check, deployment)
assert exc_info.value.response.headers["retry-after"] == str(seconds_to_midnight)
@pytest.mark.asyncio
async def test_router_serves_the_request_from_the_other_deployment_once_rpd_is_spent(self):
router = Router(
model_list=[
{
"model_name": "daily-group",
"litellm_params": {
"model": "openai/gpt-4.1-mini",
"api_key": "k",
"mock_response": "a",
"rpd": 1,
"order": 1,
},
"model_info": {"id": "spent-first"},
},
{
"model_name": "daily-group",
"litellm_params": {
"model": "openai/gpt-4.1-mini",
"api_key": "k",
"mock_response": "b",
"order": 2,
},
"model_info": {"id": "always-free"},
},
],
num_retries=1,
)
check = ModelRateLimitingCheck(dual_cache=router.cache, clock=Clock(DAY_ONE))
router.optional_callbacks = [check]
litellm.logging_callback_manager.add_litellm_callback(check)
try:
served = [
(await router.acompletion(model="daily-group", messages=[{"role": "user", "content": "hi"}]))
.choices[0]
.message.content
for _ in range(4)
]
finally:
router.discard()
assert served == ["a", "b", "b", "b"]

View file

@ -859,7 +859,7 @@ class TestModelRateLimitingCheckIOTokens:
with pytest.raises(litellm.RateLimitError):
await check.async_pre_call_check(deployment)
assert await dual_cache.async_get_cache(key=rpm_key) == 6
assert await dual_cache.async_get_cache(key=rpm_key) == 5, "the rejected request hands its rpm count back"
assert (await dual_cache.async_get_cache(key=itpm_key) or 0) == 0
assert ITPM_RESERVED_KEY not in request_kwargs["metadata"]
assert any("both limit types are enforced" in record.message for record in caplog.records)

View file

@ -154,6 +154,8 @@ OPTION_NAMES: Final = (
"model_info",
"rpm",
"tpm",
"rpd",
"tpd",
"itpm",
"otpm",
"default_api_key_rpm_limit",

View file

@ -35647,6 +35647,8 @@ export interface components {
regional_processing_uplift_multiplier_us?: number | null;
/** Responses Api Version */
responses_api_version?: string | null;
/** Rpd */
rpd?: number | null;
/** Rpm */
rpm?: number | null;
/** S3 Access Key Id */
@ -35699,6 +35701,8 @@ export interface components {
token_exchange_profile?: string | null;
/** Token Exchange Scope */
token_exchange_scope?: string | null;
/** Tpd */
tpd?: number | null;
/** Tpm */
tpm?: number | null;
/** Use Chat Completions Api */
@ -51039,6 +51043,8 @@ export interface components {
regional_processing_uplift_multiplier_us?: number | null;
/** Responses Api Version */
responses_api_version?: string | null;
/** Rpd */
rpd?: number | null;
/** Rpm */
rpm?: number | null;
/** S3 Access Key Id */
@ -51091,6 +51097,8 @@ export interface components {
token_exchange_profile?: string | null;
/** Token Exchange Scope */
token_exchange_scope?: string | null;
/** Tpd */
tpd?: number | null;
/** Tpm */
tpm?: number | null;
/** Use Chat Completions Api */