import torch import os import torch.nn as nn from typing import Optional, Tuple, Iterable, Union from transformers.models.qwen2.modeling_qwen2 import ( Qwen2ForCausalLM, Qwen2Model, Qwen2DecoderLayer, ) # ------------------------- # Low-rank adapter # ------------------------- def _get_activation(name: str): name = name.lower() if name in ("silu", "swish"): return nn.SiLU() if name == "relu": return nn.ReLU() if name == "gelu": return nn.GELU() if name == "tanh": return nn.Tanh() raise ValueError(f"Unknown activation: {name}") class LowRankAdapter(nn.Module): """ Δh = α * W_up( act(W_down(h)) ) """ def __init__(self, hidden_size: int, rank: int, alpha: float, activation: str): super().__init__() self.alpha = float(alpha) self.act = _get_activation(activation) self.down = nn.Linear(hidden_size, rank, bias=False) self.up = nn.Linear(rank, hidden_size, bias=False) # start as no-op => preserves pretrained behavior at init nn.init.zeros_(self.up.weight) def forward(self, h: torch.Tensor) -> torch.Tensor: return self.alpha * self.up(self.act(self.down(h))) # ------------------------- # Steered Decoder Layer (post-block only) # ------------------------- class Qwen2DecoderLayerPostBlockSteering(Qwen2DecoderLayer): """ Drop-in Qwen2DecoderLayer that adds an adapter AFTER the block output. apply_to: - "last": apply only to last token (B,S,H) -> only position -1 - "all": apply to all tokens """ def __init__( self, config, layer_idx: int, # Custom arguments with defaults enable: bool = True, rank: int = 8, alpha: float = 1.0, activation: str = "silu", apply_to: str = "all", **kwargs # <--- Best Practice: Catch any extra args the parent might need in future versions ): super().__init__(config, layer_idx, **kwargs) assert apply_to in ("last", "all") self.apply_to = apply_to self._adapter_enabled = True self.adapter_block = ( LowRankAdapter( hidden_size=config.hidden_size, rank=rank, alpha=alpha, activation=activation, ) if enable else None ) def set_adapter_enabled(self, enabled: bool): self._adapter_enabled = bool(enabled) def _apply_last(self, x: torch.Tensor, adapter: nn.Module) -> torch.Tensor: if x.ndim != 3: return x last = x[:, -1, :] # (B,H) new_last = (last + adapter(last)).unsqueeze(1) # (B,1,H) return torch.cat([x[:, :-1, :], new_last], dim=1) def _apply_all(self, x: torch.Tensor, adapter: nn.Module) -> torch.Tensor: if x.ndim != 3: return x b, s, h = x.shape flat = x.reshape(b * s, h) delta = adapter(flat).reshape(b, s, h) return x + delta def _apply_adapter(self, x: torch.Tensor) -> torch.Tensor: if (self.adapter_block is None) or (not self._adapter_enabled): return x if self.apply_to == "last": return self._apply_last(x, self.adapter_block) return self._apply_all(x, self.adapter_block) def forward( self, hidden_states: torch.Tensor, attention_mask: Optional[torch.Tensor] = None, position_ids: Optional[torch.LongTensor] = None, past_key_value: Optional[Tuple[torch.Tensor]] = None, # legacy name output_attentions: Optional[bool] = False, use_cache: Optional[bool] = False, cache_position: Optional[torch.LongTensor] = None, position_embeddings: Optional[Tuple[torch.Tensor, torch.FloatTensor]] = None, **kwargs, ): # Standard Qwen2 layer, inject adapter at the very end (post-block). # NOTE: transformers 4.57+ Qwen2 expects decoder layers to return a Tensor # (and optionally attn weights), NOT (hidden_states, ..., present_kv). # Cache is carried via `past_key_values` (new API) and/or handled internally. residual = hidden_states hidden_states = self.input_layernorm(hidden_states) # Avoid passing BOTH past_key_value and past_key_values to attention. past_key_values = kwargs.pop("past_key_values", None) attn_kwargs = dict(kwargs) if past_key_values is not None: attn_kwargs["past_key_values"] = past_key_values # do NOT also pass legacy past_key_value pkv_arg = {} else: pkv_arg = {"past_key_value": past_key_value} if past_key_value is not None else {} attn_out = self.self_attn( hidden_states=hidden_states, attention_mask=attention_mask, position_ids=position_ids, output_attentions=output_attentions, use_cache=use_cache, cache_position=cache_position, position_embeddings=position_embeddings, **pkv_arg, **attn_kwargs, ) # HF attention returns (attn_output,) or (attn_output, attn_weights) if isinstance(attn_out, tuple): attn_output = attn_out[0] attn_weights = attn_out[1] if (output_attentions and len(attn_out) > 1) else None else: attn_output = attn_out attn_weights = None hidden_states = residual + attn_output residual = hidden_states hidden_states = self.post_attention_layernorm(hidden_states) hidden_states = residual + self.mlp(hidden_states) # ✅ post-block steering hidden_states = self._apply_adapter(hidden_states) # Return a Tensor (or Tensor + attn weights if requested). Do NOT return cache. if output_attentions: return (hidden_states, attn_weights) return hidden_states # ------------------------- # Qwen2Model + hardcoded steering config # ------------------------- class Qwen2ModelPostBlockSteering(Qwen2Model): def __init__( self, config, layers_to_steer: Union[str, Iterable[int]] = "all", rank: int = 8, apply_to: str = "all", alpha: float = 1.0, activation: str = "silu", ): super().__init__(config) if layers_to_steer == "all": layer_ids = set(range(config.num_hidden_layers)) else: layer_ids = set(int(i) for i in layers_to_steer) new_layers = nn.ModuleList() for i in range(config.num_hidden_layers): new_layers.append( Qwen2DecoderLayerPostBlockSteering( config=config, layer_idx=i, enable=(i in layer_ids), rank=rank, alpha=alpha, activation=activation, apply_to=apply_to, ) ) self.layers = new_layers def set_adapter_enabled(self, enabled: bool): for layer in self.layers: if hasattr(layer, "set_adapter_enabled"): layer.set_adapter_enabled(enabled) # ------------------------- # Qwen2ForCausalLM with hardcoded knobs + base frozen by default # ------------------------- class Qwen2ForCausalLMPostBlockSteeringFixed(Qwen2ForCausalLM): """ Hardcoded steering config + base frozen by default. Change these class constants to match what you want globally. """ STEER_RANK: int = 8 STEER_APPLY_TO: str = "last" # "last" or "all" STEER_LAYERS: Union[str, Iterable[int]] = "all" # or e.g. [0, 5, 10] STEER_ALPHA: float = 1.0 STEER_ACTIVATION: str = "silu" def __init__(self, config): super().__init__(config) # Replace base transformer with steered one using hardcoded config self.model = Qwen2ModelPostBlockSteering( config, layers_to_steer=self.STEER_LAYERS, rank=self.STEER_RANK, apply_to=self.STEER_APPLY_TO, alpha=self.STEER_ALPHA, activation=self.STEER_ACTIVATION, ) # Freeze base by default (only steering trainable) self.freeze_base_keep_steering_trainable() # ---- freezing / params ---- def freeze_base_keep_steering_trainable(self): for n, p in self.named_parameters(): p.requires_grad = ("adapter_block" in n) def steering_parameters(self): for n, p in self.named_parameters(): if "adapter_block" in n: yield p # ---- dtype/device correctness for device_map="auto" ---- def cast_adapters_like_base(self): """ If you load with torch_dtype="auto" and/or device_map="auto", adapters are newly-created modules and need to match each layer’s dtype/device. """ for layer in self.model.layers: ref = layer.input_layernorm.weight if getattr(layer, "adapter_block", None) is not None: layer.adapter_block.to(device=ref.device, dtype=ref.dtype) @classmethod def from_pretrained(cls, *args, **kwargs): model = super().from_pretrained(*args, **kwargs) # Ensure adapters are on the right shards/dtype, then freeze base if hasattr(model, "cast_adapters_like_base"): model.cast_adapters_like_base() if hasattr(model, "freeze_base_keep_steering_trainable"): model.freeze_base_keep_steering_trainable() return model def _prepare_for_serialization(self): """ If the model was loaded with device_map/offload, Accelerate attaches hooks that can break save_pretrained for newly-added params (like adapter_block.*). This removes those hooks and consolidates to CPU. """ try: from accelerate.hooks import remove_hook_from_module remove_hook_from_module(self, recurse=True) except Exception: pass # Clean up common accelerate attributes if present for attr in ("hf_device_map", "_hf_hook"): if hasattr(self, attr): try: delattr(self, attr) except Exception: pass # Ensure all params are materialized on CPU for a normal state_dict save self.to("cpu") def _strip_accelerate_offload_hooks(self): """ Remove Accelerate's device_map/offload hooks so saving doesn't go through get_state_dict_from_offload (which doesn't know about new adapter params). """ # Best-effort official removers try: from accelerate.hooks import remove_hook_from_module remove_hook_from_module(self, recurse=True) # documented API :contentReference[oaicite:3]{index=3} except Exception: pass # Hard removal: delete _hf_hook from every submodule if still present for m in self.modules(): if hasattr(m, "_hf_hook"): # try to detach cleanly if possible try: m._hf_hook.detach_hook(m) except Exception: pass try: delattr(m, "_hf_hook") except Exception: pass # device_map bookkeeping (common on big-model inference) if hasattr(self, "hf_device_map"): try: delattr(self, "hf_device_map") except Exception: pass def save_pretrained(self, save_directory, **kwargs): os.makedirs(save_directory, exist_ok=True) # 1) remove accelerate offload hooks self._strip_accelerate_offload_hooks() # 2) consolidate to CPU (you cannot save sharded/offloaded weights “in place”) self.to("cpu") # 3) create a normal state_dict and pass it explicitly to bypass accelerate offload-saving # (save_pretrained supports state_dict=...) :contentReference[oaicite:4]{index=4} sd = {k: v.cpu() for k, v in self.state_dict().items()} return super().save_pretrained(save_directory, state_dict=sd, **kwargs)