1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

graph TD
    A[FP16 원본 모델] --> B[양자화 과정 INT4/NF4]
    B --> C[Attention Layer 오차 누적]
    B --> D[FFN Layer 오차 누적]
    B --> E[Embedding Layer 오차]
    C --> F[Context Understanding 저하]
    D --> G[지식 검색 능력 저하]
    E --> H[토큰 표현 왜곡]
    F --> I[Hallucination 발생]
    G --> I
    H --> J[Tool Calling 실패]
    I --> J

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

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

# 보정 데이터셋 로드
# Tool calling 데이터 포함 여부가 핵심
calibration_data = load_dataset("glaiveai/glaive-function-calling-v2", split="train[:512]")

tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2.5-72B-Instruct")

# 모델의 실제 사용 시나리오를 반영한 토큰 분포 확인
def analyze_token_distribution(dataset, tokenizer, num_samples=1000):
    token_counts = []
    special_token_counts = {}
    
    for example in dataset.select(range(min(num_samples, len(dataset)))):
        text = example.get("chatml", example.get("text", ""))
        tokens = tokenizer.encode(text)
        token_counts.append(len(tokens))
        
        # 특수 토큰 (도구 호출 관련) 빈도 분석
        for token_id in tokens:
            token_str = tokenizer.decode([token_id])
            if token_str in ["<|tool_calls|>", "<|function_call|>", "<|arguments|>"]:
                special_token_counts[token_str] = special_token_counts.get(token_str, 0) + 1
    
    return {
        "avg_tokens": sum(token_counts) / len(token_counts),
        "special_token_dist": special_token_counts,
        "percentiles": {
            "p50": sorted(token_counts)[len(token_counts)//2],
            "p95": sorted(token_counts)[int(len(token_counts)*0.95)]
        }
    }

# 분석 실행
analysis = analyze_token_distribution(calalibration_data, tokenizer)
print(f"Average tokens per sample: {analysis['avg_tokens']:.1f}")
print(f"Special token distribution: {analysis['special_token_dist']}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

import torch
import numpy as np
from typing import Dict, List

def layer_sensitivity_analysis(
    model: AutoModelForCausalLM,
    calibration_tokens: torch.Tensor,
    num_samples: int = 128
) -> Dict[str, float]:
    """
    각 레이어의 양자화 민감도를 분석합니다.
    낮을수록 양자화에 민감하므로 높은 정밀도가 필요합니다.
    """
    model.eval()
    sensitivities = {}
    
    # 원본 출력 확보
    with torch.no_grad():
        original_outputs = model(calibration_tokens[:num_samples])
        original_logits = original_outputs.logits
    
    # 각 레이어별로 일시적으로 양자화 후 출력 차이 측정
    for name, module in model.named_modules():
        if isinstance(module, torch.nn.Linear):
            # 원본 가중치 백업
            original_weight = module.weight.data.clone()
            
            # 간단한 min-max 양자화 적용 (4-bit)
            w_min = original_weight.min(dim=1, keepdim=True)[0]
            w_max = original_weight.max(dim=1, keepdim=True)[0]
            scale = (w_max - w_min) / 15  # 4-bit: 16 levels
            zero_point = torch.round(-w_min / scale)
            
            # 양자화 -> 역양자화
            quantized = torch.clamp(
                torch.round(original_weight / scale) + zero_point, 
                0, 15
            )
            dequantized = (quantized - zero_point) * scale
            module.weight.data = dequantized
            
            # 출력 차이 측정 (KL Divergence)
            with torch.no_grad():
                quantized_outputs = model(calibration_tokens[:num_samples])
                quantized_logits = quantized_outputs.logits
            
            # KL Divergence 계산
            kl_div = torch.nn.functional.kl_div(
                torch.nn.functional.log_softmax(quantized_logits, dim=-1),
                torch.nn.functional.softmax(original_logits, dim=-1),
                reduction='batchmean'
            )
            
            sensitivities[name] = kl_div.item()
            
            # 원본 가중치 복원
            module.weight.data = original_weight
    
    return sensitivities

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 실행 예시
# calibration_tokens = tokenizer(..., return_tensors="pt")["input_ids"]
# sensitivities = layer_sensitivity_analysis(model, calibration_tokens)

# 민감도 기반 레이어 분류
def classify_layers_by_sensitivity(sensitivities: Dict[str, float], 
                                     threshold_high: float = 0.1) -> Dict[str, List[str]]:
    """
    민감도에 따라 레이어를 분류합니다.
    """
    classification = {
        "keep_fp16": [],    # 높은 정밀도 필요
        "quantize_int4": [] # 4-bit 양자화 가능
    }
    
    for name, sensitivity in sensitivities.items():
        if sensitivity > threshold_high:
            classification["keep_fp16"].append(name)
        else:
            classification["quantize_int4"].append(name)
    
    return classification

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# mlx-lm 라이브러리를 사용한 안전한 변환 예시
# 주의: 실제 실행은 Apple Silicon 환경에서만 가능합니다

import mlx.core as mx
import mlx.nn as nn
from mlx_lm import load, generate
from mlx_lm.utils import quantize_model

def safe_mlx_quantization(
    model_path: str,
    output_path: str,
    sensitive_layers: list,
    default_bits: int = 4,
    sensitive_bits: int = 8
):
    """
    민감도 분석 결과를 반영한 MLX 양자화
    """
    # 모델 로드
    model, tokenizer = load(model_path)
    
    # 레이어별 양자화 비트 수 설정
    def get_quantization_bits(layer_name: str) -> int:
        for sensitive_layer in sensitive_layers:
            if sensitive_layer in layer_name:
                return sensitive_bits
        return default_bits
    
    # 혼합 정밀도 양자화 적용
    # MLX의 groupwise quantization 사용
    config = {
        "group_size": 64,  # 작을수록 정밀도 향상 but 메모