refactor(data): simplify packing fingerprint

Signed-off-by: yaoyu-33 <yaoyu.094@gmail.com>

Author: yaoyu-33

src/megatron/bridge/data/builders/gpt_sft.py

@@ -80,7 +80,7 @@ def _default_gpt_sft_preprocessing() -> PromptCompletionSFTPreprocessingConfig:
     )
 
 
-def _huggingface_snapshot_identity(tokenizer_path: str) -> tuple[str, str] | None:
+def _hf_snapshot_identity(tokenizer_path: str) -> tuple[str, str] | None:
     """Extract the repository cache name and revision from a Hugging Face snapshot path."""
     match = re.search(
         r"(?:^|[/\\])models--(?P<repository>[^/\\]+)[/\\]snapshots[/\\](?P<revision>[^/\\]+)",
@@ -91,67 +91,42 @@ def _huggingface_snapshot_identity(tokenizer_path: str) -> tuple[str, str] | Non
     return match.group("repository"), match.group("revision")
 
 
-def _tokenizer_packing_identity(tokenizer: MegatronTokenizer) -> dict[str, Any]:
-    """Return stable tokenizer properties that can change packed token IDs."""
-    identifiers = getattr(tokenizer, "unique_identifiers", None)
-    if isinstance(identifiers, dict):
-        normalized_identifiers = dict(identifiers)
-        tokenizer_path = normalized_identifiers.get("tokenizer_path")
-        if isinstance(tokenizer_path, str):
-            snapshot_identity = _huggingface_snapshot_identity(tokenizer_path)
-            if snapshot_identity is not None:
-                repository, revision = snapshot_identity
-                normalized_identifiers["tokenizer_path"] = f"hf-cache://{repository}@{revision}"
-        template_identifier = normalized_identifiers.pop("chat_template", None)
-        if isinstance(template_identifier, str):
-            normalized_identifiers["chat_template_sha256"] = hashlib.sha256(
-                template_identifier.encode("utf-8")
-            ).hexdigest()
-        identity: dict[str, Any] = {"unique_identifiers": normalized_identifiers}
-    else:
-        identity = {"class": f"{type(tokenizer).__module__}.{type(tokenizer).__qualname__}"}
-
-    chat_template = getattr(tokenizer, "chat_template", None)
-    if not isinstance(chat_template, str):
-        tokenizer_instance = getattr(tokenizer, "_tokenizer", None)
-        chat_template = getattr(tokenizer_instance, "chat_template", None)
-    if isinstance(chat_template, str):
-        identity["chat_template_sha256"] = hashlib.sha256(chat_template.encode("utf-8")).hexdigest()
-
-    for attribute in ("bos_id", "eos_id", "pad_id"):
-        try:
-            token_id = getattr(tokenizer, attribute, None)
-        except (AttributeError, NotImplementedError):
-            continue
-        if isinstance(token_id, int) and not isinstance(token_id, bool):
-            identity[attribute] = token_id
-
-    return identity
-
-
 def _packing_fingerprint(
     config: "GPTSFTDatasetConfig",
     dataset_kwargs: dict[str, Any] | None,
     tokenizer: MegatronTokenizer,
 ) -> str:
     """Fingerprint every setting that can change builder-managed packed rows."""
     preprocessing = resolve_gpt_sft_preprocessing(config)
+    tokenizer_identity = getattr(tokenizer, "unique_identifiers", None)
+    if not isinstance(tokenizer_identity, dict):
+        tokenizer_identity = {"class": f"{type(tokenizer).__module__}.{type(tokenizer).__qualname__}"}
+    else:
+        tokenizer_identity = dict(tokenizer_identity)
+        tokenizer_path = tokenizer_identity.get("tokenizer_path")
+        if isinstance(tokenizer_path, str) and (snapshot := _hf_snapshot_identity(tokenizer_path)) is not None:
+            tokenizer_identity["tokenizer_path"] = f"hf-cache://{snapshot[0]}@{snapshot[1]}"
+
+    chat_template = getattr(tokenizer, "chat_template", None)
+    if not isinstance(chat_template, str):
+        chat_template = getattr(getattr(tokenizer, "_tokenizer", None), "chat_template", None)
+    if not isinstance(chat_template, str):
+        chat_template = None
+
     packing_identity = {
         "seq_length": config.seq_length,
         "seed": config.seed,
-        "preprocessing_explicit": config.preprocessing is not None,
-        "preprocessing_type": type(preprocessing).__name__,
-        "preprocessing": asdict(preprocessing),
+        "preprocessing": (type(preprocessing).__name__, asdict(preprocessing)),
         "dataset_kwargs": dataset_kwargs,
-        "tokenizer": _tokenizer_packing_identity(tokenizer),
+        "tokenizer": tokenizer_identity,
+        "chat_template": chat_template,
+        "chat_eos_policy": "append_if_missing" if isinstance(preprocessing, ChatSFTPreprocessingConfig) else None,
+        "max_single_sequence_length": (
+            config.offline_packing_specs.max_single_sequence_length
+            if config.offline_packing_specs is not None
+            else None
+        ),
     }
-    if isinstance(preprocessing, ChatSFTPreprocessingConfig):
-        packing_identity["chat_eos_policy"] = "append_if_missing_v1"
-    if (
-        config.offline_packing_specs is not None
-        and config.offline_packing_specs.max_single_sequence_length is not None
-    ):
-        packing_identity["max_single_sequence_length"] = config.offline_packing_specs.max_single_sequence_length
     return hashlib.sha256(
         json.dumps(packing_identity, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
     ).hexdigest()[:12]
@@ -960,7 +935,7 @@ def _extract_tokenizer_model_name(self) -> str:
             return self.offline_packing_specs.tokenizer_model_name
         elif isinstance(tokenizer_instance, tokenizer_cls):
             name = str(self.tokenizer.path)
-            snapshot_identity = _huggingface_snapshot_identity(name)
+            snapshot_identity = _hf_snapshot_identity(name)
             if snapshot_identity is not None:
                 tokenizer_model_name, _ = snapshot_identity
                 return tokenizer_model_name