mirror of
https://github.com/BerriAI/litellm.git
synced 2026-08-28 05:25:59 +00:00
Merge pull request #33290 from mihidumh/fix/latency-routing-timedelta
fix(router_strategy): serialize latency for non-chat responses in lowest-latency routing
This commit is contained in:
tin-berri
GitHub
commit
7ac5172686
2 changed files with 197 additions and 19 deletions
|
|
@ -73,13 +73,20 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
precise_minute = f"{current_date}-{current_hour}-{current_minute}"
|
||||
|
||||
response_ms = end_time - start_time
|
||||
if isinstance(response_ms, timedelta):
|
||||
# normalize to float seconds up-front: non-chat responses
|
||||
# (embeddings, speech, image) skip the ModelResponse branch
|
||||
# below, and a raw timedelta appended to the latency list
|
||||
# breaks JSON serialization when the router cache syncs to
|
||||
# Redis (issue #33169)
|
||||
response_ms = response_ms.total_seconds()
|
||||
time_to_first_token_response_time = None
|
||||
|
||||
if kwargs.get("stream", None) is not None and kwargs["stream"] is True:
|
||||
# only log ttft for streaming request
|
||||
time_to_first_token_response_time = kwargs.get("completion_start_time", end_time) - start_time
|
||||
|
||||
final_value: Union[float, timedelta] = response_ms
|
||||
final_value: float = response_ms
|
||||
time_to_first_token: Optional[float] = None
|
||||
total_tokens = 0
|
||||
|
||||
|
|
@ -89,15 +96,12 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
completion_tokens = _usage.completion_tokens
|
||||
total_tokens = _usage.total_tokens
|
||||
|
||||
# Handle both timedelta and float response times
|
||||
if isinstance(response_ms, timedelta):
|
||||
response_seconds = response_ms.total_seconds()
|
||||
else:
|
||||
response_seconds = response_ms
|
||||
# response_ms is already normalized to float seconds above
|
||||
response_seconds = response_ms
|
||||
|
||||
final_value = safe_divide_seconds(response_seconds, completion_tokens)
|
||||
if final_value is not None:
|
||||
final_value = float(final_value)
|
||||
normalized_value = safe_divide_seconds(response_seconds, completion_tokens)
|
||||
if normalized_value is not None:
|
||||
final_value = float(normalized_value)
|
||||
else:
|
||||
final_value = response_seconds
|
||||
|
||||
|
|
@ -262,12 +266,19 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
precise_minute = f"{current_date}-{current_hour}-{current_minute}"
|
||||
|
||||
response_ms = end_time - start_time
|
||||
if isinstance(response_ms, timedelta):
|
||||
# normalize to float seconds up-front: non-chat responses
|
||||
# (embeddings, speech, image) skip the ModelResponse branch
|
||||
# below, and a raw timedelta appended to the latency list
|
||||
# breaks JSON serialization when the router cache syncs to
|
||||
# Redis (issue #33169)
|
||||
response_ms = response_ms.total_seconds()
|
||||
time_to_first_token_response_time = None
|
||||
if kwargs.get("stream", None) is not None and kwargs["stream"] is True:
|
||||
# only log ttft for streaming request
|
||||
time_to_first_token_response_time = kwargs.get("completion_start_time", end_time) - start_time
|
||||
|
||||
final_value: Union[float, timedelta] = response_ms
|
||||
final_value: float = response_ms
|
||||
total_tokens = 0
|
||||
time_to_first_token: Optional[float] = None
|
||||
|
||||
|
|
@ -277,17 +288,14 @@ class LowestLatencyLoggingHandler(CustomLogger):
|
|||
completion_tokens = _usage.completion_tokens
|
||||
total_tokens = _usage.total_tokens
|
||||
|
||||
# Handle both timedelta and float response times
|
||||
if isinstance(response_ms, timedelta):
|
||||
response_seconds = response_ms.total_seconds()
|
||||
else:
|
||||
response_seconds = response_ms
|
||||
# response_ms is already normalized to float seconds above
|
||||
response_seconds = response_ms
|
||||
|
||||
final_value = safe_divide_seconds(response_seconds, completion_tokens)
|
||||
if final_value is not None:
|
||||
final_value = float(final_value)
|
||||
normalized_value = safe_divide_seconds(response_seconds, completion_tokens)
|
||||
if normalized_value is not None:
|
||||
final_value = float(normalized_value)
|
||||
else:
|
||||
final_value = response_ms
|
||||
final_value = response_seconds
|
||||
|
||||
if time_to_first_token_response_time is not None:
|
||||
if isinstance(time_to_first_token_response_time, timedelta):
|
||||
|
|
|
|||
170
tests/test_litellm/router_strategy/test_lowest_latency.py
Normal file
170
tests/test_litellm/router_strategy/test_lowest_latency.py
Normal file
|
|
@ -0,0 +1,170 @@
|
|||
#### What this tests ####
|
||||
# Latency values recorded by lowest-latency routing must be JSON
|
||||
# serializable for non-chat responses too (embeddings/speech/image skip
|
||||
# the ModelResponse branch, so the raw timedelta used to leak into the
|
||||
# latency list and break the Redis cache sync). Issue #33169.
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(
|
||||
0, os.path.abspath("../../..")
|
||||
) # Adds the parent directory to the system path
|
||||
|
||||
import litellm
|
||||
from litellm.caching.caching import DualCache
|
||||
from litellm.router_strategy.lowest_latency import LowestLatencyLoggingHandler
|
||||
|
||||
DEPLOYMENT_ID = "9876"
|
||||
KWARGS = {
|
||||
"litellm_params": {
|
||||
"metadata": {
|
||||
"model_group": "gemini-embedding-001",
|
||||
"deployment": "vertex_ai/gemini-embedding-001",
|
||||
},
|
||||
"model_info": {"id": DEPLOYMENT_ID},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def _embedding_response():
|
||||
return litellm.EmbeddingResponse(
|
||||
model="gemini-embedding-001",
|
||||
data=[{"embedding": [0.1, 0.2], "index": 0, "object": "embedding"}],
|
||||
object="list",
|
||||
usage=litellm.Usage(prompt_tokens=5, completion_tokens=0, total_tokens=5),
|
||||
)
|
||||
|
||||
|
||||
def _recorded_latencies(cache: DualCache):
|
||||
cached = cache.get_cache(key="gemini-embedding-001_map") or {}
|
||||
return cached.get(DEPLOYMENT_ID, {}).get("latency", [])
|
||||
|
||||
|
||||
def test_sync_embedding_latency_is_json_serializable():
|
||||
"""log_success_event with datetime start/end (as the proxy passes) must not
|
||||
record a raw timedelta for non-ModelResponse results."""
|
||||
cache = DualCache()
|
||||
handler = LowestLatencyLoggingHandler(router_cache=cache)
|
||||
|
||||
start_time = datetime(2026, 1, 1, 12, 0, 0)
|
||||
end_time = datetime(2026, 1, 1, 12, 0, 2)
|
||||
|
||||
handler.log_success_event(
|
||||
response_obj=_embedding_response(),
|
||||
kwargs=KWARGS,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
)
|
||||
|
||||
latencies = _recorded_latencies(cache)
|
||||
assert latencies, "expected a latency entry to be recorded"
|
||||
assert all(
|
||||
not isinstance(value, timedelta) for value in latencies
|
||||
), f"raw timedelta leaked into latency list: {latencies}"
|
||||
assert latencies[-1] == pytest.approx(2.0)
|
||||
# the exact failure mode from production: redis cache sync json.dumps
|
||||
json.dumps({"latency": latencies})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_embedding_latency_is_json_serializable():
|
||||
"""async_log_success_event is the path the proxy actually hits."""
|
||||
cache = DualCache()
|
||||
handler = LowestLatencyLoggingHandler(router_cache=cache)
|
||||
|
||||
start_time = datetime(2026, 1, 1, 12, 0, 0)
|
||||
end_time = datetime(2026, 1, 1, 12, 0, 3)
|
||||
|
||||
await handler.async_log_success_event(
|
||||
response_obj=_embedding_response(),
|
||||
kwargs=KWARGS,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
)
|
||||
|
||||
latencies = _recorded_latencies(cache)
|
||||
assert latencies, "expected a latency entry to be recorded"
|
||||
assert all(
|
||||
not isinstance(value, timedelta) for value in latencies
|
||||
), f"raw timedelta leaked into latency list: {latencies}"
|
||||
assert latencies[-1] == pytest.approx(3.0)
|
||||
json.dumps({"latency": latencies})
|
||||
|
||||
|
||||
def _chat_response(completion_tokens: int):
|
||||
return litellm.ModelResponse(
|
||||
model="gpt-4o-mini",
|
||||
choices=[
|
||||
litellm.Choices(
|
||||
finish_reason="stop",
|
||||
index=0,
|
||||
message=litellm.Message(content="hi", role="assistant"),
|
||||
)
|
||||
],
|
||||
usage=litellm.Usage(
|
||||
prompt_tokens=10,
|
||||
completion_tokens=completion_tokens,
|
||||
total_tokens=10 + completion_tokens,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_chat_latency_normalized_per_token():
|
||||
"""Chat responses go through the per-token normalization branch — with the
|
||||
up-front timedelta conversion the stored value must be seconds/token."""
|
||||
cache = DualCache()
|
||||
handler = LowestLatencyLoggingHandler(router_cache=cache)
|
||||
|
||||
await handler.async_log_success_event(
|
||||
response_obj=_chat_response(completion_tokens=4),
|
||||
kwargs=KWARGS,
|
||||
start_time=datetime(2026, 1, 1, 12, 0, 0),
|
||||
end_time=datetime(2026, 1, 1, 12, 0, 2),
|
||||
)
|
||||
|
||||
latencies = _recorded_latencies(cache)
|
||||
assert latencies and latencies[-1] == pytest.approx(0.5) # 2s / 4 tokens
|
||||
json.dumps({"latency": latencies})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_chat_zero_completion_tokens_falls_back_to_seconds():
|
||||
"""safe_divide_seconds returns None for zero tokens — the fallback branch
|
||||
must store plain float seconds, not a timedelta."""
|
||||
cache = DualCache()
|
||||
handler = LowestLatencyLoggingHandler(router_cache=cache)
|
||||
|
||||
await handler.async_log_success_event(
|
||||
response_obj=_chat_response(completion_tokens=0),
|
||||
kwargs=KWARGS,
|
||||
start_time=datetime(2026, 1, 1, 12, 0, 0),
|
||||
end_time=datetime(2026, 1, 1, 12, 0, 3),
|
||||
)
|
||||
|
||||
latencies = _recorded_latencies(cache)
|
||||
assert latencies and latencies[-1] == pytest.approx(3.0)
|
||||
assert not isinstance(latencies[-1], timedelta)
|
||||
json.dumps({"latency": latencies})
|
||||
|
||||
|
||||
def test_sync_chat_zero_completion_tokens_falls_back_to_seconds():
|
||||
cache = DualCache()
|
||||
handler = LowestLatencyLoggingHandler(router_cache=cache)
|
||||
|
||||
handler.log_success_event(
|
||||
response_obj=_chat_response(completion_tokens=0),
|
||||
kwargs=KWARGS,
|
||||
start_time=datetime(2026, 1, 1, 12, 0, 0),
|
||||
end_time=datetime(2026, 1, 1, 12, 0, 2),
|
||||
)
|
||||
|
||||
latencies = _recorded_latencies(cache)
|
||||
assert latencies and latencies[-1] == pytest.approx(2.0)
|
||||
assert not isinstance(latencies[-1], timedelta)
|
||||
json.dumps({"latency": latencies})
|
||||
Loading…
Add table
Reference in a new issue