mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
feat(proxy): proactive model deprecation alerts and /model/deprecations endpoint
Surfaces deprecation_date metadata that is already shipped in model_prices_and_context_window.json so operators get lead time to migrate before a provider sunsets a model. - New helper litellm.proxy.common_utils.model_deprecation classifies the router's configured models into deprecated / imminent / upcoming buckets. Resolution order: explicit model_info.deprecation_date > model_info.base_model > litellm_params.model. - New GET /model/deprecations (and /v1/model/deprecations) endpoint returns a ModelDeprecationResponse, gated by user_api_key_auth. - New AlertType.model_deprecation_warnings (in DEFAULT_ALERT_TYPES) plus SlackAlerting.send_model_deprecation_alert dispatches a Slack message for deprecated/imminent models. Severity is High when any model is already past its date, Medium when only imminent. - ProxyLogging.startup_event schedules a daily background task (_run_scheduled_deprecation_check) when the alert type is enabled. The interval is configurable via LITELLM_MODEL_DEPRECATION_CHECK_INTERVAL and the warn window via LITELLM_MODEL_DEPRECATION_WARN_DAYS. - Tests: 16 unit tests for the helper plus 4 for the Slack hook in tests/test_litellm/. Co-authored-by: Mateo Wang <mateo-berri@users.noreply.github.com>
This commit is contained in:
parent
76ad1c319d
commit
f249356e16
8 changed files with 862 additions and 0 deletions
|
|
@ -1038,6 +1038,81 @@ Model Info:
|
|||
async def model_removed_alert(self, model_name: str):
|
||||
pass
|
||||
|
||||
async def send_model_deprecation_alert(
|
||||
self, llm_router: Optional[Any] = None
|
||||
) -> bool:
|
||||
"""Aggregate deprecation metadata for the configured models and alert.
|
||||
|
||||
Returns ``True`` when an alert payload was dispatched, ``False``
|
||||
otherwise. The ``send_alert`` helper itself is responsible for honoring
|
||||
the user's webhook configuration; this method only owns producing the
|
||||
message and choosing whether to send it.
|
||||
"""
|
||||
if (
|
||||
self.alerting is None
|
||||
or AlertType.model_deprecation_warnings not in self.alert_types
|
||||
):
|
||||
return False
|
||||
|
||||
from litellm.proxy.common_utils.model_deprecation import (
|
||||
collect_model_deprecations,
|
||||
format_deprecation_alert_message,
|
||||
)
|
||||
|
||||
try:
|
||||
snapshot = collect_model_deprecations(llm_router=llm_router)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.exception(
|
||||
"Error collecting model deprecation snapshot: %s", e
|
||||
)
|
||||
return False
|
||||
|
||||
message = format_deprecation_alert_message(snapshot)
|
||||
if message is None:
|
||||
return False
|
||||
|
||||
level: Literal["Low", "Medium", "High"] = (
|
||||
"High" if snapshot.deprecated else "Medium"
|
||||
)
|
||||
|
||||
await self.send_alert(
|
||||
message=message,
|
||||
level=level,
|
||||
alert_type=AlertType.model_deprecation_warnings,
|
||||
alerting_metadata={
|
||||
"deprecated_count": len(snapshot.deprecated),
|
||||
"imminent_count": len(snapshot.imminent),
|
||||
"upcoming_count": len(snapshot.upcoming),
|
||||
},
|
||||
)
|
||||
return True
|
||||
|
||||
async def _run_scheduled_deprecation_check(self, llm_router: Optional[Any] = None):
|
||||
"""Periodic background task that emits a model deprecation alert.
|
||||
|
||||
Runs immediately on startup (so operators see the current state in
|
||||
Slack) and then sleeps ``DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS``
|
||||
between runs. Exits silently if the alert type is not enabled.
|
||||
"""
|
||||
from litellm.types.proxy.model_deprecation import (
|
||||
DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
|
||||
)
|
||||
|
||||
if (
|
||||
self.alerting is None
|
||||
or AlertType.model_deprecation_warnings not in self.alert_types
|
||||
):
|
||||
return
|
||||
|
||||
while True:
|
||||
try:
|
||||
await self.send_model_deprecation_alert(llm_router=llm_router)
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.exception(
|
||||
"Error in model deprecation alert loop: %s", e
|
||||
)
|
||||
await asyncio.sleep(DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS)
|
||||
|
||||
async def send_webhook_alert(self, webhook_event: WebhookEvent) -> bool:
|
||||
"""
|
||||
Sends structured alert to webhook, if set.
|
||||
|
|
|
|||
247
litellm/proxy/common_utils/model_deprecation.py
Normal file
247
litellm/proxy/common_utils/model_deprecation.py
Normal file
|
|
@ -0,0 +1,247 @@
|
|||
"""Helpers for surfacing model deprecation/sunset information.
|
||||
|
||||
This module reads ``deprecation_date`` metadata that is bundled in
|
||||
``model_prices_and_context_window.json`` (exposed at runtime via
|
||||
``litellm.model_cost``) and classifies the proxy's configured models into
|
||||
``upcoming``, ``imminent`` and ``deprecated`` buckets. It is the single
|
||||
source of truth used by both the ``/model/deprecations`` endpoint and the
|
||||
proactive Slack alert.
|
||||
|
||||
Resolution order for a deployment's deprecation date:
|
||||
|
||||
1. ``model_info.deprecation_date`` – an explicit override on the deployment.
|
||||
2. ``model_info.base_model`` looked up in ``litellm.model_cost``.
|
||||
3. The ``litellm_params.model`` string looked up in ``litellm.model_cost``.
|
||||
|
||||
Models without any deprecation metadata are skipped silently (most models
|
||||
are not deprecated, and we don't want to pollute the response).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import date, datetime, timezone
|
||||
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.types.proxy.model_deprecation import (
|
||||
DEFAULT_DEPRECATION_WARN_DAYS,
|
||||
ModelDeprecationInfo,
|
||||
ModelDeprecationResponse,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.router import Router as _Router
|
||||
|
||||
Router = _Router
|
||||
else:
|
||||
Router = Any
|
||||
|
||||
|
||||
def _parse_deprecation_date(raw_value: Any) -> Optional[date]:
|
||||
"""Parse a ``deprecation_date`` string in YYYY-MM-DD form.
|
||||
|
||||
Returns ``None`` for missing, malformed, or sentinel placeholder values
|
||||
(the JSON map ships a documentation sentinel of the form ``"date when..."``).
|
||||
"""
|
||||
if raw_value is None:
|
||||
return None
|
||||
if isinstance(raw_value, date):
|
||||
return raw_value
|
||||
if not isinstance(raw_value, str):
|
||||
return None
|
||||
try:
|
||||
return datetime.strptime(raw_value.strip(), "%Y-%m-%d").date()
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _lookup_deprecation_date_from_cost_map(
|
||||
model_key: Optional[str],
|
||||
) -> Tuple[Optional[date], Optional[str]]:
|
||||
"""Look up a deprecation date in ``litellm.model_cost`` for ``model_key``.
|
||||
|
||||
Returns a tuple of (deprecation_date, litellm_provider).
|
||||
"""
|
||||
if not model_key:
|
||||
return None, None
|
||||
entry = litellm.model_cost.get(model_key)
|
||||
if not isinstance(entry, dict):
|
||||
return None, None
|
||||
return (
|
||||
_parse_deprecation_date(entry.get("deprecation_date")),
|
||||
entry.get("litellm_provider"),
|
||||
)
|
||||
|
||||
|
||||
def _resolve_deployment_deprecation(
|
||||
deployment: Dict[str, Any],
|
||||
) -> Tuple[Optional[date], Optional[str], Optional[str]]:
|
||||
"""Resolve a deployment's deprecation metadata.
|
||||
|
||||
Returns a tuple of (deprecation_date, litellm_model, litellm_provider).
|
||||
"""
|
||||
model_info = deployment.get("model_info") or {}
|
||||
explicit = _parse_deprecation_date(model_info.get("deprecation_date"))
|
||||
if explicit is not None:
|
||||
litellm_params = deployment.get("litellm_params") or {}
|
||||
return (
|
||||
explicit,
|
||||
litellm_params.get("model"),
|
||||
model_info.get("litellm_provider"),
|
||||
)
|
||||
|
||||
base_model = model_info.get("base_model")
|
||||
dep_date, provider = _lookup_deprecation_date_from_cost_map(base_model)
|
||||
if dep_date is not None:
|
||||
return dep_date, base_model, provider
|
||||
|
||||
litellm_params = deployment.get("litellm_params") or {}
|
||||
raw_model = litellm_params.get("model")
|
||||
dep_date, provider = _lookup_deprecation_date_from_cost_map(raw_model)
|
||||
if dep_date is not None:
|
||||
return dep_date, raw_model, provider
|
||||
|
||||
if isinstance(raw_model, str) and "/" in raw_model:
|
||||
# Try the un-prefixed lookup (e.g. "openai/gpt-4o" → "gpt-4o").
|
||||
bare = raw_model.split("/", 1)[1]
|
||||
dep_date, provider = _lookup_deprecation_date_from_cost_map(bare)
|
||||
if dep_date is not None:
|
||||
return dep_date, bare, provider
|
||||
|
||||
return None, raw_model, model_info.get("litellm_provider")
|
||||
|
||||
|
||||
def _classify(days_until: int, warn_within_days: int) -> str:
|
||||
if days_until < 0:
|
||||
return "deprecated"
|
||||
if days_until <= warn_within_days:
|
||||
return "imminent"
|
||||
return "upcoming"
|
||||
|
||||
|
||||
def _model_dump_compat(deployment: Any) -> Dict[str, Any]:
|
||||
"""Return a plain dict for both pydantic models and dicts."""
|
||||
if isinstance(deployment, dict):
|
||||
return deployment
|
||||
if hasattr(deployment, "model_dump"):
|
||||
return deployment.model_dump(exclude_none=True)
|
||||
if hasattr(deployment, "dict"):
|
||||
return deployment.dict()
|
||||
return dict(deployment)
|
||||
|
||||
|
||||
def collect_model_deprecations(
|
||||
llm_router: Optional[Router],
|
||||
warn_within_days: int = DEFAULT_DEPRECATION_WARN_DAYS,
|
||||
today: Optional[date] = None,
|
||||
) -> ModelDeprecationResponse:
|
||||
"""Aggregate deprecation info for all deployments configured on the router.
|
||||
|
||||
De-duplicates by ``(model_name, deprecation_date)`` so multi-deployment
|
||||
model groups (load-balanced across regions) only surface once per
|
||||
deprecation date.
|
||||
"""
|
||||
snapshot_time = datetime.now(timezone.utc)
|
||||
today = today or snapshot_time.date()
|
||||
|
||||
response = ModelDeprecationResponse(
|
||||
warn_within_days=warn_within_days,
|
||||
checked_at=snapshot_time,
|
||||
)
|
||||
|
||||
if llm_router is None:
|
||||
return response
|
||||
|
||||
seen: set = set()
|
||||
deployments = llm_router.get_model_list() or []
|
||||
for deployment in deployments:
|
||||
deployment_dict = _model_dump_compat(deployment)
|
||||
model_name = deployment_dict.get("model_name")
|
||||
if not model_name:
|
||||
continue
|
||||
|
||||
dep_date, litellm_model, provider = _resolve_deployment_deprecation(
|
||||
deployment_dict
|
||||
)
|
||||
if dep_date is None:
|
||||
continue
|
||||
|
||||
dedup_key = (model_name, dep_date.isoformat())
|
||||
if dedup_key in seen:
|
||||
continue
|
||||
seen.add(dedup_key)
|
||||
|
||||
days_until = (dep_date - today).days
|
||||
status = _classify(days_until, warn_within_days)
|
||||
|
||||
info = ModelDeprecationInfo(
|
||||
model_name=model_name,
|
||||
litellm_model=litellm_model,
|
||||
deprecation_date=dep_date,
|
||||
days_until_deprecation=days_until,
|
||||
status=status,
|
||||
litellm_provider=provider,
|
||||
)
|
||||
|
||||
if status == "deprecated":
|
||||
response.deprecated.append(info)
|
||||
elif status == "imminent":
|
||||
response.imminent.append(info)
|
||||
else:
|
||||
response.upcoming.append(info)
|
||||
|
||||
response.deprecated.sort(key=lambda m: m.deprecation_date)
|
||||
response.imminent.sort(key=lambda m: m.deprecation_date)
|
||||
response.upcoming.sort(key=lambda m: m.deprecation_date)
|
||||
|
||||
verbose_logger.debug(
|
||||
"model_deprecation: deprecated=%d imminent=%d upcoming=%d",
|
||||
len(response.deprecated),
|
||||
len(response.imminent),
|
||||
len(response.upcoming),
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
|
||||
def format_deprecation_alert_message(
|
||||
snapshot: ModelDeprecationResponse,
|
||||
) -> Optional[str]:
|
||||
"""Format a Slack-friendly alert message for the warning buckets.
|
||||
|
||||
Only ``deprecated`` and ``imminent`` models are included; ``upcoming``
|
||||
models are intentionally omitted to avoid alert fatigue. Returns
|
||||
``None`` when there is nothing to alert on.
|
||||
"""
|
||||
if not snapshot.deprecated and not snapshot.imminent:
|
||||
return None
|
||||
|
||||
lines: List[str] = ["*⚠️ Model Deprecation Warning*"]
|
||||
|
||||
def _format_entry(info: ModelDeprecationInfo) -> str:
|
||||
suffix = (
|
||||
f"already deprecated {abs(info.days_until_deprecation)}d ago"
|
||||
if info.days_until_deprecation < 0
|
||||
else f"in {info.days_until_deprecation}d"
|
||||
)
|
||||
return (
|
||||
f"• `{info.model_name}` "
|
||||
f"(provider: {info.litellm_provider or 'unknown'}, "
|
||||
f"deprecates {info.deprecation_date.isoformat()} – {suffix})"
|
||||
)
|
||||
|
||||
if snapshot.deprecated:
|
||||
lines.append("\n*Already deprecated:*")
|
||||
lines.extend(_format_entry(i) for i in snapshot.deprecated)
|
||||
|
||||
if snapshot.imminent:
|
||||
lines.append(f"\n*Deprecating within {snapshot.warn_within_days} days:*")
|
||||
lines.extend(_format_entry(i) for i in snapshot.imminent)
|
||||
|
||||
lines.append(
|
||||
"\nPlan migrations to a supported model. See "
|
||||
"https://docs.litellm.ai/docs/proxy/model_management for guidance."
|
||||
)
|
||||
|
||||
return "\n".join(lines)
|
||||
|
|
@ -319,6 +319,7 @@ from litellm.proxy.common_utils.load_config_utils import (
|
|||
get_config_file_contents_from_gcs,
|
||||
get_file_contents_from_s3,
|
||||
)
|
||||
from litellm.proxy.common_utils.model_deprecation import collect_model_deprecations
|
||||
from litellm.proxy.common_utils.model_listing_utils import TeamModelNameTranslator
|
||||
from litellm.proxy.common_utils.openai_endpoint_utils import (
|
||||
remove_sensitive_info_from_deployment,
|
||||
|
|
@ -624,6 +625,10 @@ from litellm.types.proxy.control_plane_endpoints import WorkerRegistryEntry
|
|||
from litellm.types.proxy.management_endpoints.model_management_endpoints import (
|
||||
ModelGroupInfoProxy,
|
||||
)
|
||||
from litellm.types.proxy.model_deprecation import (
|
||||
DEFAULT_DEPRECATION_WARN_DAYS,
|
||||
ModelDeprecationResponse,
|
||||
)
|
||||
from litellm.types.proxy.management_endpoints.ui_sso import (
|
||||
DefaultTeamSSOParams,
|
||||
LiteLLM_UpperboundKeyGenerateParams,
|
||||
|
|
@ -13436,6 +13441,52 @@ async def model_info_v1(
|
|||
return {"data": all_models}
|
||||
|
||||
|
||||
@router.get(
|
||||
"/model/deprecations",
|
||||
tags=["model management"],
|
||||
dependencies=[Depends(user_api_key_auth)],
|
||||
response_model=ModelDeprecationResponse,
|
||||
)
|
||||
@router.get(
|
||||
"/v1/model/deprecations",
|
||||
tags=["model management"],
|
||||
dependencies=[Depends(user_api_key_auth)],
|
||||
response_model=ModelDeprecationResponse,
|
||||
)
|
||||
async def model_deprecations(
|
||||
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
|
||||
warn_within_days: int = DEFAULT_DEPRECATION_WARN_DAYS,
|
||||
) -> ModelDeprecationResponse:
|
||||
"""List models with known deprecation/sunset dates, bucketed by urgency.
|
||||
|
||||
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.
|
||||
|
||||
Parameters:
|
||||
warn_within_days: Window (in days) used to bucket "imminent" models.
|
||||
Defaults to `LITELLM_MODEL_DEPRECATION_WARN_DAYS` env var (or 30).
|
||||
|
||||
Returns:
|
||||
A payload with three lists of `ModelDeprecationInfo` entries:
|
||||
|
||||
- `deprecated`: deprecation date is in the past — these requests may
|
||||
fail at any time.
|
||||
- `imminent`: deprecation date is within `warn_within_days` from today.
|
||||
- `upcoming`: deprecation date is further out.
|
||||
|
||||
Example:
|
||||
```shell
|
||||
curl -X GET 'http://localhost:4000/model/deprecations' \\
|
||||
-H 'Authorization: Bearer sk-1234'
|
||||
```
|
||||
"""
|
||||
global llm_router
|
||||
return collect_model_deprecations(
|
||||
llm_router=llm_router, warn_within_days=warn_within_days
|
||||
)
|
||||
|
||||
|
||||
def _get_model_group_info(
|
||||
llm_router: Router, all_models_str: list[str], model_group: str | None
|
||||
) -> list[ModelGroupInfoProxy]:
|
||||
|
|
|
|||
|
|
@ -442,6 +442,7 @@ class ProxyLogging:
|
|||
# Guard flags to prevent duplicate background tasks
|
||||
self.daily_report_started: bool = False
|
||||
self.hanging_requests_check_started: bool = False
|
||||
self.deprecation_check_started: bool = False
|
||||
|
||||
def startup_event(
|
||||
self,
|
||||
|
|
@ -481,6 +482,19 @@ class ProxyLogging:
|
|||
) # RUN HANGING REQUEST CHECK (if user wants to alert on hanging requests)
|
||||
self.hanging_requests_check_started = True
|
||||
|
||||
if (
|
||||
self.slack_alerting_instance is not None
|
||||
and AlertType.model_deprecation_warnings
|
||||
in self.slack_alerting_instance.alert_types
|
||||
and not self.deprecation_check_started
|
||||
):
|
||||
asyncio.create_task(
|
||||
self.slack_alerting_instance._run_scheduled_deprecation_check(
|
||||
llm_router=llm_router
|
||||
)
|
||||
) # RUN MODEL DEPRECATION ALERT LOOP (if scheduled)
|
||||
self.deprecation_check_started = True
|
||||
|
||||
def update_values(
|
||||
self,
|
||||
alerting: list | None = None,
|
||||
|
|
|
|||
|
|
@ -147,6 +147,7 @@ class AlertType(str, Enum):
|
|||
# Deployment alerts
|
||||
cooldown_deployment = "cooldown_deployment"
|
||||
new_model_added = "new_model_added"
|
||||
model_deprecation_warnings = "model_deprecation_warnings"
|
||||
|
||||
# Outage alerts
|
||||
outage_alerts = "outage_alerts"
|
||||
|
|
@ -187,6 +188,7 @@ DEFAULT_ALERT_TYPES: Final[list[AlertType]] = [
|
|||
# Deployment alerts
|
||||
AlertType.cooldown_deployment,
|
||||
AlertType.new_model_added,
|
||||
AlertType.model_deprecation_warnings,
|
||||
# Outage alerts
|
||||
AlertType.outage_alerts,
|
||||
AlertType.region_outage_alerts,
|
||||
|
|
|
|||
93
litellm/types/proxy/model_deprecation.py
Normal file
93
litellm/types/proxy/model_deprecation.py
Normal file
|
|
@ -0,0 +1,93 @@
|
|||
"""Type definitions for model deprecation tracking and proactive alerts.
|
||||
|
||||
The proxy reads deprecation/sunset metadata from
|
||||
``litellm.model_cost`` (sourced from ``model_prices_and_context_window.json``)
|
||||
and surfaces it through the ``/model/deprecations`` endpoint and Slack
|
||||
alerting. These types describe the response payload and the alert payload.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from datetime import date, datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
DEFAULT_DEPRECATION_WARN_DAYS = int(
|
||||
os.getenv("LITELLM_MODEL_DEPRECATION_WARN_DAYS", "30")
|
||||
)
|
||||
"""Number of days before the deprecation date to start raising warnings.
|
||||
|
||||
Configurable via the ``LITELLM_MODEL_DEPRECATION_WARN_DAYS`` environment
|
||||
variable. Defaults to 30 days, matching the typical migration window most
|
||||
LLM providers offer between announcement and removal.
|
||||
"""
|
||||
|
||||
DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS = int(
|
||||
os.getenv("LITELLM_MODEL_DEPRECATION_CHECK_INTERVAL", str(24 * 60 * 60))
|
||||
)
|
||||
"""How often the periodic background check runs. Defaults to once per day."""
|
||||
|
||||
|
||||
DeprecationStatusLiteral = str
|
||||
"""One of ``"upcoming"``, ``"imminent"``, ``"deprecated"``.
|
||||
|
||||
* ``upcoming`` – deprecation is scheduled but more than the warn window away.
|
||||
* ``imminent`` – deprecation date is within ``warn_within_days`` from today.
|
||||
* ``deprecated`` – deprecation date has already passed.
|
||||
"""
|
||||
|
||||
|
||||
class ModelDeprecationInfo(BaseModel):
|
||||
"""Per-model deprecation metadata returned by ``/model/deprecations``."""
|
||||
|
||||
model_name: str = Field(
|
||||
description="The public name of the model on the proxy (model_group)."
|
||||
)
|
||||
litellm_model: Optional[str] = Field(
|
||||
default=None,
|
||||
description="The underlying litellm model string the deprecation date is sourced from.",
|
||||
)
|
||||
deprecation_date: date = Field(
|
||||
description="The date (UTC) when the model becomes deprecated."
|
||||
)
|
||||
days_until_deprecation: int = Field(
|
||||
description=(
|
||||
"Days remaining until the deprecation date. Negative if the model "
|
||||
"is already deprecated."
|
||||
),
|
||||
)
|
||||
status: DeprecationStatusLiteral = Field(
|
||||
description="One of 'upcoming', 'imminent', or 'deprecated'.",
|
||||
)
|
||||
litellm_provider: Optional[str] = Field(
|
||||
default=None, description="The provider this model belongs to."
|
||||
)
|
||||
|
||||
|
||||
class ModelDeprecationResponse(BaseModel):
|
||||
"""Response payload for ``GET /model/deprecations``."""
|
||||
|
||||
deprecated: List[ModelDeprecationInfo] = Field(
|
||||
default_factory=list,
|
||||
description="Models whose deprecation date has already passed.",
|
||||
)
|
||||
imminent: List[ModelDeprecationInfo] = Field(
|
||||
default_factory=list,
|
||||
description=(
|
||||
"Models whose deprecation date is within ``warn_within_days`` from "
|
||||
"today and require immediate migration planning."
|
||||
),
|
||||
)
|
||||
upcoming: List[ModelDeprecationInfo] = Field(
|
||||
default_factory=list,
|
||||
description="Models with a future deprecation date outside the warn window.",
|
||||
)
|
||||
warn_within_days: int = Field(
|
||||
description="The window (in days) used to bucket 'imminent' models."
|
||||
)
|
||||
checked_at: datetime = Field(
|
||||
description="UTC timestamp when the deprecation snapshot was generated."
|
||||
)
|
||||
|
|
@ -0,0 +1,100 @@
|
|||
"""Tests for the Slack alerting model deprecation hook."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../../../.."))
|
||||
|
||||
import litellm
|
||||
from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting
|
||||
from litellm.proxy._types import AlertType
|
||||
|
||||
|
||||
def _make_router(deployments):
|
||||
router = MagicMock()
|
||||
router.get_model_list.return_value = deployments
|
||||
return router
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_should_skip_when_alert_type_disabled():
|
||||
alerting = SlackAlerting(
|
||||
alerting=["slack"],
|
||||
alert_types=[AlertType.llm_exceptions],
|
||||
)
|
||||
sent = await alerting.send_model_deprecation_alert(llm_router=MagicMock())
|
||||
assert sent is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_should_skip_when_no_alerting_configured():
|
||||
alerting = SlackAlerting(
|
||||
alerting=None,
|
||||
alert_types=[AlertType.model_deprecation_warnings],
|
||||
)
|
||||
sent = await alerting.send_model_deprecation_alert(llm_router=MagicMock())
|
||||
assert sent is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_should_skip_when_no_deprecations_found(monkeypatch):
|
||||
monkeypatch.setattr(litellm, "model_cost", {})
|
||||
alerting = SlackAlerting(
|
||||
alerting=["slack"],
|
||||
alert_types=[AlertType.model_deprecation_warnings],
|
||||
)
|
||||
router = _make_router(
|
||||
[
|
||||
{
|
||||
"model_name": "fresh",
|
||||
"litellm_params": {"model": "openai/gpt-4o"},
|
||||
"model_info": {"id": "x"},
|
||||
}
|
||||
]
|
||||
)
|
||||
sent = await alerting.send_model_deprecation_alert(llm_router=router)
|
||||
assert sent is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_should_dispatch_high_severity_when_deprecated(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
litellm,
|
||||
"model_cost",
|
||||
{
|
||||
"dead-model": {
|
||||
"deprecation_date": "2020-01-01",
|
||||
"litellm_provider": "openai",
|
||||
}
|
||||
},
|
||||
)
|
||||
alerting = SlackAlerting(
|
||||
alerting=["slack"],
|
||||
alert_types=[AlertType.model_deprecation_warnings],
|
||||
)
|
||||
router = _make_router(
|
||||
[
|
||||
{
|
||||
"model_name": "dead-alias",
|
||||
"litellm_params": {"model": "dead-model"},
|
||||
"model_info": {"id": "1"},
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
with patch.object(
|
||||
alerting, "send_alert", new_callable=AsyncMock
|
||||
) as mock_send_alert:
|
||||
sent = await alerting.send_model_deprecation_alert(llm_router=router)
|
||||
|
||||
assert sent is True
|
||||
mock_send_alert.assert_awaited_once()
|
||||
call_kwargs = mock_send_alert.await_args.kwargs
|
||||
assert call_kwargs["alert_type"] == AlertType.model_deprecation_warnings
|
||||
assert call_kwargs["level"] == "High"
|
||||
assert call_kwargs["alerting_metadata"]["deprecated_count"] == 1
|
||||
assert call_kwargs["alerting_metadata"]["imminent_count"] == 0
|
||||
assert "dead-alias" in call_kwargs["message"]
|
||||
280
tests/test_litellm/proxy/common_utils/test_model_deprecation.py
Normal file
280
tests/test_litellm/proxy/common_utils/test_model_deprecation.py
Normal file
|
|
@ -0,0 +1,280 @@
|
|||
"""Tests for the model deprecation helper module.
|
||||
|
||||
These tests focus on the helper itself — not on the proxy endpoint or
|
||||
Slack integration — so they can run without the full proxy stack.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from datetime import date
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../../../.."))
|
||||
|
||||
import litellm
|
||||
from litellm.proxy.common_utils.model_deprecation import (
|
||||
_classify,
|
||||
_parse_deprecation_date,
|
||||
collect_model_deprecations,
|
||||
format_deprecation_alert_message,
|
||||
)
|
||||
|
||||
|
||||
def _make_router(deployments):
|
||||
router = MagicMock()
|
||||
router.get_model_list.return_value = deployments
|
||||
return router
|
||||
|
||||
|
||||
class TestParseDeprecationDate:
|
||||
def test_should_parse_iso_string(self):
|
||||
assert _parse_deprecation_date("2026-12-31") == date(2026, 12, 31)
|
||||
|
||||
def test_should_pass_through_date_object(self):
|
||||
d = date(2026, 1, 1)
|
||||
assert _parse_deprecation_date(d) == d
|
||||
|
||||
def test_should_return_none_for_documentation_sentinel(self):
|
||||
# The JSON map ships a sentinel string under the "sample_spec" key.
|
||||
assert (
|
||||
_parse_deprecation_date(
|
||||
"date when the model becomes deprecated in the format YYYY-MM-DD"
|
||||
)
|
||||
is None
|
||||
)
|
||||
|
||||
def test_should_return_none_for_none(self):
|
||||
assert _parse_deprecation_date(None) is None
|
||||
|
||||
def test_should_return_none_for_unsupported_type(self):
|
||||
assert _parse_deprecation_date(12345) is None
|
||||
|
||||
|
||||
class TestClassify:
|
||||
def test_should_classify_past_dates_as_deprecated(self):
|
||||
assert _classify(-1, warn_within_days=30) == "deprecated"
|
||||
assert _classify(-365, warn_within_days=30) == "deprecated"
|
||||
|
||||
def test_should_classify_inside_window_as_imminent(self):
|
||||
assert _classify(0, warn_within_days=30) == "imminent"
|
||||
assert _classify(15, warn_within_days=30) == "imminent"
|
||||
assert _classify(30, warn_within_days=30) == "imminent"
|
||||
|
||||
def test_should_classify_outside_window_as_upcoming(self):
|
||||
assert _classify(31, warn_within_days=30) == "upcoming"
|
||||
assert _classify(365, warn_within_days=30) == "upcoming"
|
||||
|
||||
|
||||
class TestCollectModelDeprecations:
|
||||
def test_should_return_empty_response_when_router_is_none(self):
|
||||
snapshot = collect_model_deprecations(llm_router=None)
|
||||
assert snapshot.deprecated == []
|
||||
assert snapshot.imminent == []
|
||||
assert snapshot.upcoming == []
|
||||
|
||||
def test_should_skip_models_without_deprecation_metadata(self, monkeypatch):
|
||||
monkeypatch.setattr(litellm, "model_cost", {})
|
||||
router = _make_router(
|
||||
[
|
||||
{
|
||||
"model_name": "gpt-4o",
|
||||
"litellm_params": {"model": "openai/gpt-4o"},
|
||||
"model_info": {"id": "abc"},
|
||||
}
|
||||
]
|
||||
)
|
||||
snapshot = collect_model_deprecations(llm_router=router)
|
||||
assert snapshot.deprecated == []
|
||||
assert snapshot.imminent == []
|
||||
assert snapshot.upcoming == []
|
||||
|
||||
def test_should_classify_into_three_buckets(self, monkeypatch):
|
||||
today = date(2026, 6, 1)
|
||||
monkeypatch.setattr(
|
||||
litellm,
|
||||
"model_cost",
|
||||
{
|
||||
"deprecated-model": {
|
||||
"deprecation_date": "2026-01-01",
|
||||
"litellm_provider": "openai",
|
||||
},
|
||||
"imminent-model": {
|
||||
"deprecation_date": "2026-06-15",
|
||||
"litellm_provider": "openai",
|
||||
},
|
||||
"upcoming-model": {
|
||||
"deprecation_date": "2027-01-01",
|
||||
"litellm_provider": "openai",
|
||||
},
|
||||
},
|
||||
)
|
||||
router = _make_router(
|
||||
[
|
||||
{
|
||||
"model_name": "deprecated-alias",
|
||||
"litellm_params": {"model": "openai/deprecated-model"},
|
||||
"model_info": {"id": "1"},
|
||||
},
|
||||
{
|
||||
"model_name": "imminent-alias",
|
||||
"litellm_params": {"model": "imminent-model"},
|
||||
"model_info": {"id": "2"},
|
||||
},
|
||||
{
|
||||
"model_name": "upcoming-alias",
|
||||
"litellm_params": {"model": "openai/upcoming-model"},
|
||||
"model_info": {"id": "3"},
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
snapshot = collect_model_deprecations(
|
||||
llm_router=router, warn_within_days=30, today=today
|
||||
)
|
||||
|
||||
assert [m.model_name for m in snapshot.deprecated] == ["deprecated-alias"]
|
||||
assert [m.model_name for m in snapshot.imminent] == ["imminent-alias"]
|
||||
assert [m.model_name for m in snapshot.upcoming] == ["upcoming-alias"]
|
||||
|
||||
assert snapshot.deprecated[0].days_until_deprecation < 0
|
||||
assert snapshot.imminent[0].days_until_deprecation == 14
|
||||
assert snapshot.upcoming[0].days_until_deprecation > 30
|
||||
|
||||
def test_should_prefer_explicit_deployment_override(self, monkeypatch):
|
||||
today = date(2026, 6, 1)
|
||||
monkeypatch.setattr(
|
||||
litellm,
|
||||
"model_cost",
|
||||
{"some-model": {"deprecation_date": "2030-01-01"}},
|
||||
)
|
||||
router = _make_router(
|
||||
[
|
||||
{
|
||||
"model_name": "my-alias",
|
||||
"litellm_params": {"model": "some-model"},
|
||||
"model_info": {
|
||||
"id": "x",
|
||||
"deprecation_date": "2026-06-10",
|
||||
},
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
snapshot = collect_model_deprecations(
|
||||
llm_router=router, warn_within_days=30, today=today
|
||||
)
|
||||
|
||||
assert len(snapshot.imminent) == 1
|
||||
assert snapshot.imminent[0].deprecation_date == date(2026, 6, 10)
|
||||
|
||||
def test_should_dedupe_duplicate_deployments_in_same_group(self, monkeypatch):
|
||||
today = date(2026, 6, 1)
|
||||
monkeypatch.setattr(
|
||||
litellm,
|
||||
"model_cost",
|
||||
{"shared-model": {"deprecation_date": "2026-06-10"}},
|
||||
)
|
||||
router = _make_router(
|
||||
[
|
||||
{
|
||||
"model_name": "alias",
|
||||
"litellm_params": {"model": "shared-model"},
|
||||
"model_info": {"id": "1"},
|
||||
},
|
||||
{
|
||||
"model_name": "alias",
|
||||
"litellm_params": {"model": "shared-model"},
|
||||
"model_info": {"id": "2"},
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
snapshot = collect_model_deprecations(
|
||||
llm_router=router, warn_within_days=30, today=today
|
||||
)
|
||||
|
||||
assert len(snapshot.imminent) == 1
|
||||
|
||||
def test_should_resolve_via_base_model(self, monkeypatch):
|
||||
today = date(2026, 6, 1)
|
||||
monkeypatch.setattr(
|
||||
litellm,
|
||||
"model_cost",
|
||||
{"base-thing": {"deprecation_date": "2026-06-10"}},
|
||||
)
|
||||
router = _make_router(
|
||||
[
|
||||
{
|
||||
"model_name": "alias",
|
||||
"litellm_params": {"model": "azure/some-deployment-name"},
|
||||
"model_info": {"id": "1", "base_model": "base-thing"},
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
snapshot = collect_model_deprecations(
|
||||
llm_router=router, warn_within_days=30, today=today
|
||||
)
|
||||
|
||||
assert len(snapshot.imminent) == 1
|
||||
assert snapshot.imminent[0].litellm_model == "base-thing"
|
||||
|
||||
|
||||
class TestFormatDeprecationAlertMessage:
|
||||
def test_should_return_none_when_nothing_to_alert(self):
|
||||
snapshot = collect_model_deprecations(llm_router=None)
|
||||
assert format_deprecation_alert_message(snapshot) is None
|
||||
|
||||
def test_should_render_imminent_and_deprecated_sections(self, monkeypatch):
|
||||
today = date(2026, 6, 1)
|
||||
monkeypatch.setattr(
|
||||
litellm,
|
||||
"model_cost",
|
||||
{
|
||||
"dead-model": {
|
||||
"deprecation_date": "2026-01-01",
|
||||
"litellm_provider": "openai",
|
||||
},
|
||||
"soon-model": {
|
||||
"deprecation_date": "2026-06-15",
|
||||
"litellm_provider": "anthropic",
|
||||
},
|
||||
"later-model": {
|
||||
"deprecation_date": "2027-01-01",
|
||||
"litellm_provider": "anthropic",
|
||||
},
|
||||
},
|
||||
)
|
||||
router = _make_router(
|
||||
[
|
||||
{
|
||||
"model_name": "dead",
|
||||
"litellm_params": {"model": "dead-model"},
|
||||
"model_info": {"id": "1"},
|
||||
},
|
||||
{
|
||||
"model_name": "soon",
|
||||
"litellm_params": {"model": "soon-model"},
|
||||
"model_info": {"id": "2"},
|
||||
},
|
||||
{
|
||||
"model_name": "later",
|
||||
"litellm_params": {"model": "later-model"},
|
||||
"model_info": {"id": "3"},
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
snapshot = collect_model_deprecations(
|
||||
llm_router=router, warn_within_days=30, today=today
|
||||
)
|
||||
message = format_deprecation_alert_message(snapshot)
|
||||
|
||||
assert message is not None
|
||||
assert "Already deprecated" in message
|
||||
assert "Deprecating within 30 days" in message
|
||||
assert "`dead`" in message
|
||||
assert "`soon`" in message
|
||||
# Upcoming models must NOT be in the alert (avoid alert fatigue).
|
||||
assert "`later`" not in message
|
||||
Loading…
Add table
Reference in a new issue