scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-09 22:31:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 133

fix(router): rank streaming latency routing by raw TTFT, not TTFT per token (#40202)

Browse source 
* fix(router): rank streaming latency routing by raw TTFT, not TTFT per token

Latency-based routing divided time-to-first-token by completion_tokens
before storing it, so a deployment that streamed a long answer looked
faster to first token than one that answered briefly. TTFT is now stored
as plain seconds (first token time minus request start) in both the sync
and async success handlers, which is what the routing decision compares.

Non-streaming latency normalization per output token is unchanged.

Claude-Session: https://claude.ai/code/session_01Ttd5Q9ZhRPB4ch5guos3rj

* fix(router): store streaming TTFT under a seconds-only cache key

Workers on the previous release keep writing seconds-per-token samples
under "time_to_first_token" in the shared router cache during a rolling
deploy, so mixing the new raw-seconds samples into the same list averaged
incompatible units. Raw TTFT now lives under "time_to_first_token_seconds"
and routing reads only that key.

Also fix the regression test's token counts: with 50 tokens on the fast
deployment and 500 on the slow one the old per-token formula picks the
slow deployment, so the routing assertion now catches the bug.

Claude-Session: https://claude.ai/code/session_01Ttd5Q9ZhRPB4ch5guos3rj

* test(router): cover the TTFT sliding window from the unit-test shard

Move the TTFT list trimming checks from the CircleCI-only suite into the
mapped unit test file as one sync/async parametrized test, so the changed
lines in lowest_latency.py are exercised by the GitHub unit-test shard
that reports patch coverage.

Claude-Session: https://claude.ai/code/session_01Ttd5Q9ZhRPB4ch5guos3rj
This commit is contained in:
Clement 2026-09-09 01:05:23 +08:00 • committed by GitHub
parent 1af7a403c6
commit 95c0f9db7d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 149 additions and 168 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
litellm/router_strategy/lowest_latency.py
View file

@ -32,6 +32,12 @@ def _average_latency(samples: Sequence[float]) -> float:
return sum(samples) / len(samples)
def _ttft_seconds(elapsed: timedelta | float) -> float:
if isinstance(elapsed, timedelta):
return elapsed.total_seconds()
return float(elapsed)
class LowestLatencyLoggingHandler(CustomLogger):
test_flag: bool = False
logged_success: int = 0
@ -86,14 +92,13 @@ class LowestLatencyLoggingHandler(CustomLogger):
# 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
time_to_first_token: float | None = 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
time_to_first_token = _ttft_seconds(kwargs.get("completion_start_time", end_time) - start_time)
final_value: float = response_ms
time_to_first_token: float | None = None
total_tokens = 0
if isinstance(response_obj, ModelResponse):
@ -111,13 +116,6 @@ class LowestLatencyLoggingHandler(CustomLogger):
else:
final_value = response_seconds
if time_to_first_token_response_time is not None:
if isinstance(time_to_first_token_response_time, timedelta):
ttft_seconds = time_to_first_token_response_time.total_seconds()
else:
ttft_seconds = time_to_first_token_response_time
time_to_first_token = safe_divide_seconds(ttft_seconds, completion_tokens)
# ------------
# Update usage
# ------------
@ -138,14 +136,14 @@ class LowestLatencyLoggingHandler(CustomLogger):
## Time to first token
if time_to_first_token is not None:
if (
len(request_count_dict[id].get("time_to_first_token", []))
len(request_count_dict[id].get("time_to_first_token_seconds", []))
< self.routing_args.max_latency_list_size
):
request_count_dict[id].setdefault("time_to_first_token", []).append(time_to_first_token)
request_count_dict[id].setdefault("time_to_first_token_seconds", []).append(time_to_first_token)
else:
request_count_dict[id]["time_to_first_token"] = request_count_dict[id]["time_to_first_token"][
1:
] + [time_to_first_token]
request_count_dict[id]["time_to_first_token_seconds"] = request_count_dict[id][
"time_to_first_token_seconds"
][1:] + [time_to_first_token]
if precise_minute not in request_count_dict[id]:
request_count_dict[id][precise_minute] = {}
@ -252,7 +250,7 @@ class LowestLatencyLoggingHandler(CustomLogger):
{model_group}_map: {
id: {
"latency": [..]
"time_to_first_token": [..]
"time_to_first_token_seconds": [..]
f"{date:hour:minute}" : {"tpm": 34, "rpm": 3}
}
}
@ -273,14 +271,13 @@ class LowestLatencyLoggingHandler(CustomLogger):
# 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
time_to_first_token: float | None = 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
time_to_first_token = _ttft_seconds(kwargs.get("completion_start_time", end_time) - start_time)
final_value: float = response_ms
total_tokens = 0
time_to_first_token: float | None = None
if isinstance(response_obj, ModelResponse):
_usage: Final = getattr(response_obj, "usage", None)
@ -296,13 +293,6 @@ class LowestLatencyLoggingHandler(CustomLogger):
final_value = float(normalized_value)
else:
final_value = response_seconds
if time_to_first_token_response_time is not None:
if isinstance(time_to_first_token_response_time, timedelta):
ttft_seconds = time_to_first_token_response_time.total_seconds()
else:
ttft_seconds = time_to_first_token_response_time
time_to_first_token = safe_divide_seconds(ttft_seconds, completion_tokens)
# ------------
# Update usage
# ------------
@ -328,14 +318,14 @@ class LowestLatencyLoggingHandler(CustomLogger):
## Time to first token
if time_to_first_token is not None:
if (
len(request_count_dict[id].get("time_to_first_token", []))
len(request_count_dict[id].get("time_to_first_token_seconds", []))
< self.routing_args.max_latency_list_size
):
request_count_dict[id].setdefault("time_to_first_token", []).append(time_to_first_token)
request_count_dict[id].setdefault("time_to_first_token_seconds", []).append(time_to_first_token)
else:
request_count_dict[id]["time_to_first_token"] = request_count_dict[id]["time_to_first_token"][
1:
] + [time_to_first_token]
request_count_dict[id]["time_to_first_token_seconds"] = request_count_dict[id][
"time_to_first_token_seconds"
][1:] + [time_to_first_token]
if precise_minute not in request_count_dict[id]:
request_count_dict[id][precise_minute] = {}
@ -433,7 +423,7 @@ class LowestLatencyLoggingHandler(CustomLogger):
or float("inf")
)
item_latency = item_map.get("latency", [])
item_ttft_latency = item_map.get("time_to_first_token", [])
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)

