Module neuralops.core.model_wrapper
UnifiedTransformerModel – wraps any HuggingFace model with Hub support, attention replacement, and config serialization.
Usage -----= model = UnifiedTransformerModel.from_pretrained("distilbert-base-uncased") model.replace_attention("galerkin", rank=32) model.save_pretrained("./compressed_model")
Later:
loaded = UnifiedTransformerModel.from_pretrained("./compressed_model")
Classes
class UnifiedTransformerConfig (attention_type: str = None,
attention_rank: int | None = None,
attention_kwargs: dict[str, typing.Any] | None = None,
**kwargs)-
Expand source code
class UnifiedTransformerConfig: """Lightweight container for the custom attention configuration.""" def __init__( self, attention_type: str = None, attention_rank: int | None = None, attention_kwargs: dict[str, Any] | None = None, **kwargs, ) -> None: self.attention_type = attention_type self.attention_rank = attention_rank self.attention_kwargs = attention_kwargs or {} # Store any extra kwargs for k, v in kwargs.items(): setattr(self, k, v) def to_dict(self) -> dict[str, Any]: """Serialise to a JSON‑compatible dict.""" return { "attention_type": self.attention_type, "attention_rank": self.attention_rank, "attention_kwargs": self.attention_kwargs, } @classmethod def from_dict(cls, d: dict[str, Any]) -> "UnifiedTransformerConfig": """From dict.""" return cls(**d)Lightweight container for the custom attention configuration.
Static methods
def from_dict(d: dict[str, typing.Any]) ‑> UnifiedTransformerConfig-
From dict.
Methods
def to_dict(self) ‑> dict[str, typing.Any]-
Expand source code
def to_dict(self) -> dict[str, Any]: """Serialise to a JSON‑compatible dict.""" return { "attention_type": self.attention_type, "attention_rank": self.attention_rank, "attention_kwargs": self.attention_kwargs, }Serialise to a JSON‑compatible dict.
class UnifiedTransformerModel (model: torch.nn.modules.module.Module,
custom_config: UnifiedTransformerConfig | None = None)-
Expand source code
class UnifiedTransformerModel(nn.Module, PyTorchModelHubMixin): """ A wrapper around a HuggingFace model that supports custom attention and can be saved/loaded via the HuggingFace Hub. """ def __init__( self, model: nn.Module, custom_config: UnifiedTransformerConfig | None = None ) -> None: super().__init__() self.model = model self.config = getattr(model, "config", None) # HF model config self.arch = detect_architecture(model) self._custom_config = custom_config or UnifiedTransformerConfig() # ------------------------------------------------------------------ # Attention replacement # ------------------------------------------------------------------ def replace_attention( self, attention_type: str, rank: int | None = None, copy_weights: bool = True, **kwargs, ) -> Any: """ Inject a custom attention variant into all layers. This method also records the choice so it can be saved. """ inject_attention(self.model, attention_type, rank=rank, copy_weights=copy_weights, **kwargs) self._custom_config = UnifiedTransformerConfig( attention_type=attention_type, attention_rank=rank, attention_kwargs=kwargs, ) return self # ------------------------------------------------------------------ # Freezing / parameter count # ------------------------------------------------------------------ def freeze_layers(self, num_layers_to_freeze: int = 0) -> Any: """Freeze layers.""" layers = get_transformer_layers(self.model) for i, layer in enumerate(layers): for param in layer.parameters(): param.requires_grad = i >= num_layers_to_freeze return self def unfreeze_all(self) -> Any: """Unfreeze all.""" for param in self.model.parameters(): param.requires_grad = True return self def get_parameter_count(self) -> dict[str, int]: """Return the parameter count.""" total = sum(p.numel() for p in self.model.parameters()) trainable = sum(p.numel() for p in self.model.parameters() if p.requires_grad) return {"total": total, "trainable": trainable, "frozen": total - trainable} # ------------------------------------------------------------------ # Hub serialization – these methods are called by PyTorchModelHubMixin # ------------------------------------------------------------------ def _save_pretrained(self, save_directory: str) -> None: """ Save the underlying model and our custom config to a directory. """ # 1. Save the base HF model (attention layers are already inside) self.model.save_pretrained(save_directory) # 2. Save our custom attention config to a JSON file import json custom_cfg_path = f"{save_directory}/unified_attention_config.json" with open(custom_cfg_path, "w") as f: json.dump(self._custom_config.to_dict(), f, indent=2) @classmethod def _from_pretrained(cls, model_id: str, **kwargs) -> "UnifiedTransformerModel": """ Load a pretrained model and reconstruct the custom attention variant. This method is invoked by `from_pretrained` class method of the mixin. """ # 1. Load the custom config import json import os if os.path.isdir(model_id): cfg_path = os.path.join(model_id, "unified_attention_config.json") else: # If model_id is a Hub repo, we need to download the file from huggingface_hub import hf_hub_download cfg_path = hf_hub_download(repo_id=model_id, filename="unified_attention_config.json") with open(cfg_path) as f: custom_cfg_dict = json.load(f) custom_cfg = UnifiedTransformerConfig.from_dict(custom_cfg_dict) # 2. Load the base HF model base_model = AutoModel.from_pretrained(model_id, **kwargs) # 3. Re‑apply the attention replacement if a type was stored model = cls(base_model, custom_config=custom_cfg) if custom_cfg.attention_type: model.replace_attention( custom_cfg.attention_type, rank=custom_cfg.attention_rank, copy_weights=False, # Weights already inside the saved base model **custom_cfg.attention_kwargs, ) return model # ------------------------------------------------------------------ # Other methods # ------------------------------------------------------------------ def forward(self, *args, **kwargs) -> Any: """Forward pass — compute module output from inputs.""" return self.model(*args, **kwargs) def save_pretrained(self, save_directory: str) -> None: """Convenience wrapper – delegates to the mixin.""" super().save_pretrained(save_directory) @classmethod def from_pretrained( cls, model_id: str, task: str = "sequence_classification", num_labels: int | None = None, **kwargs, ) -> "UnifiedTransformerModel": """ Create a new UnifiedTransformerModel from a pretrained identifier. Automatically applies the custom attention if the model was saved with one. This is the main entry point for users. """ # If the model_id has a unified_attention_config.json, use the mixin's from_pretrained # which will call our _from_pretrained. Otherwise, create a fresh one. import os has_custom = False if os.path.isdir(model_id): has_custom = os.path.exists(os.path.join(model_id, "unified_attention_config.json")) if has_custom: return cls._from_pretrained(model_id, **kwargs) else: # Fresh model – no custom attention yet if task == "sequence_classification": sc_kwargs = dict(kwargs) if num_labels is not None: sc_kwargs["num_labels"] = num_labels base = AutoModelForSequenceClassification.from_pretrained(model_id, **sc_kwargs) else: base = AutoModel.from_pretrained(model_id, **kwargs) return cls(base)A wrapper around a HuggingFace model that supports custom attention and can be saved/loaded via the HuggingFace Hub.
Initialize internal Module state, shared by both nn.Module and ScriptModule.
Ancestors
- torch.nn.modules.module.Module
- huggingface_hub.hub_mixin.PyTorchModelHubMixin
- huggingface_hub.hub_mixin.ModelHubMixin
Static methods
def from_pretrained(model_id: str,
task: str = 'sequence_classification',
num_labels: int | None = None,
**kwargs) ‑> UnifiedTransformerModel-
Create a new UnifiedTransformerModel from a pretrained identifier. Automatically applies the custom attention if the model was saved with one. This is the main entry point for users.
Methods
def forward(self, *args, **kwargs) ‑> Any-
Expand source code
def forward(self, *args, **kwargs) -> Any: """Forward pass — compute module output from inputs.""" return self.model(*args, **kwargs)Forward pass — compute module output from inputs.
def freeze_layers(self, num_layers_to_freeze: int = 0) ‑> Any-
Expand source code
def freeze_layers(self, num_layers_to_freeze: int = 0) -> Any: """Freeze layers.""" layers = get_transformer_layers(self.model) for i, layer in enumerate(layers): for param in layer.parameters(): param.requires_grad = i >= num_layers_to_freeze return selfFreeze layers.
def get_parameter_count(self) ‑> dict[str, int]-
Expand source code
def get_parameter_count(self) -> dict[str, int]: """Return the parameter count.""" total = sum(p.numel() for p in self.model.parameters()) trainable = sum(p.numel() for p in self.model.parameters() if p.requires_grad) return {"total": total, "trainable": trainable, "frozen": total - trainable}Return the parameter count.
def replace_attention(self,
attention_type: str,
rank: int | None = None,
copy_weights: bool = True,
**kwargs) ‑> Any-
Expand source code
def replace_attention( self, attention_type: str, rank: int | None = None, copy_weights: bool = True, **kwargs, ) -> Any: """ Inject a custom attention variant into all layers. This method also records the choice so it can be saved. """ inject_attention(self.model, attention_type, rank=rank, copy_weights=copy_weights, **kwargs) self._custom_config = UnifiedTransformerConfig( attention_type=attention_type, attention_rank=rank, attention_kwargs=kwargs, ) return selfInject a custom attention variant into all layers. This method also records the choice so it can be saved.
def save_pretrained(self, save_directory: str) ‑> None-
Expand source code
def save_pretrained(self, save_directory: str) -> None: """Convenience wrapper – delegates to the mixin.""" super().save_pretrained(save_directory)Convenience wrapper – delegates to the mixin.
def unfreeze_all(self) ‑> Any-
Expand source code
def unfreeze_all(self) -> Any: """Unfreeze all.""" for param in self.model.parameters(): param.requires_grad = True return selfUnfreeze all.