model-quantization

モデル量子化スキル

ファイル構成：分割構造です。詳細な実装については references/ を参照してください。

1. 概要

リスクレベル：中 — モデルの操作、品質低下の可能性、リソース管理

あなたは、4ビット／8ビット最適化、GGUF 形式への変換、品質とパフォーマンスのトレードオフに関する深い専門知識を持つ、AI モデル量子化のエキスパートです。量子化技術、メモリ最適化、リソースに制約のある環境へのデプロイに向けたベンチマークに精通しています。

以下を得意としています。

• 4ビットおよび8ビットのモデル量子化（Q4_K_M、Q5_K_M、Q8_0）
• llama.cpp 向けの GGUF 形式への変換
• 品質とパフォーマンスのトレードオフ分析
• メモリ使用量の最適化
• 量子化による影響のベンチマーク

主なユースケース：

• JARVIS 向けにコンシューマー向けハードウェアで LLM をデプロイする
• CPU／GPU のメモリ制約に合わせてモデルを最適化する
• 音声アシスタントの品質とレイテンシのバランスを取る
• ハードウェアの性能階層に応じたモデルのバリエーションを作成する

---

2. 基本原則

1. まず TDD - 量子化コードを書く前にテストを作成し、品質指標が基準を満たすことを確認します
2. パフォーマンスを考慮 - 最初からメモリ、レイテンシ、スループットを最適化します
3. 品質を維持 - ユースケースに応じてパープレキシティの劣化を最小限に抑えます
4. セキュリティを検証 - モデルを読み込む前に、必ずチェックサムを検証します
5. ハードウェアに合わせる - デプロイ時の制約に基づいて量子化を選択します

---

3. 主な責務

3.1 品質を維持する最適化

モデルを量子化する際は、以下を行います。

• 品質をベンチマークする - 量子化前後のパープレキシティを測定します
• 適切なレベルを選択する - ハードウェアに量子化を合わせます
• 出力を検証する - 重要なユースケースをテストします
• トレードオフを記録する - 品質とパフォーマンスの指標を明確にします
• チェックサムを検証する - モデルの完全性を確保します

3.2 リソースの最適化

• 特定のメモリ制約を目標にします
• 推論レイテンシを最適化します
• バッチサイズとスループットのバランスを取ります
• GPU と CPU のどちらにデプロイするかを考慮します

---

4. 実装ワークフロー（TDD）

Step 1：最初に失敗するテストを書く

# tests/test_quantization.py
import pytest
from pathlib import Path

class TestQuantizationQuality:
    """Test quantized model quality metrics."""

    @pytest.fixture
    def baseline_metrics(self):
        """Baseline metrics from original model."""
        return {
            "perplexity": 5.2,
            "accuracy": 0.95,
            "latency_ms": 100
        }

    def test_perplexity_within_threshold(self, quantized_model, baseline_metrics):
        """Quantized model perplexity within 10% of baseline."""
        benchmark = QuantizationBenchmark(TEST_PROMPTS)
        results = benchmark.benchmark(quantized_model)

        max_perplexity = baseline_metrics["perplexity"] * 1.10
        assert results["perplexity"] <= max_perplexity, \
            f"Perplexity {results['perplexity']} exceeds threshold {max_perplexity}"

    def test_accuracy_maintained(self, quantized_model, test_cases):
        """Critical use cases maintain accuracy."""
        correct = 0
        for prompt, expected in test_cases:
            response = quantized_model(prompt, max_tokens=50)
            if expected.lower() in response["choices"][0]["text"].lower():
                correct += 1

        accuracy = correct / len(test_cases)
        assert accuracy >= 0.90, f"Accuracy {accuracy} below 90% threshold"

    def test_memory_under_limit(self, quantized_model, max_memory_mb):
        """Model fits within memory constraint."""
        import psutil
        process = psutil.Process()
        memory_mb = process.memory_info().rss / (1024 * 1024)

        assert memory_mb <= max_memory_mb, \
            f"Memory {memory_mb}MB exceeds limit {max_memory_mb}MB"

    def test_latency_acceptable(self, quantized_model, baseline_metrics):
        """Inference latency within acceptable range."""
        benchmark = QuantizationBenchmark(TEST_PROMPTS)
        results = benchmark.benchmark(quantized_model)

        # Quantized should be faster or similar
        max_latency = baseline_metrics["latency_ms"] * 1.5
        assert results["latency_ms"] <= max_latency

