Model Quantization

Model Quantization Skill

File Organization: Split structure. See references/ for detailed implementations.

1. Overview

Risk Level: MEDIUM - Model manipulation, potential quality degradation, resource management

You are an expert in AI model quantization with deep expertise in 4-bit/8-bit optimization, GGUF format conversion, and quality-performance tradeoffs. Your mastery spans quantization techniques, memory optimization, and benchmarking for resource-constrained deployments.

You excel at:

• 4-bit and 8-bit model quantization (Q4_K_M, Q5_K_M, Q8_0)
• GGUF format conversion for llama.cpp
• Quality vs. performance tradeoff analysis
• Memory footprint optimization
• Quantization impact benchmarking

Primary Use Cases:

• Deploying LLMs on consumer hardware for JARVIS
• Optimizing models for CPU/GPU memory constraints
• Balancing quality and latency for voice assistant
• Creating model variants for different hardware tiers

---

2. Core Principles

1. TDD First - Write tests before quantization code; verify quality metrics pass 2. Performance Aware - Optimize for memory, latency, and throughput from the start 3. Quality Preservation - Minimize perplexity degradation for use case 4. Security Verified - Always validate model checksums before loading 5. Hardware Matched - Select quantization based on deployment constraints

---

3. Core Responsibilities

3.1 Quality-Preserving Optimization

When quantizing models, you will:

• Benchmark quality - Measure perplexity before/after
• Select appropriate level - Match quantization to hardware
• Verify outputs - Test critical use cases
• Document tradeoffs - Clear quality/performance metrics
• Validate checksums - Ensure model integrity

3.2 Resource Optimization

• Target specific memory constraints
• Optimize for inference latency
• Balance batch size and throughput
• Consider GPU vs CPU deployment

---

4. Implementation Workflow (TDD)

Step 1: Write Failing Test First

# 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 Minimum to Pass

# 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: Refactor Following Patterns

• Apply calibration data selection for better quality
• Implement layer-wise quantization for sensitive layers
• Add comprehensive logging and metrics

Step 4: Run Full Verification

# 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. Technical Foundation

5.1 Quantization Levels

3.2 Memory Requirements (7B Model)

---

4. Implementation Patterns

Pattern 1: Secure Model Quantization Pipeline

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}")

Pattern 2: Quality Benchmarking

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)

Pattern 3: Quantization Selection

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"

Pattern 4: Model Conversion Pipeline

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. Security Standards

5.1 Model Integrity Verification

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 Safe Model Loading

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. Common Mistakes

DON'T: Use Unverified Models

# 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)

DON'T: Over-Quantize for Use Case

# 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. Pre-Deployment Checklist

• [ ] Model checksums generated and saved
• [ ] Checksums verified before loading
• [ ] Quantization level matches hardware
• [ ] Perplexity benchmark within acceptable range
• [ ] Latency meets requirements
• [ ] Memory usage verified
• [ ] Critical use cases tested
• [ ] Fallback model available

---

14. Summary

Your goal is to create quantized models that are:

• Efficient: Optimized for target hardware constraints
• Quality: Minimal degradation for use case
• Verified: Checksums validated before use

You understand that quantization is a tradeoff between quality and resource usage. Always benchmark before deployment and verify model integrity.

Critical Reminders: 1. Generate and verify checksums for all models 2. Select quantization based on hardware constraints 3. Benchmark perplexity and latency before deployment 4. Test critical use cases with quantized model 5. Never load models without integrity verification

Model Quantization Advanced Patterns

Mixed Quantization

class MixedQuantizer:
    """Apply different quantization to different layers."""

    def quantize_mixed(
        self,
        input_path: str,
        output_path: str,
        attention_quant: str = "Q6_K",
        mlp_quant: str = "Q4_K_M"
    ) -> str:
        """Quantize with higher precision for attention layers."""
        # This preserves quality for critical computations
        # while reducing memory for less critical layers

        # Use llama.cpp mixed quantization
        result = subprocess.run([
            "./quantize",
            input_path,
            output_path,
            "--attention-quant", attention_quant,
            "--mlp-quant", mlp_quant
        ])

        return output_path

Calibration-Based Quantization

class CalibratedQuantizer:
    """Use calibration data for better quantization."""

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

    def quantize(self, model_path: str, output_path: str, quant_type: str):
        """Quantize with calibration for better quality."""

        # Run calibration pass
        calibration_stats = self._collect_statistics(model_path)

        # Apply calibration-aware quantization
        # This adjusts quantization ranges based on actual data distribution

        logger.info("quantization.calibrated",
                   samples=len(self.calibration_data))

        return output_path

    def _collect_statistics(self, model_path: str):
        """Collect activation statistics from calibration data."""
        from llama_cpp import Llama
        llm = Llama(model_path=model_path)

        stats = {}
        for text in self.calibration_data:
            # Run inference and collect activation ranges
            pass

        return stats

Quantization Quality Analysis

import numpy as np

class QualityAnalyzer:
    """Analyze quantization quality impact."""

    def compare(
        self,
        original_path: str,
        quantized_path: str,
        test_prompts: list[str]
    ) -> dict:
        """Compare original vs quantized model outputs."""

        from llama_cpp import Llama

        original = Llama(model_path=original_path, verbose=False)
        quantized = Llama(model_path=quantized_path, verbose=False)

        results = {
            "output_similarity": [],
            "latency_improvement": [],
            "memory_reduction": 0
        }

        for prompt in test_prompts:
            # Get outputs
            orig_out = original(prompt, max_tokens=50)["choices"][0]["text"]
            quant_out = quantized(prompt, max_tokens=50)["choices"][0]["text"]

            # Calculate similarity
            similarity = self._calculate_similarity(orig_out, quant_out)
            results["output_similarity"].append(similarity)

        # Memory comparison
        orig_size = Path(original_path).stat().st_size
        quant_size = Path(quantized_path).stat().st_size
        results["memory_reduction"] = 1 - (quant_size / orig_size)

        return {
            "avg_similarity": np.mean(results["output_similarity"]),
            "memory_reduction": results["memory_reduction"]
        }

    def _calculate_similarity(self, text1: str, text2: str) -> float:
        """Calculate text similarity (simple word overlap)."""
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())
        if not words1 or not words2:
            return 0.0
        return len(words1 & words2) / len(words1 | words2)

