feat(proxy): proactive model deprecation alerts and /model/deprecations endpoint

backend/routes/allowlist.py

@@ -35,6 +35,7 @@
     # Models & routing config
     "/model/",
     "/v1/model/info",
+    "/v1/model/deprecations",
     "/v2/model/",
     "/model_group",
     "/model_access_group/",

litellm/constants.py

@@ -1485,6 +1485,7 @@
 MONTHLY_SPEND_REPORT_JOB_ID: Final = "monthly_spend_report_job"
 PROMETHEUS_FALLBACK_STATS_JOB_ID: Final = "prometheus_fallback_stats_job"
 SLACK_DAILY_REPORT_LOCK_ID: Final = "slack_daily_report"
+SLACK_MODEL_DEPRECATION_LOCK_ID: Final = "slack_model_deprecation_warning"
 SPEND_LOG_RUN_LOOPS: Final = int(os.getenv("SPEND_LOG_RUN_LOOPS", 500))
 SPEND_LOG_CLEANUP_BATCH_SIZE: Final = int(os.getenv("SPEND_LOG_CLEANUP_BATCH_SIZE", 1000))
 SPEND_LOG_CLEANUP_MAX_CONSECUTIVE_BATCH_FAILURES = int(os.getenv("SPEND_LOG_CLEANUP_MAX_CONSECUTIVE_BATCH_FAILURES", 3))

litellm/integrations/SlackAlerting/slack_alerting.py

@@ -5,6 +5,7 @@
 import os
 import random
 import time
+from collections.abc import Callable
 from datetime import timedelta
 from typing import TYPE_CHECKING, Any, Final, Literal
 
@@ -17,7 +18,11 @@
 import litellm.types
 from litellm._logging import verbose_logger, verbose_proxy_logger
 from litellm.caching.caching import DualCache
