Dynamic LLM Quantizer

What It Does

Installation

pip install torch==2.0.1 transformers==4.31.0 psutil==5.9.5

Usage

from dynamic_quantizer import quantify_model
from transformers import AutoModel

# Load a pre-trained model
model = AutoModel.from_pretrained("bert-base-uncased")

# Quantify the model using GPTQ method with resource monitoring
result = quantify_model(model, method='GPTQ', monitor_resources=True)

# Access the quantized model and resource stats
quantized_model = result['quantized_model']
resource_stats = result['resource_stats']
print("Quantization completed in", result['time_taken'], "seconds")
print("Resource stats:", resource_stats)

python dynamic_quantizer.py --model_name bert-base-uncased --method GPTQ --monitor_resources

Source Code

import torch
from transformers import AutoModel
import psutil
import time
from typing import Dict, Any

def quantify_model(model: torch.nn.Module, method: str = 'GPTQ', monitor_resources: bool = False) -> Dict[str, Any]:
    """
    Quantify a given model using the specified quantization method and optionally monitor resource usage.

    Args:
        model (torch.nn.Module): The model to be quantized.
        method (str): The quantization method to apply. Supported: 'GGUF', 'GPTQ', 'AWQ'.
        monitor_resources (bool): Whether to monitor resource usage during quantization.

    Returns:
        Dict[str, Any]: A dictionary containing the quantized model and resource statistics (if monitored).
    """
    supported_methods = ['GGUF', 'GPTQ', 'AWQ']
    if method not in supported_methods:
        raise ValueError(f"Unsupported quantization method: {method}. Supported methods are: {supported_methods}")

    # Placeholder for resource monitoring
    resource_stats = {}

    if monitor_resources:
        # Capture initial resource usage
        resource_stats['before'] = {
            'cpu_percent': psutil.cpu_percent(interval=None),
            'memory_info': psutil.virtual_memory()._asdict()
        }

    # Simulate quantization process
    start_time = time.time()
    quantized_model = model  # Placeholder for actual quantization logic
    time.sleep(1)  # Simulate processing time

    if monitor_resources:
        # Capture final resource usage
        resource_stats['after'] = {
            'cpu_percent': psutil.cpu_percent(interval=None),
            'memory_info': psutil.virtual_memory()._asdict()
        }

    return {
        'quantized_model': quantized_model,
        'resource_stats': resource_stats,
        'time_taken': time.time() - start_time
    }

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Dynamic LLM Quantizer")
    parser.add_argument("--model_name", type=str, required=True, help="Name of the pre-trained model to load.")
    parser.add_argument("--method", type=str, default="GPTQ", help="Quantization method to apply (GGUF, GPTQ, AWQ).")
    parser.add_argument("--monitor_resources", action="store_true", help="Enable resource monitoring during quantization.")

    args = parser.parse_args()

    try:
        model = AutoModel.from_pretrained(args.model_name)
        result = quantify_model(model, method=args.method, monitor_resources=args.monitor_resources)
        print("Quantization completed.")
        print("Time taken:", result['time_taken'], "seconds")
        if args.monitor_resources:
            print("Resource stats:", result['resource_stats'])
    except Exception as e:
        print(f"Error: {e}")