Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / trainer / callbacks / evaluate_callback.py: 37%

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""Scheduled placeholder for in-training evaluation.""" 

16 

17import logging 

18from typing import Any, Literal 

19 

20from .base import Callback, TrainerState 

21 

22 

23logger = logging.getLogger(__name__) 

24 

25 

26class EvaluateCallback(Callback): 

27 """Expose step and epoch evaluation hooks without running validation yet.""" 

28 

29 def __init__(self, trainer: Any) -> None: 

30 """Read evaluation intervals from the Trainer configuration. 

31 

32 Args: 

33 trainer: Trainer that owns the callback lifecycle. 

34 """ 

35 super().__init__(trainer) 

36 training_config = trainer.config.training 

37 self.eval_steps = training_config.eval_steps 

38 self.eval_epochs = training_config.eval_epochs 

39 self._last_evaluated_step: int | None = None 

40 

41 def _evaluate(self, state: TrainerState, trigger: Literal["step", "epoch"]) -> None: 

42 """Report that the scheduled validation loop is still a placeholder.""" 

43 if getattr(self.trainer, "global_rank", 0) == 0: 

44 logger.warning( 

45 "Evaluation triggered at step %s by %s, but the validation loop is not implemented yet.", 

46 state.global_step, 

47 trigger, 

48 ) 

49 

50 def _trigger_evaluation(self, state: TrainerState, trigger: Literal["step", "epoch"]) -> None: 

51 """Invoke the extension point at most once for each global step.""" 

52 if self._last_evaluated_step == state.global_step: 

53 return 

54 self._evaluate(state, trigger) 

55 self._last_evaluated_step = state.global_step 

56 

57 def on_step_end(self, state: TrainerState, **kwargs: Any) -> None: 

58 """Trigger the placeholder at the configured optimizer-step cadence.""" 

59 del kwargs 

60 if self.eval_steps > 0 and state.global_step % self.eval_steps == 0: 

61 self._trigger_evaluation(state, trigger="step") 

62 

63 def on_epoch_end(self, state: TrainerState, **kwargs: Any) -> None: 

64 """Trigger the placeholder at the configured completed-epoch cadence.""" 

65 del kwargs 

66 if self.eval_epochs > 0 and (state.epoch + 1) % self.eval_epochs == 0: 

67 self._trigger_evaluation(state, trigger="epoch")