Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / trainer / vl_trainer.py: 0%

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"""VLTrainer for native Qwen3-VL multimodal training.""" 

16import logging 

17from typing import Any, Dict, List 

18 

19import torch 

20 

21from hyper_parallel.trainer.base import BaseTrainer 

22 

23logger = logging.getLogger(__name__) 

24 

25 

26class VLTrainer: 

27 """Trainer for multimodal Qwen3-VL training (text + image/video). 

28 

29 Dataset construction delegates to :func:`hyper_parallel.data.build_dataset`. 

30 Built-in VL formats are ``vl_dummy`` (deterministic synthetic multimodal 

31 tensors) and ``preset_pt`` (replayed batches that already include 

32 ``pixel_values`` and ``image_grid_thw``). 

33 """ 

34 

35 def __init__(self, args): 

36 self.base = BaseTrainer(args) 

37 self.base._setup() 

38 self.base._build_model() 

39 self.base._freeze_model() 

40 self._build_model_assets() 

41 self._build_data_transform() 

42 self.base._build_dataset() 

43 self._build_collate_fn() 

44 self.base._build_dataloader() 

45 self.base._build_parallelized_model() 

46 self.base._build_optimizer() 

47 self.base._build_lr_scheduler() 

48 self.base._build_training_context() 

49 self.base._init_callbacks() 

50 self.base.on_init_end() 

51 

52 def _build_model_assets(self): 

53 """Load processor when a real VL dataset is configured.""" 

54 self.base.processor = None 

55 self.base.tokenizer = None 

56 data_type = self.base.args.data.type 

57 if data_type == "vl_dummy": 

58 return 

59 processor_path = ( 

60 getattr(self.base.args.data, "processor_path", None) 

61 or self.base.args.model.tokenizer_path 

62 or self.base.args.model.weights_path 

63 ) 

64 if not processor_path: 

65 raise ValueError("VL real-data mode requires data.processor_path or model.weights_path") 

66 from transformers import AutoProcessor # pylint: disable=C0415 

67 

68 self.base.processor = AutoProcessor.from_pretrained( 

69 processor_path, trust_remote_code=True, 

70 ) 

71 self.base.tokenizer = getattr(self.base.processor, "tokenizer", None) 

72 logger.info("Processor loaded from %s", processor_path) 

73 

74 def _build_data_transform(self): 

75 self.base.data_transform = None 

76 

77 @staticmethod 

78 def _stack_positions(batch: List[Dict[str, Any]], key: str): 

79 values = [item[key] for item in batch] 

80 if values[0].dim() == 1: 

81 return torch.stack(values) 

82 return torch.stack(values).transpose(0, 1).contiguous() 

83 

84 @staticmethod 

85 def _stack_or_cat_grids(batch: List[Dict[str, Any]], key: str): 

86 values = [item[key] for item in batch] 

87 if values[0].dim() == 1: 

88 return torch.stack(values) 

89 return torch.cat(values, dim=0) 

90 

91 @staticmethod 

92 def _maybe_cat_optional(batch: List[Dict[str, Any]], key: str): 

93 if key in batch[0] and batch[0].get(key) is not None: 

94 return torch.cat([item[key] for item in batch], dim=0) 

95 return None 

96 

97 def _vl_collate(self, batch: List[Dict[str, Any]]) -> Dict[str, Any]: 

98 """Collate VL tensor rows into a trainer batch.""" 

99 out = { 

100 "input_ids": torch.stack([item["input_ids"] for item in batch]), 

101 "labels": torch.stack([item["labels"] for item in batch]), 

102 "attention_mask": torch.stack([item["attention_mask"] for item in batch]), 

103 } 

104 if "num_items_in_batch" in batch[0]: 

105 out["num_items_in_batch"] = sum(int(item["num_items_in_batch"]) for item in batch) 

106 if "mm_token_type_ids" in batch[0]: 

107 out["mm_token_type_ids"] = torch.stack([item["mm_token_type_ids"] for item in batch]) 

108 if "position_ids" in batch[0]: 

109 out["position_ids"] = self._stack_positions(batch, "position_ids") 

110 for key in ("pixel_values", "pixel_values_videos"): 

111 value = self._maybe_cat_optional(batch, key) 

112 if value is not None: 

113 out[key] = value 

114 for key in ("image_grid_thw", "video_grid_thw"): 

115 if key in batch[0] and batch[0].get(key) is not None: 

116 out[key] = self._stack_or_cat_grids(batch, key) 

117 return out 

118 

119 def _build_collate_fn(self): 

120 """Build collate fn (internal).""" 

121 self.base.collate_fn = self._vl_collate 

122 

123 def train(self): 

124 """Run the full training loop by delegating to the underlying BaseTrainer.""" 

125 return self.base.train()