Add GLM-5.2 native SFT and validation outputs for PaddleFleet alignment

swift/arguments/base_args/base_args.py

@@ -346,6 +346,7 @@ def get_model_processor(self,
         res['revision'] = revision or self.model_revision
         res['task_type'] = task_type or self.task_type
         res['num_labels'] = num_labels or self.num_labels
+        res['processor_id_or_path'] = getattr(self, 'tokenizer_name_or_path', None)
 
         return get_model_processor(**res)
 

swift/arguments/base_args/data_args.py

@@ -58,6 +58,8 @@ class DataArguments:
             Example: '{"text1": "query", "text2": "response"}'. Defaults to None.
         strict (bool): If `True`, raises an error on any problematic data row. If `False`, discards the problematic
             sample and continues. Typically used for debugging. Defaults to False.
+        drop_empty_assistant_response (bool): If `True`, filters SFT rows whose assistant response is an empty string.
+            Defaults to False.
         remove_unused_columns (bool): Whether to remove columns not used by the model. If `False`, extra columns are
             passed to the trainer's `compute_loss` function, which is useful for custom loss calculations.
             Defaults to True. Note: The default is `False` for GPRO.
@@ -76,6 +78,7 @@ class DataArguments:
     val_dataset: List[str] = field(default_factory=list)
     cached_dataset: List[str] = field(default_factory=list)
     cached_val_dataset: List[str] = field(default_factory=list)
+    tokenizer_name_or_path: Optional[str] = None
     split_dataset_ratio: float = 0.
 
     data_seed: int = 42
@@ -91,6 +94,7 @@ class DataArguments:
     download_mode: Literal['force_redownload', 'reuse_dataset_if_exists'] = 'reuse_dataset_if_exists'
     columns: Optional[Union[dict, str]] = None
     strict: bool = False
+    drop_empty_assistant_response: bool = False
     remove_unused_columns: bool = True
     disable_auto_column_mapping: bool = False
     # Chinese name and English name
@@ -138,6 +142,7 @@ def get_dataset_kwargs(self):
             'download_mode': self.download_mode,
             'columns': self.columns,
             'strict': self.strict,
+            'drop_empty_assistant_response': self.drop_empty_assistant_response,
             'model_name': self.model_name,
             'model_author': self.model_author,
             'remove_unused_columns': self.remove_unused_columns,

swift/dataset/loader.py

@@ -28,6 +28,7 @@ def __init__(
         streaming: bool = False,
         hub_token: Optional[str] = None,
         strict: bool = False,
+        drop_empty_assistant_response: bool = False,
         download_mode: Literal['force_redownload', 'reuse_dataset_if_exists'] = 'reuse_dataset_if_exists',
         columns: Optional[Dict[str, str]] = None,
         remove_unused_columns: bool = True,
@@ -38,11 +39,24 @@ def __init__(
         self.streaming = streaming
         self.hub_token = hub_token
         self.strict = strict
+        self.drop_empty_assistant_response = drop_empty_assistant_response
         self.download_mode = download_mode
         self.columns = columns
         self.remove_unused_columns = remove_unused_columns
         self.disable_auto_column_mapping = disable_auto_column_mapping
 
+    @staticmethod
+    def _has_nonempty_assistant_response(row: Dict) -> bool:
+        messages = row.get('messages') or []
+        return all(
+            message.get('role') != 'assistant' or not isinstance(message.get('content'), str)
+            or bool(message['content'].strip()) for message in messages)
+
+    def _filter_empty_assistant_responses(self, dataset: HfDataset) -> HfDataset:
+        if not self.drop_empty_assistant_response:
+            return dataset
+        return dataset.filter(self._has_nonempty_assistant_response)
+
     def _load_dataset_path(
         self,
         dataset_path: str,
@@ -66,7 +80,7 @@ def _load_dataset_path(
             enable_auto_mapping=not self.disable_auto_column_mapping)
         if self.remove_unused_columns:
             dataset = RowPreprocessor.remove_useless_columns(dataset)
-        return dataset
+        return self._filter_empty_assistant_responses(dataset)
 
     def _load_repo_dataset(
         self,
@@ -136,7 +150,7 @@ def _load_repo_dataset(
                 enable_auto_mapping=not self.disable_auto_column_mapping)
             if self.remove_unused_columns:
                 dataset = RowPreprocessor.remove_useless_columns(dataset)
-            datasets.append(dataset)
+            datasets.append(self._filter_empty_assistant_responses(dataset))
         return self.concat_datasets(datasets)
 
     @staticmethod
@@ -236,6 +250,7 @@ def load_dataset(
     use_hf: Optional[bool] = None,
     hub_token: Optional[str] = None,
     strict: bool = False,
+    drop_empty_assistant_response: bool = False,
     download_mode: Literal['force_redownload', 'reuse_dataset_if_exists'] = 'reuse_dataset_if_exists',
     columns: Optional[Dict[str, str]] = None,  # columns_mapping
     remove_unused_columns: bool = True,
@@ -278,6 +293,8 @@ def load_dataset(
         hub_token: Authentication token for accessing private datasets on the hub. Default: None.
         strict: If True, raise exceptions when encountering malformed data rows.
             If False, skip invalid rows with warnings. Default: False.
+        drop_empty_assistant_response: Filter rows whose assistant response is an empty string.
+            Defaults to False.
         download_mode: How to handle existing cached datasets:
             - 'reuse_dataset_if_exists': Use cached version if available
             - 'force_redownload': Always download fresh copy
@@ -342,6 +359,7 @@ def load_dataset(
             streaming=streaming,
             hub_token=hub_token,
             strict=strict,
+            drop_empty_assistant_response=drop_empty_assistant_response,
             download_mode=download_mode,
             columns=columns,  # columns_mapping
             remove_unused_columns=remove_unused_columns,

swift/megatron/arguments/megatron_args.py

@@ -549,6 +549,7 @@ class MegatronArguments(RLHFMegatronArgumentsMixin, MegatronTunerMixin):
     start_weight_decay: Optional[float] = None
     end_weight_decay: Optional[float] = None
     clip_grad: float = 1.
+    native_unfused_adamw: bool = False
     adam_beta1: float = 0.9
     adam_beta2: float = 0.95
     adam_eps: float = 1e-8
@@ -608,6 +609,7 @@ class MegatronArguments(RLHFMegatronArgumentsMixin, MegatronTunerMixin):
     overlap_param_gather: bool = False
     overlap_param_gather_with_optimizer_step: bool = False
     align_grad_reduce: bool = True
+    deterministic_mode: bool = False
     # Eagerly create NCCL communicators before the training loop to avoid the lazy
     # first-use allocation hitting the iteration-1 memory peak (Failed to CUDA calloc async).
     nccl_comm_warmup: bool = False

swift/megatron/callbacks/print.py

@@ -1,7 +1,12 @@
 # Copyright (c) ModelScope Contributors. All rights reserved.
+import hashlib
+import json
 import os
+import platform
+import sys
 import time
 import torch
+from pathlib import Path
 from tqdm import tqdm
 
 from swift.megatron.utils import reduce_max_stat_across_model_parallel_group
@@ -11,13 +16,154 @@
 logger = get_logger()
 
 
+def raw_loss_event(step, logs):
+    """Return an unrounded training-loss event, excluding evaluation metrics."""
+    raw_losses = {
+        key: value
+        for key, value in logs.items() if key == 'loss' or (key.startswith('mtp_') and key.endswith('_loss'))
+    }
+    return {'step': step, **raw_losses} if raw_losses else None
+
+
+def _sha256_file(path):
+    return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def normalized_device():
+    """Return the device class the benchmark checker expects, not the GPU model name."""
+    return 'cuda' if torch.cuda.is_available() else 'cpu'
+
+
+def normalized_dtype(value):
+    """Return the bench dtype alias; a bare ``torch.bfloat16`` string is rejected."""
+    text = str(value or '').strip()
+    prefix = 'torch.'
+    if text.startswith(prefix):
+        text = text[len(prefix):]
+    return text.lower()
+
+
+def machine_loss_payload(events, raw_path=None, source_sha256=None):
+    """Return the machine loss artifact.
+
+    ``losses`` is the benchmark gate field: an unrounded main-loss series with one
+    entry per recorded step. ``events`` keeps the per-step diagnostic detail.
+    """
+    return {
+        'schema': 'glm52-machine-loss/v1',
+        'framework': 'torch',
+        'raw': True,
+        'stage': 'training_callback_complete',
+        'losses': [event['loss'] for event in events if 'loss' in event],
+        'event_count': len(events),
+        'steps': [event['step'] for event in events],
+        'events': events,
+        'source': raw_path,
+        'source_sha256': source_sha256,
+    }
+
+
+def _write_json(path, payload):
+    path = Path(path).expanduser().resolve()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, ensure_ascii=False, allow_nan=False, indent=2, sort_keys=True) + '\n')
+
+
+def model_repro_source_modules():
+    names = (
+        'swift.megatron.callbacks.print',
+        'swift.megatron.trainers.base',
+        'megatron.core.transformer.moe.moe_utils',
+        'megatron.core.transformer.moe.experts',
+        'megatron.core.transformer.moe.router',
+        'megatron.core.models.common.language_module.language_module',
+        'mcore_bridge.model.gpt_model',
+    )
+    records = {}
+    for name in names:
+        path = getattr(sys.modules.get(name), '__file__', None)
+        records[name] = {
+            'path': str(Path(path).resolve()) if path else None,
+            'sha256': _sha256_file(path) if path and Path(path).is_file() else None
+        }
+    return records
+
+
+def model_repro_topology(args, config):
+    from megatron.core import parallel_state
+
+    fields = ('tensor_model_parallel_size', 'pipeline_model_parallel_size', 'expert_model_parallel_size',
+              'expert_tensor_parallel_size', 'context_parallel_size', 'sequence_parallel')
+    record = {
+        'configured': {
+            name: getattr(config, name, None)
+            for name in fields
+        },
+        'world_size': torch.distributed.get_world_size(),
+        'rank': torch.distributed.get_rank(),
+        'global_batch_size': args.global_batch_size,
+        'micro_batch_size': args.micro_batch_size,
+        'groups': {}
+    }
+    getters = {
+        'tp': 'get_tensor_model_parallel_group',
+        'pp': 'get_pipeline_model_parallel_group',
+        'ep': 'get_expert_model_parallel_group',
+        'etp': 'get_expert_tensor_parallel_group',
+        'cp': 'get_context_parallel_group',
+        'data': 'get_data_parallel_group'
+    }
+    for name, getter in getters.items():
+        group = getattr(parallel_state, getter)()
+        record['groups'][name] = {
+            'size': torch.distributed.get_world_size(group),
+            'ranks': torch.distributed.get_process_group_ranks(group)
+        }
+    return record
+
+
+def model_repro_environment(args, trainer):
+    """Return the formal run-local environment receipt after model loading."""
+    config_path = os.environ.get('MODEL_REPRO_MODEL_CONFIG_PATH')
+    return {
+        'schema': 'glm52-environment/v1',
+        'framework': 'torch',
+        'framework_version': torch.__version__,
+        'python_version': platform.python_version(),
+        'device': normalized_device(),
+        'device_name': torch.cuda.get_device_name(torch.cuda.current_device()),
+        'dtype': normalized_dtype(getattr(args, 'torch_dtype', 'bfloat16')),
+        'cuda': torch.version.cuda,
+        'cudnn': torch.backends.cudnn.version(),
+        'nccl': list(torch.cuda.nccl.version()),
+        'deterministic': {
+            'algorithms_enabled': torch.are_deterministic_algorithms_enabled(),
+            'cudnn_deterministic': torch.backends.cudnn.deterministic,
+            'cudnn_benchmark': torch.backends.cudnn.benchmark,
+            'cublas_workspace_config': os.environ.get('CUBLAS_WORKSPACE_CONFIG'),
+            'nccl_algo': os.environ.get('NCCL_ALGO'),
+        },
+        'model_id': os.environ.get('MODEL_REPRO_MODEL_ID'),
+        'revision': os.environ.get('MODEL_REPRO_MODEL_REVISION'),
+        'model_config_sha256': _sha256_file(config_path) if config_path else None,
+        'weights_loaded': bool(getattr(trainer, '_model_repro_weights_source', None)),
+        'model_source': str(getattr(trainer, '_model_repro_weights_source', '')),
+        'topology': model_repro_topology(args, trainer.config),
+        'source_modules': model_repro_source_modules(),
+        'invocation_id': os.environ.get('MRK_INVOCATION_ID'),
+        'world_size': torch.distributed.get_world_size() if torch.distributed.is_initialized() else 1,
+    }
+
+
 class PrintCallback(MegatronCallback):
 
     def __init__(self, trainer):
         super().__init__(trainer)
         self.training_bar = None
         self.eval_bar = None
         self.jsonl_writer = None
+        self.raw_loss_writer = None
+        self.raw_loss_events = []
         self.is_write_rank = is_last_rank()
 
     def on_train_begin(self):
@@ -27,13 +173,35 @@ def on_train_begin(self):
         self.training_bar.update(self.state.iteration)
         self.current_step = self.state.iteration
         self.start_time = time.time()
+        self.raw_loss_events = []
+        loss_path = os.environ.get('MODEL_REPRO_LOSS_PATH')
+        if loss_path and self.is_write_rank:
+            Path(loss_path).expanduser().resolve().unlink(missing_ok=True)
         logging_path = os.path.join(self.args.output_dir, 'logging.jsonl')
         logger.info(f'logging_path: {logging_path}')
         self.jsonl_writer = JsonlWriter(logging_path, enable_async=True, write_on_rank='last')
+        raw_loss_path = os.environ.get('MODEL_REPRO_RAW_LOSS_PATH')
+        if raw_loss_path:
+            logger.info(f'raw_loss_path: {raw_loss_path}')
+            if self.is_write_rank and Path(raw_loss_path).exists():
+                Path(raw_loss_path).unlink()
+            self.raw_loss_writer = JsonlWriter(raw_loss_path, write_on_rank='last')
+        env_path = os.environ.get('MODEL_REPRO_ENV_PATH')
+        if env_path and self.is_write_rank:
+            _write_json(env_path, model_repro_environment(self.args, self.trainer))
 
     def on_train_end(self):
         self.training_bar.close()
         self.training_bar = None
+        loss_path = os.environ.get('MODEL_REPRO_LOSS_PATH')
+        if loss_path and self.is_write_rank:
+            raw_path = os.environ.get('MODEL_REPRO_RAW_LOSS_PATH')
+            payload = machine_loss_payload(
+                self.raw_loss_events,
+                raw_path=raw_path,
+                source_sha256=_sha256_file(raw_path) if raw_path and Path(raw_path).is_file() else None,
+            )
+            _write_json(loss_path, payload)
 
     def on_step_end(self):
         n_step = self.state.iteration - self.current_step
@@ -63,6 +231,11 @@ def on_log(self, logs):
         memory = reduce_max_stat_across_model_parallel_group(torch.cuda.max_memory_reserved() / 1024**3)
         logs['memory(GiB)'] = round(memory, 2)
         logs['train_speed(s/it)'] = round(train_speed, 6)
+        raw_event = raw_loss_event(state.iteration, logs)
+        if self.raw_loss_writer is not None and raw_event is not None:
+            self.raw_loss_writer.append(raw_event)
+        if raw_event is not None and self.is_write_rank and os.environ.get('MODEL_REPRO_LOSS_PATH'):
+            self.raw_loss_events.append(raw_event)
         logs = {k: round(v, 8) if isinstance(v, float) else v for k, v in logs.items()}
         self.jsonl_writer.append(logs)
         if self.is_write_rank: