No public description

PiperOrigin-RevId: 760021381

Author: haozha111

ai_edge_torch/generative/examples/amd_llama_135m/verify.py

@@ -15,15 +15,9 @@
 
 """Verifies the reauthored AMD-Llama-135M model."""
 
-import logging
-import pathlib
-
 from absl import app
 from absl import flags
-from ai_edge_torch.generative.examples.amd_llama_135m import amd_llama_135m
-from ai_edge_torch.generative.utilities import transformers_verifier
-from ai_edge_torch.generative.utilities import verifier
-import transformers
+from ai_edge_torch.generative.examples.amd_llama_135m import verify_util
 
 
 _PROMPTS = flags.DEFINE_multi_string(
@@ -39,32 +33,10 @@
 
 
 def main(_):
-  checkpoint = "amd/AMD-Llama-135m"
-  logging.info("Loading the original model from: %s", checkpoint)
-  original_model = transformers.AutoModelForCausalLM.from_pretrained(
-      checkpoint, trust_remote_code=True
-  )
-
-  # Locate the cached dir.
-  cached_config_file = transformers.utils.cached_file(
-      checkpoint, transformers.utils.CONFIG_NAME
-  )
-  reauthored_checkpoint = pathlib.Path(cached_config_file).parent
-  logging.info("Building the reauthored model from: %s", reauthored_checkpoint)
-  reauthored_model = amd_llama_135m.build_model(str(reauthored_checkpoint))
-
-  logging.info("Loading the tokenizer from: %s", checkpoint)
-  tokenizer = transformers.AutoTokenizer.from_pretrained(checkpoint)
-
-  verifier.verify_reauthored_model(
-      original_model=transformers_verifier.TransformersModelWrapper(
-          original_model
-      ),
-      reauthored_model=verifier.ReauthoredModelWrapper(reauthored_model),
-      tokenizer=verifier.TokenizerWrapper(tokenizer),
-      generate_prompts=_PROMPTS.value,
+  verify_util.verify_amd_llama_135m(
+      "amd/AMD-Llama-135m",
       max_new_tokens=_MAX_NEW_TOKENS.value,
-      atol=1e-04,
+      prompts=_PROMPTS.value,
   )
 
 

ai_edge_torch/generative/examples/amd_llama_135m/verify_util.py

@@ -0,0 +1,76 @@
+# Copyright 2025 The AI Edge Torch Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+"""Utils for verifying the AMD-Llama-135M model."""
+import logging
+import os
+import pathlib
+
+from ai_edge_torch.generative.examples.amd_llama_135m import amd_llama_135m
+from ai_edge_torch.generative.utilities import loader
+from ai_edge_torch.generative.utilities import transformers_verifier
+from ai_edge_torch.generative.utilities import verifier
+import transformers
+
+
+DEFAULT_PROMPTS = ["Tell me a story?\nOnce upon a time"]
+
+
+def verify_amd_llama_135m(
+    checkpoint_dir: str,
+    weight_filename: str = "model.safetensors",
+    max_new_tokens: int = 30,
+    initialize_from_local: bool = True,
+    prompts: list[str] | None = None,
+) -> bool:
+  """Verifies the reauthored AMD-Llama-135M model with a custom loader."""
+  logging.info("Loading the original model from: %s", checkpoint_dir)
+  original_model = transformers.AutoModelForCausalLM.from_pretrained(
+      checkpoint_dir
+  )
+
+  logging.info("Building the reauthored model from: %s", checkpoint_dir)
+  custom_loader = (
+      None
+      if initialize_from_local
+      else loader.get_custom_loader("", "safetensors")
+  )
+
+  if initialize_from_local:
+    # Locate the cached dir.
+    cached_config_file = transformers.utils.cached_file(
+        checkpoint_dir, transformers.utils.CONFIG_NAME
+    )
+    reauthored_checkpoint = pathlib.Path(cached_config_file).parent
+  else:
+    reauthored_checkpoint = os.path.join(checkpoint_dir, weight_filename)
+
+  logging.info("Building the reauthored model from: %s", reauthored_checkpoint)
+  reauthored_model = amd_llama_135m.build_model(
+      checkpoint_path=reauthored_checkpoint,
+      custom_loader=custom_loader,
+  )
+
+  logging.info("Loading the tokenizer from: %s", checkpoint_dir)
+  tokenizer = transformers.AutoTokenizer.from_pretrained(checkpoint_dir)
+  return verifier.verify_reauthored_model(
+      original_model=transformers_verifier.TransformersModelWrapper(
+          original_model
+      ),
+      reauthored_model=verifier.ReauthoredModelWrapper(reauthored_model),
+      tokenizer=verifier.TokenizerWrapper(tokenizer),
+      generate_prompts=DEFAULT_PROMPTS if prompts is None else prompts,
+      max_new_tokens=max_new_tokens,
+      atol=1e-04,
+  )

ai_edge_torch/generative/examples/deepseek/verify.py

@@ -15,15 +15,9 @@
 
 """Verifies the reauthored DeepSeek R1 distilled 1.5B model."""
 
-import logging
-import pathlib
-
 from absl import app
 from absl import flags
-from ai_edge_torch.generative.examples.deepseek import deepseek
-from ai_edge_torch.generative.utilities import transformers_verifier
-from ai_edge_torch.generative.utilities import verifier
-import transformers
+from ai_edge_torch.generative.examples.deepseek import verify_util
 
 
 _PROMPTS = flags.DEFINE_multi_string(
@@ -39,30 +33,10 @@
 
 
 def main(_):
-  checkpoint = "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B"
-  logging.info("Loading the original model from: %s", checkpoint)
-  original_model = transformers.AutoModelForCausalLM.from_pretrained(checkpoint)
-
-  # Locate the cached dir.
-  cached_config_file = transformers.utils.cached_file(
-      checkpoint, transformers.utils.CONFIG_NAME
-  )
-  reauthored_checkpoint = pathlib.Path(cached_config_file).parent
-  logging.info("Building the reauthored model from: %s", reauthored_checkpoint)
-  reauthored_model = deepseek.build_model(str(reauthored_checkpoint))
-
-  logging.info("Loading the tokenizer from: %s", checkpoint)
-  tokenizer = transformers.AutoTokenizer.from_pretrained(checkpoint)
-
-  verifier.verify_reauthored_model(
-      original_model=transformers_verifier.TransformersModelWrapper(
-          original_model
-      ),
-      reauthored_model=verifier.ReauthoredModelWrapper(reauthored_model),
-      tokenizer=verifier.TokenizerWrapper(tokenizer),
-      generate_prompts=_PROMPTS.value,
+  verify_util.verify_deepseek_r1_distill_1_5b(
+      "deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B",
       max_new_tokens=_MAX_NEW_TOKENS.value,
-      atol=1e-04,
+      prompts=_PROMPTS.value,
   )
 
 

ai_edge_torch/generative/examples/deepseek/verify_util.py

@@ -0,0 +1,76 @@
+# Copyright 2025 The AI Edge Torch Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+"""Utils for verifying the DeepSeek R1 distilled 1.5B model."""
+import logging
+import os
+import pathlib
+
+from ai_edge_torch.generative.examples.deepseek import deepseek
+from ai_edge_torch.generative.utilities import loader
+from ai_edge_torch.generative.utilities import transformers_verifier
+from ai_edge_torch.generative.utilities import verifier
+import transformers
+
+
+DEFAULT_PROMPTS = ["What is the meaning of life?"]
+
+
+def verify_deepseek_r1_distill_1_5b(
+    checkpoint_dir: str,
+    weight_filename: str = "model.safetensors",
+    max_new_tokens: int = 30,
+    initialize_from_local: bool = True,
+    prompts: list[str] | None = None,
+) -> bool:
+  """Verifies the reauthored DeepSeek R1 distilled 1.5B model with a custom loader."""
+  logging.info("Loading the original model from: %s", checkpoint_dir)
+  original_model = transformers.AutoModelForCausalLM.from_pretrained(
+      checkpoint_dir
+  )
+
+  logging.info("Building the reauthored model from: %s", checkpoint_dir)
+  custom_loader = (
+      None
+      if initialize_from_local
+      else loader.get_custom_loader("", "safetensors")
+  )
+
+  if initialize_from_local:
+    # Locate the cached dir.
+    cached_config_file = transformers.utils.cached_file(
+        checkpoint_dir, transformers.utils.CONFIG_NAME
+    )
+    reauthored_checkpoint = pathlib.Path(cached_config_file).parent
+  else:
+    reauthored_checkpoint = os.path.join(checkpoint_dir, weight_filename)
+
+  logging.info("Building the reauthored model from: %s", reauthored_checkpoint)
+  reauthored_model = deepseek.build_model(
+      checkpoint_path=reauthored_checkpoint,
+      custom_loader=custom_loader,
+  )
+
+  logging.info("Loading the tokenizer from: %s", checkpoint_dir)
+  tokenizer = transformers.AutoTokenizer.from_pretrained(checkpoint_dir)
+  return verifier.verify_reauthored_model(
+      original_model=transformers_verifier.TransformersModelWrapper(
+          original_model
+      ),
+      reauthored_model=verifier.ReauthoredModelWrapper(reauthored_model),
+      tokenizer=verifier.TokenizerWrapper(tokenizer),
+      generate_prompts=DEFAULT_PROMPTS if prompts is None else prompts,
+      max_new_tokens=max_new_tokens,
+      atol=1e-04,
+  )

ai_edge_torch/generative/examples/gemma/verify_util.py

@@ -17,11 +17,13 @@
 
 import logging
 import os
-from typing import List, Tuple
+from typing import Callable, Dict, List, Tuple
 
+from ai_edge_torch.generative.examples.gemma import gemma1
 from ai_edge_torch.generative.examples.gemma import gemma2
 import ai_edge_torch.generative.layers.attention_utils as attn_utils
 import ai_edge_torch.generative.layers.kv_cache as kv_utils
+from ai_edge_torch.generative.utilities import loader
 from ai_edge_torch.generative.utilities import verifier
 from gemma import config as gemma_config
 from gemma import model as gemma_model
@@ -107,6 +109,7 @@ def verify_reauthored_gemma_model(
     generate_prompts: List[str],
     forward_input_ids: List[List[int]],
     weight_filename: str = "model.ckpt",
+    custom_loader: Callable[[str], Dict[str, torch.Tensor]] | None = None,
     tokenizer_filename: str = "tokenizer.model",
     max_new_tokens: int = 20,
     mask_as_input: bool = False,
@@ -125,7 +128,14 @@ def verify_reauthored_gemma_model(
 
   logging.info("Loading the original model from: %s", checkpoint)
   original_model = gemma_model.GemmaForCausalLM(config).eval()
-  original_model.load_weights(os.path.join(checkpoint, weight_filename))
+  checkpoint_path = os.path.join(checkpoint, weight_filename)
+  if custom_loader is None:
+    original_model.load_weights(checkpoint_path)
+  else:
+    original_model.load_state_dict(
+        custom_loader(checkpoint_path)["model_state_dict"],
+        strict=False,
+    )
 
   return verifier.verify_reauthored_model(
       original_model=GemmaWrapper(original_model),
@@ -144,27 +154,62 @@ def verify_reauthored_gemma_model(
 
 
 def verify_gemma2(
-    gemma2_model_path: str,
+    checkpoint_dir: str,
+    weight_filename: str,
     prompts: List[str],
     max_new_tokens: int,
     mask_as_input: bool = False,
     kv_layout: kv_utils.KVLayout = kv_utils.KV_LAYOUT_DEFAULT,
+    custom_loader: Callable[[str], Dict[str, torch.Tensor]] | None = None,
 ) -> bool:
   """Verifies the reauthored Gemma2 model.
 
   Return True if the verification passes, False otherwise.
   """
-  logging.info("Building the reauthored model from: %s", gemma2_model_path)
-  reauthored_model = gemma2.build_2b_model(gemma2_model_path)
+  checkpoint_path = os.path.join(checkpoint_dir, weight_filename)
+  logging.info("Building the reauthored model from: %s", checkpoint_path)
+  reauthored_model = gemma2.build_2b_model(checkpoint_path, custom_loader)
 
   return verify_reauthored_gemma_model(
-      checkpoint=gemma2_model_path,
+      checkpoint=checkpoint_dir,
       variant="2b-v2",
       reauthored_model=reauthored_model,
       generate_prompts=prompts,
       forward_input_ids=[[2, 651, 9456, 576, 573, 3520, 3858, 603, 235248]],
+      weight_filename=weight_filename,
+      custom_loader=custom_loader,
       max_new_tokens=max_new_tokens,
       mask_as_input=mask_as_input,
       kv_layout=kv_layout,
       atol=1e-04,
   )
+
+
+def verify_gemma1_with_custom_loader(checkpoint_dir: str) -> bool:
+  """Verifies the reauthored Gemma1 model with a custom loader."""
+  weight_filename = "gemma-2b-it.ckpt"
+  checkpoint_path = os.path.join(checkpoint_dir, weight_filename)
+  custom_loader = loader.get_custom_loader(checkpoint_path)
+  reauthored_model = gemma1.build_2b_model(checkpoint_path, custom_loader)
+  return verify_reauthored_gemma_model(
+      checkpoint=checkpoint_dir,
+      variant="2b",
+      reauthored_model=reauthored_model,
+      weight_filename=weight_filename,
+      custom_loader=custom_loader,
+      generate_prompts=["What is the meaning of life?"],
+      forward_input_ids=[[1, 2, 3, 4]],
+      max_new_tokens=30,
+  )
+
+
+def verify_gemma2_with_custom_loader(checkpoint_dir: str) -> bool:
+  """Verifies the reauthored Gemma2 model with a custom loader."""
+  return verify_gemma2(
+      checkpoint_dir=checkpoint_dir,
+      weight_filename="model.ckpt",
+      prompts=["What is the meaning of life?"],
+      max_new_tokens=30,
+      mask_as_input=True,
+      custom_loader=loader.get_custom_loader("", checkpoint_format="pt"),
+  )

ai_edge_torch/generative/examples/gemma3/verify_util.py

@@ -22,6 +22,7 @@
 from ai_edge_torch.generative.examples.gemma3 import gemma3
 from ai_edge_torch.generative.layers import kv_cache as kv_utils
 import ai_edge_torch.generative.layers.attention_utils as attn_utils
+from ai_edge_torch.generative.utilities import loader
 from ai_edge_torch.generative.utilities import verifier
 from gemma import config as gemma_config
 from gemma import model as gemma_model
@@ -260,3 +261,15 @@ def verify_gemma3(
       custom_loader=custom_loader,
       atol=1e-04,
   )
+
+
+def verify_gemma3_with_custom_loader(checkpoint: str) -> bool:
+  """Verifies the reauthored Gemma3 model with a custom loader."""
+  return verify_gemma3(
+      checkpoint=checkpoint,
+      prompts=["What is the meaning of life?"],
+      max_new_tokens=30,
+      variant="1b",
+      weight_filename="model.ckpt",
+      custom_loader=loader.get_custom_loader("", checkpoint_format="pt"),
+  )