fix(router): price cost-based routing through model_info.base_model (#47617)

* fix(router): price cost-based routing through model_info.base_model

* fix(router): keep cost-based routing pricing helpers fully typed

* fix(router): drop typing.cast from cost-based routing pricing helpers

* fix(router): read the cost map through a typed accessor in cost-based routing

* fix(router): avoid a Final rebind inside the cost-based routing loop

* refactor(router): share deployment_pricing_models across spend, listing and cost-based routing

* refactor(router): keep the /v1/models listing key inline, share pricing models for spend and routing

* refactor(router): inline the deployment pricing-model wrapper in cost-based routing

* refactor(router): read deployment pricing models through one shared helper

deployment_pricing_models(deployment) now owns the model_info.base_model / litellm_params.base_model /
litellm_params.model unpacking, used by cost-based routing and the /v1/models listing. pricing_models(base_model, model)
stays for the failure-row spend estimate, which already holds the two names

* refactor(router): share the str-keyed mapping validator with cost-based routing

* ci: re-run after flaky test_aiohttp_transport read timeout in llms-providers

---------

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/utils.py

@@ -269,6 +269,7 @@ from litellm.repositories.verification_token_repository import (
     VerificationTokenRepository,
 )
 from litellm.router_utils.common_utils import resolve_model_group_alias
+from litellm.router_utils.deployment_pricing import pricing_models
 from litellm.secret_managers.main import str_to_bool
 from litellm.types.integrations.slack_alerting import DEFAULT_ALERT_TYPES
 from litellm.types.llms.openai import ResponsesAPIResponse
@@ -1245,9 +1246,9 @@ def _estimated_input_cost(
     deployment_pricing: Final = deployment_pricing_model_info(model_id=deployment_id, deployment_model=model)
     if deployment_pricing is not None:
         return _deployment_priced_input_cost(model, custom_llm_provider, prompt_tokens, deployment_pricing)
-    pricing_models: Final = (base_model, model) if base_model is not None else (model,)
+    pricing_models_to_try: Final = pricing_models(base_model, model)
     candidate_costs: Final = (
-        _priced_input_cost(pricing_model, custom_llm_provider, prompt_tokens) for pricing_model in pricing_models
+        _priced_input_cost(pricing_model, custom_llm_provider, prompt_tokens) for pricing_model in pricing_models_to_try
     )
     return next((cost for cost in candidate_costs if cost is not None), 0.0)
 

litellm/router.py

@@ -242,6 +242,7 @@ from litellm.router_utils.cooldown_handlers import (
     is_video_lookup_not_found,
     set_cooldown_deployments,
 )
+from litellm.router_utils.deployment_pricing import deployment_pricing_models
 from litellm.router_utils.embedding_batch_size_check import (
     async_log_rejection_before_provider_call,
     clear_rejected_before_provider_call,
@@ -11687,19 +11688,9 @@ class Router:
             )
             for deployment in deployments
         )
-        params: Final = tuple(deployment.get("litellm_params") or MappingProxyType({}) for deployment in deployments)
-        # base_model resolution mirrors get_router_model_info: unset or blank means the
-        # deployment's own model name is the cost-map key.
-        cost_map_keys: Final = tuple(
-            dict.fromkeys(  # deduplicates while preserving config order
-                key
-                for key in (
-                    model_info.get("base_model") or litellm_params.get("base_model") or litellm_params.get("model")
-                    for model_info, litellm_params in zip(model_infos, params)
-                )
-                if isinstance(key, str) and key
-            )
-        )
+        raw_deployments: Final[tuple[object, ...]] = deployments
+        pricing_keys: Final = (deployment_pricing_models(deployment) for deployment in raw_deployments)
+        cost_map_keys: Final = tuple(dict.fromkeys(keys[0] for keys in pricing_keys if keys))
         return DeploymentModelListingInfo(
             cost_map_keys=cost_map_keys,
             max_input_tokens=self._widest_configured_limit(model_infos, "max_input_tokens"),

litellm/router_strategy/lowest_cost.py

@@ -1,5 +1,6 @@
 #### What this does ####
 #   picks based on response time (for streaming, this is time to first token)
+from collections.abc import Mapping
 from datetime import datetime
 from typing import Final
 
@@ -10,6 +11,30 @@ from litellm._logging import verbose_router_logger
 from litellm.caching.caching import DualCache
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.router_utils.batch_utils import is_batch_retrieve_call_type
+from litellm.router_utils.deployment_pricing import deployment_pricing_models, str_keyed
+
+
+def _deployment_cost_map_rates(deployment: object) -> tuple[float | None, float | None]:
+    """Input and output per-token prices from the first cost-map row among the deployment's
+    pricing models that carries one; (None, None) when no row does, so the caller falls back to
+    its unmapped default."""
+    rows: Final = (str_keyed(_cost_map().get(model)) for model in deployment_pricing_models(deployment))
+    row: Final = next((row for row in rows if _declares_token_price(row)), None)
+    if row is None:
+        return None, None
+    return _as_rate(row.get("input_cost_per_token")), _as_rate(row.get("output_cost_per_token"))
+
+
+def _cost_map() -> Mapping[str, object]:
+    return litellm.model_cost
+
+
+def _declares_token_price(row: Mapping[str, object]) -> bool:
+    return row.get("input_cost_per_token") is not None or row.get("output_cost_per_token") is not None
+
+
+def _as_rate(value: object) -> float | None:
+    return float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) else None
 
 
 class LowestCostLoggingHandler(CustomLogger):
@@ -243,8 +268,8 @@ class LowestCostLoggingHandler(CustomLogger):
                 or _deployment.get("model_info", {}).get("rpm", None)
                 or float("inf")
             )
-            item_litellm_model_name = _deployment.get("litellm_params", {}).get("model")
-            item_litellm_model_cost_map = litellm.model_cost.get(item_litellm_model_name, {})
+            deployment_view: object = _deployment
+            map_input_cost, map_output_cost = _deployment_cost_map_rates(deployment_view)
 
             # check if user provided input_cost_per_token and output_cost_per_token in litellm_params
             item_input_cost = None
@@ -256,10 +281,10 @@ class LowestCostLoggingHandler(CustomLogger):
                 item_output_cost = _deployment.get("litellm_params", {}).get("output_cost_per_token")
 
             if item_input_cost is None:
-                item_input_cost = item_litellm_model_cost_map.get("input_cost_per_token", 5.0)
+                item_input_cost = map_input_cost if map_input_cost is not None else 5.0
 
             if item_output_cost is None:
-                item_output_cost = item_litellm_model_cost_map.get("output_cost_per_token", 5.0)
+                item_output_cost = map_output_cost if map_output_cost is not None else 5.0
 
             # if litellm["model"] is not in model_cost map -> use item_cost = $10
 

litellm/router_utils/deployment_pricing.py

@@ -0,0 +1,37 @@
+from collections.abc import Iterator, Mapping
+from types import MappingProxyType
+from typing import Final
+
+from pydantic import TypeAdapter, ValidationError
+
+_EMPTY: Final[Mapping[str, object]] = MappingProxyType({})
+_STR_KEYED: Final = TypeAdapter(dict[str, object])
+
+
+def pricing_models(base_model: object, model: object) -> tuple[str, ...]:
+    """Cost-map keys to price a deployment under, ``model_info.base_model`` first, then the
+    deployment's own model, blanks and duplicates dropped. Every spend and routing surface that
+    prices a deployment walks this order, so a ``base_model`` the cost map does not know falls
+    through to the deployment model instead of billing or ranking at the unmapped default."""
+    return tuple(dict.fromkeys(_named_models(base_model, model)))
+
+
+def deployment_pricing_models(deployment: object) -> tuple[str, ...]:
+    """``pricing_models`` read off a raw deployment dict (``model_info.base_model``, then
+    ``litellm_params.base_model``, then ``litellm_params.model``); malformed blocks count as unset."""
+    view: Final = str_keyed(deployment)
+    params: Final = str_keyed(view.get("litellm_params"))
+    info: Final = str_keyed(view.get("model_info"))
+    return pricing_models(info.get("base_model") or params.get("base_model"), params.get("model"))
+
+
+def _named_models(*candidates: object) -> Iterator[str]:
+    return (candidate for candidate in candidates if isinstance(candidate, str) and candidate)
+
+
+def str_keyed(value: object) -> Mapping[str, object]:
+    """``value`` as a read-only str-keyed mapping, empty when it is not a dict."""
+    try:
+        return MappingProxyType(_STR_KEYED.validate_python(value))
+    except ValidationError:
+        return _EMPTY

tests/unit/router_strategy/test_lowest_cost.py

@@ -112,6 +112,7 @@ def _vcr_outcome_gate(request, vcr):
     yield
     record_vcr_outcome(request, vcr)
 
+
 @pytest.fixture(scope="function")
 def isolate_litellm_state():
     """
@@ -164,6 +165,7 @@ def isolate_litellm_state():
             setattr(litellm, attr, original_value)
     _invalidate_model_cost_lowercase_map()
 
+
 _SCALAR_DEFAULTS = {
     "num_retries": getattr(litellm, "num_retries", None),
     "num_retries_per_request": getattr(litellm, "num_retries_per_request", None),
@@ -184,6 +186,7 @@ _SCALAR_DEFAULTS = {
     "api_key": getattr(litellm, "api_key", None),
 }
 
+
 @pytest.fixture(scope="module")
 def setup_and_teardown():
     """
@@ -206,6 +209,7 @@ def setup_and_teardown():
             litellm.in_memory_llm_clients_cache.flush_cache()
     yield
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "isolate_litellm_state", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_get_available_deployments():
@@ -235,6 +239,7 @@ async def test_get_available_deployments():
 
     assert selected_model["model_info"]["id"] == "groq-llama"
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "isolate_litellm_state", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_get_available_deployments_custom_price():
@@ -286,6 +291,7 @@ async def test_get_available_deployments_custom_price():
 
     assert selected_model["model_info"]["id"] == "chatgpt-v-1"
 
+
 async def _deploy(lowest_cost_logger, deployment_id, tokens_used, duration):
     kwargs = {
         "litellm_params": {
@@ -307,6 +313,7 @@ async def _deploy(lowest_cost_logger, deployment_id, tokens_used, duration):
         end_time=end_time,
     )
 
+
 @pytest.mark.usefixtures("_vcr_outcome_gate", "isolate_litellm_state", "setup_and_teardown")
 @pytest.mark.parametrize("ans_rpm", [1, 5])  # 1 should produce nothing, 10 should select first
 @pytest.mark.asyncio
@@ -357,3 +364,42 @@ async def test_get_available_endpoints_tpm_rpm_check_async(ans_rpm):
     assert (d_ans and d_ans["model_info"]["id"]) == ans
 
     print("selected deployment:", d_ans)
+
+
+_TWO_DOLLARS_PER_TOKEN = {"input_cost_per_token": 1.0, "output_cost_per_token": 1.0}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "azure_litellm_params, base_model",
+    [
+        ({"model": "azure/my-gpt4-deploy"}, "azure/gpt-4.1"),
+        ({"model": "azure/gpt-4.1"}, "azure/not-a-real-model"),
+    ],
+    ids=["base_model_priced", "unmapped_base_model_falls_back_to_deployment_model"],
+)
+async def test_cost_based_routing_prices_an_azure_deployment_through_its_base_model(
+    azure_litellm_params: dict, base_model: str
+):
+    handler = LowestCostLoggingHandler(router_cache=DualCache())
+    healthy_deployments = [
+        {
+            "model_name": "gpt-4.1",
+            "litellm_params": {"model": "openai/gpt-4.1", **_TWO_DOLLARS_PER_TOKEN},
+            "model_info": {"id": "openai-two-dollars"},
+        },
+        {
+            "model_name": "gpt-4.1",
+            "litellm_params": azure_litellm_params,
+            "model_info": {"id": "azure-cost-map-priced", "base_model": base_model},
+        },
+    ]
+
+    picked = await handler.async_get_available_deployments(
+        model_group="gpt-4.1", healthy_deployments=healthy_deployments
+    )
+
+    assert picked is not None and picked["model_info"]["id"] == "azure-cost-map-priced", (
+        "a cost-map priced Azure deployment must beat a $2 per token deployment, "
+        "it loses only when it is scored at the $10 unmapped default"
+    )

tests/unit/router_utils/test_deployment_pricing.py

@@ -0,0 +1,53 @@
+import pytest
+
+from litellm.router_utils.deployment_pricing import deployment_pricing_models, pricing_models
+
+
+@pytest.mark.parametrize(
+    "base_model, model, expected",
+    [
+        ("azure/gpt-4.1", "azure/my-gpt4-deploy", ("azure/gpt-4.1", "azure/my-gpt4-deploy")),
+        (None, "azure/my-gpt4-deploy", ("azure/my-gpt4-deploy",)),
+        ("", "azure/my-gpt4-deploy", ("azure/my-gpt4-deploy",)),
+        ("azure/gpt-4.1", "azure/gpt-4.1", ("azure/gpt-4.1",)),
+        (None, None, ()),
+    ],
+    ids=["base_model_first", "no_base_model", "blank_base_model", "same_name_once", "nothing_named"],
+)
+def test_pricing_models_walks_base_model_then_deployment_model(base_model, model, expected):
+    assert pricing_models(base_model, model) == expected
+
+
+@pytest.mark.parametrize(
+    "deployment, expected",
+    [
+        (
+            {"litellm_params": {"model": "azure/my-gpt4-deploy"}, "model_info": {"base_model": "azure/gpt-4.1"}},
+            ("azure/gpt-4.1", "azure/my-gpt4-deploy"),
+        ),
+        (
+            {"litellm_params": {"model": "azure/my-gpt4-deploy", "base_model": "azure/gpt-4.1"}},
+            ("azure/gpt-4.1", "azure/my-gpt4-deploy"),
+        ),
+        (
+            {
+                "litellm_params": {"model": "azure/my-gpt4-deploy", "base_model": "azure/gpt-4o"},
+                "model_info": {"base_model": "azure/gpt-4.1"},
+            },
+            ("azure/gpt-4.1", "azure/my-gpt4-deploy"),
+        ),
+        ({"litellm_params": {"model": "openai/gpt-4.1"}, "model_info": {}}, ("openai/gpt-4.1",)),
+        ({"litellm_params": "not-a-dict", "model_info": None}, ()),
+        ("not-a-deployment", ()),
+    ],
+    ids=[
+        "model_info_base_model",
+        "params_base_model",
+        "model_info_wins",
+        "no_base_model",
+        "malformed_blocks",
+        "not_a_dict",
+    ],
+)
+def test_deployment_pricing_models_reads_the_deployment_dict(deployment, expected):
+    assert deployment_pricing_models(deployment) == expected