Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/common_utils/model_deprecation.py: 65%

1from __future__ import annotations 

2 

3from collections.abc import Mapping, Sequence 

4from dataclasses import dataclass 

5from datetime import date, datetime, timezone 

6from itertools import groupby 

7from types import MappingProxyType 

8from typing import TYPE_CHECKING, Final 

9 

10import litellm 

11from litellm._logging import verbose_logger 

12from litellm.types.proxy.model_deprecation import ( 

13 DEFAULT_DEPRECATION_WARN_DAYS, 

14 DeprecationStatus, 

15 ModelDeprecationInfo, 

16 ModelDeprecationResponse, 

17) 

18 

19if TYPE_CHECKING: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true

20 from litellm.router import Router 

21 

22_NO_MODEL_METADATA: Final[Mapping[str, object]] = MappingProxyType({}) 

23 

24 

25@dataclass(frozen=True, slots=True) 

26class _ResolvedDeprecation: 

27 deprecation_date: date 

28 litellm_model: str | None 

29 litellm_provider: str | None 

30 

31 

32def _parse_deprecation_date(raw_value: object) -> date | None: 

33 if isinstance(raw_value, datetime): 33 ↛ 34line 33 didn't jump to line 34 because the condition on line 33 was never true

34 return raw_value.date() 

35 if isinstance(raw_value, date): 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true

36 return raw_value 

37 if not isinstance(raw_value, str): 37 ↛ 39line 37 didn't jump to line 39 because the condition on line 37 was always true

38 return None 

39 try: 

40 return date.fromisoformat(raw_value.strip()) 

41 except ValueError: 

42 return None 

43 

44 

45def _cost_map_lookup(model_key: object) -> _ResolvedDeprecation | None: 

46 if not isinstance(model_key, str) or not model_key: 

47 return None 

48 entry: Final = litellm.model_cost.get(model_key) 

49 if not isinstance(entry, Mapping): 

50 return None 

51 parsed: Final = _parse_deprecation_date(entry.get("deprecation_date")) 

52 if parsed is None: 52 ↛ 54line 52 didn't jump to line 54 because the condition on line 52 was always true

53 return None 

54 provider: Final = entry.get("litellm_provider") 

55 return _ResolvedDeprecation( 

56 deprecation_date=parsed, 

57 litellm_model=model_key, 

58 litellm_provider=provider if isinstance(provider, str) else None, 

59 ) 

60 

61 

62def _mapping_field(deployment: Mapping[str, object], key: str) -> Mapping[str, object]: 

63 value: Final = deployment.get(key) 

64 return value if isinstance(value, Mapping) else _NO_MODEL_METADATA 

65 

66 

67def _resolve_deployment_deprecation( 

68 deployment: Mapping[str, object], 

69) -> _ResolvedDeprecation | None: 

70 """Resolve a deployment's deprecation date, preferring its explicit override""" 

71 model_info: Final = _mapping_field(deployment, "model_info") 

72 raw_model: Final = _mapping_field(deployment, "litellm_params").get("model") 

73 

74 override: Final = _parse_deprecation_date(model_info.get("deprecation_date")) 

75 if override is not None: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 provider: Final = model_info.get("litellm_provider") 

77 return _ResolvedDeprecation( 

78 deprecation_date=override, 

79 litellm_model=raw_model if isinstance(raw_model, str) else None, 

80 litellm_provider=provider if isinstance(provider, str) else None, 

81 ) 

82 

83 unprefixed: Final = raw_model.split("/", 1)[1] if isinstance(raw_model, str) and "/" in raw_model else None 

84 return next( 

85 ( 

86 resolved 

87 for resolved in ( 

88 _cost_map_lookup(model_info.get("base_model")), 

89 _cost_map_lookup(raw_model), 

90 _cost_map_lookup(unprefixed), 

91 ) 

92 if resolved is not None 

93 ), 

94 None, 

95 ) 

96 

97 

98def _classify(days_until: int, warn_within_days: int) -> DeprecationStatus: 

99 if days_until < 0: 

100 return "deprecated" 

101 if days_until <= warn_within_days: 

102 return "imminent" 

103 return "upcoming" 

104 

105 

106def _build_info(deployment: Mapping[str, object], today: date, warn_within_days: int) -> ModelDeprecationInfo | None: 

107 model_name: Final = deployment.get("model_name") 

108 if not isinstance(model_name, str) or not model_name: 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true

109 return None 

110 

111 resolved: Final = _resolve_deployment_deprecation(deployment) 

112 if resolved is None: 112 ↛ 115line 112 didn't jump to line 115 because the condition on line 112 was always true

