딥러닝 모델 양자화 완전 정복: INT8, INT4, GPTQ, AWQ, GGUF 마스터하기

import torch
import numpy as np

# 각 데이터 타입의 메모리 크기 확인
x_fp32 = torch.tensor([1.5, -2.3, 0.7], dtype=torch.float32)
x_fp16 = torch.tensor([1.5, -2.3, 0.7], dtype=torch.float16)
x_bf16 = torch.tensor([1.5, -2.3, 0.7], dtype=torch.bfloat16)

print(f"FP32: {x_fp32.element_size()} bytes per element")  # 4 bytes
print(f"FP16: {x_fp16.element_size()} bytes per element")  # 2 bytes
print(f"BF16: {x_bf16.element_size()} bytes per element")  # 2 bytes

# 모델 메모리 계산 예시 (7B 파라미터 모델)
params = 7e9
fp32_memory_gb = params * 4 / 1e9
fp16_memory_gb = params * 2 / 1e9
int8_memory_gb = params * 1 / 1e9
int4_memory_gb = params * 0.5 / 1e9

print(f"\n7B 모델 메모리 요구량:")
print(f"FP32: {fp32_memory_gb:.1f} GB")   # 28.0 GB
print(f"FP16: {fp16_memory_gb:.1f} GB")   # 14.0 GB
print(f"INT8: {int8_memory_gb:.1f} GB")   # 7.0 GB
print(f"INT4: {int4_memory_gb:.1f} GB")   # 3.5 GB

q = clamp(round(x / scale) + zero_point, q_min, q_max)

x_approx = scale * (q - zero_point)

import torch
import numpy as np

def symmetric_quantize(x: torch.Tensor, num_bits: int = 8):
    """대칭 양자화 구현"""
    q_max = 2 ** (num_bits - 1) - 1  # INT8의 경우 127
    q_min = -q_max  # -127

    # 스케일 계산
    max_abs = x.abs().max()
    scale = max_abs / q_max

    # 양자화
    q = torch.clamp(torch.round(x / scale), q_min, q_max).to(torch.int8)

    return q, scale

def asymmetric_quantize(x: torch.Tensor, num_bits: int = 8):
    """비대칭 양자화 구현"""
    q_max = 2 ** num_bits - 1  # UINT8의 경우 255
    q_min = 0

    # 스케일과 zero_point 계산
    min_val = x.min()
    max_val = x.max()
    scale = (max_val - min_val) / (q_max - q_min)
    zero_point = q_min - torch.round(min_val / scale)
    zero_point = torch.clamp(zero_point, q_min, q_max).to(torch.int32)

    # 양자화
    q = torch.clamp(torch.round(x / scale) + zero_point, q_min, q_max).to(torch.uint8)

    return q, scale, zero_point

def dequantize(q: torch.Tensor, scale: torch.Tensor, zero_point: torch.Tensor = None):
    """역양자화"""
    if zero_point is None:
        return scale * q.float()
    return scale * (q.float() - zero_point.float())

# 테스트
x = torch.randn(100)
print(f"원본 데이터 범위: [{x.min():.4f}, {x.max():.4f}]")

# 대칭 양자화
q_sym, scale_sym = symmetric_quantize(x)
x_reconstructed_sym = dequantize(q_sym, scale_sym)
error_sym = (x - x_reconstructed_sym).abs().mean()
print(f"대칭 양자화 평균 오차: {error_sym:.6f}")

# 비대칭 양자화
q_asym, scale_asym, zp_asym = asymmetric_quantize(x)
x_reconstructed_asym = dequantize(q_asym, scale_asym, zp_asym)
error_asym = (x - x_reconstructed_asym).abs().mean()
print(f"비대칭 양자화 평균 오차: {error_asym:.6f}")

import torch

def per_group_quantize(weight: torch.Tensor, group_size: int = 128, num_bits: int = 4):
    """Per-Group 양자화 구현"""
    rows, cols = weight.shape

    # 그룹 단위로 분할
    weight_grouped = weight.reshape(-1, group_size)

    # 각 그룹의 최대/최소값
    max_vals = weight_grouped.max(dim=1, keepdim=True)[0]
    min_vals = weight_grouped.min(dim=1, keepdim=True)[0]

    q_max = 2 ** num_bits - 1  # 15 for INT4

    # 스케일 계산
    scales = (max_vals - min_vals) / q_max
    zero_points = torch.round(-min_vals / scales)

    # 양자화
    q = torch.clamp(torch.round(weight_grouped / scales) + zero_points, 0, q_max)

    # 역양자화
    weight_dequant = scales * (q - zero_points)
    weight_dequant = weight_dequant.reshape(rows, cols)

    return q, scales, zero_points, weight_dequant

# 예시: Transformer 가중치 양자화
weight = torch.randn(4096, 4096)  # Llama-style weight
q, scales, zp, weight_dequant = per_group_quantize(weight, group_size=128, num_bits=4)

error = (weight - weight_dequant).abs().mean()
print(f"Per-Group INT4 양자화 평균 오차: {error:.6f}")
print(f"압축률: {weight.element_size() * weight.numel() / (q.numel() / 2 + scales.numel() * 4):.2f}x")

from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
from datasets import load_dataset