-from litellm.constants import HOURS_IN_A_DAY, SLACK_DAILY_REPORT_LOCK_ID
+from litellm.constants import (
+    HOURS_IN_A_DAY,
+    SLACK_DAILY_REPORT_LOCK_ID,
+    SLACK_MODEL_DEPRECATION_LOCK_ID,
+)
 from litellm.integrations.custom_batch_logger import CustomBatchLogger
 from litellm.integrations.SlackAlerting.budget_alert_types import get_budget_alert_type
 from litellm.integrations.SlackAlerting.hanging_request_check import (
@@ -45,6 +50,10 @@
 from litellm.repositories.team_repository import TeamRepository
 from litellm.repositories.user_repository import UserRepository
 from litellm.types.integrations.slack_alerting import *
+from litellm.types.proxy.model_deprecation import (
+    DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
+    DEPRECATION_IDLE_POLL_SECONDS,
+)
 
 from ..email_templates.templates import *
 from .batching_handler import send_to_webhook, squash_payloads
@@ -59,6 +68,12 @@
     Router = Any
 
 
+def _proxy_llm_router() -> Router | None:
+    from litellm.proxy.proxy_server import llm_router
+
+    return llm_router
+
+
 class SlackAlerting(CustomBatchLogger):
     """
     Class for sending Slack Alerts
@@ -1044,6 +1059,99 @@ async def model_added_alert(self, model_name: str, litellm_model_name: str, pass
     async def model_removed_alert(self, model_name: str):
         pass
 
+    def _deprecation_alerts_enabled(self) -> bool:
+        return self.alerting is not None and AlertType.model_deprecation_warnings in self.alert_types
+
+    async def send_model_deprecation_alert(
+        self,
+        llm_router: Router | None = None,
+        pod_lock_manager: "PodLockManager | None" = None,
+    ) -> bool:
+        """Alert on the router's deprecated and imminent models, True when one was sent
+
+        The daily lock is claimed only once there is something to say, so an empty pass never blocks a
+        later real one, and a sent alert is stamped in the shared cache for a day so sibling pods stop asking
+        """
+        if not self._deprecation_alerts_enabled():
+            return False
+
+        from litellm.proxy.common_utils.model_deprecation import (
+            collect_model_deprecations,
+            format_deprecation_alert_message,
+        )
+
+        snapshot: Final = collect_model_deprecations(llm_router=llm_router)
+        message: Final = format_deprecation_alert_message(snapshot)
+        if message is None:
+            return False
+        if not await self._claimed_deprecation_alert_window(pod_lock_manager):
+            return False
+
+        level: Final[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={  # mutable-ok: send_alert takes a dict payload
+                "deprecated_count": len(snapshot.deprecated),
+                "imminent_count": len(snapshot.imminent),
+                "upcoming_count": len(snapshot.upcoming),
+            },
+        )
+        await self.internal_usage_cache.async_set_cache(
+            key=SlackAlertingCacheKeys.deprecation_alert_sent_key.value,
+            value=time.time(),
+            ttl=DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
+        )
+        return True
+
+    async def _claimed_deprecation_alert_window(self, pod_lock_manager: "PodLockManager | None") -> bool:
+        """Without a redis backed lock there is no fleet to coordinate, so a lone pod always alerts"""
+        if pod_lock_manager is None:
+            return True
+        return (
+            await pod_lock_manager.acquire_lock(
+                cronjob_id=SLACK_MODEL_DEPRECATION_LOCK_ID,
+                ttl=DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS,
+                allow_reentrant=False,
+            )
+        ) is not False
+
+    async def _deprecation_alert_sent_within_a_day(self) -> bool:
+        return (
+            await self.internal_usage_cache.async_get_cache(key=SlackAlertingCacheKeys.deprecation_alert_sent_key.value)
+        ) is not None
+
+    async def _run_deprecation_alert_pass(
+        self, llm_router: Router | None, pod_lock_manager: "PodLockManager | None"
+    ) -> bool:
+        if llm_router is None or not self._deprecation_alerts_enabled():
+            return False
+        if await self._deprecation_alert_sent_within_a_day():
+            return False
+        return await self.send_model_deprecation_alert(llm_router=llm_router, pod_lock_manager=pod_lock_manager)
+
+    async def run_scheduled_deprecation_check(
+        self,
+        get_llm_router: Callable[[], Router | None] = _proxy_llm_router,
+        pod_lock_manager: "PodLockManager | None" = None,
+    ) -> None:
+        """Poll every pass for a loaded router, the alert being on, and no alert in the last day, then alert
+
+        A pass that could not alert (no router yet, alert type off, a sibling pod holds the daily lock, or a
+        redis blip at claim time) is retried on the next poll instead of costing a day, while a pass that
+        raised (a missing webhook, say) backs off a full day so a misconfiguration logs once, not every poll
+        """
+        while True:
+            try:
+                await self._run_deprecation_alert_pass(get_llm_router(), pod_lock_manager)
+            except Exception as e:  # noqa: BLE001  # a failed alert must not kill the loop
+                verbose_proxy_logger.exception("Error in model deprecation alert loop: %s", e)
+                await asyncio.sleep(DEFAULT_DEPRECATION_CHECK_INTERVAL_SECONDS)
+                continue
+            await asyncio.sleep(DEPRECATION_IDLE_POLL_SECONDS)
+
     async def send_webhook_alert(self, webhook_event: WebhookEvent) -> bool:
         """
         Sends structured alert to webhook, if set.

litellm/proxy/common_utils/model_deprecation.py

@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import date, datetime, timezone
+from itertools import groupby
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Final
+
+import litellm
+from litellm._logging import verbose_logger
+from litellm.types.proxy.model_deprecation import (
+    DEFAULT_DEPRECATION_WARN_DAYS,
+    DeprecationStatus,
+    ModelDeprecationInfo,
+    ModelDeprecationResponse,
+)
+
+if TYPE_CHECKING:
+    from litellm.router import Router
+
+_NO_MODEL_METADATA: Final[Mapping[str, object]] = MappingProxyType({})
+
+
+@dataclass(frozen=True, slots=True)
+class _ResolvedDeprecation:
+    deprecation_date: date
+    litellm_model: str | None
+    litellm_provider: str | None
+
+
+def _parse_deprecation_date(raw_value: object) -> date | None:
+    if isinstance(raw_value, datetime):
+        return raw_value.date()
+    if isinstance(raw_value, date):
+        return raw_value
+    if not isinstance(raw_value, str):
+        return None
+    try:
+        return date.fromisoformat(raw_value.strip())
+    except ValueError:
+        return 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"))
+    if parsed is None:
+        return None
+    provider: Final = entry.get("litellm_provider")
+    return _ResolvedDeprecation(
+        deprecation_date=parsed,
+        litellm_model=model_key,
+        litellm_provider=provider if isinstance(provider, str) else None,
+    )
+
+
+def _mapping_field(deployment: Mapping[str, object], key: str) -> Mapping[str, object]:
+    value: Final = deployment.get(key)
+    return value if isinstance(value, Mapping) else _NO_MODEL_METADATA
+
+
+def _resolve_deployment_deprecation(
+    deployment: Mapping[str, object],
+) -> _ResolvedDeprecation | None:
+    """Resolve a deployment's deprecation date, preferring its explicit override"""
+    model_info: Final = _mapping_field(deployment, "model_info")
+    raw_model: Final = _mapping_field(deployment, "litellm_params").get("model")
+
+    override: Final = _parse_deprecation_date(model_info.get("deprecation_date"))
+    if override is not None:
+        provider: Final = model_info.get("litellm_provider")
+        return _ResolvedDeprecation(
+            deprecation_date=override,
+            litellm_model=raw_model if isinstance(raw_model, str) else None,
+            litellm_provider=provider if isinstance(provider, str) else None,
+        )
+
+    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,
+    )
+
+
+def _classify(days_until: int, warn_within_days: int) -> DeprecationStatus:
+    if days_until < 0:
+        return "deprecated"
+    if days_until <= warn_within_days:
+        return "imminent"
+    return "upcoming"
+
+
+def _build_info(deployment: Mapping[str, object], today: date, warn_within_days: int) -> ModelDeprecationInfo | None:
+    model_name: Final = deployment.get("model_name")
+    if not isinstance(model_name, str) or not model_name:
+        return None
+
+    resolved: Final = _resolve_deployment_deprecation(deployment)
+    if resolved is None:
+        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,
+    )
+
+
+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))
+    return tuple(
+        next(group) for _, group in groupby(ordered, key=lambda model: (model.model_name, model.deprecation_date))
+    )
+
+
+def _bucket(models: Sequence[ModelDeprecationInfo], status: DeprecationStatus) -> tuple[ModelDeprecationInfo, ...]:
+    return tuple(
+        sorted(
+            (model for model in models if model.status == status),
+            key=lambda model: model.deprecation_date,
+        )
+    )
+
+
+def collect_model_deprecations(
+    llm_router: Router | None,
+    warn_within_days: int = DEFAULT_DEPRECATION_WARN_DAYS,
+    today: date | None = None,
+) -> ModelDeprecationResponse:
+    """Bucket every deployment carrying a deprecation date by how urgent it is"""
+    snapshot_time: Final = datetime.now(timezone.utc)
+    effective_today: Final = today or snapshot_time.date()
+    deployments: Final = (llm_router.get_model_list() or ()) if llm_router is not None else ()
+
+    deduped: Final = _dedupe(
+        tuple(
+            info
+            for info in (_build_info(deployment, effective_today, warn_within_days) for deployment in deployments)
+            if info is not None
+        )
+    )
+
+    verbose_logger.debug(
+        "model_deprecation: %d/%d deployments carry a deprecation date",
+        len(deduped),
+        len(deployments),
+    )
+
+    return ModelDeprecationResponse(
+        deprecated=_bucket(deduped, "deprecated"),
+        imminent=_bucket(deduped, "imminent"),
+        upcoming=_bucket(deduped, "upcoming"),
+        warn_within_days=warn_within_days,
+        checked_at=snapshot_time,
+    )
+
+
+def _escape_slack_mrkdwn(value: str) -> str:
+    """Neutralize Slack control characters so a model name cannot forge a mention or link"""
+    return value.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
+
+
+def _format_entry(info: ModelDeprecationInfo) -> str:
+    suffix: Final = (
+        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"• `{_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})"
+    )
+
+
+def format_deprecation_alert_message(
+    snapshot: ModelDeprecationResponse,
+) -> str | None:
+    """Render the alert for the deprecated and imminent buckets, None when both are empty
+
+    Upcoming models are left out of the alert to keep it actionable.
+    """
+    if not snapshot.deprecated and not snapshot.imminent:
+        return None
+
+    deprecated_section: Final = (
+        ("\n*Already deprecated:*", *(_format_entry(i) for i in snapshot.deprecated)) if snapshot.deprecated else ()
+    )
+    imminent_section: Final = (
+        (
+            f"\n*Deprecating within {snapshot.warn_within_days} days:*",
+            *(_format_entry(i) for i in snapshot.imminent),
+        )
+        if snapshot.imminent
+        else ()
+    )
+
+    return "\n".join(
+        (
+            "*⚠️ Model Deprecation Warning*",
+            *deprecated_section,
+            *imminent_section,
+            "\nPlan migrations to a supported model. See "
+            "https://docs.litellm.ai/docs/proxy/model_management for guidance.",
+        )
+    )