import torch
from .logging_config import get_logger
logger = get_logger(__name__)
def default_can_quantize_predicate(path, module):
if not isinstance(module, torch.nn.Linear):
return False
if hasattr(module, "weight"):
weight_shape = module.weight.shape
if len(weight_shape) >= 2 and weight_shape[-1] % 64 != 0:
return False
return True
def quantize_torch_model(model, bits=8, model_name="model"):
try:
from optimum.quanto import quantize, freeze, qint4, qint8
weights = qint4 if bits == 4 else qint8
quantize(model, weights=weights)
freeze(model)
logger.info(
"%d-bit quantization applied to %s with optimum-quanto", bits, model_name
)
return True
except ImportError:
logger.warning(
"optimum-quanto not available, skipping quantization for %s", model_name
)
return False
except Exception as e:
logger.warning(
"PyTorch quantization failed for %s: %s, using full precision",
model_name,
e,
)
return False
def quantize_mlx_model(model, bits=8, model_name="model", can_quantize_predicate=None):
if can_quantize_predicate is None:
can_quantize_predicate = default_can_quantize_predicate
try:
from mlx.nn import quantize as mlx_quantize
mlx_quantize(
model,
group_size=64,
bits=bits,
class_predicate=can_quantize_predicate,
)
logger.info(
"%d-bit quantization applied to compatible layers of %s", bits, model_name
)
return True
except ImportError:
logger.warning("MLX not available, skipping quantization for %s", model_name)
return False
except Exception as e:
logger.warning(
"MLX quantization failed for %s: %s, using full precision", model_name, e
)
return False
def quantize_model(
model, bits, backend, model_name="model", can_quantize_predicate=None
):
logger.info("Applying %d-bit quantization to %s...", bits, model_name)
if backend == "torch":
return quantize_torch_model(model, bits=bits, model_name=model_name)
elif backend == "mlx":
return quantize_mlx_model(
model,
bits=bits,
model_name=model_name,
can_quantize_predicate=can_quantize_predicate,
)
else:
logger.warning("Unknown backend %s, skipping quantization", backend)
return False