136
tests/local_testing/test_lowest_latency_routing.py
View file

@ -1077,73 +1077,6 @@ async def test_latency_list_trimming_discards_oldest_entry_async():
), f"Oldest latency {oldest_latency} should have been discarded"
def test_ttft_list_trimming_discards_oldest_entry():
"""
The time_to_first_token list trims the oldest entry when full, matching
the behavior of the latency list.
"""
max_size = 3
test_cache = DualCache()
lowest_latency_logger = LowestLatencyLoggingHandler(
router_cache=test_cache, routing_args={"max_latency_list_size": max_size}
)
model_group = "gpt-3.5-turbo"
deployment_id = "test-deployment"
ttft_values = []
for i in range(max_size + 1):
start_time = time.time()
expected_ttft = float(i + 1) * 0.1 # 0.1, 0.2, 0.3, 0.4
completion_start_time = start_time + expected_ttft
end_time = start_time + float(i + 1)
ttft_values.append(expected_ttft)
kwargs = {
"litellm_params": {
"metadata": {
"model_group": model_group,
"deployment": "azure/gpt-4.1-mini",
},
"model_info": {"id": deployment_id},
},
"stream": True,
"completion_start_time": completion_start_time,
}
# TTFT is only recorded when response_obj is a ModelResponse.
response_obj = litellm.ModelResponse(
usage=litellm.Usage(completion_tokens=1, total_tokens=1)
)
lowest_latency_logger.log_success_event(
response_obj=response_obj,
kwargs=kwargs,
start_time=start_time,
end_time=end_time,
)
latency_key = f"{model_group}_map"
cached_data = test_cache.get_cache(key=latency_key)
ttft_list = cached_data[deployment_id].get("time_to_first_token", [])
assert (
len(ttft_list) == max_size
), f"Expected {max_size} entries, got {len(ttft_list)}"
newest_ttft = ttft_values[-1]
oldest_ttft = ttft_values[0]
tolerance = 0.05
assert (
abs(ttft_list[-1] - newest_ttft) < tolerance
), f"Newest TTFT {newest_ttft} should be at end of list"
for ttft in ttft_list:
assert (
abs(ttft - oldest_ttft) > tolerance
), f"Oldest TTFT {oldest_ttft} should have been discarded"
@pytest.mark.asyncio
async def test_timeout_penalty_discards_oldest_entry():
"""
@ -1269,72 +1202,3 @@ def test_list_order_preserved_after_multiple_trims():
assert (
abs(latency_list[i] - expected) < tolerance
), f"At index {i}, expected ~{expected}, got {latency_list[i]}"
@pytest.mark.asyncio
async def test_ttft_list_trimming_discards_oldest_entry_async():
"""
Async counterpart: the time_to_first_token list trims the oldest entry
when full. Exercises the async_log_success_event TTFT path, which only
runs when response_obj is a ModelResponse and the call is marked as
streaming with a completion_start_time.
"""
max_size = 3
test_cache = DualCache()
lowest_latency_logger = LowestLatencyLoggingHandler(
router_cache=test_cache, routing_args={"max_latency_list_size": max_size}
)
model_group = "gpt-3.5-turbo"
deployment_id = "test-deployment"
ttft_values = []
for i in range(max_size + 1):
start_time = time.time()
expected_ttft = float(i + 1) * 0.1 # 0.1, 0.2, 0.3, 0.4
completion_start_time = start_time + expected_ttft
end_time = start_time + float(i + 1)
ttft_values.append(expected_ttft)
kwargs = {
"litellm_params": {
"metadata": {
"model_group": model_group,
"deployment": "azure/gpt-4.1-mini",
},
"model_info": {"id": deployment_id},
},
"stream": True,
"completion_start_time": completion_start_time,
}
response_obj = litellm.ModelResponse(
usage=litellm.Usage(completion_tokens=1, total_tokens=1)
)
await lowest_latency_logger.async_log_success_event(
response_obj=response_obj,
kwargs=kwargs,
start_time=start_time,
end_time=end_time,
)
latency_key = f"{model_group}_map"
cached_data = await test_cache.async_get_cache(key=latency_key)
ttft_list = cached_data[deployment_id].get("time_to_first_token", [])
assert (
len(ttft_list) == max_size
), f"Expected {max_size} entries, got {len(ttft_list)}"
newest_ttft = ttft_values[-1]
oldest_ttft = ttft_values[0]
tolerance = 0.05
assert (
abs(ttft_list[-1] - newest_ttft) < tolerance
), f"Newest TTFT {newest_ttft} should be at end of list"
for ttft in ttft_list:
assert (
abs(ttft - oldest_ttft) > tolerance
), f"Oldest TTFT {oldest_ttft} should have been discarded"

127
tests/test_litellm/router_strategy/test_lowest_latency.py
View file

@ -165,6 +165,133 @@ def test_sync_chat_zero_completion_tokens_falls_back_to_seconds():
json.dumps({"latency": latencies})
MODEL_GROUP = "gpt-4o-mini"
FAST_TTFT_ID = "fast-ttft-short-output"
SLOW_TTFT_ID = "slow-ttft-long-output"
STREAMING_DEPLOYMENTS = [
{"model_info": {"id": FAST_TTFT_ID}, "litellm_params": {}},
{"model_info": {"id": SLOW_TTFT_ID}, "litellm_params": {}},
]
def _streaming_kwargs(deployment_id: str, start_time: datetime, ttft_seconds: float):
return {
"litellm_params": {
"metadata": {"model_group": MODEL_GROUP},
"model_info": {"id": deployment_id},
},
"stream": True,
"completion_start_time": start_time + timedelta(seconds=ttft_seconds),
}
def _recorded_ttft(cache: DualCache, deployment_id: str):
cached = cache.get_cache(key=f"{MODEL_GROUP}_map") or {}
return cached.get(deployment_id, {}).get("time_to_first_token_seconds", [])
@pytest.mark.asyncio
@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
async def test_streaming_ttft_ranking_ignores_completion_length(sync_mode: bool):
"""Deployment A: TTFT 1s, 50 completion tokens. Deployment B: TTFT 3s, 500
completion tokens. Dividing TTFT by completion tokens made B look faster
(3/500 = 0.006 beats 1/50 = 0.02); actual TTFT must win."""
cache = DualCache()
handler = LowestLatencyLoggingHandler(router_cache=cache)
start_time = datetime(2026, 1, 1, 12, 0, 0)
end_time = start_time + timedelta(seconds=10)
samples = (
(FAST_TTFT_ID, 1.0, 50),
(SLOW_TTFT_ID, 3.0, 500),
)
for deployment_id, ttft, completion_tokens in samples:
kwargs = _streaming_kwargs(deployment_id, start_time, ttft)
response_obj = _chat_response(completion_tokens=completion_tokens)
if sync_mode:
handler.log_success_event(
response_obj=response_obj, kwargs=kwargs, start_time=start_time, end_time=end_time
)
else:
await handler.async_log_success_event(
response_obj=response_obj, kwargs=kwargs, start_time=start_time, end_time=end_time
)
assert _recorded_ttft(cache, FAST_TTFT_ID) == [pytest.approx(1.0)]
assert _recorded_ttft(cache, SLOW_TTFT_ID) == [pytest.approx(3.0)]
request_kwargs = {"stream": True, "metadata": {}}
if sync_mode:
picked = handler.get_available_deployments(
model_group=MODEL_GROUP, healthy_deployments=STREAMING_DEPLOYMENTS, request_kwargs=request_kwargs
)
else:
picked = await handler.async_get_available_deployments(
model_group=MODEL_GROUP, healthy_deployments=STREAMING_DEPLOYMENTS, request_kwargs=request_kwargs
)
assert picked is not None
assert picked["model_info"]["id"] == FAST_TTFT_ID
@pytest.mark.asyncio
@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
async def test_ttft_window_keeps_newest_samples_when_full(sync_mode: bool):
"""Float timestamps, as the SDK passes them. Once max_latency_list_size
samples exist the oldest TTFT is dropped so the window slides."""
max_size = 3
cache = DualCache()
handler = LowestLatencyLoggingHandler(router_cache=cache, routing_args={"max_latency_list_size": max_size})
start_time = 1_700_000_000.0
ttfts = (0.1, 0.2, 0.3, 0.4)
for ttft in ttfts:
kwargs = {
"litellm_params": {
"metadata": {"model_group": MODEL_GROUP},
"model_info": {"id": FAST_TTFT_ID},
},
"stream": True,
"completion_start_time": start_time + ttft,
}
response_obj = _chat_response(completion_tokens=1)
if sync_mode:
handler.log_success_event(
response_obj=response_obj, kwargs=kwargs, start_time=start_time, end_time=start_time + 1.0
)
else:
await handler.async_log_success_event(
response_obj=response_obj, kwargs=kwargs, start_time=start_time, end_time=start_time + 1.0
)
assert _recorded_ttft(cache, FAST_TTFT_ID) == [pytest.approx(ttft) for ttft in ttfts[-max_size:]]
@pytest.mark.asyncio
async def test_streaming_routing_ignores_per_token_ttft_samples_from_older_workers():
"""Workers on the previous release share the Redis map and keep writing
seconds-per-token under the old "time_to_first_token" key during a rolling
deploy. Those samples favor SLOW; routing must only read the seconds key."""
cache = DualCache()
handler = LowestLatencyLoggingHandler(router_cache=cache)
cache.set_cache(
key=f"{MODEL_GROUP}_map",
value={
FAST_TTFT_ID: {"time_to_first_token": [0.02], "time_to_first_token_seconds": [1.0]},
SLOW_TTFT_ID: {"time_to_first_token": [0.006], "time_to_first_token_seconds": [3.0]},
},
)
picked = await handler.async_get_available_deployments(
model_group=MODEL_GROUP,
healthy_deployments=STREAMING_DEPLOYMENTS,
request_kwargs={"stream": True, "metadata": {}},
)
assert picked is not None
assert picked["model_info"]["id"] == FAST_TTFT_ID
@pytest.mark.asyncio
@pytest.mark.parametrize(
"cached_entry",