Merge branch 'main' into release/3.9

Author: Jintao-Huang

examples/train/embedding/train_emb.sh

@@ -28,6 +28,5 @@ swift sft \
     --gradient_accumulation_steps 4 \
     --learning_rate 6e-6 \
     --loss_type infonce \
-    --label_names labels \
     --dataloader_drop_last true \
     --deepspeed zero2

swift/llm/dataset/preprocessor/core.py

@@ -314,7 +314,6 @@ def __call__(
             dataset = sample_dataset(dataset, self.dataset_sample, True, self.random_state)
 
         map_kwargs = {'batched': True, 'batch_size': batch_size}
-        cache_file_name = None
         if isinstance(dataset, HfDataset):
             if not load_from_cache_file and is_dist() and not is_master():
                 load_from_cache_file = True
@@ -326,29 +325,28 @@ def __call__(
         dataset = RowPreprocessor.get_features_dataset(dataset)
         if 'solution' in dataset.features:
             with safe_ddp_context(None, True):
-                if not dataset.cache_files:
-                    cache_file_name = os.path.join(get_cache_dir(), 'datasets', 'map_cache',
-                                                   f'{dataset._fingerprint}.arrow')
-                dataset = dataset.map(
-                    lambda x: {'__#solution': x['solution']}, **map_kwargs, cache_file_name=cache_file_name)
+                if isinstance(dataset, HfDataset) and not dataset.cache_files:
+                    map_kwargs['cache_file_name'] = os.path.join(get_cache_dir(), 'datasets', 'map_cache',
+                                                                 f'{dataset._fingerprint}.arrow')
+                dataset = dataset.map(lambda x: {'__#solution': x['solution']}, **map_kwargs)
+                map_kwargs.pop('cache_file_name', None)
         dataset = self._rename_columns(dataset)
         dataset = self.prepare_dataset(dataset)
         dataset = self._cast_pil_image(dataset)
 
         ignore_max_length_error = True if isinstance(dataset, HfDataset) and num_proc > 1 else False
         with self._patch_arrow_writer(), safe_ddp_context(None, True):
             try:
-                if not dataset.cache_files:
-                    cache_file_name = os.path.join(get_cache_dir(), 'datasets', 'map_cache',
-                                                   f'{dataset._fingerprint}.arrow')
+                if isinstance(dataset, HfDataset) and not dataset.cache_files:
+                    map_kwargs['cache_file_name'] = os.path.join(get_cache_dir(), 'datasets', 'map_cache',
+                                                                 f'{dataset._fingerprint}.arrow')
                 dataset_mapped = dataset.map(
                     self.batched_preprocess,
                     fn_kwargs={
                         'strict': strict,
                         'ignore_max_length_error': ignore_max_length_error
                     },
                     remove_columns=list(dataset.features.keys()),
-                    cache_file_name=cache_file_name,
                     **map_kwargs)
             except NotImplementedError:
                 pass

swift/llm/template/base.py

@@ -553,7 +553,7 @@ def packing_row(self, row: List[Dict[str, Any]]) -> Dict[str, Any]:
             length.append(r['length'])
         for key in keys:
             if key in {'input_ids', 'labels', 'loss_scale'}:
-                packed[key] = sum((x[key] for x in row), start=[])
+                packed[key] = sum((x.get(key) or [] for x in row), start=[])
             elif key == 'length':
                 packed[key] = sum((x[key] for x in row))
             elif key == 'channel':

swift/megatron/model/gpt_model.py

@@ -250,9 +250,9 @@ def forward(
             logits, _ = self.output_layer(
                 hidden_states, weight=output_weight, runtime_gather_output=runtime_gather_output)
         else:
-            logits = self.output_layer(hidden_states)[0]
             if args.sequence_parallel and args.tensor_model_parallel_size > 1:
-                logits = gather_from_sequence_parallel_region(logits)
+                hidden_states = gather_from_sequence_parallel_region(hidden_states)
+            logits = self.output_layer(hidden_states)[0]
         if has_config_logger_enabled(self.config):
             payload = OrderedDict({
                 'input_ids': input_ids,