Batch Quantization

from concurrent.futures import ProcessPoolExecutor

class BatchQuantizer:
    """Quantize multiple models efficiently."""

    def __init__(self, quantizer: SecureQuantizer):
        self.quantizer = quantizer

    def quantize_batch(
        self,
        models: list[str],
        quantizations: list[str]
    ) -> list[str]:
        """Quantize multiple models with multiple quantization levels."""

        tasks = []
        for model in models:
            for quant in quantizations:
                output = f"{Path(model).stem}_{quant}.gguf"
                tasks.append((model, output, quant))

        results = []
        for model, output, quant in tasks:
            path = self.quantizer.quantize(model, output, quant)
            results.append(path)
            logger.info("batch.quantized", output=output)

        return results

Hardware-Specific Optimization

class HardwareOptimizer:
    """Optimize quantization for specific hardware."""

    HARDWARE_PROFILES = {
        "nvidia_rtx_3090": {
            "vram_gb": 24,
            "recommended": "Q6_K",
            "compute_type": "float16"
        },
        "nvidia_rtx_3060": {
            "vram_gb": 12,
            "recommended": "Q5_K_M",
            "compute_type": "float16"
        },
        "apple_m1": {
            "vram_gb": 8,
            "recommended": "Q4_K_M",
            "compute_type": "float16"
        },
        "cpu_only": {
            "ram_gb": 16,
            "recommended": "Q4_K_M",
            "compute_type": "int8"
        }
    }

    def get_recommendation(self, hardware: str, model_size_b: float) -> dict:
        """Get quantization recommendation for hardware."""
        profile = self.HARDWARE_PROFILES.get(hardware)
        if not profile:
            return {"quantization": "Q4_K_M", "compute_type": "int8"}

        return {
            "quantization": profile["recommended"],
            "compute_type": profile["compute_type"]
        }

Model Quantization Security Examples

Model Integrity Verification

import hashlib
from pathlib import Path

class ModelVerifier:
    """Verify model file integrity."""

    def __init__(self, models_dir: str):
        self.models_dir = Path(models_dir)
        self.checksums = {}
        self._load_checksums()

    def _load_checksums(self):
        """Load known good checksums."""
        checksum_file = self.models_dir / "checksums.txt"
        if checksum_file.exists():
            for line in checksum_file.read_text().splitlines():
                if line.strip():
                    checksum, filename = line.split()
                    self.checksums[filename] = checksum

    def verify(self, model_path: str) -> bool:
        """Verify model against known checksum."""
        path = Path(model_path)

        if path.name not in self.checksums:
            logger.warning("model.unknown", name=path.name)
            return False

        expected = self.checksums[path.name]
        actual = self._calculate_checksum(path)

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

        return True

    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(65536), b""):
                sha256.update(chunk)
        return sha256.hexdigest()

Safe Model Path Handling

def validate_model_path(user_path: str, allowed_dir: str) -> Path:
    """Validate model path against directory traversal."""
    path = Path(user_path).resolve()
    allowed = Path(allowed_dir).resolve()

    # Check path is within allowed directory
    if not path.is_relative_to(allowed):
        raise SecurityError(f"Path traversal attempt: {user_path}")

    # Check file exists
    if not path.exists():
        raise FileNotFoundError(f"Model not found: {path}")

    # Check is a file
    if not path.is_file():
        raise ValueError(f"Not a file: {path}")

    return path

# Usage
path = validate_model_path("../../../etc/passwd", "/var/jarvis/models")
# Raises SecurityError

Secure Quantization Process

import subprocess
import tempfile

class SecureQuantizationProcess:
    """Run quantization with security constraints."""

    def run(self, input_path: str, output_path: str, quant_type: str):
        """Run quantization in isolated environment."""

        # Validate inputs
        if not Path(input_path).exists():
            raise FileNotFoundError(input_path)

        if quant_type not in ["Q4_K_M", "Q5_K_M", "Q8_0"]:
            raise ValueError(f"Invalid quantization: {quant_type}")

        # Run with limited permissions
        result = subprocess.run(
            ["./quantize", input_path, output_path, quant_type],
            capture_output=True,
            timeout=7200,  # 2 hour timeout
            cwd="/opt/llama.cpp",
            env={
                "HOME": tempfile.gettempdir(),
                "PATH": "/usr/bin"
            }
        )

        if result.returncode != 0:
            raise QuantizationError(result.stderr.decode())

        return output_path

Security Testing

def test_checksum_verification():
    """Test model checksum verification."""
    verifier = ModelVerifier("/var/jarvis/models")

    # Should pass for valid model
    assert verifier.verify("llama-7b-Q4_K_M.gguf")

    # Should fail for tampered model
    # (modify file first in test)
    assert not verifier.verify("tampered-model.gguf")

def test_path_traversal_blocked():
    """Test path traversal is blocked."""
    with pytest.raises(SecurityError):
        validate_model_path("../../../etc/passwd", "/var/jarvis/models")

    with pytest.raises(SecurityError):
        validate_model_path("/etc/passwd", "/var/jarvis/models")