def collect_calibration_data(model_name: str, num_samples: int = 128):
    """보정 데이터 수집"""
    tokenizer = AutoTokenizer.from_pretrained(model_name)

    # WikiText-2 또는 C4 데이터셋 사용이 일반적
    dataset = load_dataset("wikitext", "wikitext-2-raw-v1", split="train")

    texts = []
    for item in dataset:
        if len(item['text'].strip()) > 100:
            texts.append(item['text'].strip())
        if len(texts) >= num_samples:
            break

    # 토크나이즈
    encoded = [
        tokenizer(text, return_tensors="pt", max_length=2048, truncation=True)
        for text in texts
    ]

    return encoded

# 보정 데이터로 활성화 통계 수집
def collect_activation_stats(model, calibration_data, layer_name: str):
    """특정 레이어의 활성화 통계 수집"""
    stats = {"min": float("inf"), "max": float("-inf"), "histogram": []}

    def hook_fn(module, input, output):
        with torch.no_grad():
            act = output.detach().float()
            stats["min"] = min(stats["min"], act.min().item())
            stats["max"] = max(stats["max"], act.max().item())

    # 훅 등록
    target_layer = dict(model.named_modules())[layer_name]
    handle = target_layer.register_forward_hook(hook_fn)

    # 보정 데이터 실행
    model.eval()
    with torch.no_grad():
        for batch in calibration_data[:32]:
            model(**batch)

    handle.remove()
    return stats

class MinMaxCalibrator:
    """최소-최대 보정기"""

    def __init__(self):
        self.min_val = float("inf")
        self.max_val = float("-inf")

    def update(self, tensor: torch.Tensor):
        self.min_val = min(self.min_val, tensor.min().item())
        self.max_val = max(self.max_val, tensor.max().item())

    def compute_scale_zp(self, num_bits: int = 8, symmetric: bool = True):
        q_max = 2 ** (num_bits - 1) - 1 if symmetric else 2 ** num_bits - 1

        if symmetric:
            max_abs = max(abs(self.min_val), abs(self.max_val))
            scale = max_abs / q_max
            zero_point = 0
        else:
            scale = (self.max_val - self.min_val) / q_max
            zero_point = -round(self.min_val / scale)

        return scale, zero_point

import numpy as np
from scipy import stats

class HistogramCalibrator:
    """히스토그램 기반 보정기 (KL Divergence 최소화)"""

    def __init__(self, num_bins: int = 2048):
        self.num_bins = num_bins
        self.histogram = None
        self.bin_edges = None

    def update(self, tensor: torch.Tensor):
        data = tensor.detach().float().numpy().flatten()

        if self.histogram is None:
            self.histogram, self.bin_edges = np.histogram(data, bins=self.num_bins)
        else:
            new_hist, _ = np.histogram(data, bins=self.bin_edges)
            self.histogram += new_hist

    def compute_optimal_range(self, num_bits: int = 8):
        """KL Divergence를 최소화하는 최적 범위 탐색"""
        num_quantized_bins = 2 ** num_bits - 1

        best_kl = float("inf")
        best_threshold = None

        # 다양한 threshold 탐색
        for i in range(num_quantized_bins, len(self.histogram)):
            # 히스토그램을 num_quantized_bins 개로 압축
            reference = self.histogram[:i].copy().astype(float)
            reference /= reference.sum()

            # KL Divergence 계산 (근사)
            quantized = np.zeros(i)
            bin_size = i / num_quantized_bins

            for j in range(num_quantized_bins):
                start = int(j * bin_size)
                end = int((j + 1) * bin_size)
                quantized[start:end] = reference[start:end].sum() / (end - start)

            # 0인 구간 처리
            quantized = np.where(quantized == 0, 1e-10, quantized)
            reference_clipped = np.where(reference == 0, 1e-10, reference)

            kl = stats.entropy(reference_clipped, quantized)

            if kl < best_kl:
                best_kl = kl
                best_threshold = self.bin_edges[i]

        return -best_threshold, best_threshold

import torch
import math
from transformers import AutoModelForCausalLM, AutoTokenizer

def compute_perplexity(model, tokenizer, text: str, device: str = "cuda"):
    """퍼플렉시티 계산"""
    encodings = tokenizer(text, return_tensors="pt")
    input_ids = encodings.input_ids.to(device)

    max_length = 1024
    stride = 512

    nlls = []
    prev_end_loc = 0

    for begin_loc in range(0, input_ids.size(1), stride):
        end_loc = min(begin_loc + max_length, input_ids.size(1))
        trg_len = end_loc - prev_end_loc

        input_ids_chunk = input_ids[:, begin_loc:end_loc]
        target_ids = input_ids_chunk.clone()
        target_ids[:, :-trg_len] = -100

        with torch.no_grad():
            outputs = model(input_ids_chunk, labels=target_ids)
            neg_log_likelihood = outputs.loss

        nlls.append(neg_log_likelihood)
        prev_end_loc = end_loc

        if end_loc == input_ids.size(1):
            break

    ppl = torch.exp(torch.stack(nlls).mean())
    return ppl.item()

# 모델별 PPL 비교 예시
# FP16: PPL ≈ 5.68
# INT8: PPL ≈ 5.71 (약 0.5% 증가)
# INT4 (GPTQ): PPL ≈ 5.89 (약 3.7% 증가)
# INT4 (naive): PPL ≈ 6.52 (약 14.8% 증가)

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

