Source code for catalyst.core.callbacks.early_stop
# flake8: noqa
# @TODO: code formatting issue for 20.07 release
from catalyst.core.callback import Callback, CallbackNode, CallbackOrder
from catalyst.core.runner import IRunner
[docs]class CheckRunCallback(Callback):
"""Execute exeperiment not fully."""
[docs] def __init__(self, num_batch_steps: int = 3, num_epoch_steps: int = 2):
"""
Args:
num_batch_steps (int): number of batches to iterate in epoch
num_epoch_steps (int): number of epoch to perform in a stage
"""
super().__init__(order=CallbackOrder.external, node=CallbackNode.all)
self.num_batch_steps = num_batch_steps
self.num_epoch_steps = num_epoch_steps
[docs] def on_epoch_end(self, runner: IRunner):
"""Check if iterated specified number of epochs.
Args:
runner (IRunner): current runner
"""
if runner.epoch >= self.num_epoch_steps:
runner.need_early_stop = True
[docs] def on_batch_end(self, runner: IRunner):
"""Check if iterated specified number of batches.
Args:
runner (IRunner): current runner
"""
if runner.loader_batch_step >= self.num_batch_steps:
runner.need_early_stop = True
[docs]class EarlyStoppingCallback(Callback):
"""Early exit based on metric.
Example of usage in notebook API:
.. code-block:: python
runner = SupervisedRunner()
runner.train(
...
callbacks=[
...
EarlyStoppingCallback(
patience=5,
metric="my_metric",
minimize=True,
)
...
]
)
...
Example of usage in config API:
.. code-block:: yaml
stages:
...
stage_N:
...
callbacks_params:
...
early_stopping:
callback: EarlyStoppingCallback
# arguments for EarlyStoppingCallback
patience: 5
metric: my_metric
minimize: true
...
"""
[docs] def __init__(
self,
patience: int,
metric: str = "loss",
minimize: bool = True,
min_delta: float = 1e-6,
):
"""
Args:
patience (int): number of epochs with no improvement
after which training will be stopped.
metric (str): metric name to use for early stopping, default
is ``"loss"``.
minimize (bool): if ``True`` then expected that metric should
decrease and early stopping will be performed only when metric
stops decreasing. If ``False`` then expected
that metric should increase. Default value ``True``.
min_delta (float): minimum change in the monitored metric
to qualify as an improvement, i.e. an absolute change
of less than min_delta, will count as no improvement,
default value is ``1e-6``.
"""
super().__init__(order=CallbackOrder.external, node=CallbackNode.all)
self.best_score = None
self.metric = metric
self.patience = patience
self.num_bad_epochs = 0
self.is_better = None
if minimize:
self.is_better = lambda score, best: score <= (best - min_delta)
else:
self.is_better = lambda score, best: score >= (best + min_delta)
[docs] def on_epoch_end(self, runner: IRunner) -> None:
"""Check if should be performed early stopping.
Args:
runner (IRunner): current runner
"""
if runner.stage_name.startswith("infer"):
return
score = runner.valid_metrics[self.metric]
if self.best_score is None or self.is_better(score, self.best_score):
self.num_bad_epochs = 0
self.best_score = score
else:
self.num_bad_epochs += 1
if self.num_bad_epochs >= self.patience:
print(f"Early stop at {runner.epoch} epoch")
runner.need_early_stop = True
__all__ = ["CheckRunCallback", "EarlyStoppingCallback"]