opsml.model.challenger

  1# Copyright (c) Shipt, Inc.
  2# This source code is licensed under the MIT license found in the
  3# LICENSE file in the root directory of this source tree.
  4from typing import Any, Dict, List, Optional, Union, cast
  5
  6from pydantic import BaseModel, ConfigDict, ValidationInfo, field_validator
  7
  8from opsml.cards.model import ModelCard
  9from opsml.cards.run import RunCard
 10from opsml.helpers.logging import ArtifactLogger
 11from opsml.registry.registry import CardRegistries
 12from opsml.types import CardInfo, Metric
 13
 14logger = ArtifactLogger.get_logger()
 15
 16# User interfaces should primarily be checked at runtime
 17
 18
 19class BattleReport(BaseModel):
 20    model_config = ConfigDict(arbitrary_types_allowed=True)
 21    champion_name: str
 22    champion_version: str
 23    champion_metric: Optional[Metric] = None
 24    challenger_metric: Optional[Metric] = None
 25    challenger_win: bool
 26
 27
 28MetricName = Union[str, List[str]]
 29MetricValue = Union[int, float, List[Union[int, float]]]
 30
 31
 32class ChallengeInputs(BaseModel):
 33    metric_name: MetricName
 34    metric_value: Optional[MetricValue] = None
 35    lower_is_better: Union[bool, List[bool]] = True
 36
 37    @property
 38    def metric_names(self) -> List[str]:
 39        return cast(List[str], self.metric_name)
 40
 41    @property
 42    def metric_values(self) -> List[Optional[Union[int, float]]]:
 43        return cast(List[Optional[Union[int, float]]], self.metric_value)
 44
 45    @property
 46    def thresholds(self) -> List[bool]:
 47        return cast(List[bool], self.lower_is_better)
 48
 49    @field_validator("metric_name")
 50    @classmethod
 51    def convert_name(cls, name: Union[List[str], str]) -> List[str]:
 52        if not isinstance(name, list):
 53            return [name]
 54        return name
 55
 56    @field_validator("metric_value")
 57    @classmethod
 58    def convert_value(cls, value: Optional[MetricValue], info: ValidationInfo) -> List[Any]:
 59        data = info.data
 60        metric = cast(MetricName, data["metric_name"])
 61        nbr_metrics = len(metric)
 62
 63        if value is not None:
 64            if not isinstance(value, list):
 65                metric_value = [value]
 66            else:
 67                metric_value = value
 68        else:
 69            metric_value = [None] * nbr_metrics  # type: ignore
 70
 71        if len(metric_value) != nbr_metrics:
 72            raise ValueError("List of metric values must be the same length as metric names")
 73
 74        return metric_value
 75
 76    @field_validator("lower_is_better")
 77    @classmethod
 78    def convert_threshold(cls, threshold: Union[bool, List[bool]], info: ValidationInfo) -> List[bool]:
 79        data = info.data
 80        metric = cast(MetricName, data["metric_name"])
 81        nbr_metrics = len(metric)
 82
 83        if not isinstance(threshold, list):
 84            _threshold = [threshold] * nbr_metrics
 85        else:
 86            _threshold = threshold
 87
 88        if len(_threshold) != nbr_metrics:
 89            if len(_threshold) == 1:
 90                _threshold = _threshold * nbr_metrics
 91            else:
 92                raise ValueError("Length of lower_is_better must be the same length as number of metrics")
 93
 94        return _threshold
 95
 96
 97class ModelChallenger:
 98    def __init__(self, challenger: ModelCard):
 99        """
100        Instantiates ModelChallenger class
101
102        Args:
103            challenger:
104                ModelCard of challenger
105
106        """
107        self._challenger = challenger
108        self._challenger_metric: Optional[Metric] = None
109        self._registries = CardRegistries()
110
111    @property
112    def challenger_metric(self) -> Metric:
113        if self._challenger_metric is not None:
114            return self._challenger_metric
115        raise ValueError("Challenger metric not set")
116
117    @challenger_metric.setter
118    def challenger_metric(self, metric: Metric) -> None:
119        self._challenger_metric = metric
120
121    def _get_last_champion_record(self) -> Optional[Dict[str, Any]]:
122        """Gets the previous champion record"""
123
124        champion_records = self._registries.model.list_cards(
125            name=self._challenger.name,
126            repository=self._challenger.repository,
127        )
128
129        if not bool(champion_records):
130            return None
131
132        # indicates challenger has been registered
133        if self._challenger.version is not None and len(champion_records) > 1:
134            return champion_records[1]
135
136        # account for cases where challenger is only model in registry
137        champion_record = champion_records[0]
138        if champion_record.get("version") == self._challenger.version:
139            return None
140
141        return champion_record
142
143    def _get_runcard_metric(self, runcard_uid: str, metric_name: str) -> Metric:
144        """
145        Loads a RunCard from uid
146
147        Args:
148            runcard_uid:
149                RunCard uid
150            metric_name:
151                Name of metric
152
153        """
154        runcard = cast(RunCard, self._registries.run.load_card(uid=runcard_uid))
155        metric = runcard.get_metric(name=metric_name)
156
157        if isinstance(metric, list):
158            metric = metric[0]
159
160        return metric
161
162    def _battle(self, champion: CardInfo, champion_metric: Metric, lower_is_better: bool) -> BattleReport:
163        """
164        Runs a battle between champion and current challenger
165
166        Args:
167            champion:
168                Champion record
169            champion_metric:
170                Champion metric from a runcard
171            lower_is_better:
172                Whether lower metric is preferred
173
174        Returns:
175            `BattleReport`
176
177        """
178        if lower_is_better:
179            challenger_win = self.challenger_metric.value < champion_metric.value
180        else:
181            challenger_win = self.challenger_metric.value > champion_metric.value
182        return BattleReport.model_construct(
183            champion_name=str(champion.name),
184            champion_version=str(champion.version),
185            champion_metric=champion_metric,
186            challenger_metric=self.challenger_metric.model_copy(deep=True),
187            challenger_win=challenger_win,
188        )
189
190    def _battle_last_model_version(self, metric_name: str, lower_is_better: bool) -> BattleReport:
191        """Compares the last champion model to the current challenger"""
192
193        champion_record = self._get_last_champion_record()
194
195        if champion_record is None:
196            logger.info("No previous model found. Challenger wins")
197
198            return BattleReport(
199                champion_name="No model",
200                champion_version="No version",
201                challenger_win=True,
202            )
203
204        runcard_id = champion_record.get("runcard_uid")
205        if runcard_id is None:
206            raise ValueError(f"No RunCard is associated with champion: {champion_record}")
207
208        champion_metric = self._get_runcard_metric(runcard_uid=runcard_id, metric_name=metric_name)
209
210        return self._battle(
211            champion=CardInfo(
212                name=champion_record.get("name"),
213                version=champion_record.get("version"),
214            ),
215            champion_metric=champion_metric,
216            lower_is_better=lower_is_better,
217        )
218
219    def _battle_champions(
220        self,
221        champions: List[CardInfo],
222        metric_name: str,
223        lower_is_better: bool,
224    ) -> List[BattleReport]:
225        """Loops through and creates a `BattleReport` for each champion"""
226        battle_reports = []
227
228        for champion in champions:
229            champion_record = self._registries.model.list_cards(
230                info=champion,
231            )
232
233            if not bool(champion_record):
234                raise ValueError(f"Champion model does not exist. {champion}")
235
236            champion_card = champion_record[0]
237            runcard_uid = champion_card.get("runcard_uid")
238            if runcard_uid is None:
239                raise ValueError(f"No RunCard associated with champion: {champion}")
240
241            champion_metric = self._get_runcard_metric(
242                runcard_uid=runcard_uid,
243                metric_name=metric_name,
244            )
245
246            # update name, repository and version in case of None
247            champion.name = champion.name or champion_card.get("name")
248            champion.repository = champion.repository or champion_card.get("repository")
249            champion.version = champion.version or champion_card.get("version")
250
251            battle_reports.append(
252                self._battle(
253                    champion=champion,
254                    champion_metric=champion_metric,
255                    lower_is_better=lower_is_better,
256                )
257            )
258        return battle_reports
259
260    def challenge_champion(
261        self,
262        metric_name: MetricName,
263        metric_value: Optional[MetricValue] = None,
264        champions: Optional[List[CardInfo]] = None,
265        lower_is_better: Union[bool, List[bool]] = True,
266    ) -> Dict[str, List[BattleReport]]:
267        """
268        Challenges n champion models against the challenger model. If no champion is provided,
269        the latest model version is used as a champion.
270
271        Args:
272            champions:
273                Optional list of champion CardInfo
274            metric_name:
275                Name of metric to evaluate
276            metric_value:
277                Challenger metric value
278            lower_is_better:
279                Whether a lower metric value is better or not
280
281        Returns
282            `BattleReport`
283        """
284
285        # validate inputs
286        inputs = ChallengeInputs(
287            metric_name=metric_name,
288            metric_value=metric_value,
289            lower_is_better=lower_is_better,
290        )
291
292        report_dict = {}
293
294        for name, value, _lower_is_better in zip(
295            inputs.metric_names,
296            inputs.metric_values,
297            inputs.thresholds,
298        ):
299            # get challenger metric
300            if value is None:
301                if self._challenger.metadata.runcard_uid is not None:
302                    self.challenger_metric = self._get_runcard_metric(
303                        self._challenger.metadata.runcard_uid, metric_name=name
304                    )
305                else:
306                    raise ValueError("Challenger and champions must be associated with a registered RunCard")
307            else:
308                self.challenger_metric = Metric(name=name, value=value)
309
310            if champions is None:
311                report_dict[name] = [
312                    self._battle_last_model_version(
313                        metric_name=name,
314                        lower_is_better=_lower_is_better,
315                    )
316                ]
317
318            else:
319                report_dict[name] = self._battle_champions(
320                    champions=champions,
321                    metric_name=name,
322                    lower_is_better=_lower_is_better,
323                )
324
325        return report_dict