Skip to content

llmcompressor.observers.mse

Classes:

MemorylessMSEObserver

MemorylessMSEObserver(*args, **kwargs)

Bases: Observer

Compute quantization parameters by finding the optimal min/max values which minimize the mean of quantization error squared.

Source code in src/llmcompressor/observers/mse.py
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    observer_kwargs = self.args.observer_kwargs
    self.maxshrink = observer_kwargs.get("maxshrink", 0.20)
    self.patience = observer_kwargs.get("patience", 5)
    self.grid = observer_kwargs.get("grid", 100.0)
    self.norm = observer_kwargs.get("norm", 2.4)
    self.chunk_size = observer_kwargs.get("chunk_size", 5)
    if self.chunk_size <= 0:
        raise ValueError(f"chunk_size must be positive, got {self.chunk_size}")

    # Pre-create token_args to avoid patch_attr context manager
    # which causes torch.compile graph breaks
    self._token_args = self.args.model_copy(
        update={"strategy": QuantizationStrategy.TOKEN}
    )

MovingAverageMSEObserver

MovingAverageMSEObserver(*args, **kwargs)

Bases: Observer

Compute quantization parameters by finding the optimal min/max values which minimize the mean of quantization error squared, with moving average smoothing.

Source code in src/llmcompressor/observers/mse.py
def __init__(self, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.avg_constant = self.args.observer_kwargs.get("averaging_constant", 0.01)
    observer_kwargs = self.args.observer_kwargs
    self.maxshrink = observer_kwargs.get("maxshrink", 0.20)
    self.patience = observer_kwargs.get("patience", 5)
    self.grid = observer_kwargs.get("grid", 100.0)
    self.norm = observer_kwargs.get("norm", 2.4)
    self.chunk_size = observer_kwargs.get("chunk_size", 5)
    if self.chunk_size <= 0:
        raise ValueError(f"chunk_size must be positive, got {self.chunk_size}")

    # Pre-create token_args to avoid patch_attr context manager
    # which causes torch.compile graph breaks
    self._token_args = self.args.model_copy(
        update={"strategy": QuantizationStrategy.TOKEN}
    )