Challenger

One of the benefits to linking and tracking ModelCards along with various Runcard metrics is that it's relatively easy to compare different model versions via the ModelChallenger class.

Examples

For examples please refer to examples/challenger for examples on how to compare registered model and un-registered models.

---

Docs

opsml.model.challenger.ModelChallenger

Source code in opsml/model/challenger.py

class ModelChallenger:
    def __init__(self, challenger: ModelCard):
        """
        Instantiates ModelChallenger class

        Args:
            challenger:
                ModelCard of challenger

        """
        self._challenger = challenger
        self._challenger_metric: Optional[Metric] = None
        self._registries = CardRegistries()

    @property
    def challenger_metric(self) -> Metric:
        if self._challenger_metric is not None:
            return self._challenger_metric
        raise ValueError("Challenger metric not set")

    @challenger_metric.setter
    def challenger_metric(self, metric: Metric) -> None:
        self._challenger_metric = metric

    def _get_last_champion_record(self) -> Optional[Dict[str, Any]]:
        """Gets the previous champion record"""

        champion_records = self._registries.model.list_cards(
            name=self._challenger.name,
            repository=self._challenger.repository,
        )

        if not champion_records:
            return None

        for record in champion_records:
            if record.get("runcard_uid") is not None and record.get("version") != self._challenger.version:
                return record

        logger.info("No previous version associated with a RunCard. Challenger wins")
        return None

    def _get_runcard_metric(self, runcard_uid: str, metric_name: str) -> Optional[Metric]:
        """
        Loads a RunCard from uid

        Args:
            runcard_uid:
                RunCard uid
            metric_name:
                Name of metric

        """
        runcard = cast(RunCard, self._registries.run.load_card(uid=runcard_uid))
        metric = runcard.get_metric(name=metric_name)

        if not metric:
            return None

        return metric[0]

    def _battle(self, champion: CardInfo, champion_metric: Metric, lower_is_better: bool) -> BattleReport:
        """
        Runs a battle between champion and current challenger

        Args:
            champion:
                Champion record
            champion_metric:
                Champion metric from a runcard
            lower_is_better:
                Whether lower metric is preferred

        Returns:
            `BattleReport`

        """
        if lower_is_better:
            challenger_win = self.challenger_metric.value < champion_metric.value
        else:
            challenger_win = self.challenger_metric.value > champion_metric.value
        return BattleReport.model_construct(
            champion_name=str(champion.name),
            champion_version=str(champion.version),
            champion_metric=champion_metric,
            challenger_metric=self.challenger_metric.model_copy(deep=True),
            challenger_win=challenger_win,
        )

    def _battle_last_model_version(self, metric_name: str, lower_is_better: bool) -> BattleReport:
        """Compares the last champion model to the current challenger"""

        champion_record = self._get_last_champion_record()

        if champion_record is None:
            logger.info("No previous model found. Challenger wins")

            return BattleReport(
                champion_name="No model",
                champion_version="No version",
                challenger_win=True,
            )

        runcard_id = champion_record.get("runcard_uid")
        assert runcard_id is not None, "RunCard uid is None"
        champion_metric = self._get_runcard_metric(runcard_uid=runcard_id, metric_name=metric_name)

        if champion_metric is None:
            logger.info("Metric not found in RunCard. Challenger wins")
            return BattleReport(
                champion_name=champion_record.get("name"),
                champion_version=champion_record.get("version"),
                challenger_win=True,
            )

        return self._battle(
            champion=CardInfo(
                name=champion_record.get("name"),
                version=champion_record.get("version"),
            ),
            champion_metric=champion_metric,
            lower_is_better=lower_is_better,
        )

    def _battle_champions(
        self,
        champions: List[CardInfo],
        metric_name: str,
        lower_is_better: bool,
    ) -> List[BattleReport]:
        """Loops through and creates a `BattleReport` for each champion"""
        battle_reports = []

        for champion in champions:
            champion_record = self._registries.model.list_cards(info=champion)

            if not bool(champion_record):
                logger.error(f"Champion model does not exist. {champion.name} {champion.version}")
                battle_reports.append(
                    BattleReport(
                        champion_name=champion.name,
                        champion_version="model not found",
                        challenger_win=True,
                    )
                )
                continue

            champion_card = champion_record[0]
            runcard_uid = champion_card.get("runcard_uid")

            if runcard_uid is None:
                logger.error(f"No RunCard associated with champion: {champion.name} {champion.version}")
                battle_reports.append(
                    BattleReport(
                        champion_name=champion.name,
                        champion_version="no runcard",
                        challenger_win=True,
                    )
                )
                continue

            champion_metric = self._get_runcard_metric(
                runcard_uid=runcard_uid,
                metric_name=metric_name,
            )

            if champion_metric is None:
                logger.error(f"No metric found in RunCard associated with champion: {champion.name} {champion.version}")
                battle_reports.append(
                    BattleReport(
                        champion_name=champion.name,
                        champion_version="metric not found",
                        challenger_win=True,
                    )
                )
                continue

            # update name, repository and version in case of None
            champion.name = champion.name or champion_card.get("name")
            champion.repository = champion.repository or champion_card.get("repository")
            champion.version = champion.version or champion_card.get("version")

            battle_reports.append(
                self._battle(
                    champion=champion,
                    champion_metric=champion_metric,
                    lower_is_better=lower_is_better,
                )
            )
        return battle_reports

    def challenge_champion(
        self,
        metric_name: MetricName,
        metric_value: Optional[MetricValue] = None,
        champions: Optional[List[CardInfo]] = None,
        lower_is_better: Union[bool, List[bool]] = True,
    ) -> Dict[str, List[BattleReport]]:
        """
        Challenges n champion models against the challenger model. If no champion is provided,
        the latest model version is used as a champion.

        Args:
            champions:
                Optional list of champion CardInfo
            metric_name:
                Name of metric to evaluate
            metric_value:
                Challenger metric value
            lower_is_better:
                Whether a lower metric value is better or not

        Returns
            `BattleReport`
        """

        # validate inputs
        inputs = ChallengeInputs(
            metric_name=metric_name,
            metric_value=metric_value,
            lower_is_better=lower_is_better,
        )

        report_dict = {}

        for name, value, _lower_is_better in zip(
            inputs.metric_names,
            inputs.metric_values,
            inputs.thresholds,
        ):
            # get challenger metric
            if value is None:
                if self._challenger.metadata.runcard_uid is not None:
                    metric = self._get_runcard_metric(self._challenger.metadata.runcard_uid, metric_name=name)

                    assert metric is not None, "Challenger metric not found"
                    self.challenger_metric = metric
                else:
                    raise ValueError("Challenger and champions must be associated with a registered RunCard")
            else:
                self.challenger_metric = Metric(name=name, value=value)

            if champions is None:
                report_dict[name] = [
                    self._battle_last_model_version(
                        metric_name=name,
                        lower_is_better=_lower_is_better,
                    )
                ]

            else:
                report_dict[name] = self._battle_champions(
                    champions=champions,
                    metric_name=name,
                    lower_is_better=_lower_is_better,
                )

        return report_dict

