feat(cost-map): record the provider's recommended replacement as successor_model on deprecating rows and report it from /model/deprecations (#49789)

* feat(cost-map): record the provider's recommended replacement as successor_model on deprecating rows and report it from /model/deprecations

* fix(cost-map): pick the alias row's successor by its own shutdown date when the family has several

* chore(ui): regenerate schema.d.ts for the /model/deprecations successor_model wording

* fix(proxy): prefer a configured successor_model over the cost map's when deployments share a name, and keep the cost map's under a hand-set deprecation_date

* test(cost-map): allow successor_model in the cost map row schema

* feat(cost-map): infer successor_model as the next version in the same family for 136 dated rows with no published replacement

---------

Co-authored-by: devin-ai-integration[bot] <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Kerry Lu 2026-11-03 13:31:12 -08:00 • committed by GitHub
parent 6c6594f1d0
commit c7a6657449
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 1038 additions and 81 deletions

File diff suppressed because it is too large Load diff

View file

@ -32,6 +32,7 @@ class _ResolvedDeprecation:
litellm_model: str | None
litellm_provider: str | None
copied_from_another_row: bool = False
successor_model: str | None = None
def _parse_deprecation_date(raw_value: object) -> date | None:
@ -58,10 +59,11 @@ def _cost_map_lookup(model_key: str | None, catalog: Mapping[str, object]) -> _R
entry: Final = litellm.model_cost.get(model_key)
if not isinstance(entry, Mapping):
return None
parsed: Final = _parse_deprecation_date(entry.get("deprecation_date"))
fields: Final[Mapping[str, object]] = entry
parsed: Final = _parse_deprecation_date(fields.get("deprecation_date"))
if parsed is None:
return None
provider: Final = entry.get("litellm_provider")
provider: Final = fields.get("litellm_provider")
return _ResolvedDeprecation(
deprecation_date=parsed,
litellm_model=model_key,
@ -69,6 +71,7 @@ def _cost_map_lookup(model_key: str | None, catalog: Mapping[str, object]) -> _R
copied_from_another_row=isinstance(entry, AliasedCostEntry)
and "deprecation_date" not in entry.registered_fields
and model_key not in catalog,
successor_model=_successor_model(fields.get("successor_model")),
)
@ -127,15 +130,6 @@ def _resolve_deployment_deprecation(
litellm_params: Final = _mapping_field(deployment, "litellm_params")
model_key: Final = _as_model_key(litellm_params.get("model"))
override: Final = _parse_deprecation_date(model_info.get("deprecation_date"))
if override is not None:
override_provider: Final = model_info.get("litellm_provider")
return _ResolvedDeprecation(
deprecation_date=override,
litellm_model=model_key,
litellm_provider=override_provider if isinstance(override_provider, str) else None,
)
provider: Final = _deployment_provider(litellm_params)
base_model: Final = _as_model_key(model_info.get("base_model"))
model_name: Final = model_key.split("/", 1)[1] if model_key is not None and "/" in model_key else model_key
@ -150,7 +144,18 @@ def _resolve_deployment_deprecation(
if pick is not None
)
own_rows: Final = tuple(pick for pick in picks if not pick.copied_from_another_row)
return min(own_rows or picks, key=lambda pick: pick.deprecation_date, default=None)
from_catalog: Final = min(own_rows or picks, key=lambda pick: pick.deprecation_date, default=None)
override: Final = _parse_deprecation_date(model_info.get("deprecation_date"))
if override is None:
return from_catalog
override_provider: Final = model_info.get("litellm_provider")
return _ResolvedDeprecation(
deprecation_date=override,
litellm_model=model_key,
litellm_provider=override_provider if isinstance(override_provider, str) else None,
successor_model=from_catalog.successor_model if from_catalog is not None else None,
)
def _classify(days_until: int, warn_within_days: int) -> DeprecationStatus:
@ -182,7 +187,7 @@ def _build_info(
today: date,
warn_within_days: int,
catalog: Mapping[str, object],
) -> ModelDeprecationInfo | None:
) -> _DeploymentInfo | None:
model_name: Final = deployment.get("model_name")
if not isinstance(model_name, str) or not model_name:
return None
@ -192,21 +197,40 @@ def _build_info(
return None
days_until: Final = (resolved.deprecation_date - today).days
return ModelDeprecationInfo(
model_name=model_name,
litellm_model=resolved.litellm_model,
deprecation_date=resolved.deprecation_date,
days_until_deprecation=days_until,
status=_classify(days_until, warn_within_days),
litellm_provider=resolved.litellm_provider,
successor_model=_checked_successor(model_name, _mapping_field(deployment, "model_info"), llm_router),
configured: Final = _checked_successor(model_name, _mapping_field(deployment, "model_info"), llm_router)
return _DeploymentInfo(
info=ModelDeprecationInfo(
model_name=model_name,
litellm_model=resolved.litellm_model,
deprecation_date=resolved.deprecation_date,
days_until_deprecation=days_until,
status=_classify(days_until, warn_within_days),
litellm_provider=resolved.litellm_provider,
successor_model=configured or resolved.successor_model,
),
configured_successor=configured is not None,
)
def _merge_group(group: Sequence[ModelDeprecationInfo]) -> ModelDeprecationInfo:
"""Keep the group's first deployment in config order, naming the first successor any of its deployments names"""
first: Final = group[0]
successors: Final = tuple(model.successor_model for model in group if model.successor_model is not None)
@dataclass(frozen=True, slots=True)
class _DeploymentInfo:
info: ModelDeprecationInfo
configured_successor: bool
def _merge_group(group: Sequence[_DeploymentInfo]) -> ModelDeprecationInfo:
"""Keep the group's first deployment in config order, naming the first successor any of its deployments names,
a successor configured in model_info ahead of one the cost map supplies"""
first: Final = group[0].info
tiers: Final = tuple(
tuple(
deployment.info.successor_model
for deployment in group
if deployment.configured_successor is configured and deployment.info.successor_model is not None
)
for configured in (True, False)
)
successors: Final = next((tier for tier in tiers if tier), ())
if not successors:
return first
if len(frozenset(successors)) > 1:
@ -220,13 +244,13 @@ def _merge_group(group: Sequence[ModelDeprecationInfo]) -> ModelDeprecationInfo:
def _dedupe(
models: Sequence[ModelDeprecationInfo],
models: Sequence[_DeploymentInfo],
) -> tuple[ModelDeprecationInfo, ...]:
"""Report a model group carrying the same date on several deployments once"""
ordered: Final = sorted(models, key=lambda model: (model.model_name, model.deprecation_date))
ordered: Final = sorted(models, key=lambda model: (model.info.model_name, model.info.deprecation_date))
return tuple(
_merge_group(tuple(group))
for _, group in groupby(ordered, key=lambda model: (model.model_name, model.deprecation_date))
for _, group in groupby(ordered, key=lambda model: (model.info.model_name, model.info.deprecation_date))
)

View file

@ -18611,8 +18611,9 @@ async def model_deprecations(
Reads `deprecation_date` metadata from `model_prices_and_context_window.json`
(and any per-deployment `model_info.deprecation_date` overrides) for the
models configured on this proxy. An entry also names its `successor_model`
when the deployment sets `model_info.successor_model`.
models configured on this proxy. An entry also names its `successor_model`:
`model_info.successor_model` when the deployment sets one, else the
replacement the cost map records from the provider's deprecation notice.
Parameters:
warn_within_days: Window (in days) used to bucket "imminent" models,

View file

@ -34,7 +34,10 @@ class ModelDeprecationInfo(LiteLLMBaseModel):
litellm_provider: str | None = Field(default=None, description="The provider this model belongs to.")
successor_model: str | None = Field(
default=None,
description=("The model to migrate to, as set in model_info.successor_model on the deployment."),
description=(
"The model to migrate to: model_info.successor_model on the deployment when set, "
"else the successor_model the cost map names for the deprecated model."
),
)

File diff suppressed because it is too large Load diff

View file

@ -64,12 +64,79 @@ def test_should_bucket_configured_models_by_urgency(authenticated_client):
def test_should_rebucket_with_warn_within_days_override(authenticated_client):
response = authenticated_client.get(
"/v1/model/deprecations", params={"warn_within_days": 40000}
)
response = authenticated_client.get("/v1/model/deprecations", params={"warn_within_days": 40000})
assert response.status_code == 200
payload = response.json()
assert [m["model_name"] for m in payload["imminent"]] == ["future-alias"]
assert payload["upcoming"] == []
assert payload["warn_within_days"] == 40000
@pytest.fixture
def client_with_cost_map_successors(monkeypatch):
app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN, api_key="sk-9876"
)
monkeypatch.setattr(
litellm,
"model_cost",
{
"retiring-model": {
"deprecation_date": "2020-01-01",
"litellm_provider": "openai",
"successor_model": "fresh-model",
},
"orphan-model": {"deprecation_date": "2020-01-01", "litellm_provider": "openai"},
},
)
router = route_entry_points_to_stubs(MagicMock())
router.get_model_list.return_value = [
{
"model_name": "retiring-alias",
"litellm_params": {"model": "retiring-model"},
"model_info": {"id": "1"},
},
{
"model_name": "retiring-overridden",
"litellm_params": {"model": "retiring-model"},
"model_info": {"id": "2", "successor_model": "my-own-pick"},
},
{
"model_name": "shared-name",
"litellm_params": {"model": "retiring-model", "api_key": "fake"},
"model_info": {"id": "4"},
},
{
"model_name": "shared-name",
"litellm_params": {"model": "retiring-model", "api_key": "fake"},
"model_info": {"id": "5", "successor_model": "my-own-pick"},
},
{
"model_name": "dated-by-hand",
"litellm_params": {"model": "retiring-model", "api_key": "fake"},
"model_info": {"id": "6", "deprecation_date": "2020-06-01"},
},
{
"model_name": "orphan-alias",
"litellm_params": {"model": "orphan-model"},
"model_info": {"id": "3"},
},
]
monkeypatch.setattr(proxy_server, "llm_router", router)
yield client
app.dependency_overrides.pop(user_api_key_auth, None)
def test_cost_map_successor_is_reported_unless_the_deployment_names_its_own(client_with_cost_map_successors):
response = client_with_cost_map_successors.get("/model/deprecations")
assert response.status_code == 200, response.text
successors = {m["model_name"]: m["successor_model"] for m in response.json()["deprecated"]}
assert successors == {
"retiring-alias": "fresh-model",
"retiring-overridden": "my-own-pick",
"shared-name": "my-own-pick",
"dated-by-hand": "fresh-model",
"orphan-alias": None,
}

View file

@ -10066,8 +10066,9 @@ export interface paths {
*
* Reads `deprecation_date` metadata from `model_prices_and_context_window.json`
* (and any per-deployment `model_info.deprecation_date` overrides) for the
* models configured on this proxy. An entry also names its `successor_model`
* when the deployment sets `model_info.successor_model`.
* models configured on this proxy. An entry also names its `successor_model`:
* `model_info.successor_model` when the deployment sets one, else the
* replacement the cost map records from the provider's deprecation notice.
*
* Parameters:
* warn_within_days: Window (in days) used to bucket "imminent" models,
@ -21599,8 +21600,9 @@ export interface paths {
*
* Reads `deprecation_date` metadata from `model_prices_and_context_window.json`
* (and any per-deployment `model_info.deprecation_date` overrides) for the
* models configured on this proxy. An entry also names its `successor_model`
* when the deployment sets `model_info.successor_model`.
* models configured on this proxy. An entry also names its `successor_model`:
* `model_info.successor_model` when the deployment sets one, else the
* replacement the cost map records from the provider's deprecation notice.
*
* Parameters:
* warn_within_days: Window (in days) used to bucket "imminent" models,
@ -38990,7 +38992,7 @@ export interface components {
status: "upcoming" | "imminent" | "deprecated";
/**
* Successor Model
* @description The model to migrate to, as set in model_info.successor_model on the deployment.
* @description The model to migrate to: model_info.successor_model on the deployment when set, else the successor_model the cost map names for the deprecated model.
*/
successor_model?: string | null;
};