113 return None 

114 

115 days_until: Final = (resolved.deprecation_date - today).days 

116 return ModelDeprecationInfo( 

117 model_name=model_name, 

118 litellm_model=resolved.litellm_model, 

119 deprecation_date=resolved.deprecation_date, 

120 days_until_deprecation=days_until, 

121 status=_classify(days_until, warn_within_days), 

122 litellm_provider=resolved.litellm_provider, 

123 ) 

124 

125 

126def _dedupe( 

127 models: Sequence[ModelDeprecationInfo], 

128) -> tuple[ModelDeprecationInfo, ...]: 

129 """Report a model group carrying the same date on several deployments once""" 

130 ordered: Final = sorted(models, key=lambda model: (model.model_name, model.deprecation_date)) 

131 return tuple( 

132 next(group) for _, group in groupby(ordered, key=lambda model: (model.model_name, model.deprecation_date)) 

133 ) 

134 

135 

136def _bucket(models: Sequence[ModelDeprecationInfo], status: DeprecationStatus) -> tuple[ModelDeprecationInfo, ...]: 

137 return tuple( 

138 sorted( 

139 (model for model in models if model.status == status), 

140 key=lambda model: model.deprecation_date, 

141 ) 

142 ) 

143 

144 

145def collect_model_deprecations( 

146 llm_router: Router | None, 

147 warn_within_days: int = DEFAULT_DEPRECATION_WARN_DAYS, 

148 today: date | None = None, 

149) -> ModelDeprecationResponse: 

150 """Bucket every deployment carrying a deprecation date by how urgent it is""" 

151 snapshot_time: Final = datetime.now(timezone.utc) 

152 effective_today: Final = today or snapshot_time.date() 

153 deployments: Final = (llm_router.get_model_list() or ()) if llm_router is not None else () 

154 

155 deduped: Final = _dedupe( 

156 tuple( 

157 info 

158 for info in (_build_info(deployment, effective_today, warn_within_days) for deployment in deployments) 

159 if info is not None 

160 ) 

161 ) 

162 

163 verbose_logger.debug( 

164 "model_deprecation: %d/%d deployments carry a deprecation date", 

165 len(deduped), 

166 len(deployments), 

167 ) 

168 

169 return ModelDeprecationResponse( 

170 deprecated=_bucket(deduped, "deprecated"), 

171 imminent=_bucket(deduped, "imminent"), 

172 upcoming=_bucket(deduped, "upcoming"), 

173 warn_within_days=warn_within_days, 

174 checked_at=snapshot_time, 

175 ) 

176 

177 

178def _escape_slack_mrkdwn(value: str) -> str: 

179 """Neutralize Slack control characters so a model name cannot forge a mention or link""" 

180 return value.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;") 

181 

182 

183def _format_entry(info: ModelDeprecationInfo) -> str: 

184 suffix: Final = ( 

185 f"already deprecated {abs(info.days_until_deprecation)}d ago" 

186 if info.days_until_deprecation < 0 

187 else f"in {info.days_until_deprecation}d" 

188 ) 

189 return ( 

190 f"• `{_escape_slack_mrkdwn(info.model_name)}` " 

191 f"(provider: {_escape_slack_mrkdwn(info.litellm_provider) if info.litellm_provider else 'unknown'}, " 

192 f"deprecates {info.deprecation_date.isoformat()}, {suffix})" 

193 ) 

194 

195 

196def format_deprecation_alert_message( 

197 snapshot: ModelDeprecationResponse, 

198) -> str | None: 

199 """Render the alert for the deprecated and imminent buckets, None when both are empty 

200 

201 Upcoming models are left out of the alert to keep it actionable. 

202 """ 

203 if not snapshot.deprecated and not snapshot.imminent: 

204 return None 

205 

206 deprecated_section: Final = ( 

207 ("\n*Already deprecated:*", *(_format_entry(i) for i in snapshot.deprecated)) if snapshot.deprecated else () 

208 ) 

209 imminent_section: Final = ( 

210 ( 

211 f"\n*Deprecating within {snapshot.warn_within_days} days:*", 

212 *(_format_entry(i) for i in snapshot.imminent), 

213 ) 

214 if snapshot.imminent 

215 else () 

216 ) 

217 

218 return "\n".join( 

219 ( 

220 "*⚠️ Model Deprecation Warning*", 

221 *deprecated_section, 

222 *imminent_section, 

223 "\nPlan migrations to a supported model. See " 

224 "https://docs.litellm.ai/docs/proxy/model_management for guidance.", 

225 ) 

226 )