__init__(challenger)

Instantiates ModelChallenger class

Parameters:

Name	Type	Description	Default
challenger	ModelCard	ModelCard of challenger	required

Source code in opsml/model/challenger.py

def __init__(self, challenger: ModelCard):
    """
    Instantiates ModelChallenger class

    Args:
        challenger:
            ModelCard of challenger

    """
    self._challenger = challenger
    self._challenger_metric: Optional[Metric] = None
    self._registries = CardRegistries()

challenge_champion(metric_name, metric_value=None, champions=None, lower_is_better=True)

Challenges n champion models against the challenger model. If no champion is provided, the latest model version is used as a champion.

Parameters:

Name	Type	Description	Default
champions	Optional[List[CardInfo]]	Optional list of champion CardInfo	None
metric_name	MetricName	Name of metric to evaluate	required
metric_value	Optional[MetricValue]	Challenger metric value	None
lower_is_better	Union[bool, List[bool]]	Whether a lower metric value is better or not	True

Returns BattleReport

Source code in opsml/model/challenger.py

def challenge_champion(
    self,
    metric_name: MetricName,
    metric_value: Optional[MetricValue] = None,
    champions: Optional[List[CardInfo]] = None,
    lower_is_better: Union[bool, List[bool]] = True,
) -> Dict[str, List[BattleReport]]:
    """
    Challenges n champion models against the challenger model. If no champion is provided,
    the latest model version is used as a champion.

    Args:
        champions:
            Optional list of champion CardInfo
        metric_name:
            Name of metric to evaluate
        metric_value:
            Challenger metric value
        lower_is_better:
            Whether a lower metric value is better or not

    Returns
        `BattleReport`
    """

    # validate inputs
    inputs = ChallengeInputs(
        metric_name=metric_name,
        metric_value=metric_value,
        lower_is_better=lower_is_better,
    )

    report_dict = {}

    for name, value, _lower_is_better in zip(
        inputs.metric_names,
        inputs.metric_values,
        inputs.thresholds,
    ):
        # get challenger metric
        if value is None:
            if self._challenger.metadata.runcard_uid is not None:
                metric = self._get_runcard_metric(self._challenger.metadata.runcard_uid, metric_name=name)

                assert metric is not None, "Challenger metric not found"
                self.challenger_metric = metric
            else:
                raise ValueError("Challenger and champions must be associated with a registered RunCard")
        else:
            self.challenger_metric = Metric(name=name, value=value)

        if champions is None:
            report_dict[name] = [
                self._battle_last_model_version(
                    metric_name=name,
                    lower_is_better=_lower_is_better,
                )
            ]

        else:
            report_dict[name] = self._battle_champions(
                champions=champions,
                metric_name=name,
                lower_is_better=_lower_is_better,
            )

    return report_dict