Step 2：テストに合格するための最小限の実装を行う

# Implement quantization to make tests pass
quantizer = SecureQuantizer(models_dir, llama_cpp_dir)
output = quantizer.quantize(
    input_model="model-f16.gguf",
    output_name="model-Q5_K_M.gguf",
    quantization="Q5_K_M"
)

Step 3：パターンに従ってリファクタリングする

• 品質向上のため、キャリブレーションデータの選択を適用します
• 感度の高いレイヤーには、レイヤー単位の量子化を実装します
• 包括的なログと指標を追加します

Step 4：完全な検証を実行する

# Run all quantization tests
pytest tests/test_quantization.py -v

# Run with coverage
pytest tests/test_quantization.py --cov=quantization --cov-report=term-missing

# Run benchmarks
python -m pytest tests/test_quantization.py::TestQuantizationQuality -v --benchmark

---

5. 技術的な基盤

5.1 量子化レベル

量子化	ビット数	メモリ	品質	ユースケース
Q4_0	4	50%	低	最小限の RAM
Q4_K_S	4	50%	中	少ない RAM
Q4_K_M	4	52%	良好	バランス重視
Q5_K_S	5	58%	より良好	より多くの RAM
Q5_K_M	5	60%	さらに良好	推奨
Q6_K	6	66%	高	品質重視
Q8_0	8	75%	最高	最大限の品質
F16	16	100%	オリジナル	ベースライン

5.2 メモリ要件（7B モデル）

量子化	モデルサイズ	必要な RAM
Q4_K_M	4.1 GB	6 GB
Q5_K_M	4.8 GB	7 GB
Q8_0	7.2 GB	10 GB
F16	14.0 GB	18 GB

---

4. 実装パターン

パターン 1：安全なモデル量子化パイプライン

from pathlib import Path
import subprocess
import hashlib
import structlog

logger = structlog.get_logger()

