This commit is contained in:
rameswaran-mohan 2026-09-29 16:13:28 +07:00 • committed by GitHub
commit 07ebbaa1b2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 193 additions and 21 deletions

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from dataclasses import dataclass, replace
from datetime import date, datetime, timezone
from itertools import groupby
from types import MappingProxyType
@ -27,6 +27,7 @@ class _ResolvedDeprecation:
deprecation_date: date
litellm_model: str | None
litellm_provider: str | None
successor_model: str | None = None
def _parse_deprecation_date(raw_value: object) -> date | None:
@ -42,20 +43,27 @@ def _parse_deprecation_date(raw_value: object) -> date | None:
return None
def _successor_model(raw_value: object) -> str | None:
"""A non-empty successor name, or None when the field is missing, blank, or not a string"""
return raw_value.strip() or None if isinstance(raw_value, str) else None
def _cost_map_lookup(model_key: object) -> _ResolvedDeprecation | None:
if not isinstance(model_key, str) or not model_key:
return None
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,
litellm_provider=provider if isinstance(provider, str) else None,
successor_model=_successor_model(fields.get("successor_model")),
)
@ -67,9 +75,25 @@ def _mapping_field(deployment: Mapping[str, object], key: str) -> Mapping[str, o
def _resolve_deployment_deprecation(
deployment: Mapping[str, object],
) -> _ResolvedDeprecation | None:
"""Resolve a deployment's deprecation date, preferring its explicit override"""
"""Resolve a deployment's deprecation date and successor, preferring its explicit overrides field by field"""
model_info: Final = _mapping_field(deployment, "model_info")
raw_model: Final = _mapping_field(deployment, "litellm_params").get("model")
unprefixed: Final = raw_model.split("/", 1)[1] if isinstance(raw_model, str) and "/" in raw_model else None
from_cost_map: Final = next(
(
candidate
for candidate in (
_cost_map_lookup(model_info.get("base_model")),
_cost_map_lookup(raw_model),
_cost_map_lookup(unprefixed),
)
if candidate is not None
),
None,
)
successor: Final = _successor_model(model_info.get("successor_model")) or (
from_cost_map.successor_model if from_cost_map is not None else None
)
override: Final = _parse_deprecation_date(model_info.get("deprecation_date"))
if override is not None:
@ -78,21 +102,9 @@ def _resolve_deployment_deprecation(
deprecation_date=override,
litellm_model=raw_model if isinstance(raw_model, str) else None,
litellm_provider=provider if isinstance(provider, str) else None,
successor_model=successor,
)
unprefixed: Final = raw_model.split("/", 1)[1] if isinstance(raw_model, str) and "/" in raw_model else None
return next(
(
resolved
for resolved in (
_cost_map_lookup(model_info.get("base_model")),
_cost_map_lookup(raw_model),
_cost_map_lookup(unprefixed),
)
if resolved is not None
),
None,
)
return None if from_cost_map is None else replace(from_cost_map, successor_model=successor)
def _classify(days_until: int, warn_within_days: int) -> DeprecationStatus:
@ -120,14 +132,17 @@ def _build_info(deployment: Mapping[str, object], today: date, warn_within_days:
days_until_deprecation=days_until,
status=_classify(days_until, warn_within_days),
litellm_provider=resolved.litellm_provider,
successor_model=resolved.successor_model,
)
def _dedupe(
models: Sequence[ModelDeprecationInfo],
) -> 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))
"""Report a model group carrying the same date on several deployments once, keeping one that names a successor"""
ordered: Final = sorted(
models, key=lambda model: (model.model_name, model.deprecation_date, model.successor_model is None)
)
return tuple(
next(group) for _, group in groupby(ordered, key=lambda model: (model.model_name, model.deprecation_date))
)
@ -186,10 +201,11 @@ def _format_entry(info: ModelDeprecationInfo) -> str:
if info.days_until_deprecation < 0
else f"in {info.days_until_deprecation}d"
)
migration: Final = f", migrate to `{_escape_slack_mrkdwn(info.successor_model)}`" if info.successor_model else ""
return (
f"• `{_escape_slack_mrkdwn(info.model_name)}` "
f"(provider: {_escape_slack_mrkdwn(info.litellm_provider) if info.litellm_provider else 'unknown'}, "
f"deprecates {info.deprecation_date.isoformat()}, {suffix})"
f"deprecates {info.deprecation_date.isoformat()}, {suffix}{migration})"
)

View file

@ -30,6 +30,13 @@ class ModelDeprecationInfo(BaseModel):
),
)
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, taken from model_info.successor_model on the deployment "
"or from the model cost map entry."
),
)
class ModelDeprecationResponse(BaseModel):

View file