class FakeQuantize(nn.Module):
    """가짜 양자화 모듈"""

    def __init__(self, num_bits: int = 8, symmetric: bool = True):
        super().__init__()
        self.num_bits = num_bits
        self.symmetric = symmetric

        self.register_buffer('scale', torch.tensor(1.0))
        self.register_buffer('zero_point', torch.tensor(0))
        self.register_buffer('fake_quant_enabled', torch.tensor(1))

        if symmetric:
            self.q_min = -(2 ** (num_bits - 1))
            self.q_max = 2 ** (num_bits - 1) - 1
        else:
            self.q_min = 0
            self.q_max = 2 ** num_bits - 1

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        if self.fake_quant_enabled[0] == 0:
            return x

        # 스케일 업데이트 (이동 평균)
        if self.training:
            with torch.no_grad():
                if self.symmetric:
                    max_abs = x.abs().max()
                    new_scale = max_abs / self.q_max
                else:
                    new_scale = (x.max() - x.min()) / (self.q_max - self.q_min)

                # 지수 이동 평균으로 스케일 업데이트
                self.scale.copy_(0.9 * self.scale + 0.1 * new_scale)

        # 가짜 양자화: 양자화 후 역양자화
        x_scaled = x / self.scale
        x_clipped = torch.clamp(x_scaled, self.q_min, self.q_max)
        x_rounded = torch.round(x_clipped)
        x_dequant = x_rounded * self.scale

        return x_dequant


### 3.2 STE (Straight-Through Estimator)