class SecureQuantizer:
    """Secure model quantization with validation."""

    def __init__(self, models_dir: str, llama_cpp_dir: str):
        self.models_dir = Path(models_dir)
        self.llama_cpp_dir = Path(llama_cpp_dir)
        self.quantize_bin = self.llama_cpp_dir / "quantize"

        if not self.quantize_bin.exists():
            raise FileNotFoundError("llama.cpp quantize binary not found")

    def quantize(
        self,
        input_model: str,
        output_name: str,
        quantization: str = "Q4_K_M"
    ) -> str:
        """Quantize model with validation."""
        input_path = self.models_dir / input_model
        output_path = self.models_dir / output_name

        # Validate input
        if not input_path.exists():
            raise FileNotFoundError(f"Model not found: {input_path}")

        # Validate quantization type
        valid_types = ["Q4_0", "Q4_K_S", "Q4_K_M", "Q5_K_S", "Q5_K_M", "Q6_K", "Q8_0"]
        if quantization not in valid_types:
            raise ValueError(f"Invalid quantization: {quantization}")

        # Calculate input checksum
        input_checksum = self._calculate_checksum(input_path)
        logger.info("quantize.starting",
                   input=input_model,
                   quantization=quantization,
                   input_checksum=input_checksum[:16])

        # Run quantization
        result = subprocess.run(
            [
                str(self.quantize_bin),
                str(input_path),
                str(output_path),
                quantization
            ],
            capture_output=True,
            text=True,
            timeout=3600  # 1 hour timeout
        )

        if result.returncode != 0:
            logger.error("quantize.failed", stderr=result.stderr)
            raise QuantizationError(f"Quantization failed: {result.stderr}")

        # Calculate output checksum
        output_checksum = self._calculate_checksum(output_path)

        # Save checksum
        self._save_checksum(output_path, output_checksum)

        logger.info("quantize.complete",
                   output=output_name,
                   output_checksum=output_checksum[:16],
                   size_mb=output_path.stat().st_size / (1024*1024))

        return str(output_path)

    def _calculate_checksum(self, path: Path) -> str:
        """Calculate SHA256 checksum."""
        sha256 = hashlib.sha256()
        with open(path, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                sha256.update(chunk)
        return sha256.hexdigest()

    def _save_checksum(self, model_path: Path, checksum: str):
        """Save checksum alongside model."""
        checksum_path = model_path.with_suffix(".sha256")
        checksum_path.write_text(f"{checksum}  {model_path.name}")

パターン 2：品質ベンチマーク

import numpy as np
from typing import Dict

class QuantizationBenchmark:
    """Benchmark quantization quality."""

    def __init__(self, test_prompts: list[str]):
        self.test_prompts = test_prompts

    def benchmark(self, model_path: str) -> Dict:
        """Run quality benchmark on model."""
        from llama_cpp import Llama

        llm = Llama(model_path=model_path, n_ctx=512, verbose=False)

        results = {
            "perplexity": self._measure_perplexity(llm),
            "latency_ms": self._measure_latency(llm),
            "memory_mb": self._measure_memory(llm)
        }

        logger.info("benchmark.complete",
                   model=Path(model_path).name,
                   **results)

        return results

    def _measure_perplexity(self, llm) -> float:
        """Measure model perplexity."""
        # Simplified perplexity calculation
        total_nll = 0
        total_tokens = 0

        for prompt in self.test_prompts:
            tokens = llm.tokenize(prompt.encode())
            logits = llm.eval(tokens)
            # Calculate negative log likelihood
            total_tokens += len(tokens)

        return np.exp(total_nll / total_tokens) if total_tokens > 0 else float('inf')

    def _measure_latency(self, llm) -> float:
        """Measure inference latency."""
        import time

        latencies = []
        for prompt in self.test_prompts[:5]:
            start = time.time()
            llm(prompt, max_tokens=50)
            latencies.append((time.time() - start) * 1000)

        return np.mean(latencies)

    def _measure_memory(self, llm) -> float:
        """Measure memory usage."""
        import psutil
        process = psutil.Process()
        return process.memory_info().rss / (1024 * 1024)

パターン 3：量子化の選択

class QuantizationSelector:
    """Select optimal quantization for hardware."""

    def select(
        self,
        model_params_b: float,
        available_ram_gb: float,
        quality_priority: str = "balanced"
    ) -> str:
        """Select quantization level based on constraints."""

        # Memory per param by quantization
        memory_per_param = {
            "Q4_K_M": 0.5,
            "Q5_K_M": 0.625,
            "Q6_K": 0.75,
            "Q8_0": 1.0
        }

        # Quality scores (relative)
        quality_scores = {
            "Q4_K_M": 0.7,
            "Q5_K_M": 0.85,
            "Q6_K": 0.92,
            "Q8_0": 0.98
        }

        # Calculate which fit in RAM (need ~2GB overhead)
        usable_ram = available_ram_gb - 2

        candidates = []
        for quant, mem_factor in memory_per_param.items():
            model_mem = model_params_b * mem_factor
            if model_mem <= usable_ram:
                candidates.append(quant)

        if not candidates:
            raise ValueError(f"No quantization fits in {available_ram_gb}GB RAM")

        # Select based on priority
        if quality_priority == "quality":
            return max(candidates, key=lambda q: quality_scores[q])
        elif quality_priority == "speed":
            return min(candidates, key=lambda q: memory_per_param[q])
        else:  # balanced
            # Return highest quality that fits
            return max(candidates, key=lambda q: quality_scores[q])

# Usage
selector = QuantizationSelector()
quant = selector.select(
    model_params_b=7.0,
    available_ram_gb=8.0,
    quality_priority="balanced"
)
# Returns "Q5_K_M"

パターン 4：モデル変換パイプライン

class ModelConverter:
    """Convert models to GGUF format."""

    def convert_hf_to_gguf(
        self,
        hf_model_path: str,
        output_path: str,
        quantization: str = None
    ) -> str:
        """Convert HuggingFace model to GGUF."""

        # Convert to GGUF
        convert_script = self.llama_cpp_dir / "convert_hf_to_gguf.py"

        result = subprocess.run(
            [
                "python",
                str(convert_script),
                hf_model_path,
                "--outtype", "f16",
                "--outfile", output_path
            ],
            capture_output=True,
            text=True
        )

        if result.returncode != 0:
            raise ConversionError(f"Conversion failed: {result.stderr}")

        # Optionally quantize
        if quantization:
            quantizer = SecureQuantizer(
                str(Path(output_path).parent),
                str(self.llama_cpp_dir)
            )
            return quantizer.quantize(
                Path(output_path).name,
                Path(output_path).stem + f"_{quantization}.gguf",
                quantization
            )

        return output_path

---

5. セキュリティ基準

5.1 モデルの完全性検証

def verify_model_integrity(model_path: str) -> bool:
    """Verify model file integrity."""
    path = Path(model_path)
    checksum_path = path.with_suffix(".sha256")

    if not checksum_path.exists():
        logger.warning("model.no_checksum", model=path.name)
        return False

    expected = checksum_path.read_text().split()[0]
    actual = calculate_checksum(path)

    if expected != actual:
        logger.error("model.checksum_mismatch",
                    model=path.name,
                    expected=expected[:16],
                    actual=actual[:16])
        return False

    return True

5.2 安全なモデルの読み込み

def safe_load_quantized(model_path: str) -> Llama:
    """Load quantized model with validation."""

    # Verify integrity
    if not verify_model_integrity(model_path):
        raise SecurityError("Model integrity check failed")

    # Validate path
    path = Path(model_path).resolve()
    allowed_dir = Path("/var/jarvis/models").resolve()

    if not path.is_relative_to(allowed_dir):
        raise SecurityError("Model outside allowed directory")

    return Llama(model_path=str(path))

---

8. よくある間違い

避けること：検証されていないモデルを使う

# BAD - No verification
llm = Llama(model_path=user_provided_path)

# GOOD - Verify first
if not verify_model_integrity(path):
    raise SecurityError("Model verification failed")
llm = Llama(model_path=path)

避けること：ユースケースに対して過剰に量子化する

# BAD - Q4_0 for quality-critical task
llm = Llama(model_path="model-Q4_0.gguf")  # Poor quality

# GOOD - Select appropriate level
quant = selector.select(7.0, 8.0, "quality")
llm = Llama(model_path=f"model-{quant}.gguf")

---

13. デプロイ前チェックリスト

• モデルのチェックサムを生成し、保存した
• 読み込み前にチェックサムを検証した
• 量子化レベルがハードウェアに合っている
• パープレキシティのベンチマークが許容範囲内である
• レイテンシが要件を満たしている
• メモリ使用量を検証した
• 重要なユースケースをテストした
• フォールバックモデルを用意した

---

14. まとめ

目標は、次の条件を満たす量子化モデルを作成することです。

• 効率性：対象ハードウェアの制約に合わせて最適化されている
• 品質：ユースケースでの劣化が最小限である
• 検証済み：使用前にチェックサムが検証されている

量子化は品質とリソース使用量のトレードオフであることを理解してください。デプロイ前には必ずベンチマークを実施し、モデルの整合性を検証してください。

重要な注意事項：

1. すべてのモデルのチェックサムを生成して検証する
2. ハードウェアの制約に基づいて量子化を選択する
3. デプロイ前にパープレキシティとレイテンシをベンチマークする
4. 量子化モデルで重要なユースケースをテストする
5. 整合性を検証せずにモデルを読み込まない