quantization.py · YSamchuk/Hetman-2b at main

Hetman-2b / quantization.py

feat: publish official Hetman-2.0B architecture, tokenizer, and model card

0c5303b verified 2 days ago

7.45 kB

"""
BitNet b1.58 Ternary Quantization & Fast Walsh-Hadamard Transform (FWHT)
"""

import math
import torch
import torch.nn as nn
import torch.nn.functional as F

def fast_walsh_hadamard_transform_1d(x: torch.Tensor) -> torch.Tensor:
    """
    Vectorized Fast Walsh-Hadamard Transform (FWHT) for tensor of length N (power of 2).
    Computed in float32 for high numerical precision and stability, then converted back to input dtype.
    Orthogonal rotation with scale factor 1/sqrt(N).
    """
    orig_dtype = x.dtype
    compute_dtype = torch.float32 if x.dtype in (torch.float16, torch.bfloat16) else x.dtype
    out = x.to(compute_dtype)
    N = out.shape[-1]
    assert (N & (N - 1) == 0) and N > 0, f"FWHT length N must be a power of 2, got {N}. For arbitrary dimensions, use blockwise_fwht()."
    lead_shape = out.shape[:-1]
    out = out.reshape(-1, N)
    
    h = 1
    while h < N:
        out = out.reshape(-1, N // (2 * h), 2, h)
        a = out[:, :, 0, :]
        b = out[:, :, 1, :]
        out = torch.stack([a + b, a - b], dim=2).reshape(-1, N)
        h *= 2
        
    out = out.reshape(*lead_shape, N) / math.sqrt(N)
    return out.to(orig_dtype)

def blockwise_fwht(x: torch.Tensor, block_size: int = 512) -> torch.Tensor:
    """
    Applies FWHT block-by-block across the hidden dimension d.
    If hidden_dim = 1536 and block_size = 512, applies 3 independent 512-point rotations.
    """
    orig_shape = x.shape
    d = orig_shape[-1]
    if d % block_size != 0:
        return x  # Fallback if not divisible
    
    x_reshaped = x.reshape(-1, d // block_size, block_size)
    x_rotated = fast_walsh_hadamard_transform_1d(x_reshaped)
    return x_rotated.reshape(orig_shape)

class BitLinear(nn.Module):
    """
    BitNet b1.58 Linear Layer with Straight-Through Estimator (STE)
    and online Fast Walsh-Hadamard Transform for outlier suppression.
    """
    def __init__(self, in_features: int, out_features: int, bias: bool = False, use_fwht: bool = True):
        super().__init__()
        self.in_features = in_features
        self.out_features = out_features
        self.use_fwht = use_fwht and (in_features % 512 == 0)
        
        # Latent master weights in FP32/BF16
        self.weight = nn.Parameter(torch.empty(out_features, in_features))
        if bias:
            self.bias = nn.Parameter(torch.zeros(out_features))
        else:
            self.register_parameter('bias', None)
            
        self.rms_norm = nn.RMSNorm(in_features, eps=1e-6)
        self.reset_parameters()

    def reset_parameters(self):
        # Kaiming uniform initialization scaled for ternary regime
        nn.init.kaiming_uniform_(self.weight, a=math.sqrt(5))

    def quantize_weights(self, w: torch.Tensor) -> torch.Tensor:
        # Scale factor gamma: mean absolute value
        gamma = torch.mean(torch.abs(w)).clamp(min=1e-6)
        # Scaled round to {-1, 0, +1}
        w_scaled = w / gamma
        w_quant = torch.clamp(torch.round(w_scaled), -1.0, 1.0)
        # Straight-Through Estimator (STE)
        return w + (w_quant * gamma - w).detach()

    def quantize_activations(self, x: torch.Tensor) -> torch.Tensor:
        # 8-bit dynamic activation quantization
        eta = torch.max(torch.abs(x), dim=-1, keepdim=True).values.clamp(min=1e-6)
        scale = 127.0 / eta
        x_quant = torch.clamp(torch.round(x * scale), -128.0, 127.0)
        return x + (x_quant / scale - x).detach()

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # 1. Sub-Layer RMSNorm
        x_norm = self.rms_norm(x)
        
        # 2. Online FWHT to eliminate activation outliers
        if self.use_fwht:
            x_rot = blockwise_fwht(x_norm, block_size=512)
        else:
            x_rot = x_norm
            
        # 3. Quantize activations and weights
        x_q = self.quantize_activations(x_rot)
        w_q = self.quantize_weights(self.weight)
        
        # 4. Add-Only / Bit-parallel linear contraction
        out = F.linear(x_q, w_q, self.bias)
        return out


if __name__ == "__main__":
    print("=" * 60)
    print("  BitNet b1.58 & FWHT Self-Verification Test Suite")
    print("=" * 60)

    # 1. FWHT Orthogonality and Isometry
    print("[1/5] Verifying FWHT orthogonality and norm preservation...")
    eye512 = torch.eye(512)
    H512 = fast_walsh_hadamard_transform_1d(eye512)
    ortho_err = torch.max(torch.abs(H512 @ H512.T - eye512)).item()
    symm_err = torch.max(torch.abs(H512 - H512.T)).item()
    assert ortho_err < 1e-5, f"FWHT not orthogonal: error {ortho_err}"
    assert symm_err < 1e-5, f"FWHT not symmetric: error {symm_err}"

    x_test = torch.randn(8, 1536)
    x_rot = blockwise_fwht(x_test, block_size=512)
    norm_err = torch.max(torch.abs(torch.norm(x_test, p=2, dim=-1) - torch.norm(x_rot, p=2, dim=-1))).item()
    assert norm_err < 1e-4, f"L2 norm not preserved: error {norm_err}"
    print(f"      ✓ Orthogonality err={ortho_err:.2e}, Symmetry err={symm_err:.2e}, Norm err={norm_err:.2e}")

    # 2. Outlier Suppression Test
    print("[2/5] Verifying Outlier Suppression via FWHT...")
    x_outlier = torch.randn(1, 512)
    x_outlier[0, 100] = 100.0  # single channel spike
    x_suppressed = fast_walsh_hadamard_transform_1d(x_outlier)
    max_orig = torch.max(torch.abs(x_outlier)).item()
    max_supp = torch.max(torch.abs(x_suppressed)).item()
    reduction = max_orig / max_supp
    assert reduction > 10.0, f"Outlier suppression insufficient: {reduction:.2f}x"
    print(f"      ✓ Outlier reduced from {max_orig:.1f} to {max_supp:.2f} ({reduction:.1f}x reduction, ~sqrt(512))")

    # 3. Ternary Weight Quantization & Kaiming Stats
    print("[3/5] Verifying Weight Ternarization {-1, 0, +1} & STE...")
    layer = BitLinear(1536, 6144, bias=True, use_fwht=True)
    w_q = layer.quantize_weights(layer.weight)
    gamma = torch.mean(torch.abs(layer.weight)).item()
    # Normalize by gamma to get pure discrete ternary values
    ternary_vals = torch.unique(torch.round(w_q / gamma))
    expected_vals = torch.tensor([-1.0, 0.0, 1.0])
    assert torch.equal(torch.sort(ternary_vals).values, expected_vals), f"Ternary states invalid: {ternary_vals}"
    print(f"      ✓ Pure ternary states confirmed: {ternary_vals.tolist()} with gamma={gamma:.5f}")

    # 4. Activation INT8 Dynamic Quantization
    print("[4/5] Verifying INT8 Activation Quantization [-128, 127]...")
    act_dummy = torch.randn(4, 16, 1536) * 5.0
    act_q = layer.quantize_activations(act_dummy)
    assert not torch.isnan(act_q).any(), "NaN in quantized activations"
    print("      ✓ INT8 dynamic per-token scaling verified without saturation")

    # 5. End-to-end Gradient Flow (STE) & Dtype Support
    print("[5/5] Verifying STE Gradient Backpropagation across Dtypes...")
    for dt in [torch.float32, torch.bfloat16]:
        l = BitLinear(512, 1024, bias=True, use_fwht=True).to(dt)
        inp = torch.randn(2, 4, 512, dtype=dt, requires_grad=True)
        out = l(inp)
        loss = out.sum()
        loss.backward()
        assert inp.grad is not None and not torch.isnan(inp.grad).any(), f"Gradient error for {dt}"
        assert l.weight.grad is not None and not torch.isnan(l.weight.grad).any(), f"Weight grad error for {dt}"
        print(f"      ✓ {dt}: Forward & STE Backward successfully verified (no NaNs)")

    print("=" * 60)
    print("  ALL BITNET B1.58 & FWHT CHECKS PASSED SUCCESSFULLY!")
    print("=" * 60)

	"""
	BitNet b1.58 Ternary Quantization & Fast Walsh-Hadamard Transform (FWHT)
	"""

	import math
	import torch
	import torch.nn as nn
	import torch.nn.functional as F

	def fast_walsh_hadamard_transform_1d(x: torch.Tensor) -> torch.Tensor:
	"""
	Vectorized Fast Walsh-Hadamard Transform (FWHT) for tensor of length N (power of 2).
	Computed in float32 for high numerical precision and stability, then converted back to input dtype.
	Orthogonal rotation with scale factor 1/sqrt(N).
	"""
	orig_dtype = x.dtype
	compute_dtype = torch.float32 if x.dtype in (torch.float16, torch.bfloat16) else x.dtype
	out = x.to(compute_dtype)
	N = out.shape[-1]
	assert (N & (N - 1) == 0) and N > 0, f"FWHT length N must be a power of 2, got {N}. For arbitrary dimensions, use blockwise_fwht()."
	lead_shape = out.shape[:-1]
	out = out.reshape(-1, N)

	h = 1
	while h < N:
	out = out.reshape(-1, N // (2 * h), 2, h)
	a = out[:, :, 0, :]
	b = out[:, :, 1, :]
	out = torch.stack([a + b, a - b], dim=2).reshape(-1, N)
	h *= 2

	out = out.reshape(*lead_shape, N) / math.sqrt(N)
	return out.to(orig_dtype)

	def blockwise_fwht(x: torch.Tensor, block_size: int = 512) -> torch.Tensor:
	"""
	Applies FWHT block-by-block across the hidden dimension d.
	If hidden_dim = 1536 and block_size = 512, applies 3 independent 512-point rotations.
	"""
	orig_shape = x.shape
	d = orig_shape[-1]
	if d % block_size != 0:
	return x # Fallback if not divisible

	x_reshaped = x.reshape(-1, d // block_size, block_size)
	x_rotated = fast_walsh_hadamard_transform_1d(x_reshaped)
	return x_rotated.reshape(orig_shape)

	class BitLinear(nn.Module):
	"""
	BitNet b1.58 Linear Layer with Straight-Through Estimator (STE)
	and online Fast Walsh-Hadamard Transform for outlier suppression.
	"""
	def __init__(self, in_features: int, out_features: int, bias: bool = False, use_fwht: bool = True):
	super().__init__()
	self.in_features = in_features
	self.out_features = out_features
	self.use_fwht = use_fwht and (in_features % 512 == 0)

	# Latent master weights in FP32/BF16
	self.weight = nn.Parameter(torch.empty(out_features, in_features))
	if bias:
	self.bias = nn.Parameter(torch.zeros(out_features))
	else:
	self.register_parameter('bias', None)

	self.rms_norm = nn.RMSNorm(in_features, eps=1e-6)
	self.reset_parameters()

	def reset_parameters(self):
	# Kaiming uniform initialization scaled for ternary regime
	nn.init.kaiming_uniform_(self.weight, a=math.sqrt(5))

	def quantize_weights(self, w: torch.Tensor) -> torch.Tensor:
	# Scale factor gamma: mean absolute value
	gamma = torch.mean(torch.abs(w)).clamp(min=1e-6)
	# Scaled round to {-1, 0, +1}
	w_scaled = w / gamma
	w_quant = torch.clamp(torch.round(w_scaled), -1.0, 1.0)
	# Straight-Through Estimator (STE)
	return w + (w_quant * gamma - w).detach()

	def quantize_activations(self, x: torch.Tensor) -> torch.Tensor:
	# 8-bit dynamic activation quantization
	eta = torch.max(torch.abs(x), dim=-1, keepdim=True).values.clamp(min=1e-6)
	scale = 127.0 / eta
	x_quant = torch.clamp(torch.round(x * scale), -128.0, 127.0)
	return x + (x_quant / scale - x).detach()

	def forward(self, x: torch.Tensor) -> torch.Tensor:
	# 1. Sub-Layer RMSNorm
	x_norm = self.rms_norm(x)

	# 2. Online FWHT to eliminate activation outliers
	if self.use_fwht:
	x_rot = blockwise_fwht(x_norm, block_size=512)
	else:
	x_rot = x_norm

	# 3. Quantize activations and weights
	x_q = self.quantize_activations(x_rot)
	w_q = self.quantize_weights(self.weight)

	# 4. Add-Only / Bit-parallel linear contraction
	out = F.linear(x_q, w_q, self.bias)
	return out


	if __name__ == "__main__":
	print("=" * 60)
	print(" BitNet b1.58 & FWHT Self-Verification Test Suite")
	print("=" * 60)

	# 1. FWHT Orthogonality and Isometry
	print("[1/5] Verifying FWHT orthogonality and norm preservation...")
	eye512 = torch.eye(512)
	H512 = fast_walsh_hadamard_transform_1d(eye512)
	ortho_err = torch.max(torch.abs(H512 @ H512.T - eye512)).item()
	symm_err = torch.max(torch.abs(H512 - H512.T)).item()
	assert ortho_err < 1e-5, f"FWHT not orthogonal: error {ortho_err}"
	assert symm_err < 1e-5, f"FWHT not symmetric: error {symm_err}"

	x_test = torch.randn(8, 1536)
	x_rot = blockwise_fwht(x_test, block_size=512)
	norm_err = torch.max(torch.abs(torch.norm(x_test, p=2, dim=-1) - torch.norm(x_rot, p=2, dim=-1))).item()
	assert norm_err < 1e-4, f"L2 norm not preserved: error {norm_err}"
	print(f" ✓ Orthogonality err={ortho_err:.2e}, Symmetry err={symm_err:.2e}, Norm err={norm_err:.2e}")

	# 2. Outlier Suppression Test
	print("[2/5] Verifying Outlier Suppression via FWHT...")
	x_outlier = torch.randn(1, 512)
	x_outlier[0, 100] = 100.0 # single channel spike
	x_suppressed = fast_walsh_hadamard_transform_1d(x_outlier)
	max_orig = torch.max(torch.abs(x_outlier)).item()
	max_supp = torch.max(torch.abs(x_suppressed)).item()
	reduction = max_orig / max_supp
	assert reduction > 10.0, f"Outlier suppression insufficient: {reduction:.2f}x"
	print(f" ✓ Outlier reduced from {max_orig:.1f} to {max_supp:.2f} ({reduction:.1f}x reduction, ~sqrt(512))")

	# 3. Ternary Weight Quantization & Kaiming Stats
	print("[3/5] Verifying Weight Ternarization {-1, 0, +1} & STE...")
	layer = BitLinear(1536, 6144, bias=True, use_fwht=True)
	w_q = layer.quantize_weights(layer.weight)
	gamma = torch.mean(torch.abs(layer.weight)).item()
	# Normalize by gamma to get pure discrete ternary values
	ternary_vals = torch.unique(torch.round(w_q / gamma))
	expected_vals = torch.tensor([-1.0, 0.0, 1.0])
	assert torch.equal(torch.sort(ternary_vals).values, expected_vals), f"Ternary states invalid: {ternary_vals}"
	print(f" ✓ Pure ternary states confirmed: {ternary_vals.tolist()} with gamma={gamma:.5f}")

	# 4. Activation INT8 Dynamic Quantization
	print("[4/5] Verifying INT8 Activation Quantization [-128, 127]...")
	act_dummy = torch.randn(4, 16, 1536) * 5.0
	act_q = layer.quantize_activations(act_dummy)
	assert not torch.isnan(act_q).any(), "NaN in quantized activations"
	print(" ✓ INT8 dynamic per-token scaling verified without saturation")

	# 5. End-to-end Gradient Flow (STE) & Dtype Support
	print("[5/5] Verifying STE Gradient Backpropagation across Dtypes...")
	for dt in [torch.float32, torch.bfloat16]:
	l = BitLinear(512, 1024, bias=True, use_fwht=True).to(dt)
	inp = torch.randn(2, 4, 512, dtype=dt, requires_grad=True)
	out = l(inp)
	loss = out.sum()
	loss.backward()
	assert inp.grad is not None and not torch.isnan(inp.grad).any(), f"Gradient error for {dt}"
	assert l.weight.grad is not None and not torch.isnan(l.weight.grad).any(), f"Weight grad error for {dt}"
	print(f" ✓ {dt}: Forward & STE Backward successfully verified (no NaNs)")

	print("=" * 60)
	print(" ALL BITNET B1.58 & FWHT CHECKS PASSED SUCCESSFULLY!")
	print("=" * 60)