@ -357,3 +357,147 @@ class TestFormatDeprecationAlertMessage:
assert "<https://evil.example|openai>" not in message
assert "&lt;!channel&gt; pwned" in message
assert "&lt;https://evil.example|openai&gt; &amp; co" in message
class TestSuccessorModel:
def test_should_surface_the_successor_set_on_the_deployment(self, monkeypatch):
monkeypatch.setattr(litellm, "model_cost", {})
router = _make_router(
[
{
"model_name": "old",
"litellm_params": {"model": "openai/old-model"},
"model_info": {"id": "1", "deprecation_date": "2026-06-10", "successor_model": "new"},
}
]
)
snapshot = collect_model_deprecations(llm_router=router, warn_within_days=30, today=date(2026, 6, 1))
assert snapshot.imminent[0].successor_model == "new"
def test_should_take_the_successor_from_the_cost_map_when_the_deployment_has_none(self, monkeypatch):
monkeypatch.setattr(
litellm,
"model_cost",
{"old-model": {"deprecation_date": "2026-06-10", "successor_model": "new-model"}},
)
router = _make_router(
[{"model_name": "old", "litellm_params": {"model": "old-model"}, "model_info": {"id": "1"}}]
)
snapshot = collect_model_deprecations(llm_router=router, warn_within_days=30, today=date(2026, 6, 1))
assert snapshot.imminent[0].successor_model == "new-model"
def test_should_prefer_the_deployment_successor_over_the_cost_map(self, monkeypatch):
monkeypatch.setattr(
litellm,
"model_cost",
{"old-model": {"deprecation_date": "2026-06-10", "successor_model": "cost-map-pick"}},
)
router = _make_router(
[
{
"model_name": "old",
"litellm_params": {"model": "old-model"},
"model_info": {"id": "1", "successor_model": "my-pick"},
}
]
)
snapshot = collect_model_deprecations(llm_router=router, warn_within_days=30, today=date(2026, 6, 1))
assert snapshot.imminent[0].successor_model == "my-pick"
def test_should_ignore_blank_or_non_string_successors(self, monkeypatch):
monkeypatch.setattr(litellm, "model_cost", {})
router = _make_router(
[
{
"model_name": "blank",
"litellm_params": {"model": "openai/a"},
"model_info": {"id": "1", "deprecation_date": "2026-06-10", "successor_model": " "},
},
{
"model_name": "number",
"litellm_params": {"model": "openai/b"},
"model_info": {"id": "2", "deprecation_date": "2026-06-10", "successor_model": 7},
},
]
)
snapshot = collect_model_deprecations(llm_router=router, warn_within_days=30, today=date(2026, 6, 1))
assert [m.successor_model for m in snapshot.imminent] == [None, None]
def test_should_keep_the_deployment_naming_a_successor_when_deduping(self, monkeypatch):
monkeypatch.setattr(litellm, "model_cost", {})
router = _make_router(
[
{
"model_name": "alias",
"litellm_params": {"model": "openai/a"},
"model_info": {"id": "1", "deprecation_date": "2026-06-10"},
},
{
"model_name": "alias",
"litellm_params": {"model": "openai/b"},
"model_info": {"id": "2", "deprecation_date": "2026-06-10", "successor_model": "alias-v2"},
},
]
)
snapshot = collect_model_deprecations(llm_router=router, warn_within_days=30, today=date(2026, 6, 1))
assert [m.successor_model for m in snapshot.imminent] == ["alias-v2"]
def test_should_name_the_successor_in_the_slack_entry_and_escape_it(self, monkeypatch):
monkeypatch.setattr(litellm, "model_cost", {})
router = _make_router(
[
{
"model_name": "old",
"litellm_params": {"model": "openai/old-model"},
"model_info": {
"id": "1",
"deprecation_date": "2026-06-10",
"successor_model": "<https://evil.example|new>",
},
},
{
"model_name": "orphan",
"litellm_params": {"model": "openai/orphan-model"},
"model_info": {"id": "2", "deprecation_date": "2026-06-10"},
},
]
)
snapshot = collect_model_deprecations(llm_router=router, warn_within_days=30, today=date(2026, 6, 1))
message = format_deprecation_alert_message(snapshot)
assert message is not None
assert "migrate to `&lt;https://evil.example|new&gt;`" in message
assert "<https://evil.example|new>" not in message
assert message.count("migrate to") == 1
def test_should_take_the_cost_map_successor_when_the_deployment_overrides_only_the_date(self, monkeypatch):
monkeypatch.setattr(
litellm,
"model_cost",
{"old-model": {"deprecation_date": "2030-01-01", "successor_model": "new-model"}},
)
router = _make_router(
[
{
"model_name": "old",
"litellm_params": {"model": "old-model"},
"model_info": {"id": "1", "deprecation_date": "2026-06-10"},
}
]
)
snapshot = collect_model_deprecations(llm_router=router, warn_within_days=30, today=date(2026, 6, 1))
assert snapshot.imminent[0].deprecation_date == date(2026, 6, 10)
assert snapshot.imminent[0].successor_model == "new-model"

View file

@ -35888,6 +35888,11 @@ export interface components {
* @enum {string}
*/
status: "upcoming" | "imminent" | "deprecated";
/**
* Successor Model
* @description The model to migrate to, taken from model_info.successor_model on the deployment or from the model cost map entry.
*/
successor_model?: string | null;
};
/** ModelDeprecationResponse */
ModelDeprecationResponse: {