355 lines
12 KiB
Python
355 lines
12 KiB
Python
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)
|