scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-05 07:37:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 360

fix(router): tolerate null deployment metadata in cost callbacks

Browse source
This commit is contained in:
Tanishq 2026-10-26 01:34:04 +05:30
parent dddbaa803d
commit cf5082d807
2 changed files with 35 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm/router_strategy/lowest_cost.py
View file

@ -58,7 +58,7 @@ class LowestCostLoggingHandler(CustomLogger):
else:
model_group: Final = kwargs["litellm_params"]["metadata"].get("model_group", None)
id = kwargs["litellm_params"].get("model_info", {}).get("id", None)
id = (kwargs["litellm_params"].get("model_info") or {}).get("id", None)
if model_group is None or id is None:
return
elif isinstance(id, int):
@ -134,7 +134,7 @@ class LowestCostLoggingHandler(CustomLogger):
else:
model_group: Final = kwargs["litellm_params"]["metadata"].get("model_group", None)
id = kwargs["litellm_params"].get("model_info", {}).get("id", None)
id = (kwargs["litellm_params"].get("model_info") or {}).get("id", None)
if model_group is None or id is None:
return
elif isinstance(id, int):

33
tests/unit/router_strategy/test_lowest_cost.py
View file

@ -1,5 +1,7 @@
import asyncio, copy, importlib, os, time
import logging
from datetime import datetime
from typing import Final
import pytest
@ -36,6 +38,37 @@ def _recorded_minute_counters(cache: DualCache) -> dict[str, int]:
return next(iter(minute_buckets.values()))
@pytest.mark.asyncio
@pytest.mark.parametrize("use_async", [False, True], ids=["sync", "async"])
@pytest.mark.parametrize(
"deployment_params",
[{}, {"model_info": None}, {"model_info": {}}, {"model_info": {"id": None}}],
ids=["omitted", "null", "empty", "null-id"],
)
async def test_success_without_a_deployment_id_skips_cost_accounting_without_errors(
caplog: pytest.LogCaptureFixture,
use_async: bool,
deployment_params: dict[str, dict[str, str | None] | None],
) -> None:
cache: Final = DualCache()
handler: Final = LowestCostLoggingHandler(router_cache=cache)
call_args: Final = {
"kwargs": {"litellm_params": {"metadata": {"model_group": "gpt-5.5-pool"}, **deployment_params}},
"response_obj": _chat_response_with_no_completion_tokens(),
"start_time": datetime(2026, 1, 1, 12, 0, 0),
"end_time": datetime(2026, 1, 1, 12, 0, 2),
}
with caplog.at_level(logging.ERROR, logger="LiteLLM"):
if use_async:
await handler.async_log_success_event(**call_args)
else:
handler.log_success_event(**call_args)
assert cache.get_cache(key=COST_KEY) is None
assert caplog.record_tuples == []
def test_log_success_event_counts_a_response_with_no_completion_tokens():
cache = DualCache()
handler = LowestCostLoggingHandler(router_cache=cache)