mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-29 06:33:28 +00:00
fix(router): normalize native Responses latency per output token (#47727)
* fix(router): normalize native Responses latency per output token * fix(router): isolate normalized latency samples during upgrades --------- Co-authored-by: Joshua Valluru <326636767+joshua-berri@users.noreply.github.com>
This commit is contained in:
joshua-berri
GitHub
parent
7ae46f2768
commit
230f653523
3 changed files with 249 additions and 38 deletions
|
|
@ -14,7 +14,9 @@ from litellm._internal_context import with_service_target
|
|||
from litellm.caching.caching import DualCache
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.litellm_core_utils.core_helpers import get_parent_otel_span_from_kwargs, safe_divide_seconds
|
||||
from litellm.responses.utils import ResponseAPILoggingUtils
|
||||
from litellm.router_utils.batch_utils import is_batch_retrieve_call_type
|
||||
from litellm.types.llms.openai import ResponsesAPIResponse
|
||||
from litellm.types.utils import LiteLLMPydanticObjectBase
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -86,7 +88,7 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
{
|
||||
{model_group}_map: {
|
||||
id: {
|
||||
"latency": [..]
|
||||
"latency_normalized": [..]
|
||||
f"{date:hour:minute}" : {"tpm": 34, "rpm": 3}
|
||||
}
|
||||
}
|
||||
|
|
@ -116,11 +118,16 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
final_value: float = response_ms
|
||||
total_tokens = 0
|
||||
|
||||
if isinstance(response_obj, ModelResponse):
|
||||
_usage: Final = getattr(response_obj, "usage", None)
|
||||
if isinstance(response_obj, (ModelResponse, ResponsesAPIResponse)):
|
||||
_usage: Final = (
|
||||
ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(response_obj.usage)
|
||||
if isinstance(response_obj, ResponsesAPIResponse)
|
||||
else getattr(response_obj, "usage", None)
|
||||
)
|
||||
if _usage is not None:
|
||||
completion_tokens: Final = _usage.completion_tokens
|
||||
total_tokens = _usage.total_tokens
|
||||
if isinstance(response_obj, ModelResponse):
|
||||
total_tokens = _usage.total_tokens
|
||||
|
||||
# response_ms is already normalized to float seconds above
|
||||
response_seconds: Final = response_ms
|
||||
|
|
@ -143,10 +150,12 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
request_count_dict[id] = {}
|
||||
|
||||
## Latency
|
||||
if len(request_count_dict[id].get("latency", [])) < self.routing_args.max_latency_list_size:
|
||||
request_count_dict[id].setdefault("latency", []).append(final_value)
|
||||
if len(request_count_dict[id].get("latency_normalized", [])) < self.routing_args.max_latency_list_size:
|
||||
request_count_dict[id].setdefault("latency_normalized", []).append(final_value)
|
||||
else:
|
||||
request_count_dict[id]["latency"] = request_count_dict[id]["latency"][1:] + [final_value]
|
||||
request_count_dict[id]["latency_normalized"] = request_count_dict[id]["latency_normalized"][1:] + [
|
||||
final_value
|
||||
]
|
||||
|
||||
## Time to first token
|
||||
if time_to_first_token is not None:
|
||||
|
|
@ -212,7 +221,7 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
{
|
||||
{model_group}_map: {
|
||||
id: {
|
||||
"latency": [..]
|
||||
"latency_normalized": [..]
|
||||
f"{date:hour:minute}" : {"tpm": 34, "rpm": 3}
|
||||
}
|
||||
}
|
||||
|
|
@ -225,10 +234,15 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
request_count_dict[id] = {}
|
||||
|
||||
## Latency - give 1000s penalty for failing
|
||||
if len(request_count_dict[id].get("latency", [])) < self.routing_args.max_latency_list_size:
|
||||
request_count_dict[id].setdefault("latency", []).append(1000.0)
|
||||
if (
|
||||
len(request_count_dict[id].get("latency_normalized", []))
|
||||
< self.routing_args.max_latency_list_size
|
||||
):
|
||||
request_count_dict[id].setdefault("latency_normalized", []).append(1000.0)
|
||||
else:
|
||||
request_count_dict[id]["latency"] = request_count_dict[id]["latency"][1:] + [1000.0]
|
||||
request_count_dict[id]["latency_normalized"] = request_count_dict[id]["latency_normalized"][
|
||||
1:
|
||||
] + [1000.0]
|
||||
|
||||
await self.router_cache.async_set_cache(
|
||||
key=latency_key,
|
||||
|
|
@ -270,7 +284,7 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
{
|
||||
{model_group}_map: {
|
||||
id: {
|
||||
"latency": [..]
|
||||
"latency_normalized": [..]
|
||||
"time_to_first_token_seconds": [..]
|
||||
f"{date:hour:minute}" : {"tpm": 34, "rpm": 3}
|
||||
}
|
||||
|
|
@ -300,11 +314,16 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
final_value: float = response_ms
|
||||
total_tokens = 0
|
||||
|
||||
if isinstance(response_obj, ModelResponse):
|
||||
_usage: Final = getattr(response_obj, "usage", None)
|
||||
if isinstance(response_obj, (ModelResponse, ResponsesAPIResponse)):
|
||||
_usage: Final = (
|
||||
ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(response_obj.usage)
|
||||
if isinstance(response_obj, ResponsesAPIResponse)
|
||||
else getattr(response_obj, "usage", None)
|
||||
)
|
||||
if _usage is not None:
|
||||
completion_tokens: Final = _usage.completion_tokens
|
||||
total_tokens = _usage.total_tokens
|
||||
if isinstance(response_obj, ModelResponse):
|
||||
total_tokens = _usage.total_tokens
|
||||
|
||||
# response_ms is already normalized to float seconds above
|
||||
response_seconds: Final = response_ms
|
||||
|
|
@ -331,10 +350,12 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
request_count_dict[id] = {}
|
||||
|
||||
## Latency
|
||||
if len(request_count_dict[id].get("latency", [])) < self.routing_args.max_latency_list_size:
|
||||
request_count_dict[id].setdefault("latency", []).append(final_value)
|
||||
if len(request_count_dict[id].get("latency_normalized", [])) < self.routing_args.max_latency_list_size:
|
||||
request_count_dict[id].setdefault("latency_normalized", []).append(final_value)
|
||||
else:
|
||||
request_count_dict[id]["latency"] = request_count_dict[id]["latency"][1:] + [final_value]
|
||||
request_count_dict[id]["latency_normalized"] = request_count_dict[id]["latency_normalized"][1:] + [
|
||||
final_value
|
||||
]
|
||||
|
||||
## Time to first token
|
||||
if time_to_first_token is not None:
|
||||
|
|
@ -403,7 +424,7 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
## if healthy deployment not yet used
|
||||
if d["model_info"]["id"] not in all_deployments:
|
||||
all_deployments[d["model_info"]["id"]] = {
|
||||
"latency": [0],
|
||||
"latency_normalized": [0],
|
||||
precise_minute: {"tpm": 0, "rpm": 0},
|
||||
}
|
||||
|
||||
|
|
@ -443,7 +464,7 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
or _deployment.get("model_info", {}).get("rpm", None)
|
||||
or float("inf")
|
||||
)
|
||||
item_latency = item_map.get("latency", [])
|
||||
item_latency = item_map.get("latency_normalized", [])
|
||||
item_ttft_latency = item_map.get("time_to_first_token_seconds", [])
|
||||
item_rpm = item_map.get(precise_minute, {}).get("rpm", 0)
|
||||
item_tpm = item_map.get(precise_minute, {}).get("tpm", 0)
|
||||
|
|
|
|||
|
|
@ -17,8 +17,10 @@ from pydantic import ValidationError
|
|||
import litellm
|
||||
from litellm.caching.caching import DualCache
|
||||
from litellm.caching.in_memory_cache import InMemoryCache
|
||||
from litellm.responses.utils import ResponseAPILoggingUtils
|
||||
from litellm.router import Router
|
||||
from litellm.router_strategy.lowest_latency import LowestLatencyLoggingHandler, RoutingArgs
|
||||
from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
|
||||
|
||||
DEPLOYMENT_ID = "9876"
|
||||
KWARGS = {
|
||||
|
|
@ -43,7 +45,7 @@ def _embedding_response() -> litellm.EmbeddingResponse:
|
|||
|
||||
def _recorded_latencies(cache: DualCache):
|
||||
cached = cache.get_cache(key="gemini-embedding-001_map") or {}
|
||||
return cached.get(DEPLOYMENT_ID, {}).get("latency", [])
|
||||
return cached.get(DEPLOYMENT_ID, {}).get("latency_normalized", [])
|
||||
|
||||
|
||||
def test_sync_embedding_latency_is_json_serializable():
|
||||
|
|
@ -69,7 +71,7 @@ def test_sync_embedding_latency_is_json_serializable():
|
|||
)
|
||||
assert latencies[-1] == pytest.approx(2.0)
|
||||
# the exact failure mode from production: redis cache sync json.dumps
|
||||
json.dumps({"latency": latencies})
|
||||
json.dumps({"latency_normalized": latencies})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -94,7 +96,7 @@ async def test_async_embedding_latency_is_json_serializable():
|
|||
f"raw timedelta leaked into latency list: {latencies}"
|
||||
)
|
||||
assert latencies[-1] == pytest.approx(3.0)
|
||||
json.dumps({"latency": latencies})
|
||||
json.dumps({"latency_normalized": latencies})
|
||||
|
||||
|
||||
def _chat_response(completion_tokens: int) -> litellm.ModelResponse:
|
||||
|
|
@ -131,7 +133,7 @@ async def test_async_chat_latency_normalized_per_token():
|
|||
|
||||
latencies = _recorded_latencies(cache)
|
||||
assert latencies and latencies[-1] == pytest.approx(0.5) # 2s / 4 tokens
|
||||
json.dumps({"latency": latencies})
|
||||
json.dumps({"latency_normalized": latencies})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -151,7 +153,7 @@ async def test_async_chat_zero_completion_tokens_falls_back_to_seconds():
|
|||
latencies = _recorded_latencies(cache)
|
||||
assert latencies and latencies[-1] == pytest.approx(3.0)
|
||||
assert not isinstance(latencies[-1], timedelta)
|
||||
json.dumps({"latency": latencies})
|
||||
json.dumps({"latency_normalized": latencies})
|
||||
|
||||
|
||||
def test_sync_chat_zero_completion_tokens_falls_back_to_seconds():
|
||||
|
|
@ -168,7 +170,7 @@ def test_sync_chat_zero_completion_tokens_falls_back_to_seconds():
|
|||
latencies = _recorded_latencies(cache)
|
||||
assert latencies and latencies[-1] == pytest.approx(2.0)
|
||||
assert not isinstance(latencies[-1], timedelta)
|
||||
json.dumps({"latency": latencies})
|
||||
json.dumps({"latency_normalized": latencies})
|
||||
|
||||
|
||||
MODEL_GROUP = "gpt-4o-mini"
|
||||
|
|
@ -362,8 +364,8 @@ async def test_ttft_percentile_does_not_change_non_streaming_routing():
|
|||
cache.set_cache(
|
||||
key=f"{MODEL_GROUP}_map",
|
||||
value={
|
||||
FAST_TTFT_ID: {"latency": [1.0], "time_to_first_token_seconds": [0.1]},
|
||||
SLOW_TTFT_ID: {"latency": [0.2], "time_to_first_token_seconds": [1.5]},
|
||||
FAST_TTFT_ID: {"latency_normalized": [1.0], "time_to_first_token_seconds": [0.1]},
|
||||
SLOW_TTFT_ID: {"latency_normalized": [0.2], "time_to_first_token_seconds": [1.5]},
|
||||
},
|
||||
)
|
||||
|
||||
|
|
@ -380,7 +382,7 @@ async def test_ttft_percentile_does_not_change_non_streaming_routing():
|
|||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"cached_entry",
|
||||
[{"latency": []}, {"2026-09-05-15-39": {"tpm": 28, "rpm": 1}}],
|
||||
[{"latency_normalized": []}, {"2026-09-05-15-39": {"tpm": 28, "rpm": 1}}],
|
||||
ids=["empty_latency_list", "minute_bucket_only_as_cost_based_routing_writes"],
|
||||
)
|
||||
async def test_async_get_available_deployments_treats_missing_samples_as_zero_latency(cached_entry):
|
||||
|
|
@ -388,7 +390,7 @@ async def test_async_get_available_deployments_treats_missing_samples_as_zero_la
|
|||
handler = LowestLatencyLoggingHandler(router_cache=cache)
|
||||
cache.set_cache(
|
||||
key="gemini-embedding-001_map",
|
||||
value={DEPLOYMENT_ID: cached_entry, "slower": {"latency": [0.5]}},
|
||||
value={DEPLOYMENT_ID: cached_entry, "slower": {"latency_normalized": [0.5]}},
|
||||
)
|
||||
healthy_deployments = [
|
||||
{"model_info": {"id": DEPLOYMENT_ID}, "litellm_params": {}},
|
||||
|
|
@ -605,7 +607,7 @@ def test_latency_updated():
|
|||
end_time=end_time,
|
||||
)
|
||||
latency_key = f"{model_group}_map"
|
||||
assert end_time - start_time == test_cache.get_cache(key=latency_key)[deployment_id]["latency"][0]
|
||||
assert end_time - start_time == test_cache.get_cache(key=latency_key)[deployment_id]["latency_normalized"][0]
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("frozen_latency_clock")
|
||||
|
|
@ -1002,7 +1004,7 @@ def test_latency_list_trimming_discards_oldest_entry():
|
|||
|
||||
latency_key = f"{model_group}_map"
|
||||
cached_data = test_cache.get_cache(key=latency_key)
|
||||
latency_list = cached_data[deployment_id]["latency"]
|
||||
latency_list = cached_data[deployment_id]["latency_normalized"]
|
||||
|
||||
assert len(latency_list) == max_size, f"Expected {max_size} entries, got {len(latency_list)}"
|
||||
|
||||
|
|
@ -1062,7 +1064,7 @@ async def test_latency_list_trimming_discards_oldest_entry_async():
|
|||
|
||||
latency_key = f"{model_group}_map"
|
||||
cached_data = await test_cache.async_get_cache(key=latency_key)
|
||||
latency_list = cached_data[deployment_id]["latency"]
|
||||
latency_list = cached_data[deployment_id]["latency_normalized"]
|
||||
|
||||
assert len(latency_list) == max_size
|
||||
|
||||
|
|
@ -1129,7 +1131,7 @@ async def test_timeout_penalty_discards_oldest_entry():
|
|||
|
||||
latency_key = f"{model_group}_map"
|
||||
cached_data = await test_cache.async_get_cache(key=latency_key)
|
||||
latency_list = cached_data[deployment_id]["latency"]
|
||||
latency_list = cached_data[deployment_id]["latency_normalized"]
|
||||
|
||||
assert len(latency_list) == max_size
|
||||
|
||||
|
|
@ -1181,7 +1183,7 @@ def test_list_order_preserved_after_multiple_trims():
|
|||
|
||||
latency_key = f"{model_group}_map"
|
||||
cached_data = test_cache.get_cache(key=latency_key)
|
||||
latency_list = cached_data[deployment_id]["latency"]
|
||||
latency_list = cached_data[deployment_id]["latency_normalized"]
|
||||
|
||||
assert len(latency_list) == max_size
|
||||
|
||||
|
|
@ -1423,7 +1425,7 @@ def test_latency_cache_honors_custom_ttl() -> None:
|
|||
latency_key: Final = "ttl-group_map"
|
||||
cached: Final = cache.get_cache(key=latency_key)
|
||||
assert isinstance(cached, dict)
|
||||
assert cached["deployment"]["latency"] == [1.0]
|
||||
assert cached["deployment"]["latency_normalized"] == [1.0]
|
||||
assert cache.in_memory_cache.ttl_dict[latency_key] == 103.0
|
||||
|
||||
clock.return_value = 104.0
|
||||
|
|
@ -1461,3 +1463,191 @@ async def test_router_model_group_usage_increases_by_logged_tokens() -> None:
|
|||
updated_usage: Final = await router.get_model_group_usage(model_group=model_group)
|
||||
|
||||
assert updated_usage[0] == initial_usage[0] + 15
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_callback", [False, True])
|
||||
@pytest.mark.parametrize("normalized_usage", [False, True])
|
||||
@pytest.mark.parametrize("native_seconds", [2, 3])
|
||||
async def test_responses_and_chat_latency_use_comparable_token_units(
|
||||
async_callback: bool, normalized_usage: bool, native_seconds: int
|
||||
) -> None:
|
||||
cache: Final = DualCache()
|
||||
handler: Final = LowestLatencyLoggingHandler(router_cache=cache)
|
||||
usage: Final = ResponseAPIUsage(input_tokens=10, output_tokens=100, total_tokens=110)
|
||||
native: Final = ResponsesAPIResponse(id="resp-latency", created_at=0, output=[], usage=usage)
|
||||
logged_native: Final = (
|
||||
native.model_copy(update={"usage": ResponseAPILoggingUtils.transform_response_api_usage_to_chat_usage(usage)})
|
||||
if normalized_usage
|
||||
else native
|
||||
)
|
||||
original_usage: Final = logged_native.usage
|
||||
assert original_usage is not None
|
||||
original_fields: Final = original_usage.model_dump()
|
||||
start: Final = datetime(2026, 1, 1)
|
||||
for deployment_id, response, seconds in (
|
||||
("native", logged_native, native_seconds),
|
||||
("bridge", _chat_response(100), 3),
|
||||
):
|
||||
kwargs: Final = {
|
||||
"call_type": "aresponses",
|
||||
"litellm_params": {"metadata": {"model_group": "mixed"}, "model_info": {"id": deployment_id}},
|
||||
}
|
||||
if async_callback:
|
||||
await handler.async_log_success_event(kwargs, response, start, start + timedelta(seconds=seconds))
|
||||
else:
|
||||
handler.log_success_event(kwargs, response, start, start + timedelta(seconds=seconds))
|
||||
recorded: Final = cache.get_cache(key="mixed_map")
|
||||
assert recorded["native"]["latency_normalized"] == pytest.approx([native_seconds / 100])
|
||||
assert recorded["bridge"]["latency_normalized"] == pytest.approx([3 / 100])
|
||||
assert logged_native.usage is original_usage
|
||||
assert logged_native.usage.model_dump() == original_fields
|
||||
assert native.usage is usage
|
||||
deployments: Final = [
|
||||
{"model_name": "mixed", "model_info": {"id": name}, "litellm_params": {}} for name in ("native", "bridge")
|
||||
]
|
||||
picked: Final = handler.get_available_deployments(
|
||||
model_group="mixed", healthy_deployments=deployments, request_kwargs={"stream": False, "metadata": {}}
|
||||
)
|
||||
assert picked["model_info"]["id"] in ({"native"} if native_seconds < 3 else {"native", "bridge"})
|
||||
native_counts: Final = tuple(value for key, value in recorded["native"].items() if key != "latency_normalized")
|
||||
assert native_counts == ({"tpm": 0, "rpm": 1},)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_callback", [False, True])
|
||||
@pytest.mark.parametrize(
|
||||
"response",
|
||||
[
|
||||
litellm.ModelResponse(),
|
||||
ResponsesAPIResponse(id="resp-missing", created_at=0, output=[], usage=None),
|
||||
ResponsesAPIResponse(
|
||||
id="resp-zero", created_at=0, output=[],
|
||||
usage=ResponseAPIUsage(input_tokens=10, output_tokens=0, total_tokens=10),
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_responses_without_output_tokens_keep_elapsed_seconds(
|
||||
async_callback: bool, response: litellm.ModelResponse | ResponsesAPIResponse
|
||||
) -> None:
|
||||
cache: Final = DualCache()
|
||||
handler: Final = LowestLatencyLoggingHandler(router_cache=cache)
|
||||
original_usage: Final = getattr(response, "usage", None)
|
||||
if async_callback:
|
||||
await handler.async_log_success_event(KWARGS, response, 1.0, 3.0)
|
||||
else:
|
||||
handler.log_success_event(KWARGS, response, 1.0, 3.0)
|
||||
assert _recorded_latencies(cache) == [2.0]
|
||||
assert getattr(response, "usage", None) is original_usage
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("frozen_latency_clock")
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True])
|
||||
@pytest.mark.parametrize("preserve_new_samples", [False, True])
|
||||
async def test_legacy_latency_does_not_starve_deployments_after_upgrade(
|
||||
async_mode: bool, preserve_new_samples: bool
|
||||
) -> None:
|
||||
cache: Final = DualCache()
|
||||
handler: Final = LowestLatencyLoggingHandler(router_cache=cache)
|
||||
minute: Final = FROZEN_NOW.strftime("%Y-%m-%d-%H-%M")
|
||||
legacy: Final = {
|
||||
"native": {"latency": [2.0], "time_to_first_token_seconds": [0.1], minute: {"tpm": 0, "rpm": 1}},
|
||||
"bridge": {"latency": [0.03], "time_to_first_token_seconds": [0.2], minute: {"tpm": 110, "rpm": 1}},
|
||||
}
|
||||
cache.set_cache(key="mixed_map", value=copy.deepcopy(legacy))
|
||||
deployments: Final = [
|
||||
{"model_name": "mixed", "model_info": {"id": name}, "litellm_params": {}} for name in ("native", "bridge")
|
||||
]
|
||||
native: Final = ResponsesAPIResponse(
|
||||
id="resp-upgrade", created_at=0, output=[],
|
||||
usage=ResponseAPIUsage(input_tokens=10, output_tokens=100, total_tokens=110),
|
||||
)
|
||||
for iteration in range(2):
|
||||
bridge_kwargs: Final = {
|
||||
"litellm_params": {"metadata": {"model_group": "mixed"}, "model_info": {"id": "bridge"}},
|
||||
}
|
||||
if async_mode:
|
||||
await handler.async_log_success_event(bridge_kwargs, _chat_response(100), 0.0, 3.0)
|
||||
else:
|
||||
handler.log_success_event(bridge_kwargs, _chat_response(100), 0.0, 3.0)
|
||||
selected: Final = (
|
||||
await handler.async_get_available_deployments(model_group="mixed", healthy_deployments=deployments)
|
||||
if async_mode
|
||||
else handler.get_available_deployments(model_group="mixed", healthy_deployments=deployments)
|
||||
)
|
||||
assert selected["model_info"]["id"] == "native"
|
||||
native_kwargs: Final = {
|
||||
"litellm_params": {"metadata": {"model_group": "mixed"}, "model_info": {"id": "native"}},
|
||||
}
|
||||
if async_mode:
|
||||
await handler.async_log_success_event(native_kwargs, native, 0.0, 2.0)
|
||||
else:
|
||||
handler.log_success_event(native_kwargs, native, 0.0, 2.0)
|
||||
recorded: Final = copy.deepcopy(cache.get_cache(key="mixed_map"))
|
||||
assert recorded["native"]["latency_normalized"] == pytest.approx(
|
||||
[0.02] * (iteration + 1 if preserve_new_samples else 1)
|
||||
)
|
||||
assert recorded["native"]["latency"] == [2.0]
|
||||
assert recorded["native"]["time_to_first_token_seconds"] == [0.1]
|
||||
assert recorded["bridge"]["time_to_first_token_seconds"] == [0.2]
|
||||
assert recorded["native"][minute] == {"tpm": 0, "rpm": iteration + 2}
|
||||
assert recorded["bridge"][minute] == {"tpm": 110 + (iteration + 1) * 110, "rpm": iteration + 2}
|
||||
streaming: Final = handler.get_available_deployments(
|
||||
model_group="mixed", healthy_deployments=deployments, request_kwargs={"stream": True}
|
||||
)
|
||||
assert streaming["model_info"]["id"] == "native"
|
||||
cache.set_cache(
|
||||
key="mixed_map",
|
||||
value={
|
||||
name: {**(recorded[name] if preserve_new_samples else legacy[name]), minute: recorded[name][minute]}
|
||||
for name in ("native", "bridge")
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True])
|
||||
async def test_normalized_latency_window_evicts_samples_without_changing_legacy_data(async_mode: bool) -> None:
|
||||
cache: Final = DualCache()
|
||||
cache.set_cache(key="gemini-embedding-001_map", value={DEPLOYMENT_ID: {"latency": [2.0]}})
|
||||
handler: Final = LowestLatencyLoggingHandler(router_cache=cache, routing_args={"max_latency_list_size": 2})
|
||||
for elapsed, expected in ((2.0, [0.02]), (3.0, [0.02, 0.03]), (4.0, [0.03, 0.04])):
|
||||
if async_mode:
|
||||
await handler.async_log_success_event(KWARGS, _chat_response(100), 0.0, elapsed)
|
||||
else:
|
||||
handler.log_success_event(KWARGS, _chat_response(100), 0.0, elapsed)
|
||||
recorded: Final = cache.get_cache(key="gemini-embedding-001_map")[DEPLOYMENT_ID]
|
||||
assert recorded["latency_normalized"] == pytest.approx(expected)
|
||||
assert recorded["latency"] == [2.0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("existing_samples", [(), (0.02, 0.03)])
|
||||
async def test_timeout_penalty_updates_normalized_latency_and_avoids_failed_deployment(
|
||||
existing_samples: tuple[float, ...],
|
||||
) -> None:
|
||||
cache: Final = DualCache()
|
||||
handler: Final = LowestLatencyLoggingHandler(router_cache=cache, routing_args={"max_latency_list_size": 2})
|
||||
cache.set_cache(
|
||||
key="mixed_map",
|
||||
value={
|
||||
"native": {"latency": [0.001], "latency_normalized": list(existing_samples)},
|
||||
"bridge": {"latency": [2.0], "latency_normalized": [0.04]},
|
||||
},
|
||||
)
|
||||
await handler.async_log_failure_event(
|
||||
kwargs={
|
||||
"exception": litellm.Timeout(message="request timed out", model="test", llm_provider="openai"),
|
||||
"litellm_params": {"metadata": {"model_group": "mixed"}, "model_info": {"id": "native"}},
|
||||
},
|
||||
response_obj=None, start_time=0.0, end_time=3.0,
|
||||
)
|
||||
recorded: Final = cache.get_cache(key="mixed_map")["native"]
|
||||
assert recorded["latency_normalized"] == [*existing_samples[-1:], 1000.0]
|
||||
assert recorded["latency"] == [0.001]
|
||||
selected: Final = await handler.async_get_available_deployments(
|
||||
model_group="mixed",
|
||||
healthy_deployments=[{"model_info": {"id": name}, "litellm_params": {}} for name in ("native", "bridge")],
|
||||
)
|
||||
assert selected["model_info"]["id"] == "bridge"
|
||||
|
|
|
|||
|
|
@ -1249,7 +1249,7 @@ async def test_latency_based_override_learns_from_the_overriding_requests():
|
|||
)
|
||||
|
||||
def samples(deployment_id: str) -> list[float]:
|
||||
recorded = (router.cache.get_cache("grp_map") or {}).get(deployment_id, {}).get("latency", [])
|
||||
recorded = (router.cache.get_cache("grp_map") or {}).get(deployment_id, {}).get("latency_normalized", [])
|
||||
return [latency for latency in recorded if latency > 0]
|
||||
|
||||
async def overriding_call() -> str:
|
||||
|
|
@ -1316,7 +1316,7 @@ async def test_override_matching_a_routing_group_strategy_records_each_request_o
|
|||
deployment_id = response._hidden_params["model_id"]
|
||||
await _async_wait_until(lambda: (router.cache.get_cache("grp_map") or {}).get(deployment_id) is not None)
|
||||
|
||||
assert len(router.cache.get_cache("grp_map")[deployment_id]["latency"]) == 1
|
||||
assert len(router.cache.get_cache("grp_map")[deployment_id]["latency_normalized"]) == 1
|
||||
assert request["litellm_logging_obj"].dynamic_success_callbacks is None
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue