fix split in trainer

Author: Difers

legacy/examples/RLHF/trainer_utils.py

@@ -340,11 +340,8 @@ def full_training_step(self: Trainer, inputs: Dict[str, paddle.Tensor], **kwargs
             fused_allreduce_gradients(list(model.parameters()), None)
 
         # Pipeline parallel mode,  handle gradient reduce here to overlap
-        pipeline_parallel_config = (
-            set(args.pipeline_parallel_config.split(" ")) if args.pipeline_parallel_degree > 1 else set()
-        )
-        enable_dp_comm_overlap = "enable_dp_comm_overlap" in pipeline_parallel_config
-        enable_release_grads = "enable_release_grads" in pipeline_parallel_config
+        enable_dp_comm_overlap = "enable_dp_comm_overlap" in args.pipeline_parallel_config
+        enable_release_grads = "enable_release_grads" in args.pipeline_parallel_config
 
         # Case 3: Pipeline parallel mode, overlap with dp
         if isinstance(self.optimizer, HybridParallelOptimizer) and not self.do_grad_scaling:

llm/alignment/ppo/trainer_utils.py

@@ -340,11 +340,8 @@ def full_training_step(self: Trainer, inputs: Dict[str, paddle.Tensor], **kwargs
             fused_allreduce_gradients(list(model.parameters()), None)
 
         # Pipeline parallel mode,  handle gradient reduce here to overlap
-        pipeline_parallel_config = (
-            set(args.pipeline_parallel_config.split(" ")) if args.pipeline_parallel_degree > 1 else set()
-        )
-        enable_dp_comm_overlap = "enable_dp_comm_overlap" in pipeline_parallel_config
-        enable_release_grads = "enable_release_grads" in pipeline_parallel_config
+        enable_dp_comm_overlap = "enable_dp_comm_overlap" in args.pipeline_parallel_config
+        enable_release_grads = "enable_release_grads" in args.pipeline_parallel_config
 
         # Case 3: Pipeline parallel mode, overlap with dp
         if isinstance(self.optimizer, HybridParallelOptimizer) and not self.do_grad_scaling:

paddlenlp/trainer/trainer.py

@@ -1083,16 +1083,10 @@ def fused_allreduce_gradients_no_sync(paramlist, hcg):
                         fused_allreduce_gradients_no_sync(list(model.parameters()), None)
 
                     # Pipeline parallel mode,  handle gradient reduce here to overlap
-                    pipeline_parallel_config = (
-                        set(args.pipeline_parallel_config.split(" ")) if args.pipeline_parallel_degree > 1 else set()
-                    )
-                    sharding_parallel_config = (
-                        set(args.sharding_parallel_config.split(" ")) if args.sharding_parallel_degree > 1 else set()
-                    )
-                    enable_dp_comm_overlap = "enable_dp_comm_overlap" in pipeline_parallel_config
+                    enable_dp_comm_overlap = "enable_dp_comm_overlap" in args.pipeline_parallel_config
                     enable_release_grads = (
-                        "enable_release_grads" in pipeline_parallel_config
-                        or "enable_release_grads" in sharding_parallel_config
+                        "enable_release_grads" in args.pipeline_parallel_config
+                        or "enable_release_grads" in args.sharding_parallel_config
                     )
 
                     # Case 3: Pipeline parallel mode, overlap with dp
@@ -1992,8 +1986,7 @@ def get_expected_keys(inputs, keys):
                         "please upgrade your paddle (using nightly version)."
                     )
 
-                sharding_parallel_config = set(self.args.sharding_parallel_config.split(" "))
-                if level == "os_g" and "enable_stage2_overlap" in sharding_parallel_config:
+                if level == "os_g" and "enable_stage2_overlap" in self.args.sharding_parallel_config:
                     model._set_reduce_overlap(True)
                     optimizer._set_broadcast_overlap(True, model)
 
@@ -2133,9 +2126,9 @@ def compute_loss(self, model, inputs, return_outputs=False):
     def _enable_delay_scale_loss(self):
         key = "enable_delay_scale_loss"
         if self.args.pipeline_parallel_degree > 1:
-            return key in self.args.pipeline_parallel_config.split(" ")
+            return key in self.args.pipeline_parallel_config
         elif self.args.tensor_parallel_degree > 1:
-            return key in self.args.tensor_parallel_config.split(" ")
+            return key in self.args.tensor_parallel_config
         else:
             return False
 

paddlenlp/trainer/training_args.py

@@ -1039,10 +1039,10 @@ def __post_init__(self):
                 strategy = fleet.DistributedStrategy()
                 assert self.data_parallel_config == "", "data_parallle_config is not supported in hybrid parallel"
                 if self.pipeline_parallel_degree > 1:
-                    if " " in self.pipeline_parallel_config:
-                        pipeline_parallel_config = set(self.pipeline_parallel_config.split(" "))
-                    else:
+                    if "," in self.pipeline_parallel_config:
                         pipeline_parallel_config = set(self.pipeline_parallel_config.split(","))
+                    else:
+                        pipeline_parallel_config = set(self.pipeline_parallel_config.split(" "))
                     for x in pipeline_parallel_config:
                         if len(x) > 0:
                             if x not in [
@@ -1116,10 +1116,10 @@ def __post_init__(self):
                 if self.tensor_parallel_degree > 1:
                     strategy.tensor_parallel_configs = {"tensor_init_seed": self.seed}
 
-                    if " " in self.tensor_parallel_config:
-                        mp_config = set(self.tensor_parallel_config.split(" "))
-                    else:
+                    if "," in self.tensor_parallel_config:
                         mp_config = set(self.tensor_parallel_config.split(","))
+                    else:
+                        mp_config = set(self.tensor_parallel_config.split(" "))
 
                     for x in mp_config:
                         if len(x) > 0:
@@ -1225,10 +1225,10 @@ def is_segment_parallel_supported():
                 strategy.hybrid_configs = hybrid_configs
 
                 if self.sharding_parallel_degree > 1:
-                    if " " in self.sharding_parallel_config:
-                        sharding_parallel_config = set(self.sharding_parallel_config.split(" "))
-                    else:
+                    if "," in self.sharding_parallel_config:
                         sharding_parallel_config = set(self.sharding_parallel_config.split(","))
+                    else:
+                        sharding_parallel_config = set(self.sharding_parallel_config.split(" "))
                     for x in sharding_parallel_config:
                         if len(x) > 0:
                             if x not in [
@@ -1384,10 +1384,10 @@ def is_segment_parallel_supported():
 
             # navie-pp: pipeline_parallel_degree > 1 and gradient_accumulation_steps == 1
             if self.pipeline_parallel_degree > 1 and self.gradient_accumulation_steps > 1:
-                if " " in self.pipeline_parallel_config:
-                    pipeline_parallel_config = set(self.pipeline_parallel_config.split(" "))
-                else:
+                if "," in self.pipeline_parallel_config:
                     pipeline_parallel_config = set(self.pipeline_parallel_config.split(","))
+                else:
+                    pipeline_parallel_config = set(self.pipeline_parallel_config.split(" "))
                 for x in pipeline_parallel_config:
                     if len(x) > 0:
                         if x not in [
@@ -1437,10 +1437,10 @@ def is_segment_parallel_supported():
             if self.tensor_parallel_degree > 1:
                 mp_optimization = strategy.mp_optimization
 
-                if " " in self.tensor_parallel_config:
-                    mp_config = set(self.tensor_parallel_config.split(" "))
-                else:
+                if "," in self.tensor_parallel_config:
                     mp_config = set(self.tensor_parallel_config.split(","))
+                else:
+                    mp_config = set(self.tensor_parallel_config.split(" "))
 
                 for x in mp_config:
                     if len(x) > 0:
@@ -1473,10 +1473,10 @@ def is_segment_parallel_supported():
                 elif ShardingOption.FULL_SHARD in self.sharding:
                     sharding.stage = 3
 
-                if " " in self.sharding_parallel_config:
-                    sharding_parallel_config = set(self.sharding_parallel_config.split(" "))
-                else:
+                if "," in self.sharding_parallel_config:
                     sharding_parallel_config = set(self.sharding_parallel_config.split(","))
+                else:
+                    sharding_parallel_config = set(self.sharding_parallel_config.split(" "))
                 for x in sharding_parallel_config:
                     if len(x) > 0:
                         if x not in [