Implementation:BerriAI Litellm Lowest TPM RPM V2 Strategy

Attribute	Value
Sources	litellm/router_strategy/lowest_tpm_rpm_v2.py
Domains	Router, Strategy, Rate Limiting, Load Balancing
last_updated	2026-02-15 16:00 GMT

Overview

The Lowest TPM/RPM V2 Strategy is an updated deployment selection strategy that routes requests to the deployment with the lowest tokens-per-minute (TPM) usage, designed to work across multiple instances using Redis-backed batch operations.

Description

This module provides the LowestTPMLoggingHandler_v2 class, which extends both BaseRoutingStrategy and CustomLogger. Unlike the V1 strategy that stores aggregate model group counts, V2 caches individual model deployment TPM/RPM values with separate cache keys per deployment and minute window. This design enables efficient cross-instance coordination via redis.mget for batch reads and redis.incr for atomic increments. The class includes pre-call RPM limit enforcement (raising RateLimitError before the call is made), post-call TPM tracking, and deployment selection that picks the deployment with the lowest current TPM while respecting both TPM and RPM limits.

Usage

Import this class when configuring the LiteLLM Router with routing_strategy="usage-based-routing-v2" or the default usage-based routing. The router automatically registers it for logging callbacks and deployment selection.

Code Reference

Source Location

litellm/router_strategy/lowest_tpm_rpm_v2.py

Classes

class RoutingArgs(LiteLLMPydanticObjectBase):
    ttl: int = 1 * 60  # 1min (RPM/TPM expire key)

class LowestTPMLoggingHandler_v2(BaseRoutingStrategy, CustomLogger):
    test_flag: bool = False
    logged_success: int = 0
    logged_failure: int = 0
    default_cache_time_seconds: int = 1 * 60 * 60  # 1 hour

    def __init__(self, router_cache: DualCache, routing_args: dict = {}):

Key Methods

Method	Signature	Description
`pre_call_check`	`def pre_call_check(self, deployment: Dict) -> Optional[Dict]`	Sync pre-call RPM limit check; increments RPM counter and raises `RateLimitError` if over limit
`async_pre_call_check`	`async def async_pre_call_check(self, deployment: Dict, parent_otel_span: Optional[Span]) -> Optional[Dict]`	Async pre-call RPM limit check with OpenTelemetry span support
`log_success_event`	`def log_success_event(self, kwargs, response_obj, start_time, end_time)`	Sync callback that increments TPM counter after successful call
`async_log_success_event`	`async def async_log_success_event(self, kwargs, response_obj, start_time, end_time)`	Async callback that increments TPM counter after successful call
`get_available_deployments`	`def get_available_deployments(self, model_group: str, healthy_deployments: list, messages=None, input=None, parent_otel_span=None)`	Sync deployment selection using batch cache reads
`async_get_available_deployments`	`async def async_get_available_deployments(self, model_group: str, healthy_deployments: list, messages=None, input=None)`	Async deployment selection using batch cache reads

Import

from litellm.router_strategy.lowest_tpm_rpm_v2 import LowestTPMLoggingHandler_v2

I/O Contract

Inputs

Parameter	Type	Description
`router_cache`	`DualCache`	Shared dual cache (in-memory + optional Redis) for TPM/RPM counters
`routing_args`	`dict`	Configuration with optional `ttl` (default 60s) for cache key expiration
`model_group`	`str`	The model group name to select a deployment for
`healthy_deployments`	`list`	List of deployment dictionaries currently considered healthy
`messages`	`Optional[List[Dict[str, str]]]`	Messages for estimating input token count
`input`	`Optional[Union[str, List]]`	Text input for estimating input token count

Outputs

Return Type	Description
`dict`	The selected deployment dictionary with the lowest TPM usage
Raises `RateLimitError`	When no deployments are available within their TPM/RPM limits (async variant) or raises `ValueError` (sync variant)

Usage Examples

from litellm.caching.caching import DualCache
from litellm.router_strategy.lowest_tpm_rpm_v2 import LowestTPMLoggingHandler_v2

# Initialize with a shared router cache
cache = DualCache()
handler = LowestTPMLoggingHandler_v2(router_cache=cache, routing_args={"ttl": 60})

# Pre-call check to enforce RPM limits (raises RateLimitError if exceeded)
deployment = {"model_info": {"id": "abc123"}, "litellm_params": {"model": "gpt-4"}, "rpm": 100}
checked = handler.pre_call_check(deployment=deployment)

# Select the deployment with the lowest TPM usage
selected = handler.get_available_deployments(
    model_group="gpt-4",
    healthy_deployments=[
        {"model_info": {"id": "abc123"}, "litellm_params": {"model": "gpt-4"}, "tpm": 100000, "rpm": 500},
        {"model_info": {"id": "def456"}, "litellm_params": {"model": "gpt-4"}, "tpm": 200000, "rpm": 1000},
    ],
    messages=[{"role": "user", "content": "Hello"}],
)

Related Pages

BerriAI_Litellm_Lowest_TPM_RPM_Strategy - Original V1 TPM/RPM routing strategy
BerriAI_Litellm_Lowest_Cost_Strategy - Cost-based deployment selection strategy
BerriAI_Litellm_Least_Busy_Strategy - Least busy deployment selection strategy
BerriAI_Litellm_Simple_Shuffle_Strategy - Random/weighted shuffle deployment selection

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment