fix(cost): port the registration lookup and flat-price helpers #47419 calls to stable/1.105.x

Prerequisite for #47419 on stable/1.105.x. Hand-ports from main (#41523, #46647):
ModelInfoLookupMiss and lookup_model_info_for_registration, the flat-price override helpers
(_without_overridden_price_variants and its rate-table helpers), and runtime_model_cost_registrations.
runtime_model_cost_registrations reads the line's _runtime_registered_model_cost store instead of
main's AliasedCostEntry.registered_fields, so cost-map alias entries stay plain dicts. Nothing on the
line calls these until #47419 lands, and register_model's behavior is unchanged.
This commit is contained in:
Mrinal Chanshetty 2026-10-20 16:36:40 -07:00
parent bff83d9fc0
commit de796475ae

View file

@ -45,7 +45,7 @@ from httpx import Proxy
from httpx._utils import get_environment_proxies
from openai.lib import _parsing, _pydantic
from openai.types.chat.completion_create_params import ResponseFormat
from pydantic import BaseModel
from pydantic import BaseModel, TypeAdapter, ValidationError
import litellm
import litellm.litellm_core_utils
@ -3253,6 +3253,100 @@ def _get_builtin_model_info_for_registration(model: str) -> ModelInfo | None:
_runtime_registered_model_cost: Final[dict[str, dict[str, object]]] = {} # mutable-ok: replayed on reload
@dataclass(frozen=True, slots=True)
class ModelInfoLookupMiss:
tripped: bool
def lookup_model_info_for_registration(model: str, custom_llm_provider: str | None) -> ModelInfo | ModelInfoLookupMiss:
try:
info: Final = get_model_info(model=model, custom_llm_provider=custom_llm_provider)
except ModelNotMappedError:
return ModelInfoLookupMiss(tripped=False)
except Exception:
return ModelInfoLookupMiss(tripped=True)
return ModelInfoLookupMiss(tripped=False) if is_generalized_model_info(info) else info
def runtime_model_cost_registrations(
model: str, custom_llm_provider: str
) -> tuple[tuple[str, Mapping[str, object]], ...]:
names: Final = _get_potential_model_names(model, custom_llm_provider)
candidates: Final = (
model,
names["combined_model_name"],
f"{custom_llm_provider}/{names['split_model']}",
names["split_model"],
names["stripped_model_name"],
)
return tuple(
(key, MappingProxyType({name: value for name, value in entry.items() if value is not None}))
for key, entry in ((key, _runtime_registered_model_cost.get(key)) for key in dict.fromkeys(candidates))
if entry
)
_TOKEN_COST_FIELDS: Final = ("input_cost_per_token", "output_cost_per_token")
_DIRECTION_RATE_FIELDS: Final = MappingProxyType(
{
"input_cost_per_token": ("input_cost_per_token",),
"output_cost_per_token": ("output_cost_per_token", "output_cost_per_reasoning_token"),
}
)
_RATE_TABLE: Final = TypeAdapter(Mapping[str, object])
_RATE_TIERS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
def _is_overridden_rate(name: str, overridden_fields: tuple[str, ...]) -> bool:
is_variant: Final = name.startswith(tuple(f"{rate_field}_" for rate_field in overridden_fields))
return name in overridden_fields or (is_variant and "_cache_" not in name)
def _without_overridden_rates(rates: Mapping[str, object], overridden_fields: tuple[str, ...]) -> Mapping[str, object]:
return MappingProxyType(
{name: rate for name, rate in rates.items() if not _is_overridden_rate(name, overridden_fields)}
)
def _rate_table_with_flat_prices(
name: str, table: object, overridden_fields: tuple[str, ...], flat_prices: Mapping[str, object]
) -> object:
try:
if name == "tiered_pricing":
return [
{**_without_overridden_rates(tier, overridden_fields), **flat_prices}
for tier in _RATE_TIERS.validate_python(table)
]
if name == "off_peak_pricing":
return dict(_without_overridden_rates(_RATE_TABLE.validate_python(table), overridden_fields))
except ValidationError:
return table
return table
def _without_overridden_price_variants(
entry: Mapping[str, object], value: Mapping[str, object]
) -> Mapping[str, object]:
"""A flat ``input_cost_per_token`` / ``output_cost_per_token`` replaces every inherited rate that would stand in
for it (token thresholds, service tiers, batches, ``tiered_pricing`` tiers, off-peak windows, and the reasoning
rate for output)."""
flat_prices: Final = MappingProxyType(
{cost_field: value[cost_field] for cost_field in _TOKEN_COST_FIELDS if value.get(cost_field) is not None}
)
overridden_fields: Final = tuple(
itertools.chain.from_iterable(_DIRECTION_RATE_FIELDS[cost_field] for cost_field in flat_prices)
)
if not overridden_fields:
return entry
return MappingProxyType(
{
name: _rate_table_with_flat_prices(name, field_value, overridden_fields, flat_prices)
for name, field_value in entry.items()
if not _is_overridden_rate(name, overridden_fields)
}
)
class _LiveDeploymentReplay:
"""Single-slot holder for the callback that rebuilds live router deployments.