```python
class STERound(torch.autograd.Function):
    """Straight-Through Estimator for round()"""

    @staticmethod
    def forward(ctx, x):
        return torch.round(x)

    @staticmethod
    def backward(ctx, grad_output):
        # 역전파 시 round()를 통과하여 gradient 전달 (항등함수로 근사)
        return grad_output

class STEClamp(torch.autograd.Function):
    """Straight-Through Estimator for clamp()"""

    @staticmethod
    def forward(ctx, x, min_val, max_val):
        ctx.save_for_backward(x)
        ctx.min_val = min_val
        ctx.max_val = max_val
        return torch.clamp(x, min_val, max_val)

    @staticmethod
    def backward(ctx, grad_output):
        x, = ctx.saved_tensors
        # clamp 범위 내에서만 gradient 전달
        grad = grad_output * ((x >= ctx.min_val) & (x <= ctx.max_val)).float()
        return grad, None, None

class QATLinear(nn.Module):
    """QAT를 적용한 Linear 레이어"""

    def __init__(self, in_features, out_features, num_bits=8):
        super().__init__()
        self.linear = nn.Linear(in_features, out_features)
        self.weight_fake_quant = FakeQuantize(num_bits=num_bits)
        self.act_fake_quant = FakeQuantize(num_bits=num_bits, symmetric=False)

    def forward(self, x):
        # 활성화 양자화
        x_q = self.act_fake_quant(x)
        # 가중치 양자화
        w_q = self.weight_fake_quant(self.linear.weight)
        # FP32 연산 (실제로는 INT8)
        return F.linear(x_q, w_q, self.linear.bias)

# QAT 학습 워크플로
import torch.optim as optim
from torch.quantization import prepare_qat, convert

def train_qat_model(model, train_loader, num_epochs=10):
    """QAT 학습 예시"""

    # QAT 준비
    model.qconfig = torch.quantization.get_default_qat_qconfig('fbgemm')
    model_prepared = prepare_qat(model.train())

    optimizer = optim.Adam(model_prepared.parameters(), lr=1e-5)

    for epoch in range(num_epochs):
        for batch in train_loader:
            inputs, labels = batch
            outputs = model_prepared(inputs)
            loss = F.cross_entropy(outputs, labels)

            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

    # INT8 모델로 변환
    model_prepared.eval()
    model_quantized = convert(model_prepared)

    return model_quantized

import torch
from torch.ao.quantization import (
    get_default_qconfig,
    get_default_qat_qconfig,
    prepare,
    prepare_qat,
    convert
)

# 정적 양자화 (PTQ)
def static_quantization_example():
    """정적 양자화 예시"""
    model = MyModel()
    model.eval()

    # 백엔드 설정 (fbgemm: x86, qnnpack: ARM)
    model.qconfig = get_default_qconfig('fbgemm')

    # 보정 준비
    model_prepared = prepare(model)

    # 보정 데이터로 통계 수집
    with torch.no_grad():
        for data in calibration_loader:
            model_prepared(data)

    # INT8 모델로 변환
    model_quantized = convert(model_prepared)

    return model_quantized

# 동적 양자화 (LSTM, Linear에 효과적)
def dynamic_quantization_example():
    """동적 양자화 예시"""
    model = MyModel()

    model_quantized = torch.quantization.quantize_dynamic(
        model,
        {nn.Linear, nn.LSTM},  # 양자화할 레이어 타입
        dtype=torch.qint8
    )

    return model_quantized

from torch.ao.quantization.quantize_fx import prepare_fx, convert_fx
from torch.ao.quantization import QConfigMapping

def fx_quantization_example(model, calibration_data):
    """FX Graph Mode 양자화"""
    model.eval()

    # QConfig 설정
    qconfig_mapping = QConfigMapping().set_global(
        get_default_qconfig('fbgemm')
    )

    # 예시 입력
    example_inputs = (torch.randn(1, 3, 224, 224),)

    # FX 그래프 기반 준비
    model_prepared = prepare_fx(
        model,
        qconfig_mapping,
        example_inputs
    )

    # 보정
    with torch.no_grad():
        for batch in calibration_data:
            model_prepared(batch)

    # 변환
    model_quantized = convert_fx(model_prepared)

    return model_quantized

argmin_Q ||WX - QX||_F^2

# GPTQ 핵심 알고리즘 구현 (단순화 버전)
import torch
import math

def gptq_quantize_weight(weight: torch.Tensor,
                          hessian: torch.Tensor,
                          num_bits: int = 4,
                          group_size: int = 128,
                          damp_percent: float = 0.01):
    """
    GPTQ 알고리즘으로 가중치 양자화

    Args:
        weight: [out_features, in_features] 가중치 행렬
        hessian: [in_features, in_features] 헤시안 행렬 (H = 2 * X @ X.T)
        num_bits: 양자화 비트 수
        group_size: 그룹 크기
        damp_percent: 헤시안 안정화를 위한 댐핑 비율
    """
    W = weight.clone().float()
    n_rows, n_cols = W.shape

    # 헤시안 댐핑 (수치 안정성)
    H = hessian.clone().float()
    dead_cols = torch.diag(H) == 0
    H[dead_cols, dead_cols] = 1
    W[:, dead_cols] = 0

    damp = damp_percent * H.diag().mean()
    H.diagonal().add_(damp)

    # 헤시안 역행렬 (Cholesky 분해 이용)
    H_inv = torch.linalg.cholesky(H)
    H_inv = torch.cholesky_inverse(H_inv)
    H_inv = torch.linalg.cholesky(H_inv, upper=True)

    Q = torch.zeros_like(W)
    Losses = torch.zeros_like(W)

    q_max = 2 ** (num_bits - 1) - 1

    for col_idx in range(n_cols):
        w_col = W[:, col_idx]  # 현재 컬럼의 가중치
        h_inv_diag = H_inv[col_idx, col_idx]  # 헤시안 역행렬의 대각 요소

        # 그룹별 스케일 계산
        if group_size != -1 and col_idx % group_size == 0:
            group_end = min(col_idx + group_size, n_cols)
            w_group = W[:, col_idx:group_end]
            max_abs = w_group.abs().max(dim=1)[0].unsqueeze(1)
            scale = max_abs / q_max
            scale = torch.clamp(scale, min=1e-8)

        # 양자화
        q_col = torch.clamp(torch.round(w_col / scale.squeeze()), -q_max, q_max)
        q_col = q_col * scale.squeeze()
        Q[:, col_idx] = q_col

        # 양자화 오차
        err = (w_col - q_col) / h_inv_diag
        Losses[:, col_idx] = err ** 2 / 2

        # 오차를 나머지 가중치에 전파 (핵심!)
        W[:, col_idx + 1:] -= err.unsqueeze(1) * H_inv[col_idx, col_idx + 1:].unsqueeze(0)

    return Q, Losses


def collect_hessian(model_layer, calibration_data, device='cuda'):
    """보정 데이터로 헤시안 수집"""
    hessians = {}

    def make_hook(name):
        def hook(module, input, output):
            inp = input[0].detach().float()
            if inp.dim() == 3:
                inp = inp.reshape(-1, inp.size(-1))

            if name not in hessians:
                hessians[name] = torch.zeros(inp.size(1), inp.size(1), device=device)

            hessians[name] += 2 * inp.T @ inp
        return hook

    handles = []
    for name, module in model_layer.named_modules():
        if isinstance(module, torch.nn.Linear):
            handles.append(module.register_forward_hook(make_hook(name)))

    with torch.no_grad():
        for batch in calibration_data:
            model_layer(batch.to(device))

    for h in handles:
        h.remove()

    return hessians

from auto_gptq import AutoGPTQForCausalLM, BaseQuantizeConfig
from transformers import AutoTokenizer
import torch

def quantize_with_gptq(
    model_name: str,
    output_dir: str,
    bits: int = 4,
    group_size: int = 128
):
    """AutoGPTQ로 모델 양자화"""

    tokenizer = AutoTokenizer.from_pretrained(model_name, use_fast=True)

    # 양자화 설정
    quantize_config = BaseQuantizeConfig(
        bits=bits,              # 4 또는 8
        group_size=group_size,  # 128 권장
        damp_percent=0.01,      # 헤시안 댐핑
        desc_act=False,         # 활성화 재정렬 (품질 향상, 속도 감소)
        sym=True,               # 대칭 양자화
        true_sequential=True    # 순차적 레이어 양자화
    )

    # 모델 로드
    model = AutoGPTQForCausalLM.from_pretrained(
        model_name,
        quantize_config=quantize_config
    )

    # 보정 데이터 준비
    from datasets import load_dataset
    dataset = load_dataset("wikitext", "wikitext-2-raw-v1", split="train")

    calibration_data = []
    for text in dataset["text"][:128]:
        if len(text.strip()) > 50:
            encoded = tokenizer(
                text.strip(),
                return_tensors="pt",
                max_length=2048,
                truncation=True
            )
            calibration_data.append(encoded["input_ids"].squeeze())

    # GPTQ 양자화 실행
    print(f"GPTQ {bits}bit 양자화 시작...")
    model.quantize(calibration_data)

    # 저장
    model.save_quantized(output_dir, use_safetensors=True)
    tokenizer.save_pretrained(output_dir)

    print(f"양자화 완료: {output_dir}")
    return model, tokenizer


def load_gptq_model(model_dir: str, device: str = "cuda"):
    """GPTQ 양자화 모델 로드"""

    model = AutoGPTQForCausalLM.from_quantized(
        model_dir,
        device=device,
        use_triton=False,       # Triton 커널 사용 여부
        disable_exllama=False,  # ExLlama 커널 사용 (속도 향상)
        inject_fused_attention=True,
        inject_fused_mlp=True
    )

    tokenizer = AutoTokenizer.from_pretrained(model_dir)

    return model, tokenizer

# 사용 예시
# model, tokenizer = quantize_with_gptq("meta-llama/Llama-2-7b-hf", "./llama2-7b-gptq-4bit")
# model, tokenizer = load_gptq_model("./llama2-7b-gptq-4bit")

from awq import AutoAWQForCausalLM
from transformers import AutoTokenizer

def quantize_with_awq(
    model_name: str,
    output_dir: str,
    bits: int = 4,
    group_size: int = 128
):
    """AutoAWQ로 모델 양자화"""

    tokenizer = AutoTokenizer.from_pretrained(
        model_name,
        trust_remote_code=True
    )

    model = AutoAWQForCausalLM.from_pretrained(
        model_name,
        low_cpu_mem_usage=True,
        use_cache=False
    )

    # AWQ 양자화 설정
    quant_config = {
        "zero_point": True,   # 비대칭 양자화
        "q_group_size": group_size,
        "w_bit": bits,
        "version": "GEMM"     # GEMM 또는 GEMV (작은 배치에 최적)
    }

    # 양자화 실행
    print(f"AWQ {bits}bit 양자화 시작...")
    model.quantize(tokenizer, quant_config=quant_config)

    # 저장
    model.save_quantized(output_dir)
    tokenizer.save_pretrained(output_dir)

    print(f"AWQ 양자화 완료: {output_dir}")
    return model

def load_awq_model(model_dir: str, device: str = "cuda"):
    """AWQ 양자화 모델 로드"""

    model = AutoAWQForCausalLM.from_quantized(
        model_dir,
        fuse_layers=True,       # 레이어 퓨전으로 속도 향상
        trust_remote_code=True,
        safetensors=True
    )

    tokenizer = AutoTokenizer.from_pretrained(model_dir)

    return model, tokenizer

# Hugging Face transformers와 통합
from transformers import AutoModelForCausalLM

def load_awq_with_transformers(model_dir: str):
    """transformers로 AWQ 모델 로드"""
    model = AutoModelForCausalLM.from_pretrained(
        model_dir,
        device_map="auto"
    )
    tokenizer = AutoTokenizer.from_pretrained(model_dir)
    return model, tokenizer

GGUF 파일 구조:
┌─────────────────────────────┐
│ 매직 넘버 (GGUF)            │
│ 버전                        │
│ 텐서 개수                   │
│ 메타데이터 KV 쌍            │
│  - 모델 아키텍처            │
│  - 컨텍스트 길이            │
│  - 어텐션 헤드 수           │
│  - 임베딩 차원              │
├─────────────────────────────┤
│ 텐서 인포 (이름, 타입, 형태) │
├─────────────────────────────┤
│ 텐서 데이터                 │
└─────────────────────────────┘

# llama.cpp 클론 및 빌드
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp

# CUDA 지원 빌드
cmake -B build -DGGML_CUDA=ON
cmake --build build --config Release -j $(nproc)

# CPU 전용 빌드
cmake -B build
cmake --build build --config Release -j $(nproc)

# HuggingFace 모델을 GGUF로 변환
python convert_hf_to_gguf.py \
    --model meta-llama/Llama-2-7b-hf \
    --outfile llama2-7b-f16.gguf \
    --outtype f16

# GGUF 양자화 (Q4_K_M)
./build/bin/llama-quantize \
    llama2-7b-f16.gguf \
    llama2-7b-q4_k_m.gguf \
    Q4_K_M

# 추론 실행
./build/bin/llama-cli \
    -m llama2-7b-q4_k_m.gguf \
    -p "The future of AI is" \
    -n 100 \
    --ctx-size 4096 \
    --threads 8 \
    --n-gpu-layers 35

from llama_cpp import Llama

# 모델 로드
llm = Llama(
    model_path="./llama2-7b-q4_k_m.gguf",
    n_ctx=4096,          # 컨텍스트 길이
    n_gpu_layers=35,     # GPU로 오프로드할 레이어 수 (-1: 전체)
    n_threads=8,         # CPU 스레드 수
    verbose=False
)

# 텍스트 생성
output = llm(
    "Once upon a time",
    max_tokens=200,
    temperature=0.7,
    top_p=0.9,
    stop=["</s>", "\n\n"]
)

print(output["choices"][0]["text"])

# 채팅 완성 형식
response = llm.create_chat_completion(
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is machine learning?"}
    ],
    max_tokens=500,
    temperature=0.7
)

print(response["choices"][0]["message"]["content"])

# 스트리밍 출력
for chunk in llm.create_chat_completion(
    messages=[{"role": "user", "content": "Tell me a joke"}],
    stream=True
):
    delta = chunk["choices"][0].get("delta", {})
    if "content" in delta:
        print(delta["content"], end="", flush=True)

from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

# INT8 모델 로드
model_8bit = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    load_in_8bit=True,
    device_map="auto"
)

# 메모리 사용량 확인
def print_model_size(model, label):
    """모델 메모리 사용량 출력"""
    total_params = sum(p.numel() for p in model.parameters())
    total_bytes = sum(
        p.numel() * p.element_size() for p in model.parameters()
    )
    print(f"{label}: {total_params/1e9:.2f}B params, {total_bytes/1e9:.2f} GB")

print_model_size(model_8bit, "INT8 모델")
# INT8 모델: 6.74B params, ~7.0 GB

import bitsandbytes as bnb
from transformers import BitsAndBytesConfig

# NF4 양자화 설정 (QLoRA)
bnb_config_nf4 = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",           # NF4 또는 FP4
    bnb_4bit_compute_dtype=torch.bfloat16, # 연산 시 데이터 타입
    bnb_4bit_use_double_quant=True,       # 이중 양자화 (양자화 상수도 양자화)
)

# FP4 양자화 설정
bnb_config_fp4 = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="fp4",
    bnb_4bit_compute_dtype=torch.float16,
)

# 모델 로드
model_4bit = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b-hf",
    quantization_config=bnb_config_nf4,
    device_map="auto"
)

print_model_size(model_4bit, "NF4 모델")
# NF4 모델: 6.74B params, ~4.0 GB (이중 양자화 포함)

# QLoRA 파인튜닝 설정
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training

model_4bit = prepare_model_for_kbit_training(model_4bit)

lora_config = LoraConfig(
    r=64,
    lora_alpha=16,
    target_modules=["q_proj", "v_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)

model_lora = get_peft_model(model_4bit, lora_config)
model_lora.print_trainable_parameters()
# trainable params: 4,194,304 || all params: 3,504,607,232 || trainable%: 0.1197

import numpy as np
import matplotlib.pyplot as plt

# NF4 양자화 포인트 시각화
def get_nf4_quantization_points():
    """NF4 16개 양자화 포인트"""
    # 정규 분포의 1/16 분위수
    nf4_points = []
    for i in range(16):
        quantile = (i + 0.5) / 16
        nf4_points.append(scipy.stats.norm.ppf(quantile))

    # 정규화
    max_val = max(abs(p) for p in nf4_points)
    nf4_points = [p / max_val for p in nf4_points]

    return nf4_points

# NF4: [-1.0, -0.6961, -0.5250, -0.3949, -0.2844, -0.1848, -0.0911, 0.0000,
#        0.0796, 0.1609, 0.2461, 0.3379, 0.4407, 0.5626, 0.7230, 1.0]

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

def analyze_activation_outliers(model, tokenizer, text: str, threshold: float = 100.0):
    """활성화 아웃라이어 분석"""

    activations = {}

    def make_hook(name):
        def hook(module, input, output):
            act = output.detach().float()
            max_val = act.abs().max().item()
            outlier_ratio = (act.abs() > threshold).float().mean().item()
            activations[name] = {
                "max": max_val,
                "outlier_ratio": outlier_ratio,
                "std": act.std().item()
            }
        return hook

    handles = []
    for name, module in model.named_modules():
        if isinstance(module, torch.nn.Linear):
            handles.append(module.register_forward_hook(make_hook(name)))

    input_ids = tokenizer(text, return_tensors="pt").input_ids.cuda()

    with torch.no_grad():
        model(input_ids)

    for h in handles:
        h.remove()

    # 아웃라이어가 많은 레이어 순으로 정렬
    sorted_acts = sorted(
        activations.items(),
        key=lambda x: x[1]["max"],
        reverse=True
    )

    print("아웃라이어가 큰 레이어 Top 10:")
    for name, stats in sorted_acts[:10]:
        print(f"  {name}: max={stats['max']:.1f}, outlier_ratio={stats['outlier_ratio']:.3%}")

    return activations

Y = (X * diag(s)^(-1)) * (diag(s) * W)
  = X_smooth * W_smooth

def smooth_quantize(
    model,
    calibration_samples,
    alpha: float = 0.5
):
    """
    SmoothQuant 적용

    Args:
        alpha: 마이그레이션 강도 (0=가중치만, 1=활성화만)
               권장값: 0.5 (균등 분배)
    """

    # 활성화 통계 수집
    act_scales = {}

    def collect_scales(name):
        def hook(module, input, output):
            inp = input[0].detach()
            if inp.dim() == 3:
                inp = inp.reshape(-1, inp.size(-1))

            channel_max = inp.abs().max(dim=0)[0]

            if name not in act_scales:
                act_scales[name] = channel_max
            else:
                act_scales[name] = torch.maximum(act_scales[name], channel_max)
        return hook

    handles = []
    for name, module in model.named_modules():
        if isinstance(module, torch.nn.Linear):
            handles.append(module.register_forward_hook(collect_scales(name)))

    with torch.no_grad():
        for sample in calibration_samples:
            model(**sample)

    for h in handles:
        h.remove()

    # 스케일 계산 및 적용
    for name, module in model.named_modules():
        if isinstance(module, torch.nn.Linear) and name in act_scales:
            act_scale = act_scales[name]
            weight_scale = module.weight.abs().max(dim=0)[0]

            # 마이그레이션 스케일 계산
            smooth_scale = (act_scale ** alpha) / (weight_scale ** (1 - alpha))
            smooth_scale = torch.clamp(smooth_scale, min=1e-5)

            # 가중치에 스케일 적용
            module.weight.data = module.weight.data / smooth_scale.unsqueeze(0)

            # 이전 레이어 (LayerNorm 등)의 출력 스케일에 역스케일 적용
            # (실제 구현에서는 이전 레이어를 찾아서 수정)

    return model, act_scales

import torch

def spqr_quantize(weight: torch.Tensor,
                   num_bits: int = 3,
                   outlier_threshold_percentile: float = 1.0):
    """
    SpQR 양자화 (단순화 버전)

    핵심: 상위 p% 아웃라이어를 FP16으로 저장, 나머지는 저비트 양자화
    """

    # 아웃라이어 임계값 계산
    threshold = torch.quantile(weight.abs(), 1 - outlier_threshold_percentile / 100)

    # 아웃라이어 마스크
    outlier_mask = weight.abs() > threshold

    # 아웃라이어 저장 (FP16)
    outlier_values = weight.clone()
    outlier_values[~outlier_mask] = 0

    # 나머지 양자화
    regular_weight = weight.clone()
    regular_weight[outlier_mask] = 0

    # Per-group 양자화 적용
    q_max = 2 ** (num_bits - 1) - 1
    group_size = 16

    rows, cols = regular_weight.shape
    regular_grouped = regular_weight.reshape(-1, group_size)

    max_abs = regular_grouped.abs().max(dim=1, keepdim=True)[0]
    scales = max_abs / q_max
    scales = torch.clamp(scales, min=1e-8)

    q = torch.clamp(torch.round(regular_grouped / scales), -q_max, q_max).to(torch.int8)
    regular_dequant = (scales * q.float()).reshape(rows, cols)

    # 최종 재구성
    reconstructed = regular_dequant + outlier_values

    error = (weight - reconstructed).abs().mean().item()

    # 메모리 사용량 계산
    outlier_memory = outlier_mask.sum().item() * 2  # FP16 = 2 bytes
    regular_memory = (~outlier_mask).sum().item() * (num_bits / 8)
    total_memory = outlier_memory + regular_memory
    original_memory = weight.numel() * weight.element_size()
    compression_ratio = original_memory / total_memory

    print(f"아웃라이어 비율: {outlier_mask.float().mean():.2%}")
    print(f"평균 재구성 오차: {error:.6f}")
    print(f"압축률: {compression_ratio:.2f}x")

    return q, scales, outlier_values, outlier_mask

import time
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import psutil
import GPUtil

def benchmark_quantization(model, tokenizer, device="cuda", num_runs=50):
    """양자화 모델 벤치마크"""

    prompt = "The history of artificial intelligence began"
    inputs = tokenizer(prompt, return_tensors="pt").to(device)

    # 메모리 사용량
    if device == "cuda":
        torch.cuda.synchronize()
        gpu = GPUtil.getGPUs()[0]
        memory_used_gb = gpu.memoryUsed / 1024
    else:
        memory_used_gb = psutil.virtual_memory().used / 1e9

    # 워밍업
    with torch.no_grad():
        for _ in range(5):
            outputs = model.generate(
                **inputs,
                max_new_tokens=50,
                do_sample=False
            )

    # 속도 측정
    if device == "cuda":
        torch.cuda.synchronize()
    start = time.time()

    with torch.no_grad():
        for _ in range(num_runs):
            outputs = model.generate(
                **inputs,
                max_new_tokens=50,
                do_sample=False
            )

    if device == "cuda":
        torch.cuda.synchronize()
    elapsed = time.time() - start

    avg_time = elapsed / num_runs
    tokens_per_second = 50 / avg_time

    return {
        "memory_gb": memory_used_gb,
        "avg_time_ms": avg_time * 1000,
        "tokens_per_second": tokens_per_second
    }

# 결과 예시 (A100 80GB 기준, Llama-2-7B)
benchmark_results = {
    "FP16": {"memory_gb": 13.5, "tokens_per_second": 52.3, "ppl": 5.68},
    "INT8 (bitsandbytes)": {"memory_gb": 7.8, "tokens_per_second": 38.1, "ppl": 5.71},
    "INT4 GPTQ": {"memory_gb": 4.5, "tokens_per_second": 65.2, "ppl": 5.89},
    "INT4 AWQ": {"memory_gb": 4.3, "tokens_per_second": 68.7, "ppl": 5.86},
    "Q4_K_M (GGUF)": {"memory_gb": 4.1, "tokens_per_second": 45.2, "ppl": 5.91},  # CPU
    "INT4 NF4": {"memory_gb": 4.0, "tokens_per_second": 31.5, "ppl": 5.94},
}

print("=" * 80)
print(f"{'방법':<25} {'메모리(GB)':<12} {'토큰/초':<12} {'PPL':<8}")
print("=" * 80)
for method, stats in benchmark_results.items():
    print(f"{method:<25} {stats['memory_gb']:<12.1f} {stats['tokens_per_second']:<12.1f} {stats['ppl']:<8.2f}")

def recommend_quantization(
    task: str,
    model_size_b: float,
    gpu_memory_gb: float,
    cpu_only: bool = False,
    fine_tuning_needed: bool = False
):
    """태스크와 환경에 따른 양자화 추천"""

    recommendations = []

    if cpu_only:
        recommendations.append({
            "method": "GGUF Q4_K_M",
            "reason": "CPU 추론에 최적화, llama.cpp 기반",
            "library": "llama-cpp-python"
        })
        return recommendations

    if fine_tuning_needed:
        recommendations.append({
            "method": "bitsandbytes NF4 + QLoRA",
            "reason": "파인튜닝 가능, 4GB 추가 메모리로 LoRA 어댑터 학습",
            "library": "bitsandbytes + peft"
        })
        return recommendations

    # 메모리 요구량 계산
    fp16_memory = model_size_b * 2  # FP16 = 2 bytes per param
    int8_memory = model_size_b * 1  # INT8 = 1 byte per param
    int4_memory = model_size_b * 0.5  # INT4 = 0.5 bytes per param

    if fp16_memory <= gpu_memory_gb * 0.8:
        recommendations.append({
            "method": "FP16 (기본)",
            "reason": "메모리 여유 있음, 최고 품질",
            "memory_gb": fp16_memory
        })

    if int8_memory <= gpu_memory_gb * 0.8:
        if task in ["chat", "completion", "summarization"]:
            recommendations.append({
                "method": "AWQ INT8",
                "reason": "품질과 속도의 최적 균형",
                "library": "autoawq",
                "memory_gb": int8_memory
            })

    if int4_memory <= gpu_memory_gb * 0.8:
        recommendations.append({
            "method": "AWQ INT4",
            "reason": "고속 추론, 우수한 품질",
            "library": "autoawq",
            "memory_gb": int4_memory
        })
        recommendations.append({
            "method": "GPTQ INT4",
            "reason": "최고의 INT4 품질, 느린 양자화",
            "library": "auto-gptq",
            "memory_gb": int4_memory
        })

    return recommendations

# 사용 예시
recommendations = recommend_quantization(
    task="chat",
    model_size_b=7.0,
    gpu_memory_gb=16.0,
    fine_tuning_needed=False
)

for rec in recommendations:
    print(f"\n방법: {rec['method']}")
    print(f"이유: {rec['reason']}")
    if 'library' in rec:
        print(f"라이브러리: {rec['library']}")
    if 'memory_gb' in rec:
        print(f"예상 메모리: {rec['memory_gb']:.1f} GB")

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from auto_gptq import AutoGPTQForCausalLM, BaseQuantizeConfig
from awq import AutoAWQForCausalLM
import json
import os

class QuantizationPipeline:
    """통합 양자화 파이프라인"""

    def __init__(self, model_name: str, output_base_dir: str):
        self.model_name = model_name
        self.output_base_dir = output_base_dir
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)

        os.makedirs(output_base_dir, exist_ok=True)

    def quantize_gptq(self, bits: int = 4, group_size: int = 128):
        """GPTQ 양자화"""
        output_dir = os.path.join(self.output_base_dir, f"gptq-{bits}bit")

        config = BaseQuantizeConfig(
            bits=bits,
            group_size=group_size,
            sym=True,
            desc_act=False
        )

        model = AutoGPTQForCausalLM.from_pretrained(
            self.model_name,
            quantize_config=config
        )

        # 보정 데이터
        calibration_data = self._prepare_calibration_data()

        model.quantize(calibration_data)
        model.save_quantized(output_dir)
        self.tokenizer.save_pretrained(output_dir)

        print(f"GPTQ {bits}bit 저장: {output_dir}")
        return output_dir

    def quantize_awq(self, bits: int = 4, group_size: int = 128):
        """AWQ 양자화"""
        output_dir = os.path.join(self.output_base_dir, f"awq-{bits}bit")

        model = AutoAWQForCausalLM.from_pretrained(
            self.model_name,
            low_cpu_mem_usage=True
        )

        quant_config = {
            "zero_point": True,
            "q_group_size": group_size,
            "w_bit": bits,
            "version": "GEMM"
        }

        model.quantize(self.tokenizer, quant_config=quant_config)
        model.save_quantized(output_dir)
        self.tokenizer.save_pretrained(output_dir)

        print(f"AWQ {bits}bit 저장: {output_dir}")
        return output_dir

    def _prepare_calibration_data(self, num_samples: int = 128):
        """보정 데이터 준비"""
        from datasets import load_dataset

        dataset = load_dataset("wikitext", "wikitext-2-raw-v1", split="train")

        data = []
        for text in dataset["text"]:
            if len(text.strip()) > 50:
                encoded = self.tokenizer(
                    text.strip(),
                    return_tensors="pt",
                    max_length=2048,
                    truncation=True
                )
                data.append(encoded["input_ids"].squeeze())
                if len(data) >= num_samples:
                    break

        return data

    def evaluate_all(self, test_text: str = None):
        """모든 양자화 모델 평가"""
        if test_text is None:
            from datasets import load_dataset
            dataset = load_dataset("wikitext", "wikitext-2-raw-v1", split="test")
            test_text = " ".join(dataset["text"][:10])

        results = {}

        # FP16 기준선
        model_fp16 = AutoModelForCausalLM.from_pretrained(
            self.model_name,
            torch_dtype=torch.float16,
            device_map="auto"
        )

        # PPL 계산
        from transformers import pipeline

        # 각 모델별 평가 결과 출력
        print("\n=== 양자화 평가 결과 ===")
        print(f"모델: {self.model_name}")
        print(f"{'방법':<20} {'PPL':<10} {'메모리(GB)':<12}")
        print("-" * 42)

        return results


# 전체 파이프라인 실행
pipeline = QuantizationPipeline(
    model_name="meta-llama/Llama-2-7b-hf",
    output_base_dir="./quantized_models"
)

# GPTQ 4bit 양자화
gptq_dir = pipeline.quantize_gptq(bits=4)

# AWQ 4bit 양자화
awq_dir = pipeline.quantize_awq(bits=4)