Repository: huggingface/transformers
File: src/transformers/generation/utils.py (Python)

Code before:
                cache_implementation=generation_config.cache_implementation,
                batch_size=max(generation_config.num_beams, generation_config.num_return_sequences) * batch_size,
                max_cache_len=max_cache_length,
                model_kwargs=model_kwargs,
            )
        elif generation_config.cache_implementation == "quantized":
            if self.config.is_encoder_decoder or not self._supports_default_dynamic_cache():
                raise ValueError(
                    "This model does not support the quantized cache. If you want your model to support quantized "
                    "cache, please open an issue and tag @zucchini-nlp."
                )

            cache_config = generation_config.cache_config if generation_config.cache_config is not None else {}
            cache_config.setdefault("config", self.config.get_text_config(decoder=True))
            backend = cache_config.pop("backend", "quanto")
            if backend == "quanto" and not is_optimum_quanto_available():
                raise ImportError(
                    "You need to install optimum-quanto in order to use KV cache quantization with optimum-quanto "
                    "backend. Please install it via  with `pip install optimum-quanto`"
                )
            elif backend == "HQQ" and not is_hqq_available():
                raise ImportError(
                    "You need to install `HQQ` in order to use KV cache quantization with HQQ backend. "
                    "Please install it via  with `pip install hqq`"
                )
            model_kwargs["past_key_values"] = QuantizedCache(backend=backend, **cache_config)
        # i.e. `cache_implementation` in [None, "dynamic", "offloaded", "dynamic_full"]
        # TODO: prepare linear cache from a single API, instead of creating in modeling code
        else:
            model_kwargs["past_key_values"] = DynamicCache(**dynamic_cache_kwargs)

        if (
            self.config.is_encoder_decoder
            and "past_key_values" in model_kwargs
            and not isinstance(model_kwargs["past_key_values"], EncoderDecoderCache)
        ):
            model_kwargs["past_key_values"] = EncoderDecoderCache(
                model_kwargs["past_key_values"],  # self-attention cache
                DynamicCache(**dynamic_cache_kwargs),  # cross-attention cache
            )

    def _supports_logits_to_keep(self) -> bool:
        """
        Return True if the current model supports the keyword argument `logits_to_keep` in forward()
        to save memory. Checking it in this way allows to avoid using a new model attribute.
        """
        return "logits_to_keep" in set(inspect.signature(self.forward).parameters.keys())

    def _prepare_special_tokens(
        self,
        generation_config: GenerationConfig,

Review comment (performance):
Imo, this error should be raised within the constructor of the cache - we should not have to check this ourselves here

Code after:
                model_kwargs=model_kwargs,
            )
        elif generation_config.cache_implementation == "quantized":
            if self.config.is_encoder_decoder or not self._supports_default_dynamic_cache():
                raise ValueError(
                    "This model does not support the quantized cache. If you want your model to support quantized "
                    "cache, please open an issue and tag @zucchini-nlp."
                )

            cache_config = generation_config.cache_config if generation_config.cache_config is not None else {}
            cache_config.setdefault("config", self.config.get_text_config(decoder=True))
            backend = cache_config.pop("backend", "quanto")
            model_kwargs["past_key_values"] = QuantizedCache(backend=backend, **cache_config)
        # i.e. `cache_implementation` in [None, "dynamic", "offloaded", "dynamic_full"]
        # TODO: prepare linear cache from a single API, instead of creating in modeling code
        else:
            model_kwargs["past_key_values"] = DynamicCache(**dynamic_cache_kwargs)

        if (
            self.config.is_encoder_decoder
            and "past_key_values" in model_kwargs
            and not isinstance(model_kwargs["past_key_values"], EncoderDecoderCache)
        ):
            model_kwargs["past_key_values"] = EncoderDecoderCache(
                model_kwargs["past_key_values"],  # self-attention cache
                DynamicCache(**dynamic_cache_kwargs),  # cross-attention cache
            )

    def _supports_logits_to_keep(self) -> bool:
        """
        Return True if the current model supports the keyword argument `logits_to_keep` in forward()
        to save memory. Checking it in this way allows to avoid using a new model attribute.
        """
        return "logits_to_keep" in set(inspect.signature(self.forward).parameters.keys())

    def _prepare_special_tokens(
        self,
        generation_config: GenerationConfig,
        kwargs_has_attention_mask: bool | None = None,
        device: torch.device | str | None = None,
    ):
        """
        Prepares the special tokens for generation, overwriting the generation config with their processed versions
        converted to tensor.

        Note that `generation_config` is changed in place and stops being serializable after this method is called.
        That is no problem if called within `generate` (`generation_config` is a local copy that doesn't leave the
        function). However, if called outside `generate`, consider creating a copy of `generation_config` first.
        """

        # Convert special tokens to tensors