LoRA

LoRA is a fast and lightweight training method that inserts and trains a significantly smaller number of parameters instead of all the model parameters. This produces a smaller file (~100 MBs) and makes it easier to quickly train a model to learn a new concept. LoRA weights are typically loaded into the denoiser, text encoder or both. The denoiser usually corresponds to a UNet (UNet2DConditionModel, for example) or a Transformer (SD3Transformer2DModel, for example). There are several classes for loading LoRA weights:

• StableDiffusionLoraLoaderMixin provides functions for loading and unloading, fusing and unfusing, enabling and disabling, and more functions for managing LoRA weights. This class can be used with any model.
• StableDiffusionXLLoraLoaderMixin is a Stable Diffusion (SDXL) version of the StableDiffusionLoraLoaderMixin class for loading and saving LoRA weights. It can only be used with the SDXL model.
• SD3LoraLoaderMixin provides similar functions for Stable Diffusion 3
• LoraBaseMixin provides a base class with several utility methods to fuse, unfuse, unload, LoRAs and more.

Tip

To learn more about how to load LoRA weights, see the LoRA loading guide.

mindone.diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin

Bases: LoraBaseMixin

Load LoRA layers into Stable Diffusion [UNet2DConditionModel] and CLIPTextModel.

Source code in mindone/diffusers/loaders/lora_pipeline.py

class StableDiffusionLoraLoaderMixin(LoraBaseMixin):
    r"""
    Load LoRA layers into Stable Diffusion [`UNet2DConditionModel`] and
    [`CLIPTextModel`](https://huggingface.co/docs/transformers/model_doc/clip#transformers.CLIPTextModel).
    """

    _lora_loadable_modules = ["unet", "text_encoder"]
    unet_name = UNET_NAME
    text_encoder_name = TEXT_ENCODER_NAME

    def load_lora_weights(
        self, pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]], adapter_name=None, **kwargs
    ):
        """
        Load LoRA weights specified in `pretrained_model_name_or_path_or_dict` into `self.unet` and
        `self.text_encoder`.

        All kwargs are forwarded to `self.lora_state_dict`.

        See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`] for more details on how the state dict is
        loaded.

        See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_unet`] for more details on how the state dict is
        loaded into `self.unet`.

        See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder`] for more details on how the state
        dict is loaded into `self.text_encoder`.

        Parameters:
            pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
                See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
            kwargs (`dict`, *optional*):
                See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
            adapter_name (`str`, *optional*):
                Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
                `default_{i}` where i is the total number of adapters being loaded.
        """
        # if a dict is passed, copy it instead of modifying it inplace
        if isinstance(pretrained_model_name_or_path_or_dict, dict):
            pretrained_model_name_or_path_or_dict = pretrained_model_name_or_path_or_dict.copy()

        # First, ensure that the checkpoint is a compatible one and can be successfully loaded.
        state_dict, network_alphas = self.lora_state_dict(pretrained_model_name_or_path_or_dict, **kwargs)

        is_correct_format = all("lora" in key or "dora_scale" in key for key in state_dict.keys())
        if not is_correct_format:
            raise ValueError("Invalid LoRA checkpoint.")

        self.load_lora_into_unet(
            state_dict,
            network_alphas=network_alphas,
            unet=getattr(self, self.unet_name) if not hasattr(self, "unet") else self.unet,
            adapter_name=adapter_name,
            _pipeline=self,
        )
        self.load_lora_into_text_encoder(
            state_dict,
            network_alphas=network_alphas,
            text_encoder=getattr(self, self.text_encoder_name)
            if not hasattr(self, "text_encoder")
            else self.text_encoder,
            lora_scale=self.lora_scale,
            adapter_name=adapter_name,
            _pipeline=self,
        )

    @classmethod
    @validate_hf_hub_args
    def lora_state_dict(
        cls,
        pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]],
        **kwargs,
    ):
        r"""
        Return state dict for lora weights and the network alphas.

        <Tip warning={true}>

        We support loading A1111 formatted LoRA checkpoints in a limited capacity.

        This function is experimental and might change in the future.

        </Tip>

        Parameters:
            pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
                Can be either:

                    - A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on
                      the Hub.
                    - A path to a *directory* (for example `./my_model_directory`) containing the model weights saved
                      with [`ModelMixin.save_pretrained`].
                    - A MindSpore state dict

            cache_dir (`Union[str, os.PathLike]`, *optional*):
                Path to a directory where a downloaded pretrained model configuration is cached if the standard cache
                is not used.
            force_download (`bool`, *optional*, defaults to `False`):
                Whether or not to force the (re-)download of the model weights and configuration files, overriding the
                cached versions if they exist.

            proxies (`Dict[str, str]`, *optional*):
                A dictionary of proxy servers to use by protocol or endpoint, for example, `{'http': 'foo.bar:3128',
                'http://hostname': 'foo.bar:4012'}`. The proxies are used on each request.
            local_files_only (`bool`, *optional*, defaults to `False`):
                Whether to only load local model weights and configuration files or not. If set to `True`, the model
                won't be downloaded from the Hub.
            token (`str` or *bool*, *optional*):
                The token to use as HTTP bearer authorization for remote files. If `True`, the token generated from
                `diffusers-cli login` (stored in `~/.huggingface`) is used.
            revision (`str`, *optional*, defaults to `"main"`):
                The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier
                allowed by Git.
            subfolder (`str`, *optional*, defaults to `""`):
                The subfolder location of a model file within a larger model repository on the Hub or locally.
            weight_name (`str`, *optional*, defaults to None):
                Name of the serialized state dict file.
        """
        # Load the main state dict first which has the LoRA layers for either of
        # UNet and text encoder or both.
        cache_dir = kwargs.pop("cache_dir", None)
        force_download = kwargs.pop("force_download", False)
        proxies = kwargs.pop("proxies", None)
        local_files_only = kwargs.pop("local_files_only", None)
        token = kwargs.pop("token", None)
        revision = kwargs.pop("revision", None)
        subfolder = kwargs.pop("subfolder", None)
        weight_name = kwargs.pop("weight_name", None)
        unet_config = kwargs.pop("unet_config", None)
        use_safetensors = kwargs.pop("use_safetensors", None)

        allow_pickle = False
        if use_safetensors is None:
            use_safetensors = True
            allow_pickle = True

        user_agent = {
            "file_type": "attn_procs_weights",
            "framework": "pytorch",
        }

        state_dict = cls._fetch_state_dict(
            pretrained_model_name_or_path_or_dict=pretrained_model_name_or_path_or_dict,
            weight_name=weight_name,
            use_safetensors=use_safetensors,
            local_files_only=local_files_only,
            cache_dir=cache_dir,
            force_download=force_download,
            proxies=proxies,
            token=token,
            revision=revision,
            subfolder=subfolder,
            user_agent=user_agent,
            allow_pickle=allow_pickle,
        )

        network_alphas = None
        # TODO: replace it with a method from `state_dict_utils`
        if all(
            (
                k.startswith("lora_te_")
                or k.startswith("lora_unet_")
                or k.startswith("lora_te1_")
                or k.startswith("lora_te2_")
            )
            for k in state_dict.keys()
        ):
            # Map SDXL blocks correctly.
            if unet_config is not None:
                # use unet config to remap block numbers
                state_dict = _maybe_map_sgm_blocks_to_diffusers(state_dict, unet_config)
            state_dict, network_alphas = _convert_non_diffusers_lora_to_diffusers(state_dict)

        return state_dict, network_alphas

    @classmethod
    def load_lora_into_unet(cls, state_dict, network_alphas, unet, adapter_name=None, _pipeline=None):
        """
        This will load the LoRA layers specified in `state_dict` into `unet`.

        Parameters:
            state_dict (`dict`):
                A standard state dict containing the lora layer parameters. The keys can either be indexed directly
                into the unet or prefixed with an additional `unet` which can be used to distinguish between text
                encoder lora layers.
            network_alphas (`Dict[str, float]`):
                The value of the network alpha used for stable learning and preventing underflow. This value has the
                same meaning as the `--network_alpha` option in the kohya-ss trainer script. Refer to [this
                link](https://github.com/darkstorm2150/sd-scripts/blob/main/docs/train_network_README-en.md#execute-learning).
            unet (`UNet2DConditionModel`):
                The UNet model to load the LoRA layers into.
            adapter_name (`str`, *optional*):
                Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
                `default_{i}` where i is the total number of adapters being loaded.
        """
        # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
        # then the `state_dict` keys should have `cls.unet_name` and/or `cls.text_encoder_name` as
        # their prefixes.
        keys = list(state_dict.keys())
        only_text_encoder = all(key.startswith(cls.text_encoder_name) for key in keys)
        if not only_text_encoder:
            # Load the layers corresponding to UNet.
            logger.info(f"Loading {cls.unet_name}.")
            unet.load_attn_procs(
                state_dict, network_alphas=network_alphas, adapter_name=adapter_name, _pipeline=_pipeline
            )

    @classmethod
    def load_lora_into_text_encoder(
        cls,
        state_dict,
        network_alphas,
        text_encoder,
        prefix=None,
        lora_scale=1.0,
        adapter_name=None,
        _pipeline=None,
    ):
        """
        This will load the LoRA layers specified in `state_dict` into `text_encoder`

        Parameters:
            state_dict (`dict`):
                A standard state dict containing the lora layer parameters. The key should be prefixed with an
                additional `text_encoder` to distinguish between unet lora layers.
            network_alphas (`Dict[str, float]`):
                See `LoRALinearLayer` for more details.
            text_encoder (`CLIPTextModel`):
                The text encoder model to load the LoRA layers into.
            prefix (`str`):
                Expected prefix of the `text_encoder` in the `state_dict`.
            lora_scale (`float`):
                How much to scale the output of the lora linear layer before it is added with the output of the regular
                lora layer.
            adapter_name (`str`, *optional*):
                Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
                `default_{i}` where i is the total number of adapters being loaded.
        """
        from mindone.diffusers._peft import LoraConfig

        # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
        # then the `state_dict` keys should have `self.unet_name` and/or `self.text_encoder_name` as
        # their prefixes.
        keys = list(state_dict.keys())
        prefix = cls.text_encoder_name if prefix is None else prefix

        # Safe prefix to check with.
        if any(cls.text_encoder_name in key for key in keys):
            # Load the layers corresponding to text encoder and make necessary adjustments.
            text_encoder_keys = [k for k in keys if k.startswith(prefix) and k.split(".")[0] == prefix]
            text_encoder_lora_state_dict = {
                k.replace(f"{prefix}.", ""): v for k, v in state_dict.items() if k in text_encoder_keys
            }

            if len(text_encoder_lora_state_dict) > 0:
                logger.info(f"Loading {prefix}.")
                rank = {}
                text_encoder_lora_state_dict = convert_state_dict_to_diffusers(text_encoder_lora_state_dict)

                # convert state dict
                text_encoder_lora_state_dict = convert_state_dict_to_peft(text_encoder_lora_state_dict)

                for name, _ in text_encoder_attn_modules(text_encoder):
                    for module in ("out_proj", "q_proj", "k_proj", "v_proj"):
                        rank_key = f"{name}.{module}.lora_B.weight"
                        if rank_key not in text_encoder_lora_state_dict:
                            continue
                        rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

                for name, _ in text_encoder_mlp_modules(text_encoder):
                    for module in ("fc1", "fc2"):
                        rank_key = f"{name}.{module}.lora_B.weight"
                        if rank_key not in text_encoder_lora_state_dict:
                            continue
                        rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

                if network_alphas is not None:
                    alpha_keys = [
                        k for k in network_alphas.keys() if k.startswith(prefix) and k.split(".")[0] == prefix
                    ]
                    network_alphas = {
                        k.replace(f"{prefix}.", ""): v for k, v in network_alphas.items() if k in alpha_keys
                    }

                lora_config_kwargs = get_peft_kwargs(rank, network_alphas, text_encoder_lora_state_dict, is_unet=False)
                if "use_dora" in lora_config_kwargs:
                    if lora_config_kwargs["use_dora"]:
                        if is_peft_version("<", "0.9.0"):
                            raise ValueError(
                                "You need `peft` 0.9.0 at least to use DoRA-enabled LoRAs. Please upgrade your installation of `peft`."
                            )
                    else:
                        if is_peft_version("<", "0.9.0"):
                            lora_config_kwargs.pop("use_dora")
                lora_config = LoraConfig(**lora_config_kwargs)

                # adapter_name
                if adapter_name is None:
                    adapter_name = get_adapter_name(text_encoder)

                # inject LoRA layers and load the state dict
                # in transformers we automatically check whether the adapter name is already in use or not
                text_encoder.load_adapter(
                    adapter_name=adapter_name,
                    adapter_state_dict=text_encoder_lora_state_dict,
                    peft_config=lora_config,
                )

                # scale LoRA layers with `lora_scale`
                scale_lora_layers(text_encoder, weight=lora_scale)

                text_encoder.to(dtype=text_encoder.dtype)

    @classmethod
    def save_lora_weights(
        cls,
        save_directory: Union[str, os.PathLike],
        unet_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
        text_encoder_lora_layers: Dict[str, nn.Cell] = None,
        is_main_process: bool = True,
        weight_name: str = None,
        save_function: Callable = None,
        safe_serialization: bool = True,
    ):
        r"""
        Save the LoRA parameters corresponding to the UNet and text encoder.

        Arguments:
            save_directory (`str` or `os.PathLike`):
                Directory to save LoRA parameters to. Will be created if it doesn't exist.
            unet_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
                State dict of the LoRA layers corresponding to the `unet`.
            text_encoder_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
                State dict of the LoRA layers corresponding to the `text_encoder`. Must explicitly pass the text
                encoder LoRA state dict because it comes from 🤗 Transformers.
            is_main_process (`bool`, *optional*, defaults to `True`):
                Whether the process calling this is the main process or not. Useful during distributed training and you
                need to call this function on all processes. In this case, set `is_main_process=True` only on the main
                process to avoid race conditions.
            save_function (`Callable`):
                The function to use to save the state dictionary. Useful during distributed training when you need to
                replace `mindspore.save_checkpoint` with another method. Can be configured with the environment variable
                `DIFFUSERS_SAVE_MODE`.
            safe_serialization (`bool`, *optional*, defaults to `True`):
                Whether to save the model using `safetensors` or the traditional MindSpore way.
        """
        state_dict = {}

        if not (unet_lora_layers or text_encoder_lora_layers):
            raise ValueError("You must pass at least one of `unet_lora_layers` and `text_encoder_lora_layers`.")

        if unet_lora_layers:
            state_dict.update(cls.pack_weights(unet_lora_layers, cls.unet_name))

        if text_encoder_lora_layers:
            state_dict.update(cls.pack_weights(text_encoder_lora_layers, cls.text_encoder_name))

        # Save the model
        cls.write_lora_layers(
            state_dict=state_dict,
            save_directory=save_directory,
            is_main_process=is_main_process,
            weight_name=weight_name,
            save_function=save_function,
            safe_serialization=safe_serialization,
        )

    def fuse_lora(
        self,
        components: List[str] = ["unet", "text_encoder"],
        lora_scale: float = 1.0,
        safe_fusing: bool = False,
        adapter_names: Optional[List[str]] = None,
        **kwargs,
    ):
        r"""
        Fuses the LoRA parameters into the original parameters of the corresponding blocks.

        <Tip warning={true}>

        This is an experimental API.

        </Tip>

        Args:
            components: (`List[str]`): List of LoRA-injectable components to fuse the LoRAs into.
            lora_scale (`float`, defaults to 1.0):
                Controls how much to influence the outputs with the LoRA parameters.
            safe_fusing (`bool`, defaults to `False`):
                Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them.
            adapter_names (`List[str]`, *optional*):
                Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused.

        Example:

        ```py
        from mindone.diffusers import DiffusionPipeline
        import mindspore

        pipeline = DiffusionPipeline.from_pretrained(
            "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
        )
        pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
        pipeline.fuse_lora(lora_scale=0.7)
        ```
        """
        super().fuse_lora(
            components=components, lora_scale=lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names
        )

    def unfuse_lora(self, components: List[str] = ["unet", "text_encoder"], **kwargs):
        r"""
        Reverses the effect of
        [`pipe.fuse_lora()`](https://huggingface.co/docs/diffusers/main/en/api/loaders#diffusers.loaders.LoraBaseMixin.fuse_lora).

        <Tip warning={true}>

        This is an experimental API.

        </Tip>

        Args:
            components (`List[str]`): List of LoRA-injectable components to unfuse LoRA from.
            unfuse_unet (`bool`, defaults to `True`): Whether to unfuse the UNet LoRA parameters.
            unfuse_text_encoder (`bool`, defaults to `True`):
                Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the
                LoRA parameters then it won't have any effect.
        """
        super().unfuse_lora(components=components)

mindone.diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.fuse_lora(components=['unet', 'text_encoder'], lora_scale=1.0, safe_fusing=False, adapter_names=None, **kwargs)

Fuses the LoRA parameters into the original parameters of the corresponding blocks.

This is an experimental API.

PARAMETER	DESCRIPTION
components	(List[str]): List of LoRA-injectable components to fuse the LoRAs into. TYPE: List[str] DEFAULT: ['unet', 'text_encoder']
lora_scale	Controls how much to influence the outputs with the LoRA parameters. TYPE: `float`, defaults to 1.0 DEFAULT: 1.0
safe_fusing	Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them. TYPE: `bool`, defaults to `False` DEFAULT: False
adapter_names	Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused. TYPE: `List[str]`, *optional* DEFAULT: None

from mindone.diffusers import DiffusionPipeline
import mindspore

pipeline = DiffusionPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
)
pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
pipeline.fuse_lora(lora_scale=0.7)

Source code in mindone/diffusers/loaders/lora_pipeline.py

def fuse_lora(
    self,
    components: List[str] = ["unet", "text_encoder"],
    lora_scale: float = 1.0,
    safe_fusing: bool = False,
    adapter_names: Optional[List[str]] = None,
    **kwargs,
):
    r"""
    Fuses the LoRA parameters into the original parameters of the corresponding blocks.

    <Tip warning={true}>

    This is an experimental API.

    </Tip>

    Args:
        components: (`List[str]`): List of LoRA-injectable components to fuse the LoRAs into.
        lora_scale (`float`, defaults to 1.0):
            Controls how much to influence the outputs with the LoRA parameters.
        safe_fusing (`bool`, defaults to `False`):
            Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them.
        adapter_names (`List[str]`, *optional*):
            Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused.

    Example:

    ```py
    from mindone.diffusers import DiffusionPipeline
    import mindspore

    pipeline = DiffusionPipeline.from_pretrained(
        "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
    )
    pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
    pipeline.fuse_lora(lora_scale=0.7)
    ```
    """
    super().fuse_lora(
        components=components, lora_scale=lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names
    )

mindone.diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder(state_dict, network_alphas, text_encoder, prefix=None, lora_scale=1.0, adapter_name=None, _pipeline=None) classmethod

This will load the LoRA layers specified in state_dict into text_encoder

PARAMETER	DESCRIPTION
state_dict	A standard state dict containing the lora layer parameters. The key should be prefixed with an additional text_encoder to distinguish between unet lora layers. TYPE: `dict`
network_alphas	See LoRALinearLayer for more details. TYPE: `Dict[str, float]`
text_encoder	The text encoder model to load the LoRA layers into. TYPE: `CLIPTextModel`
prefix	Expected prefix of the text_encoder in the state_dict. TYPE: `str` DEFAULT: None
lora_scale	How much to scale the output of the lora linear layer before it is added with the output of the regular lora layer. TYPE: `float` DEFAULT: 1.0
adapter_name	Adapter name to be used for referencing the loaded adapter model. If not specified, it will use default_{i} where i is the total number of adapters being loaded. TYPE: `str`, *optional* DEFAULT: None

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
def load_lora_into_text_encoder(
    cls,
    state_dict,
    network_alphas,
    text_encoder,
    prefix=None,
    lora_scale=1.0,
    adapter_name=None,
    _pipeline=None,
):
    """
    This will load the LoRA layers specified in `state_dict` into `text_encoder`

    Parameters:
        state_dict (`dict`):
            A standard state dict containing the lora layer parameters. The key should be prefixed with an
            additional `text_encoder` to distinguish between unet lora layers.
        network_alphas (`Dict[str, float]`):
            See `LoRALinearLayer` for more details.
        text_encoder (`CLIPTextModel`):
            The text encoder model to load the LoRA layers into.
        prefix (`str`):
            Expected prefix of the `text_encoder` in the `state_dict`.
        lora_scale (`float`):
            How much to scale the output of the lora linear layer before it is added with the output of the regular
            lora layer.
        adapter_name (`str`, *optional*):
            Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
            `default_{i}` where i is the total number of adapters being loaded.
    """
    from mindone.diffusers._peft import LoraConfig

    # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
    # then the `state_dict` keys should have `self.unet_name` and/or `self.text_encoder_name` as
    # their prefixes.
    keys = list(state_dict.keys())
    prefix = cls.text_encoder_name if prefix is None else prefix

    # Safe prefix to check with.
    if any(cls.text_encoder_name in key for key in keys):
        # Load the layers corresponding to text encoder and make necessary adjustments.
        text_encoder_keys = [k for k in keys if k.startswith(prefix) and k.split(".")[0] == prefix]
        text_encoder_lora_state_dict = {
            k.replace(f"{prefix}.", ""): v for k, v in state_dict.items() if k in text_encoder_keys
        }

        if len(text_encoder_lora_state_dict) > 0:
            logger.info(f"Loading {prefix}.")
            rank = {}
            text_encoder_lora_state_dict = convert_state_dict_to_diffusers(text_encoder_lora_state_dict)

            # convert state dict
            text_encoder_lora_state_dict = convert_state_dict_to_peft(text_encoder_lora_state_dict)

            for name, _ in text_encoder_attn_modules(text_encoder):
                for module in ("out_proj", "q_proj", "k_proj", "v_proj"):
                    rank_key = f"{name}.{module}.lora_B.weight"
                    if rank_key not in text_encoder_lora_state_dict:
                        continue
                    rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

            for name, _ in text_encoder_mlp_modules(text_encoder):
                for module in ("fc1", "fc2"):
                    rank_key = f"{name}.{module}.lora_B.weight"
                    if rank_key not in text_encoder_lora_state_dict:
                        continue
                    rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

            if network_alphas is not None:
                alpha_keys = [
                    k for k in network_alphas.keys() if k.startswith(prefix) and k.split(".")[0] == prefix
                ]
                network_alphas = {
                    k.replace(f"{prefix}.", ""): v for k, v in network_alphas.items() if k in alpha_keys
                }

            lora_config_kwargs = get_peft_kwargs(rank, network_alphas, text_encoder_lora_state_dict, is_unet=False)
            if "use_dora" in lora_config_kwargs:
                if lora_config_kwargs["use_dora"]:
                    if is_peft_version("<", "0.9.0"):
                        raise ValueError(
                            "You need `peft` 0.9.0 at least to use DoRA-enabled LoRAs. Please upgrade your installation of `peft`."
                        )
                else:
                    if is_peft_version("<", "0.9.0"):
                        lora_config_kwargs.pop("use_dora")
            lora_config = LoraConfig(**lora_config_kwargs)

            # adapter_name
            if adapter_name is None:
                adapter_name = get_adapter_name(text_encoder)

            # inject LoRA layers and load the state dict
            # in transformers we automatically check whether the adapter name is already in use or not
            text_encoder.load_adapter(
                adapter_name=adapter_name,
                adapter_state_dict=text_encoder_lora_state_dict,
                peft_config=lora_config,
            )

            # scale LoRA layers with `lora_scale`
            scale_lora_layers(text_encoder, weight=lora_scale)

            text_encoder.to(dtype=text_encoder.dtype)

mindone.diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.load_lora_into_unet(state_dict, network_alphas, unet, adapter_name=None, _pipeline=None) classmethod

This will load the LoRA layers specified in state_dict into unet.

PARAMETER	DESCRIPTION
state_dict	A standard state dict containing the lora layer parameters. The keys can either be indexed directly into the unet or prefixed with an additional unet which can be used to distinguish between text encoder lora layers. TYPE: `dict`
network_alphas	The value of the network alpha used for stable learning and preventing underflow. This value has the same meaning as the --network_alpha option in the kohya-ss trainer script. Refer to this link. TYPE: `Dict[str, float]`
unet	The UNet model to load the LoRA layers into. TYPE: `UNet2DConditionModel`
adapter_name	Adapter name to be used for referencing the loaded adapter model. If not specified, it will use default_{i} where i is the total number of adapters being loaded. TYPE: `str`, *optional* DEFAULT: None

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
def load_lora_into_unet(cls, state_dict, network_alphas, unet, adapter_name=None, _pipeline=None):
    """
    This will load the LoRA layers specified in `state_dict` into `unet`.

    Parameters:
        state_dict (`dict`):
            A standard state dict containing the lora layer parameters. The keys can either be indexed directly
            into the unet or prefixed with an additional `unet` which can be used to distinguish between text
            encoder lora layers.
        network_alphas (`Dict[str, float]`):
            The value of the network alpha used for stable learning and preventing underflow. This value has the
            same meaning as the `--network_alpha` option in the kohya-ss trainer script. Refer to [this
            link](https://github.com/darkstorm2150/sd-scripts/blob/main/docs/train_network_README-en.md#execute-learning).
        unet (`UNet2DConditionModel`):
            The UNet model to load the LoRA layers into.
        adapter_name (`str`, *optional*):
            Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
            `default_{i}` where i is the total number of adapters being loaded.
    """
    # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
    # then the `state_dict` keys should have `cls.unet_name` and/or `cls.text_encoder_name` as
    # their prefixes.
    keys = list(state_dict.keys())
    only_text_encoder = all(key.startswith(cls.text_encoder_name) for key in keys)
    if not only_text_encoder:
        # Load the layers corresponding to UNet.
        logger.info(f"Loading {cls.unet_name}.")
        unet.load_attn_procs(
            state_dict, network_alphas=network_alphas, adapter_name=adapter_name, _pipeline=_pipeline
        )

mindone.diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.load_lora_weights(pretrained_model_name_or_path_or_dict, adapter_name=None, **kwargs)

Load LoRA weights specified in pretrained_model_name_or_path_or_dict into self.unet and self.text_encoder.

All kwargs are forwarded to self.lora_state_dict.

See [~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict] for more details on how the state dict is loaded.

See [~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_unet] for more details on how the state dict is loaded into self.unet.

See [~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder] for more details on how the state dict is loaded into self.text_encoder.

PARAMETER	DESCRIPTION
pretrained_model_name_or_path_or_dict	See [~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict]. TYPE: `str` or `os.PathLike` or `dict`
kwargs	See [~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict]. TYPE: `dict`, *optional* DEFAULT: {}
adapter_name	Adapter name to be used for referencing the loaded adapter model. If not specified, it will use default_{i} where i is the total number of adapters being loaded. TYPE: `str`, *optional* DEFAULT: None

Source code in mindone/diffusers/loaders/lora_pipeline.py

def load_lora_weights(
    self, pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]], adapter_name=None, **kwargs
):
    """
    Load LoRA weights specified in `pretrained_model_name_or_path_or_dict` into `self.unet` and
    `self.text_encoder`.

    All kwargs are forwarded to `self.lora_state_dict`.

    See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`] for more details on how the state dict is
    loaded.

    See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_unet`] for more details on how the state dict is
    loaded into `self.unet`.

    See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder`] for more details on how the state
    dict is loaded into `self.text_encoder`.

    Parameters:
        pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
            See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
        kwargs (`dict`, *optional*):
            See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
        adapter_name (`str`, *optional*):
            Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
            `default_{i}` where i is the total number of adapters being loaded.
    """
    # if a dict is passed, copy it instead of modifying it inplace
    if isinstance(pretrained_model_name_or_path_or_dict, dict):
        pretrained_model_name_or_path_or_dict = pretrained_model_name_or_path_or_dict.copy()

    # First, ensure that the checkpoint is a compatible one and can be successfully loaded.
    state_dict, network_alphas = self.lora_state_dict(pretrained_model_name_or_path_or_dict, **kwargs)

    is_correct_format = all("lora" in key or "dora_scale" in key for key in state_dict.keys())
    if not is_correct_format:
        raise ValueError("Invalid LoRA checkpoint.")

    self.load_lora_into_unet(
        state_dict,
        network_alphas=network_alphas,
        unet=getattr(self, self.unet_name) if not hasattr(self, "unet") else self.unet,
        adapter_name=adapter_name,
        _pipeline=self,
    )
    self.load_lora_into_text_encoder(
        state_dict,
        network_alphas=network_alphas,
        text_encoder=getattr(self, self.text_encoder_name)
        if not hasattr(self, "text_encoder")
        else self.text_encoder,
        lora_scale=self.lora_scale,
        adapter_name=adapter_name,
        _pipeline=self,
    )

mindone.diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.lora_state_dict(pretrained_model_name_or_path_or_dict, **kwargs) classmethod

Return state dict for lora weights and the network alphas.

We support loading A1111 formatted LoRA checkpoints in a limited capacity.

This function is experimental and might change in the future.

PARAMETER	DESCRIPTION
pretrained_model_name_or_path_or_dict	Can be either: - A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on the Hub. - A path to a *directory* (for example `./my_model_directory`) containing the model weights saved with [`ModelMixin.save_pretrained`]. - A MindSpore state dict TYPE: `str` or `os.PathLike` or `dict`
cache_dir	Path to a directory where a downloaded pretrained model configuration is cached if the standard cache is not used. TYPE: `Union[str, os.PathLike]`, *optional*
force_download	Whether or not to force the (re-)download of the model weights and configuration files, overriding the cached versions if they exist. TYPE: `bool`, *optional*, defaults to `False`
proxies	A dictionary of proxy servers to use by protocol or endpoint, for example, {'http': 'foo.bar:3128', 'http://hostname': 'foo.bar:4012'}. The proxies are used on each request. TYPE: `Dict[str, str]`, *optional*
local_files_only	Whether to only load local model weights and configuration files or not. If set to True, the model won't be downloaded from the Hub. TYPE: `bool`, *optional*, defaults to `False`
token	The token to use as HTTP bearer authorization for remote files. If True, the token generated from diffusers-cli login (stored in ~/.huggingface) is used. TYPE: `str` or *bool*, *optional*
revision	The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier allowed by Git. TYPE: `str`, *optional*, defaults to `"main"`
subfolder	The subfolder location of a model file within a larger model repository on the Hub or locally. TYPE: `str`, *optional*, defaults to `""`
weight_name	Name of the serialized state dict file. TYPE: `str`, *optional*, defaults to None

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
@validate_hf_hub_args
def lora_state_dict(
    cls,
    pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]],
    **kwargs,
):
    r"""
    Return state dict for lora weights and the network alphas.

    <Tip warning={true}>

    We support loading A1111 formatted LoRA checkpoints in a limited capacity.

    This function is experimental and might change in the future.

    </Tip>

    Parameters:
        pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
            Can be either:

                - A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on
                  the Hub.
                - A path to a *directory* (for example `./my_model_directory`) containing the model weights saved
                  with [`ModelMixin.save_pretrained`].
                - A MindSpore state dict

        cache_dir (`Union[str, os.PathLike]`, *optional*):
            Path to a directory where a downloaded pretrained model configuration is cached if the standard cache
            is not used.
        force_download (`bool`, *optional*, defaults to `False`):
            Whether or not to force the (re-)download of the model weights and configuration files, overriding the
            cached versions if they exist.

        proxies (`Dict[str, str]`, *optional*):
            A dictionary of proxy servers to use by protocol or endpoint, for example, `{'http': 'foo.bar:3128',
            'http://hostname': 'foo.bar:4012'}`. The proxies are used on each request.
        local_files_only (`bool`, *optional*, defaults to `False`):
            Whether to only load local model weights and configuration files or not. If set to `True`, the model
            won't be downloaded from the Hub.
        token (`str` or *bool*, *optional*):
            The token to use as HTTP bearer authorization for remote files. If `True`, the token generated from
            `diffusers-cli login` (stored in `~/.huggingface`) is used.
        revision (`str`, *optional*, defaults to `"main"`):
            The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier
            allowed by Git.
        subfolder (`str`, *optional*, defaults to `""`):
            The subfolder location of a model file within a larger model repository on the Hub or locally.
        weight_name (`str`, *optional*, defaults to None):
            Name of the serialized state dict file.
    """
    # Load the main state dict first which has the LoRA layers for either of
    # UNet and text encoder or both.
    cache_dir = kwargs.pop("cache_dir", None)
    force_download = kwargs.pop("force_download", False)
    proxies = kwargs.pop("proxies", None)
    local_files_only = kwargs.pop("local_files_only", None)
    token = kwargs.pop("token", None)
    revision = kwargs.pop("revision", None)
    subfolder = kwargs.pop("subfolder", None)
    weight_name = kwargs.pop("weight_name", None)
    unet_config = kwargs.pop("unet_config", None)
    use_safetensors = kwargs.pop("use_safetensors", None)

    allow_pickle = False
    if use_safetensors is None:
        use_safetensors = True
        allow_pickle = True

    user_agent = {
        "file_type": "attn_procs_weights",
        "framework": "pytorch",
    }

    state_dict = cls._fetch_state_dict(
        pretrained_model_name_or_path_or_dict=pretrained_model_name_or_path_or_dict,
        weight_name=weight_name,
        use_safetensors=use_safetensors,
        local_files_only=local_files_only,
        cache_dir=cache_dir,
        force_download=force_download,
        proxies=proxies,
        token=token,
        revision=revision,
        subfolder=subfolder,
        user_agent=user_agent,
        allow_pickle=allow_pickle,
    )

    network_alphas = None
    # TODO: replace it with a method from `state_dict_utils`
    if all(
        (
            k.startswith("lora_te_")
            or k.startswith("lora_unet_")
            or k.startswith("lora_te1_")
            or k.startswith("lora_te2_")
        )
        for k in state_dict.keys()
    ):
        # Map SDXL blocks correctly.
        if unet_config is not None:
            # use unet config to remap block numbers
            state_dict = _maybe_map_sgm_blocks_to_diffusers(state_dict, unet_config)
        state_dict, network_alphas = _convert_non_diffusers_lora_to_diffusers(state_dict)

    return state_dict, network_alphas

mindone.diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.save_lora_weights(save_directory, unet_lora_layers=None, text_encoder_lora_layers=None, is_main_process=True, weight_name=None, save_function=None, safe_serialization=True) classmethod

Save the LoRA parameters corresponding to the UNet and text encoder.

PARAMETER	DESCRIPTION
save_directory	Directory to save LoRA parameters to. Will be created if it doesn't exist. TYPE: `str` or `os.PathLike`
unet_lora_layers	State dict of the LoRA layers corresponding to the unet. TYPE: `Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]` DEFAULT: None
text_encoder_lora_layers	State dict of the LoRA layers corresponding to the text_encoder. Must explicitly pass the text encoder LoRA state dict because it comes from 🤗 Transformers. TYPE: `Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]` DEFAULT: None
is_main_process	Whether the process calling this is the main process or not. Useful during distributed training and you need to call this function on all processes. In this case, set is_main_process=True only on the main process to avoid race conditions. TYPE: `bool`, *optional*, defaults to `True` DEFAULT: True
save_function	The function to use to save the state dictionary. Useful during distributed training when you need to replace mindspore.save_checkpoint with another method. Can be configured with the environment variable DIFFUSERS_SAVE_MODE. TYPE: `Callable` DEFAULT: None
safe_serialization	Whether to save the model using safetensors or the traditional MindSpore way. TYPE: `bool`, *optional*, defaults to `True` DEFAULT: True

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
def save_lora_weights(
    cls,
    save_directory: Union[str, os.PathLike],
    unet_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
    text_encoder_lora_layers: Dict[str, nn.Cell] = None,
    is_main_process: bool = True,
    weight_name: str = None,
    save_function: Callable = None,
    safe_serialization: bool = True,
):
    r"""
    Save the LoRA parameters corresponding to the UNet and text encoder.

    Arguments:
        save_directory (`str` or `os.PathLike`):
            Directory to save LoRA parameters to. Will be created if it doesn't exist.
        unet_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
            State dict of the LoRA layers corresponding to the `unet`.
        text_encoder_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
            State dict of the LoRA layers corresponding to the `text_encoder`. Must explicitly pass the text
            encoder LoRA state dict because it comes from 🤗 Transformers.
        is_main_process (`bool`, *optional*, defaults to `True`):
            Whether the process calling this is the main process or not. Useful during distributed training and you
            need to call this function on all processes. In this case, set `is_main_process=True` only on the main
            process to avoid race conditions.
        save_function (`Callable`):
            The function to use to save the state dictionary. Useful during distributed training when you need to
            replace `mindspore.save_checkpoint` with another method. Can be configured with the environment variable
            `DIFFUSERS_SAVE_MODE`.
        safe_serialization (`bool`, *optional*, defaults to `True`):
            Whether to save the model using `safetensors` or the traditional MindSpore way.
    """
    state_dict = {}

    if not (unet_lora_layers or text_encoder_lora_layers):
        raise ValueError("You must pass at least one of `unet_lora_layers` and `text_encoder_lora_layers`.")

    if unet_lora_layers:
        state_dict.update(cls.pack_weights(unet_lora_layers, cls.unet_name))

    if text_encoder_lora_layers:
        state_dict.update(cls.pack_weights(text_encoder_lora_layers, cls.text_encoder_name))

    # Save the model
    cls.write_lora_layers(
        state_dict=state_dict,
        save_directory=save_directory,
        is_main_process=is_main_process,
        weight_name=weight_name,
        save_function=save_function,
        safe_serialization=safe_serialization,
    )

mindone.diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.unfuse_lora(components=['unet', 'text_encoder'], **kwargs)

Reverses the effect of pipe.fuse_lora().

This is an experimental API.

PARAMETER	DESCRIPTION
components	List of LoRA-injectable components to unfuse LoRA from. TYPE: `List[str]` DEFAULT: ['unet', 'text_encoder']
unfuse_unet	Whether to unfuse the UNet LoRA parameters. TYPE: `bool`, defaults to `True`
unfuse_text_encoder	Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the LoRA parameters then it won't have any effect. TYPE: `bool`, defaults to `True`

Source code in mindone/diffusers/loaders/lora_pipeline.py

def unfuse_lora(self, components: List[str] = ["unet", "text_encoder"], **kwargs):
    r"""
    Reverses the effect of
    [`pipe.fuse_lora()`](https://huggingface.co/docs/diffusers/main/en/api/loaders#diffusers.loaders.LoraBaseMixin.fuse_lora).

    <Tip warning={true}>

    This is an experimental API.

    </Tip>

    Args:
        components (`List[str]`): List of LoRA-injectable components to unfuse LoRA from.
        unfuse_unet (`bool`, defaults to `True`): Whether to unfuse the UNet LoRA parameters.
        unfuse_text_encoder (`bool`, defaults to `True`):
            Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the
            LoRA parameters then it won't have any effect.
    """
    super().unfuse_lora(components=components)

mindone.diffusers.loaders.lora_pipeline.StableDiffusionXLLoraLoaderMixin

Bases: LoraBaseMixin

Load LoRA layers into Stable Diffusion XL [UNet2DConditionModel], CLIPTextModel, and CLIPTextModelWithProjection.

Source code in mindone/diffusers/loaders/lora_pipeline.py

class StableDiffusionXLLoraLoaderMixin(LoraBaseMixin):
    r"""
    Load LoRA layers into Stable Diffusion XL [`UNet2DConditionModel`],
    [`CLIPTextModel`](https://huggingface.co/docs/transformers/model_doc/clip#transformers.CLIPTextModel), and
    [`CLIPTextModelWithProjection`](https://huggingface.co/docs/transformers/model_doc/clip#transformers.CLIPTextModelWithProjection).
    """

    _lora_loadable_modules = ["unet", "text_encoder", "text_encoder_2"]
    unet_name = UNET_NAME
    text_encoder_name = TEXT_ENCODER_NAME

    def load_lora_weights(
        self,
        pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]],
        adapter_name: Optional[str] = None,
        **kwargs,
    ):
        """
        Load LoRA weights specified in `pretrained_model_name_or_path_or_dict` into `self.unet` and
        `self.text_encoder`.

        All kwargs are forwarded to `self.lora_state_dict`.

        See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`] for more details on how the state dict is
        loaded.

        See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_unet`] for more details on how the state dict is
        loaded into `self.unet`.

        See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder`] for more details on how the state
        dict is loaded into `self.text_encoder`.

        Parameters:
            pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
                See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
            adapter_name (`str`, *optional*):
                Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
                `default_{i}` where i is the total number of adapters being loaded.
            kwargs (`dict`, *optional*):
                See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
        """
        # We could have accessed the unet config from `lora_state_dict()` too. We pass
        # it here explicitly to be able to tell that it's coming from an SDXL
        # pipeline.

        # if a dict is passed, copy it instead of modifying it inplace
        if isinstance(pretrained_model_name_or_path_or_dict, dict):
            pretrained_model_name_or_path_or_dict = pretrained_model_name_or_path_or_dict.copy()

        # First, ensure that the checkpoint is a compatible one and can be successfully loaded.
        state_dict, network_alphas = self.lora_state_dict(
            pretrained_model_name_or_path_or_dict,
            unet_config=self.unet.config,
            **kwargs,
        )
        is_correct_format = all("lora" in key or "dora_scale" in key for key in state_dict.keys())
        if not is_correct_format:
            raise ValueError("Invalid LoRA checkpoint.")

        self.load_lora_into_unet(
            state_dict, network_alphas=network_alphas, unet=self.unet, adapter_name=adapter_name, _pipeline=self
        )
        text_encoder_state_dict = {k: v for k, v in state_dict.items() if "text_encoder." in k}
        if len(text_encoder_state_dict) > 0:
            self.load_lora_into_text_encoder(
                text_encoder_state_dict,
                network_alphas=network_alphas,
                text_encoder=self.text_encoder,
                prefix="text_encoder",
                lora_scale=self.lora_scale,
                adapter_name=adapter_name,
                _pipeline=self,
            )

        text_encoder_2_state_dict = {k: v for k, v in state_dict.items() if "text_encoder_2." in k}
        if len(text_encoder_2_state_dict) > 0:
            self.load_lora_into_text_encoder(
                text_encoder_2_state_dict,
                network_alphas=network_alphas,
                text_encoder=self.text_encoder_2,
                prefix="text_encoder_2",
                lora_scale=self.lora_scale,
                adapter_name=adapter_name,
                _pipeline=self,
            )

    @classmethod
    @validate_hf_hub_args
    # Copied from diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.lora_state_dict
    def lora_state_dict(
        cls,
        pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]],
        **kwargs,
    ):
        r"""
        Return state dict for lora weights and the network alphas.

        <Tip warning={true}>

        We support loading A1111 formatted LoRA checkpoints in a limited capacity.

        This function is experimental and might change in the future.

        </Tip>

        Parameters:
            pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
                Can be either:

                    - A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on
                      the Hub.
                    - A path to a *directory* (for example `./my_model_directory`) containing the model weights saved
                      with [`ModelMixin.save_pretrained`].
                    - A MindSpore state dict.

            cache_dir (`Union[str, os.PathLike]`, *optional*):
                Path to a directory where a downloaded pretrained model configuration is cached if the standard cache
                is not used.
            force_download (`bool`, *optional*, defaults to `False`):
                Whether or not to force the (re-)download of the model weights and configuration files, overriding the
                cached versions if they exist.

            proxies (`Dict[str, str]`, *optional*):
                A dictionary of proxy servers to use by protocol or endpoint, for example, `{'http': 'foo.bar:3128',
                'http://hostname': 'foo.bar:4012'}`. The proxies are used on each request.
            local_files_only (`bool`, *optional*, defaults to `False`):
                Whether to only load local model weights and configuration files or not. If set to `True`, the model
                won't be downloaded from the Hub.
            token (`str` or *bool*, *optional*):
                The token to use as HTTP bearer authorization for remote files. If `True`, the token generated from
                `diffusers-cli login` (stored in `~/.huggingface`) is used.
            revision (`str`, *optional*, defaults to `"main"`):
                The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier
                allowed by Git.
            subfolder (`str`, *optional*, defaults to `""`):
                The subfolder location of a model file within a larger model repository on the Hub or locally.
            weight_name (`str`, *optional*, defaults to None):
                Name of the serialized state dict file.
        """
        # Load the main state dict first which has the LoRA layers for either of
        # UNet and text encoder or both.
        cache_dir = kwargs.pop("cache_dir", None)
        force_download = kwargs.pop("force_download", False)
        proxies = kwargs.pop("proxies", None)
        local_files_only = kwargs.pop("local_files_only", None)
        token = kwargs.pop("token", None)
        revision = kwargs.pop("revision", None)
        subfolder = kwargs.pop("subfolder", None)
        weight_name = kwargs.pop("weight_name", None)
        unet_config = kwargs.pop("unet_config", None)
        use_safetensors = kwargs.pop("use_safetensors", None)

        allow_pickle = False
        if use_safetensors is None:
            use_safetensors = True
            allow_pickle = True

        user_agent = {
            "file_type": "attn_procs_weights",
            "framework": "pytorch",
        }

        state_dict = cls._fetch_state_dict(
            pretrained_model_name_or_path_or_dict=pretrained_model_name_or_path_or_dict,
            weight_name=weight_name,
            use_safetensors=use_safetensors,
            local_files_only=local_files_only,
            cache_dir=cache_dir,
            force_download=force_download,
            proxies=proxies,
            token=token,
            revision=revision,
            subfolder=subfolder,
            user_agent=user_agent,
            allow_pickle=allow_pickle,
        )

        network_alphas = None
        # TODO: replace it with a method from `state_dict_utils`
        if all(
            (
                k.startswith("lora_te_")
                or k.startswith("lora_unet_")
                or k.startswith("lora_te1_")
                or k.startswith("lora_te2_")
            )
            for k in state_dict.keys()
        ):
            # Map SDXL blocks correctly.
            if unet_config is not None:
                # use unet config to remap block numbers
                state_dict = _maybe_map_sgm_blocks_to_diffusers(state_dict, unet_config)
            state_dict, network_alphas = _convert_non_diffusers_lora_to_diffusers(state_dict)

        return state_dict, network_alphas

    @classmethod
    # Copied from diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.load_lora_into_unet
    def load_lora_into_unet(cls, state_dict, network_alphas, unet, adapter_name=None, _pipeline=None):
        """
        This will load the LoRA layers specified in `state_dict` into `unet`.

        Parameters:
            state_dict (`dict`):
                A standard state dict containing the lora layer parameters. The keys can either be indexed directly
                into the unet or prefixed with an additional `unet` which can be used to distinguish between text
                encoder lora layers.
            network_alphas (`Dict[str, float]`):
                The value of the network alpha used for stable learning and preventing underflow. This value has the
                same meaning as the `--network_alpha` option in the kohya-ss trainer script. Refer to [this
                link](https://github.com/darkstorm2150/sd-scripts/blob/main/docs/train_network_README-en.md#execute-learning).
            unet (`UNet2DConditionModel`):
                The UNet model to load the LoRA layers into.
            adapter_name (`str`, *optional*):
                Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
                `default_{i}` where i is the total number of adapters being loaded.
        """
        # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
        # then the `state_dict` keys should have `cls.unet_name` and/or `cls.text_encoder_name` as
        # their prefixes.
        keys = list(state_dict.keys())
        only_text_encoder = all(key.startswith(cls.text_encoder_name) for key in keys)
        if not only_text_encoder:
            # Load the layers corresponding to UNet.
            logger.info(f"Loading {cls.unet_name}.")
            unet.load_attn_procs(
                state_dict, network_alphas=network_alphas, adapter_name=adapter_name, _pipeline=_pipeline
            )

    @classmethod
    # Copied from diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder
    def load_lora_into_text_encoder(
        cls,
        state_dict,
        network_alphas,
        text_encoder,
        prefix=None,
        lora_scale=1.0,
        adapter_name=None,
        _pipeline=None,
    ):
        """
        This will load the LoRA layers specified in `state_dict` into `text_encoder`

        Parameters:
            state_dict (`dict`):
                A standard state dict containing the lora layer parameters. The key should be prefixed with an
                additional `text_encoder` to distinguish between unet lora layers.
            network_alphas (`Dict[str, float]`):
                See `LoRALinearLayer` for more details.
            text_encoder (`CLIPTextModel`):
                The text encoder model to load the LoRA layers into.
            prefix (`str`):
                Expected prefix of the `text_encoder` in the `state_dict`.
            lora_scale (`float`):
                How much to scale the output of the lora linear layer before it is added with the output of the regular
                lora layer.
            adapter_name (`str`, *optional*):
                Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
                `default_{i}` where i is the total number of adapters being loaded.
        """
        from mindone.diffusers._peft import LoraConfig

        # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
        # then the `state_dict` keys should have `self.unet_name` and/or `self.text_encoder_name` as
        # their prefixes.
        keys = list(state_dict.keys())
        prefix = cls.text_encoder_name if prefix is None else prefix

        # Safe prefix to check with.
        if any(cls.text_encoder_name in key for key in keys):
            # Load the layers corresponding to text encoder and make necessary adjustments.
            text_encoder_keys = [k for k in keys if k.startswith(prefix) and k.split(".")[0] == prefix]
            text_encoder_lora_state_dict = {
                k.replace(f"{prefix}.", ""): v for k, v in state_dict.items() if k in text_encoder_keys
            }

            if len(text_encoder_lora_state_dict) > 0:
                logger.info(f"Loading {prefix}.")
                rank = {}
                text_encoder_lora_state_dict = convert_state_dict_to_diffusers(text_encoder_lora_state_dict)

                # convert state dict
                text_encoder_lora_state_dict = convert_state_dict_to_peft(text_encoder_lora_state_dict)

                for name, _ in text_encoder_attn_modules(text_encoder):
                    for module in ("out_proj", "q_proj", "k_proj", "v_proj"):
                        rank_key = f"{name}.{module}.lora_B.weight"
                        if rank_key not in text_encoder_lora_state_dict:
                            continue
                        rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

                for name, _ in text_encoder_mlp_modules(text_encoder):
                    for module in ("fc1", "fc2"):
                        rank_key = f"{name}.{module}.lora_B.weight"
                        if rank_key not in text_encoder_lora_state_dict:
                            continue
                        rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

                if network_alphas is not None:
                    alpha_keys = [
                        k for k in network_alphas.keys() if k.startswith(prefix) and k.split(".")[0] == prefix
                    ]
                    network_alphas = {
                        k.replace(f"{prefix}.", ""): v for k, v in network_alphas.items() if k in alpha_keys
                    }

                lora_config_kwargs = get_peft_kwargs(rank, network_alphas, text_encoder_lora_state_dict, is_unet=False)
                if "use_dora" in lora_config_kwargs:
                    if lora_config_kwargs["use_dora"]:
                        if is_peft_version("<", "0.9.0"):
                            raise ValueError(
                                "You need `peft` 0.9.0 at least to use DoRA-enabled LoRAs. Please upgrade your installation of `peft`."
                            )
                    else:
                        if is_peft_version("<", "0.9.0"):
                            lora_config_kwargs.pop("use_dora")
                lora_config = LoraConfig(**lora_config_kwargs)

                # adapter_name
                if adapter_name is None:
                    adapter_name = get_adapter_name(text_encoder)

                # inject LoRA layers and load the state dict
                # in transformers we automatically check whether the adapter name is already in use or not
                text_encoder.load_adapter(
                    adapter_name=adapter_name,
                    adapter_state_dict=text_encoder_lora_state_dict,
                    peft_config=lora_config,
                )

                # scale LoRA layers with `lora_scale`
                scale_lora_layers(text_encoder, weight=lora_scale)

                text_encoder.to(dtype=text_encoder.dtype)

    @classmethod
    def save_lora_weights(
        cls,
        save_directory: Union[str, os.PathLike],
        unet_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
        text_encoder_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
        text_encoder_2_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
        is_main_process: bool = True,
        weight_name: str = None,
        save_function: Callable = None,
        safe_serialization: bool = True,
    ):
        r"""
        Save the LoRA parameters corresponding to the UNet and text encoder.

        Arguments:
            save_directory (`str` or `os.PathLike`):
                Directory to save LoRA parameters to. Will be created if it doesn't exist.
            unet_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
                State dict of the LoRA layers corresponding to the `unet`.
            text_encoder_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
                State dict of the LoRA layers corresponding to the `text_encoder`. Must explicitly pass the text
                encoder LoRA state dict because it comes from 🤗 Transformers.
            text_encoder_2_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
                State dict of the LoRA layers corresponding to the `text_encoder_2`. Must explicitly pass the text
                encoder LoRA state dict because it comes from 🤗 Transformers.
            is_main_process (`bool`, *optional*, defaults to `True`):
                Whether the process calling this is the main process or not. Useful during distributed training and you
                need to call this function on all processes. In this case, set `is_main_process=True` only on the main
                process to avoid race conditions.
            save_function (`Callable`):
                The function to use to save the state dictionary. Useful during distributed training when you need to
                replace `mindspore.save_checkpoint` with another method. Can be configured with the environment variable
                `DIFFUSERS_SAVE_MODE`.
            safe_serialization (`bool`, *optional*, defaults to `True`):
                Whether to save the model using `safetensors` or the traditional MindSpore way.
        """
        state_dict = {}

        if not (unet_lora_layers or text_encoder_lora_layers or text_encoder_2_lora_layers):
            raise ValueError(
                "You must pass at least one of `unet_lora_layers`, `text_encoder_lora_layers` or `text_encoder_2_lora_layers`."
            )

        if unet_lora_layers:
            state_dict.update(cls.pack_weights(unet_lora_layers, "unet"))

        if text_encoder_lora_layers:
            state_dict.update(cls.pack_weights(text_encoder_lora_layers, "text_encoder"))

        if text_encoder_2_lora_layers:
            state_dict.update(cls.pack_weights(text_encoder_2_lora_layers, "text_encoder_2"))

        cls.write_lora_layers(
            state_dict=state_dict,
            save_directory=save_directory,
            is_main_process=is_main_process,
            weight_name=weight_name,
            save_function=save_function,
            safe_serialization=safe_serialization,
        )

    def fuse_lora(
        self,
        components: List[str] = ["unet", "text_encoder", "text_encoder_2"],
        lora_scale: float = 1.0,
        safe_fusing: bool = False,
        adapter_names: Optional[List[str]] = None,
        **kwargs,
    ):
        r"""
        Fuses the LoRA parameters into the original parameters of the corresponding blocks.

        <Tip warning={true}>

        This is an experimental API.

        </Tip>

        Args:
            components: (`List[str]`): List of LoRA-injectable components to fuse the LoRAs into.
            lora_scale (`float`, defaults to 1.0):
                Controls how much to influence the outputs with the LoRA parameters.
            safe_fusing (`bool`, defaults to `False`):
                Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them.
            adapter_names (`List[str]`, *optional*):
                Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused.

        Example:

        ```py
        from mindone.diffusers import DiffusionPipeline
        import mindspore

        pipeline = DiffusionPipeline.from_pretrained(
            "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
        )
        pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
        pipeline.fuse_lora(lora_scale=0.7)
        ```
        """
        super().fuse_lora(
            components=components, lora_scale=lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names
        )

    def unfuse_lora(self, components: List[str] = ["unet", "text_encoder", "text_encoder_2"], **kwargs):
        r"""
        Reverses the effect of
        [`pipe.fuse_lora()`](https://huggingface.co/docs/diffusers/main/en/api/loaders#diffusers.loaders.LoraBaseMixin.fuse_lora).

        <Tip warning={true}>

        This is an experimental API.

        </Tip>

        Args:
            components (`List[str]`): List of LoRA-injectable components to unfuse LoRA from.
            unfuse_unet (`bool`, defaults to `True`): Whether to unfuse the UNet LoRA parameters.
            unfuse_text_encoder (`bool`, defaults to `True`):
                Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the
                LoRA parameters then it won't have any effect.
        """
        super().unfuse_lora(components=components)

mindone.diffusers.loaders.lora_pipeline.StableDiffusionXLLoraLoaderMixin.fuse_lora(components=['unet', 'text_encoder', 'text_encoder_2'], lora_scale=1.0, safe_fusing=False, adapter_names=None, **kwargs)

Fuses the LoRA parameters into the original parameters of the corresponding blocks.

This is an experimental API.

PARAMETER	DESCRIPTION
components	(List[str]): List of LoRA-injectable components to fuse the LoRAs into. TYPE: List[str] DEFAULT: ['unet', 'text_encoder', 'text_encoder_2']
lora_scale	Controls how much to influence the outputs with the LoRA parameters. TYPE: `float`, defaults to 1.0 DEFAULT: 1.0
safe_fusing	Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them. TYPE: `bool`, defaults to `False` DEFAULT: False
adapter_names	Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused. TYPE: `List[str]`, *optional* DEFAULT: None

from mindone.diffusers import DiffusionPipeline
import mindspore

pipeline = DiffusionPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
)
pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
pipeline.fuse_lora(lora_scale=0.7)

Source code in mindone/diffusers/loaders/lora_pipeline.py

def fuse_lora(
    self,
    components: List[str] = ["unet", "text_encoder", "text_encoder_2"],
    lora_scale: float = 1.0,
    safe_fusing: bool = False,
    adapter_names: Optional[List[str]] = None,
    **kwargs,
):
    r"""
    Fuses the LoRA parameters into the original parameters of the corresponding blocks.

    <Tip warning={true}>

    This is an experimental API.

    </Tip>

    Args:
        components: (`List[str]`): List of LoRA-injectable components to fuse the LoRAs into.
        lora_scale (`float`, defaults to 1.0):
            Controls how much to influence the outputs with the LoRA parameters.
        safe_fusing (`bool`, defaults to `False`):
            Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them.
        adapter_names (`List[str]`, *optional*):
            Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused.

    Example:

    ```py
    from mindone.diffusers import DiffusionPipeline
    import mindspore

    pipeline = DiffusionPipeline.from_pretrained(
        "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
    )
    pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
    pipeline.fuse_lora(lora_scale=0.7)
    ```
    """
    super().fuse_lora(
        components=components, lora_scale=lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names
    )

mindone.diffusers.loaders.lora_pipeline.StableDiffusionXLLoraLoaderMixin.load_lora_into_text_encoder(state_dict, network_alphas, text_encoder, prefix=None, lora_scale=1.0, adapter_name=None, _pipeline=None) classmethod

This will load the LoRA layers specified in state_dict into text_encoder

PARAMETER	DESCRIPTION
state_dict	A standard state dict containing the lora layer parameters. The key should be prefixed with an additional text_encoder to distinguish between unet lora layers. TYPE: `dict`
network_alphas	See LoRALinearLayer for more details. TYPE: `Dict[str, float]`
text_encoder	The text encoder model to load the LoRA layers into. TYPE: `CLIPTextModel`
prefix	Expected prefix of the text_encoder in the state_dict. TYPE: `str` DEFAULT: None
lora_scale	How much to scale the output of the lora linear layer before it is added with the output of the regular lora layer. TYPE: `float` DEFAULT: 1.0
adapter_name	Adapter name to be used for referencing the loaded adapter model. If not specified, it will use default_{i} where i is the total number of adapters being loaded. TYPE: `str`, *optional* DEFAULT: None

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
# Copied from diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder
def load_lora_into_text_encoder(
    cls,
    state_dict,
    network_alphas,
    text_encoder,
    prefix=None,
    lora_scale=1.0,
    adapter_name=None,
    _pipeline=None,
):
    """
    This will load the LoRA layers specified in `state_dict` into `text_encoder`

    Parameters:
        state_dict (`dict`):
            A standard state dict containing the lora layer parameters. The key should be prefixed with an
            additional `text_encoder` to distinguish between unet lora layers.
        network_alphas (`Dict[str, float]`):
            See `LoRALinearLayer` for more details.
        text_encoder (`CLIPTextModel`):
            The text encoder model to load the LoRA layers into.
        prefix (`str`):
            Expected prefix of the `text_encoder` in the `state_dict`.
        lora_scale (`float`):
            How much to scale the output of the lora linear layer before it is added with the output of the regular
            lora layer.
        adapter_name (`str`, *optional*):
            Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
            `default_{i}` where i is the total number of adapters being loaded.
    """
    from mindone.diffusers._peft import LoraConfig

    # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
    # then the `state_dict` keys should have `self.unet_name` and/or `self.text_encoder_name` as
    # their prefixes.
    keys = list(state_dict.keys())
    prefix = cls.text_encoder_name if prefix is None else prefix

    # Safe prefix to check with.
    if any(cls.text_encoder_name in key for key in keys):
        # Load the layers corresponding to text encoder and make necessary adjustments.
        text_encoder_keys = [k for k in keys if k.startswith(prefix) and k.split(".")[0] == prefix]
        text_encoder_lora_state_dict = {
            k.replace(f"{prefix}.", ""): v for k, v in state_dict.items() if k in text_encoder_keys
        }

        if len(text_encoder_lora_state_dict) > 0:
            logger.info(f"Loading {prefix}.")
            rank = {}
            text_encoder_lora_state_dict = convert_state_dict_to_diffusers(text_encoder_lora_state_dict)

            # convert state dict
            text_encoder_lora_state_dict = convert_state_dict_to_peft(text_encoder_lora_state_dict)

            for name, _ in text_encoder_attn_modules(text_encoder):
                for module in ("out_proj", "q_proj", "k_proj", "v_proj"):
                    rank_key = f"{name}.{module}.lora_B.weight"
                    if rank_key not in text_encoder_lora_state_dict:
                        continue
                    rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

            for name, _ in text_encoder_mlp_modules(text_encoder):
                for module in ("fc1", "fc2"):
                    rank_key = f"{name}.{module}.lora_B.weight"
                    if rank_key not in text_encoder_lora_state_dict:
                        continue
                    rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

            if network_alphas is not None:
                alpha_keys = [
                    k for k in network_alphas.keys() if k.startswith(prefix) and k.split(".")[0] == prefix
                ]
                network_alphas = {
                    k.replace(f"{prefix}.", ""): v for k, v in network_alphas.items() if k in alpha_keys
                }

            lora_config_kwargs = get_peft_kwargs(rank, network_alphas, text_encoder_lora_state_dict, is_unet=False)
            if "use_dora" in lora_config_kwargs:
                if lora_config_kwargs["use_dora"]:
                    if is_peft_version("<", "0.9.0"):
                        raise ValueError(
                            "You need `peft` 0.9.0 at least to use DoRA-enabled LoRAs. Please upgrade your installation of `peft`."
                        )
                else:
                    if is_peft_version("<", "0.9.0"):
                        lora_config_kwargs.pop("use_dora")
            lora_config = LoraConfig(**lora_config_kwargs)

            # adapter_name
            if adapter_name is None:
                adapter_name = get_adapter_name(text_encoder)

            # inject LoRA layers and load the state dict
            # in transformers we automatically check whether the adapter name is already in use or not
            text_encoder.load_adapter(
                adapter_name=adapter_name,
                adapter_state_dict=text_encoder_lora_state_dict,
                peft_config=lora_config,
            )

            # scale LoRA layers with `lora_scale`
            scale_lora_layers(text_encoder, weight=lora_scale)

            text_encoder.to(dtype=text_encoder.dtype)

mindone.diffusers.loaders.lora_pipeline.StableDiffusionXLLoraLoaderMixin.load_lora_into_unet(state_dict, network_alphas, unet, adapter_name=None, _pipeline=None) classmethod

This will load the LoRA layers specified in state_dict into unet.

PARAMETER	DESCRIPTION
state_dict	A standard state dict containing the lora layer parameters. The keys can either be indexed directly into the unet or prefixed with an additional unet which can be used to distinguish between text encoder lora layers. TYPE: `dict`
network_alphas	The value of the network alpha used for stable learning and preventing underflow. This value has the same meaning as the --network_alpha option in the kohya-ss trainer script. Refer to this link. TYPE: `Dict[str, float]`
unet	The UNet model to load the LoRA layers into. TYPE: `UNet2DConditionModel`
adapter_name	Adapter name to be used for referencing the loaded adapter model. If not specified, it will use default_{i} where i is the total number of adapters being loaded. TYPE: `str`, *optional* DEFAULT: None

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
# Copied from diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.load_lora_into_unet
def load_lora_into_unet(cls, state_dict, network_alphas, unet, adapter_name=None, _pipeline=None):
    """
    This will load the LoRA layers specified in `state_dict` into `unet`.

    Parameters:
        state_dict (`dict`):
            A standard state dict containing the lora layer parameters. The keys can either be indexed directly
            into the unet or prefixed with an additional `unet` which can be used to distinguish between text
            encoder lora layers.
        network_alphas (`Dict[str, float]`):
            The value of the network alpha used for stable learning and preventing underflow. This value has the
            same meaning as the `--network_alpha` option in the kohya-ss trainer script. Refer to [this
            link](https://github.com/darkstorm2150/sd-scripts/blob/main/docs/train_network_README-en.md#execute-learning).
        unet (`UNet2DConditionModel`):
            The UNet model to load the LoRA layers into.
        adapter_name (`str`, *optional*):
            Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
            `default_{i}` where i is the total number of adapters being loaded.
    """
    # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
    # then the `state_dict` keys should have `cls.unet_name` and/or `cls.text_encoder_name` as
    # their prefixes.
    keys = list(state_dict.keys())
    only_text_encoder = all(key.startswith(cls.text_encoder_name) for key in keys)
    if not only_text_encoder:
        # Load the layers corresponding to UNet.
        logger.info(f"Loading {cls.unet_name}.")
        unet.load_attn_procs(
            state_dict, network_alphas=network_alphas, adapter_name=adapter_name, _pipeline=_pipeline
        )

mindone.diffusers.loaders.lora_pipeline.StableDiffusionXLLoraLoaderMixin.load_lora_weights(pretrained_model_name_or_path_or_dict, adapter_name=None, **kwargs)

Load LoRA weights specified in pretrained_model_name_or_path_or_dict into self.unet and self.text_encoder.

All kwargs are forwarded to self.lora_state_dict.

See [~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict] for more details on how the state dict is loaded.

See [~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_unet] for more details on how the state dict is loaded into self.unet.

See [~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder] for more details on how the state dict is loaded into self.text_encoder.

PARAMETER	DESCRIPTION
pretrained_model_name_or_path_or_dict	See [~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict]. TYPE: `str` or `os.PathLike` or `dict`
adapter_name	Adapter name to be used for referencing the loaded adapter model. If not specified, it will use default_{i} where i is the total number of adapters being loaded. TYPE: `str`, *optional* DEFAULT: None
kwargs	See [~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict]. TYPE: `dict`, *optional* DEFAULT: {}

Source code in mindone/diffusers/loaders/lora_pipeline.py

def load_lora_weights(
    self,
    pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]],
    adapter_name: Optional[str] = None,
    **kwargs,
):
    """
    Load LoRA weights specified in `pretrained_model_name_or_path_or_dict` into `self.unet` and
    `self.text_encoder`.

    All kwargs are forwarded to `self.lora_state_dict`.

    See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`] for more details on how the state dict is
    loaded.

    See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_unet`] for more details on how the state dict is
    loaded into `self.unet`.

    See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder`] for more details on how the state
    dict is loaded into `self.text_encoder`.

    Parameters:
        pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
            See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
        adapter_name (`str`, *optional*):
            Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
            `default_{i}` where i is the total number of adapters being loaded.
        kwargs (`dict`, *optional*):
            See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
    """
    # We could have accessed the unet config from `lora_state_dict()` too. We pass
    # it here explicitly to be able to tell that it's coming from an SDXL
    # pipeline.

    # if a dict is passed, copy it instead of modifying it inplace
    if isinstance(pretrained_model_name_or_path_or_dict, dict):
        pretrained_model_name_or_path_or_dict = pretrained_model_name_or_path_or_dict.copy()

    # First, ensure that the checkpoint is a compatible one and can be successfully loaded.
    state_dict, network_alphas = self.lora_state_dict(
        pretrained_model_name_or_path_or_dict,
        unet_config=self.unet.config,
        **kwargs,
    )
    is_correct_format = all("lora" in key or "dora_scale" in key for key in state_dict.keys())
    if not is_correct_format:
        raise ValueError("Invalid LoRA checkpoint.")

    self.load_lora_into_unet(
        state_dict, network_alphas=network_alphas, unet=self.unet, adapter_name=adapter_name, _pipeline=self
    )
    text_encoder_state_dict = {k: v for k, v in state_dict.items() if "text_encoder." in k}
    if len(text_encoder_state_dict) > 0:
        self.load_lora_into_text_encoder(
            text_encoder_state_dict,
            network_alphas=network_alphas,
            text_encoder=self.text_encoder,
            prefix="text_encoder",
            lora_scale=self.lora_scale,
            adapter_name=adapter_name,
            _pipeline=self,
        )

    text_encoder_2_state_dict = {k: v for k, v in state_dict.items() if "text_encoder_2." in k}
    if len(text_encoder_2_state_dict) > 0:
        self.load_lora_into_text_encoder(
            text_encoder_2_state_dict,
            network_alphas=network_alphas,
            text_encoder=self.text_encoder_2,
            prefix="text_encoder_2",
            lora_scale=self.lora_scale,
            adapter_name=adapter_name,
            _pipeline=self,
        )

mindone.diffusers.loaders.lora_pipeline.StableDiffusionXLLoraLoaderMixin.lora_state_dict(pretrained_model_name_or_path_or_dict, **kwargs) classmethod

Return state dict for lora weights and the network alphas.

We support loading A1111 formatted LoRA checkpoints in a limited capacity.

This function is experimental and might change in the future.

PARAMETER	DESCRIPTION
pretrained_model_name_or_path_or_dict	Can be either: - A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on the Hub. - A path to a *directory* (for example `./my_model_directory`) containing the model weights saved with [`ModelMixin.save_pretrained`]. - A MindSpore state dict. TYPE: `str` or `os.PathLike` or `dict`
cache_dir	Path to a directory where a downloaded pretrained model configuration is cached if the standard cache is not used. TYPE: `Union[str, os.PathLike]`, *optional*
force_download	Whether or not to force the (re-)download of the model weights and configuration files, overriding the cached versions if they exist. TYPE: `bool`, *optional*, defaults to `False`
proxies	A dictionary of proxy servers to use by protocol or endpoint, for example, {'http': 'foo.bar:3128', 'http://hostname': 'foo.bar:4012'}. The proxies are used on each request. TYPE: `Dict[str, str]`, *optional*
local_files_only	Whether to only load local model weights and configuration files or not. If set to True, the model won't be downloaded from the Hub. TYPE: `bool`, *optional*, defaults to `False`
token	The token to use as HTTP bearer authorization for remote files. If True, the token generated from diffusers-cli login (stored in ~/.huggingface) is used. TYPE: `str` or *bool*, *optional*
revision	The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier allowed by Git. TYPE: `str`, *optional*, defaults to `"main"`
subfolder	The subfolder location of a model file within a larger model repository on the Hub or locally. TYPE: `str`, *optional*, defaults to `""`
weight_name	Name of the serialized state dict file. TYPE: `str`, *optional*, defaults to None

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
@validate_hf_hub_args
# Copied from diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.lora_state_dict
def lora_state_dict(
    cls,
    pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]],
    **kwargs,
):
    r"""
    Return state dict for lora weights and the network alphas.

    <Tip warning={true}>

    We support loading A1111 formatted LoRA checkpoints in a limited capacity.

    This function is experimental and might change in the future.

    </Tip>

    Parameters:
        pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
            Can be either:

                - A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on
                  the Hub.
                - A path to a *directory* (for example `./my_model_directory`) containing the model weights saved
                  with [`ModelMixin.save_pretrained`].
                - A MindSpore state dict.

        cache_dir (`Union[str, os.PathLike]`, *optional*):
            Path to a directory where a downloaded pretrained model configuration is cached if the standard cache
            is not used.
        force_download (`bool`, *optional*, defaults to `False`):
            Whether or not to force the (re-)download of the model weights and configuration files, overriding the
            cached versions if they exist.

        proxies (`Dict[str, str]`, *optional*):
            A dictionary of proxy servers to use by protocol or endpoint, for example, `{'http': 'foo.bar:3128',
            'http://hostname': 'foo.bar:4012'}`. The proxies are used on each request.
        local_files_only (`bool`, *optional*, defaults to `False`):
            Whether to only load local model weights and configuration files or not. If set to `True`, the model
            won't be downloaded from the Hub.
        token (`str` or *bool*, *optional*):
            The token to use as HTTP bearer authorization for remote files. If `True`, the token generated from
            `diffusers-cli login` (stored in `~/.huggingface`) is used.
        revision (`str`, *optional*, defaults to `"main"`):
            The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier
            allowed by Git.
        subfolder (`str`, *optional*, defaults to `""`):
            The subfolder location of a model file within a larger model repository on the Hub or locally.
        weight_name (`str`, *optional*, defaults to None):
            Name of the serialized state dict file.
    """
    # Load the main state dict first which has the LoRA layers for either of
    # UNet and text encoder or both.
    cache_dir = kwargs.pop("cache_dir", None)
    force_download = kwargs.pop("force_download", False)
    proxies = kwargs.pop("proxies", None)
    local_files_only = kwargs.pop("local_files_only", None)
    token = kwargs.pop("token", None)
    revision = kwargs.pop("revision", None)
    subfolder = kwargs.pop("subfolder", None)
    weight_name = kwargs.pop("weight_name", None)
    unet_config = kwargs.pop("unet_config", None)
    use_safetensors = kwargs.pop("use_safetensors", None)

    allow_pickle = False
    if use_safetensors is None:
        use_safetensors = True
        allow_pickle = True

    user_agent = {
        "file_type": "attn_procs_weights",
        "framework": "pytorch",
    }

    state_dict = cls._fetch_state_dict(
        pretrained_model_name_or_path_or_dict=pretrained_model_name_or_path_or_dict,
        weight_name=weight_name,
        use_safetensors=use_safetensors,
        local_files_only=local_files_only,
        cache_dir=cache_dir,
        force_download=force_download,
        proxies=proxies,
        token=token,
        revision=revision,
        subfolder=subfolder,
        user_agent=user_agent,
        allow_pickle=allow_pickle,
    )

    network_alphas = None
    # TODO: replace it with a method from `state_dict_utils`
    if all(
        (
            k.startswith("lora_te_")
            or k.startswith("lora_unet_")
            or k.startswith("lora_te1_")
            or k.startswith("lora_te2_")
        )
        for k in state_dict.keys()
    ):
        # Map SDXL blocks correctly.
        if unet_config is not None:
            # use unet config to remap block numbers
            state_dict = _maybe_map_sgm_blocks_to_diffusers(state_dict, unet_config)
        state_dict, network_alphas = _convert_non_diffusers_lora_to_diffusers(state_dict)

    return state_dict, network_alphas

mindone.diffusers.loaders.lora_pipeline.StableDiffusionXLLoraLoaderMixin.save_lora_weights(save_directory, unet_lora_layers=None, text_encoder_lora_layers=None, text_encoder_2_lora_layers=None, is_main_process=True, weight_name=None, save_function=None, safe_serialization=True) classmethod

Save the LoRA parameters corresponding to the UNet and text encoder.

PARAMETER	DESCRIPTION
save_directory	Directory to save LoRA parameters to. Will be created if it doesn't exist. TYPE: `str` or `os.PathLike`
unet_lora_layers	State dict of the LoRA layers corresponding to the unet. TYPE: `Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]` DEFAULT: None
text_encoder_lora_layers	State dict of the LoRA layers corresponding to the text_encoder. Must explicitly pass the text encoder LoRA state dict because it comes from 🤗 Transformers. TYPE: `Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]` DEFAULT: None
text_encoder_2_lora_layers	State dict of the LoRA layers corresponding to the text_encoder_2. Must explicitly pass the text encoder LoRA state dict because it comes from 🤗 Transformers. TYPE: `Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]` DEFAULT: None
is_main_process	Whether the process calling this is the main process or not. Useful during distributed training and you need to call this function on all processes. In this case, set is_main_process=True only on the main process to avoid race conditions. TYPE: `bool`, *optional*, defaults to `True` DEFAULT: True
save_function	The function to use to save the state dictionary. Useful during distributed training when you need to replace mindspore.save_checkpoint with another method. Can be configured with the environment variable DIFFUSERS_SAVE_MODE. TYPE: `Callable` DEFAULT: None
safe_serialization	Whether to save the model using safetensors or the traditional MindSpore way. TYPE: `bool`, *optional*, defaults to `True` DEFAULT: True

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
def save_lora_weights(
    cls,
    save_directory: Union[str, os.PathLike],
    unet_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
    text_encoder_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
    text_encoder_2_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
    is_main_process: bool = True,
    weight_name: str = None,
    save_function: Callable = None,
    safe_serialization: bool = True,
):
    r"""
    Save the LoRA parameters corresponding to the UNet and text encoder.

    Arguments:
        save_directory (`str` or `os.PathLike`):
            Directory to save LoRA parameters to. Will be created if it doesn't exist.
        unet_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
            State dict of the LoRA layers corresponding to the `unet`.
        text_encoder_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
            State dict of the LoRA layers corresponding to the `text_encoder`. Must explicitly pass the text
            encoder LoRA state dict because it comes from 🤗 Transformers.
        text_encoder_2_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
            State dict of the LoRA layers corresponding to the `text_encoder_2`. Must explicitly pass the text
            encoder LoRA state dict because it comes from 🤗 Transformers.
        is_main_process (`bool`, *optional*, defaults to `True`):
            Whether the process calling this is the main process or not. Useful during distributed training and you
            need to call this function on all processes. In this case, set `is_main_process=True` only on the main
            process to avoid race conditions.
        save_function (`Callable`):
            The function to use to save the state dictionary. Useful during distributed training when you need to
            replace `mindspore.save_checkpoint` with another method. Can be configured with the environment variable
            `DIFFUSERS_SAVE_MODE`.
        safe_serialization (`bool`, *optional*, defaults to `True`):
            Whether to save the model using `safetensors` or the traditional MindSpore way.
    """
    state_dict = {}

    if not (unet_lora_layers or text_encoder_lora_layers or text_encoder_2_lora_layers):
        raise ValueError(
            "You must pass at least one of `unet_lora_layers`, `text_encoder_lora_layers` or `text_encoder_2_lora_layers`."
        )

    if unet_lora_layers:
        state_dict.update(cls.pack_weights(unet_lora_layers, "unet"))

    if text_encoder_lora_layers:
        state_dict.update(cls.pack_weights(text_encoder_lora_layers, "text_encoder"))

    if text_encoder_2_lora_layers:
        state_dict.update(cls.pack_weights(text_encoder_2_lora_layers, "text_encoder_2"))

    cls.write_lora_layers(
        state_dict=state_dict,
        save_directory=save_directory,
        is_main_process=is_main_process,
        weight_name=weight_name,
        save_function=save_function,
        safe_serialization=safe_serialization,
    )

mindone.diffusers.loaders.lora_pipeline.StableDiffusionXLLoraLoaderMixin.unfuse_lora(components=['unet', 'text_encoder', 'text_encoder_2'], **kwargs)

Reverses the effect of pipe.fuse_lora().

This is an experimental API.

PARAMETER	DESCRIPTION
components	List of LoRA-injectable components to unfuse LoRA from. TYPE: `List[str]` DEFAULT: ['unet', 'text_encoder', 'text_encoder_2']
unfuse_unet	Whether to unfuse the UNet LoRA parameters. TYPE: `bool`, defaults to `True`
unfuse_text_encoder	Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the LoRA parameters then it won't have any effect. TYPE: `bool`, defaults to `True`

Source code in mindone/diffusers/loaders/lora_pipeline.py

def unfuse_lora(self, components: List[str] = ["unet", "text_encoder", "text_encoder_2"], **kwargs):
    r"""
    Reverses the effect of
    [`pipe.fuse_lora()`](https://huggingface.co/docs/diffusers/main/en/api/loaders#diffusers.loaders.LoraBaseMixin.fuse_lora).

    <Tip warning={true}>

    This is an experimental API.

    </Tip>

    Args:
        components (`List[str]`): List of LoRA-injectable components to unfuse LoRA from.
        unfuse_unet (`bool`, defaults to `True`): Whether to unfuse the UNet LoRA parameters.
        unfuse_text_encoder (`bool`, defaults to `True`):
            Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the
            LoRA parameters then it won't have any effect.
    """
    super().unfuse_lora(components=components)

mindone.diffusers.loaders.lora_pipeline.SD3LoraLoaderMixin

Bases: LoraBaseMixin

Load LoRA layers into [SD3Transformer2DModel], CLIPTextModel, and CLIPTextModelWithProjection.

Specific to [StableDiffusion3Pipeline].

Source code in mindone/diffusers/loaders/lora_pipeline.py

class SD3LoraLoaderMixin(LoraBaseMixin):
    r"""
    Load LoRA layers into [`SD3Transformer2DModel`],
    [`CLIPTextModel`](https://huggingface.co/docs/transformers/model_doc/clip#transformers.CLIPTextModel), and
    [`CLIPTextModelWithProjection`](https://huggingface.co/docs/transformers/model_doc/clip#transformers.CLIPTextModelWithProjection).

    Specific to [`StableDiffusion3Pipeline`].
    """

    _lora_loadable_modules = ["transformer", "text_encoder", "text_encoder_2"]
    transformer_name = TRANSFORMER_NAME
    text_encoder_name = TEXT_ENCODER_NAME

    @classmethod
    @validate_hf_hub_args
    def lora_state_dict(
        cls,
        pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]],
        **kwargs,
    ):
        r"""
        Return state dict for lora weights and the network alphas.

        <Tip warning={true}>

        We support loading A1111 formatted LoRA checkpoints in a limited capacity.

        This function is experimental and might change in the future.

        </Tip>

        Parameters:
            pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
                Can be either:

                    - A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on
                      the Hub.
                    - A path to a *directory* (for example `./my_model_directory`) containing the model weights saved
                      with [`ModelMixin.save_pretrained`].
                    - A MindSpore state dict.

            cache_dir (`Union[str, os.PathLike]`, *optional*):
                Path to a directory where a downloaded pretrained model configuration is cached if the standard cache
                is not used.
            force_download (`bool`, *optional*, defaults to `False`):
                Whether or not to force the (re-)download of the model weights and configuration files, overriding the
                cached versions if they exist.

            proxies (`Dict[str, str]`, *optional*):
                A dictionary of proxy servers to use by protocol or endpoint, for example, `{'http': 'foo.bar:3128',
                'http://hostname': 'foo.bar:4012'}`. The proxies are used on each request.
            local_files_only (`bool`, *optional*, defaults to `False`):
                Whether to only load local model weights and configuration files or not. If set to `True`, the model
                won't be downloaded from the Hub.
            token (`str` or *bool*, *optional*):
                The token to use as HTTP bearer authorization for remote files. If `True`, the token generated from
                `diffusers-cli login` (stored in `~/.huggingface`) is used.
            revision (`str`, *optional*, defaults to `"main"`):
                The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier
                allowed by Git.
            subfolder (`str`, *optional*, defaults to `""`):
                The subfolder location of a model file within a larger model repository on the Hub or locally.

        """
        # Load the main state dict first which has the LoRA layers for either of
        # transformer and text encoder or both.
        cache_dir = kwargs.pop("cache_dir", None)
        force_download = kwargs.pop("force_download", False)
        proxies = kwargs.pop("proxies", None)
        local_files_only = kwargs.pop("local_files_only", None)
        token = kwargs.pop("token", None)
        revision = kwargs.pop("revision", None)
        subfolder = kwargs.pop("subfolder", None)
        weight_name = kwargs.pop("weight_name", None)
        use_safetensors = kwargs.pop("use_safetensors", None)

        allow_pickle = False
        if use_safetensors is None:
            use_safetensors = True
            allow_pickle = True

        user_agent = {
            "file_type": "attn_procs_weights",
            "framework": "pytorch",
        }

        state_dict = cls._fetch_state_dict(
            pretrained_model_name_or_path_or_dict=pretrained_model_name_or_path_or_dict,
            weight_name=weight_name,
            use_safetensors=use_safetensors,
            local_files_only=local_files_only,
            cache_dir=cache_dir,
            force_download=force_download,
            proxies=proxies,
            token=token,
            revision=revision,
            subfolder=subfolder,
            user_agent=user_agent,
            allow_pickle=allow_pickle,
        )

        return state_dict

    def load_lora_weights(
        self, pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]], adapter_name=None, **kwargs
    ):
        """
        Load LoRA weights specified in `pretrained_model_name_or_path_or_dict` into `self.unet` and
        `self.text_encoder`.

        All kwargs are forwarded to `self.lora_state_dict`.

        See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`] for more details on how the state dict is
        loaded.

        See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_transformer`] for more details on how the state
        dict is loaded into `self.transformer`.

        Parameters:
            pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
                See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
            kwargs (`dict`, *optional*):
                See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
            adapter_name (`str`, *optional*):
                Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
                `default_{i}` where i is the total number of adapters being loaded.
        """
        # if a dict is passed, copy it instead of modifying it inplace
        if isinstance(pretrained_model_name_or_path_or_dict, dict):
            pretrained_model_name_or_path_or_dict = pretrained_model_name_or_path_or_dict.copy()

        # First, ensure that the checkpoint is a compatible one and can be successfully loaded.
        state_dict = self.lora_state_dict(pretrained_model_name_or_path_or_dict, **kwargs)

        is_correct_format = all("lora" in key or "dora_scale" in key for key in state_dict.keys())
        if not is_correct_format:
            raise ValueError("Invalid LoRA checkpoint.")

        self.load_lora_into_transformer(
            state_dict,
            transformer=getattr(self, self.transformer_name) if not hasattr(self, "transformer") else self.transformer,
            adapter_name=adapter_name,
            _pipeline=self,
        )

        text_encoder_state_dict = {k: v for k, v in state_dict.items() if "text_encoder." in k}
        if len(text_encoder_state_dict) > 0:
            self.load_lora_into_text_encoder(
                text_encoder_state_dict,
                network_alphas=None,
                text_encoder=self.text_encoder,
                prefix="text_encoder",
                lora_scale=self.lora_scale,
                adapter_name=adapter_name,
                _pipeline=self,
            )

        text_encoder_2_state_dict = {k: v for k, v in state_dict.items() if "text_encoder_2." in k}
        if len(text_encoder_2_state_dict) > 0:
            self.load_lora_into_text_encoder(
                text_encoder_2_state_dict,
                network_alphas=None,
                text_encoder=self.text_encoder_2,
                prefix="text_encoder_2",
                lora_scale=self.lora_scale,
                adapter_name=adapter_name,
                _pipeline=self,
            )

    @classmethod
    def load_lora_into_transformer(cls, state_dict, transformer, adapter_name=None, _pipeline=None):
        """
        This will load the LoRA layers specified in `state_dict` into `transformer`.

        Parameters:
            state_dict (`dict`):
                A standard state dict containing the lora layer parameters. The keys can either be indexed directly
                into the unet or prefixed with an additional `unet` which can be used to distinguish between text
                encoder lora layers.
            transformer (`SD3Transformer2DModel`):
                The Transformer model to load the LoRA layers into.
            adapter_name (`str`, *optional*):
                Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
                `default_{i}` where i is the total number of adapters being loaded.
        """
        from mindone.diffusers._peft import LoraConfig, inject_adapter_in_model, set_peft_model_state_dict

        keys = list(state_dict.keys())

        transformer_keys = [k for k in keys if k.startswith(cls.transformer_name)]
        state_dict = {
            k.replace(f"{cls.transformer_name}.", ""): v for k, v in state_dict.items() if k in transformer_keys
        }

        if len(state_dict.keys()) > 0:
            # check with first key if is not in peft format
            first_key = next(iter(state_dict.keys()))
            if "lora_A" not in first_key:
                state_dict = convert_unet_state_dict_to_peft(state_dict)

            if adapter_name in getattr(transformer, "peft_config", {}):
                raise ValueError(
                    f"Adapter name {adapter_name} already in use in the transformer - please select a new adapter name."
                )

            rank = {}
            for key, val in state_dict.items():
                if "lora_B" in key:
                    rank[key] = val.shape[1]

            lora_config_kwargs = get_peft_kwargs(rank, network_alpha_dict=None, peft_state_dict=state_dict)
            if "use_dora" in lora_config_kwargs:
                if lora_config_kwargs["use_dora"] and is_peft_version("<", "0.9.0"):
                    raise ValueError(
                        "You need `peft` 0.9.0 at least to use DoRA-enabled LoRAs. Please upgrade your installation of `peft`."
                    )
                else:
                    lora_config_kwargs.pop("use_dora")
            lora_config = LoraConfig(**lora_config_kwargs)

            # adapter_name
            if adapter_name is None:
                adapter_name = get_adapter_name(transformer)

            inject_adapter_in_model(lora_config, transformer, adapter_name=adapter_name)
            incompatible_keys = set_peft_model_state_dict(transformer, state_dict, adapter_name)

            if incompatible_keys is not None:
                # check only for unexpected keys
                unexpected_keys = getattr(incompatible_keys, "unexpected_keys", None)
                if unexpected_keys:
                    logger.warning(
                        f"Loading adapter weights from state_dict led to unexpected keys not found in the model: "
                        f" {unexpected_keys}. "
                    )

    @classmethod
    # Copied from diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder
    def load_lora_into_text_encoder(
        cls,
        state_dict,
        network_alphas,
        text_encoder,
        prefix=None,
        lora_scale=1.0,
        adapter_name=None,
        _pipeline=None,
    ):
        """
        This will load the LoRA layers specified in `state_dict` into `text_encoder`

        Parameters:
            state_dict (`dict`):
                A standard state dict containing the lora layer parameters. The key should be prefixed with an
                additional `text_encoder` to distinguish between unet lora layers.
            network_alphas (`Dict[str, float]`):
                See `LoRALinearLayer` for more details.
            text_encoder (`CLIPTextModel`):
                The text encoder model to load the LoRA layers into.
            prefix (`str`):
                Expected prefix of the `text_encoder` in the `state_dict`.
            lora_scale (`float`):
                How much to scale the output of the lora linear layer before it is added with the output of the regular
                lora layer.
            adapter_name (`str`, *optional*):
                Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
                `default_{i}` where i is the total number of adapters being loaded.
        """
        from mindone.diffusers._peft import LoraConfig

        # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
        # then the `state_dict` keys should have `self.unet_name` and/or `self.text_encoder_name` as
        # their prefixes.
        keys = list(state_dict.keys())
        prefix = cls.text_encoder_name if prefix is None else prefix

        # Safe prefix to check with.
        if any(cls.text_encoder_name in key for key in keys):
            # Load the layers corresponding to text encoder and make necessary adjustments.
            text_encoder_keys = [k for k in keys if k.startswith(prefix) and k.split(".")[0] == prefix]
            text_encoder_lora_state_dict = {
                k.replace(f"{prefix}.", ""): v for k, v in state_dict.items() if k in text_encoder_keys
            }

            if len(text_encoder_lora_state_dict) > 0:
                logger.info(f"Loading {prefix}.")
                rank = {}
                text_encoder_lora_state_dict = convert_state_dict_to_diffusers(text_encoder_lora_state_dict)

                # convert state dict
                text_encoder_lora_state_dict = convert_state_dict_to_peft(text_encoder_lora_state_dict)

                for name, _ in text_encoder_attn_modules(text_encoder):
                    for module in ("out_proj", "q_proj", "k_proj", "v_proj"):
                        rank_key = f"{name}.{module}.lora_B.weight"
                        if rank_key not in text_encoder_lora_state_dict:
                            continue
                        rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

                for name, _ in text_encoder_mlp_modules(text_encoder):
                    for module in ("fc1", "fc2"):
                        rank_key = f"{name}.{module}.lora_B.weight"
                        if rank_key not in text_encoder_lora_state_dict:
                            continue
                        rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

                if network_alphas is not None:
                    alpha_keys = [
                        k for k in network_alphas.keys() if k.startswith(prefix) and k.split(".")[0] == prefix
                    ]
                    network_alphas = {
                        k.replace(f"{prefix}.", ""): v for k, v in network_alphas.items() if k in alpha_keys
                    }

                lora_config_kwargs = get_peft_kwargs(rank, network_alphas, text_encoder_lora_state_dict, is_unet=False)
                if "use_dora" in lora_config_kwargs:
                    if lora_config_kwargs["use_dora"]:
                        if is_peft_version("<", "0.9.0"):
                            raise ValueError(
                                "You need `peft` 0.9.0 at least to use DoRA-enabled LoRAs. Please upgrade your installation of `peft`."
                            )
                    else:
                        if is_peft_version("<", "0.9.0"):
                            lora_config_kwargs.pop("use_dora")
                lora_config = LoraConfig(**lora_config_kwargs)

                # adapter_name
                if adapter_name is None:
                    adapter_name = get_adapter_name(text_encoder)

                # inject LoRA layers and load the state dict
                # in transformers we automatically check whether the adapter name is already in use or not
                text_encoder.load_adapter(
                    adapter_name=adapter_name,
                    adapter_state_dict=text_encoder_lora_state_dict,
                    peft_config=lora_config,
                )

                # scale LoRA layers with `lora_scale`
                scale_lora_layers(text_encoder, weight=lora_scale)

                text_encoder.to(dtype=text_encoder.dtype)

    @classmethod
    def save_lora_weights(
        cls,
        save_directory: Union[str, os.PathLike],
        transformer_lora_layers: Dict[str, nn.Cell] = None,
        text_encoder_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
        text_encoder_2_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
        is_main_process: bool = True,
        weight_name: str = None,
        save_function: Callable = None,
        safe_serialization: bool = True,
    ):
        r"""
        Save the LoRA parameters corresponding to the UNet and text encoder.

        Arguments:
            save_directory (`str` or `os.PathLike`):
                Directory to save LoRA parameters to. Will be created if it doesn't exist.
            transformer_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
                State dict of the LoRA layers corresponding to the `transformer`.
            text_encoder_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
                State dict of the LoRA layers corresponding to the `text_encoder`. Must explicitly pass the text
                encoder LoRA state dict because it comes from 🤗 Transformers.
            text_encoder_2_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
                State dict of the LoRA layers corresponding to the `text_encoder_2`. Must explicitly pass the text
                encoder LoRA state dict because it comes from 🤗 Transformers.
            is_main_process (`bool`, *optional*, defaults to `True`):
                Whether the process calling this is the main process or not. Useful during distributed training and you
                need to call this function on all processes. In this case, set `is_main_process=True` only on the main
                process to avoid race conditions.
            save_function (`Callable`):
                The function to use to save the state dictionary. Useful during distributed training when you need to
                replace `mindspore.save_checkpoint` with another method. Can be configured with the environment variable
                `DIFFUSERS_SAVE_MODE`.
            safe_serialization (`bool`, *optional*, defaults to `True`):
                Whether to save the model using `safetensors` or the traditional MindSpore way.
        """
        state_dict = {}

        if not (transformer_lora_layers or text_encoder_lora_layers or text_encoder_2_lora_layers):
            raise ValueError(
                "You must pass at least one of `transformer_lora_layers`, `text_encoder_lora_layers`, `text_encoder_2_lora_layers`."
            )

        if transformer_lora_layers:
            state_dict.update(cls.pack_weights(transformer_lora_layers, cls.transformer_name))

        if text_encoder_lora_layers:
            state_dict.update(cls.pack_weights(text_encoder_lora_layers, "text_encoder"))

        if text_encoder_2_lora_layers:
            state_dict.update(cls.pack_weights(text_encoder_2_lora_layers, "text_encoder_2"))

        # Save the model
        cls.write_lora_layers(
            state_dict=state_dict,
            save_directory=save_directory,
            is_main_process=is_main_process,
            weight_name=weight_name,
            save_function=save_function,
            safe_serialization=safe_serialization,
        )

    def fuse_lora(
        self,
        components: List[str] = ["transformer", "text_encoder", "text_encoder_2"],
        lora_scale: float = 1.0,
        safe_fusing: bool = False,
        adapter_names: Optional[List[str]] = None,
        **kwargs,
    ):
        r"""
        Fuses the LoRA parameters into the original parameters of the corresponding blocks.

        <Tip warning={true}>

        This is an experimental API.

        </Tip>

        Args:
            components: (`List[str]`): List of LoRA-injectable components to fuse the LoRAs into.
            lora_scale (`float`, defaults to 1.0):
                Controls how much to influence the outputs with the LoRA parameters.
            safe_fusing (`bool`, defaults to `False`):
                Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them.
            adapter_names (`List[str]`, *optional*):
                Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused.

        Example:

        ```py
        from mindone.diffusers import DiffusionPipeline
        import mindspore

        pipeline = DiffusionPipeline.from_pretrained(
            "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
        )
        pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
        pipeline.fuse_lora(lora_scale=0.7)
        ```
        """
        super().fuse_lora(
            components=components, lora_scale=lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names
        )

    def unfuse_lora(self, components: List[str] = ["transformer", "text_encoder", "text_encoder_2"], **kwargs):
        r"""
        Reverses the effect of
        [`pipe.fuse_lora()`](https://huggingface.co/docs/diffusers/main/en/api/loaders#diffusers.loaders.LoraBaseMixin.fuse_lora).

        <Tip warning={true}>

        This is an experimental API.

        </Tip>

        Args:
            components (`List[str]`): List of LoRA-injectable components to unfuse LoRA from.
            unfuse_unet (`bool`, defaults to `True`): Whether to unfuse the UNet LoRA parameters.
            unfuse_text_encoder (`bool`, defaults to `True`):
                Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the
                LoRA parameters then it won't have any effect.
        """
        super().unfuse_lora(components=components)

mindone.diffusers.loaders.lora_pipeline.SD3LoraLoaderMixin.fuse_lora(components=['transformer', 'text_encoder', 'text_encoder_2'], lora_scale=1.0, safe_fusing=False, adapter_names=None, **kwargs)

Fuses the LoRA parameters into the original parameters of the corresponding blocks.

This is an experimental API.

PARAMETER	DESCRIPTION
components	(List[str]): List of LoRA-injectable components to fuse the LoRAs into. TYPE: List[str] DEFAULT: ['transformer', 'text_encoder', 'text_encoder_2']
lora_scale	Controls how much to influence the outputs with the LoRA parameters. TYPE: `float`, defaults to 1.0 DEFAULT: 1.0
safe_fusing	Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them. TYPE: `bool`, defaults to `False` DEFAULT: False
adapter_names	Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused. TYPE: `List[str]`, *optional* DEFAULT: None

from mindone.diffusers import DiffusionPipeline
import mindspore

pipeline = DiffusionPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
)
pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
pipeline.fuse_lora(lora_scale=0.7)

Source code in mindone/diffusers/loaders/lora_pipeline.py

def fuse_lora(
    self,
    components: List[str] = ["transformer", "text_encoder", "text_encoder_2"],
    lora_scale: float = 1.0,
    safe_fusing: bool = False,
    adapter_names: Optional[List[str]] = None,
    **kwargs,
):
    r"""
    Fuses the LoRA parameters into the original parameters of the corresponding blocks.

    <Tip warning={true}>

    This is an experimental API.

    </Tip>

    Args:
        components: (`List[str]`): List of LoRA-injectable components to fuse the LoRAs into.
        lora_scale (`float`, defaults to 1.0):
            Controls how much to influence the outputs with the LoRA parameters.
        safe_fusing (`bool`, defaults to `False`):
            Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them.
        adapter_names (`List[str]`, *optional*):
            Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused.

    Example:

    ```py
    from mindone.diffusers import DiffusionPipeline
    import mindspore

    pipeline = DiffusionPipeline.from_pretrained(
        "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
    )
    pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
    pipeline.fuse_lora(lora_scale=0.7)
    ```
    """
    super().fuse_lora(
        components=components, lora_scale=lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names
    )

mindone.diffusers.loaders.lora_pipeline.SD3LoraLoaderMixin.load_lora_into_text_encoder(state_dict, network_alphas, text_encoder, prefix=None, lora_scale=1.0, adapter_name=None, _pipeline=None) classmethod

This will load the LoRA layers specified in state_dict into text_encoder

PARAMETER	DESCRIPTION
state_dict	A standard state dict containing the lora layer parameters. The key should be prefixed with an additional text_encoder to distinguish between unet lora layers. TYPE: `dict`
network_alphas	See LoRALinearLayer for more details. TYPE: `Dict[str, float]`
text_encoder	The text encoder model to load the LoRA layers into. TYPE: `CLIPTextModel`
prefix	Expected prefix of the text_encoder in the state_dict. TYPE: `str` DEFAULT: None
lora_scale	How much to scale the output of the lora linear layer before it is added with the output of the regular lora layer. TYPE: `float` DEFAULT: 1.0
adapter_name	Adapter name to be used for referencing the loaded adapter model. If not specified, it will use default_{i} where i is the total number of adapters being loaded. TYPE: `str`, *optional* DEFAULT: None

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
# Copied from diffusers.loaders.lora_pipeline.StableDiffusionLoraLoaderMixin.load_lora_into_text_encoder
def load_lora_into_text_encoder(
    cls,
    state_dict,
    network_alphas,
    text_encoder,
    prefix=None,
    lora_scale=1.0,
    adapter_name=None,
    _pipeline=None,
):
    """
    This will load the LoRA layers specified in `state_dict` into `text_encoder`

    Parameters:
        state_dict (`dict`):
            A standard state dict containing the lora layer parameters. The key should be prefixed with an
            additional `text_encoder` to distinguish between unet lora layers.
        network_alphas (`Dict[str, float]`):
            See `LoRALinearLayer` for more details.
        text_encoder (`CLIPTextModel`):
            The text encoder model to load the LoRA layers into.
        prefix (`str`):
            Expected prefix of the `text_encoder` in the `state_dict`.
        lora_scale (`float`):
            How much to scale the output of the lora linear layer before it is added with the output of the regular
            lora layer.
        adapter_name (`str`, *optional*):
            Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
            `default_{i}` where i is the total number of adapters being loaded.
    """
    from mindone.diffusers._peft import LoraConfig

    # If the serialization format is new (introduced in https://github.com/huggingface/diffusers/pull/2918),
    # then the `state_dict` keys should have `self.unet_name` and/or `self.text_encoder_name` as
    # their prefixes.
    keys = list(state_dict.keys())
    prefix = cls.text_encoder_name if prefix is None else prefix

    # Safe prefix to check with.
    if any(cls.text_encoder_name in key for key in keys):
        # Load the layers corresponding to text encoder and make necessary adjustments.
        text_encoder_keys = [k for k in keys if k.startswith(prefix) and k.split(".")[0] == prefix]
        text_encoder_lora_state_dict = {
            k.replace(f"{prefix}.", ""): v for k, v in state_dict.items() if k in text_encoder_keys
        }

        if len(text_encoder_lora_state_dict) > 0:
            logger.info(f"Loading {prefix}.")
            rank = {}
            text_encoder_lora_state_dict = convert_state_dict_to_diffusers(text_encoder_lora_state_dict)

            # convert state dict
            text_encoder_lora_state_dict = convert_state_dict_to_peft(text_encoder_lora_state_dict)

            for name, _ in text_encoder_attn_modules(text_encoder):
                for module in ("out_proj", "q_proj", "k_proj", "v_proj"):
                    rank_key = f"{name}.{module}.lora_B.weight"
                    if rank_key not in text_encoder_lora_state_dict:
                        continue
                    rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

            for name, _ in text_encoder_mlp_modules(text_encoder):
                for module in ("fc1", "fc2"):
                    rank_key = f"{name}.{module}.lora_B.weight"
                    if rank_key not in text_encoder_lora_state_dict:
                        continue
                    rank[rank_key] = text_encoder_lora_state_dict[rank_key].shape[1]

            if network_alphas is not None:
                alpha_keys = [
                    k for k in network_alphas.keys() if k.startswith(prefix) and k.split(".")[0] == prefix
                ]
                network_alphas = {
                    k.replace(f"{prefix}.", ""): v for k, v in network_alphas.items() if k in alpha_keys
                }

            lora_config_kwargs = get_peft_kwargs(rank, network_alphas, text_encoder_lora_state_dict, is_unet=False)
            if "use_dora" in lora_config_kwargs:
                if lora_config_kwargs["use_dora"]:
                    if is_peft_version("<", "0.9.0"):
                        raise ValueError(
                            "You need `peft` 0.9.0 at least to use DoRA-enabled LoRAs. Please upgrade your installation of `peft`."
                        )
                else:
                    if is_peft_version("<", "0.9.0"):
                        lora_config_kwargs.pop("use_dora")
            lora_config = LoraConfig(**lora_config_kwargs)

            # adapter_name
            if adapter_name is None:
                adapter_name = get_adapter_name(text_encoder)

            # inject LoRA layers and load the state dict
            # in transformers we automatically check whether the adapter name is already in use or not
            text_encoder.load_adapter(
                adapter_name=adapter_name,
                adapter_state_dict=text_encoder_lora_state_dict,
                peft_config=lora_config,
            )

            # scale LoRA layers with `lora_scale`
            scale_lora_layers(text_encoder, weight=lora_scale)

            text_encoder.to(dtype=text_encoder.dtype)

mindone.diffusers.loaders.lora_pipeline.SD3LoraLoaderMixin.load_lora_into_transformer(state_dict, transformer, adapter_name=None, _pipeline=None) classmethod

This will load the LoRA layers specified in state_dict into transformer.

PARAMETER	DESCRIPTION
state_dict	A standard state dict containing the lora layer parameters. The keys can either be indexed directly into the unet or prefixed with an additional unet which can be used to distinguish between text encoder lora layers. TYPE: `dict`
transformer	The Transformer model to load the LoRA layers into. TYPE: `SD3Transformer2DModel`
adapter_name	Adapter name to be used for referencing the loaded adapter model. If not specified, it will use default_{i} where i is the total number of adapters being loaded. TYPE: `str`, *optional* DEFAULT: None

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
def load_lora_into_transformer(cls, state_dict, transformer, adapter_name=None, _pipeline=None):
    """
    This will load the LoRA layers specified in `state_dict` into `transformer`.

    Parameters:
        state_dict (`dict`):
            A standard state dict containing the lora layer parameters. The keys can either be indexed directly
            into the unet or prefixed with an additional `unet` which can be used to distinguish between text
            encoder lora layers.
        transformer (`SD3Transformer2DModel`):
            The Transformer model to load the LoRA layers into.
        adapter_name (`str`, *optional*):
            Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
            `default_{i}` where i is the total number of adapters being loaded.
    """
    from mindone.diffusers._peft import LoraConfig, inject_adapter_in_model, set_peft_model_state_dict

    keys = list(state_dict.keys())

    transformer_keys = [k for k in keys if k.startswith(cls.transformer_name)]
    state_dict = {
        k.replace(f"{cls.transformer_name}.", ""): v for k, v in state_dict.items() if k in transformer_keys
    }

    if len(state_dict.keys()) > 0:
        # check with first key if is not in peft format
        first_key = next(iter(state_dict.keys()))
        if "lora_A" not in first_key:
            state_dict = convert_unet_state_dict_to_peft(state_dict)

        if adapter_name in getattr(transformer, "peft_config", {}):
            raise ValueError(
                f"Adapter name {adapter_name} already in use in the transformer - please select a new adapter name."
            )

        rank = {}
        for key, val in state_dict.items():
            if "lora_B" in key:
                rank[key] = val.shape[1]

        lora_config_kwargs = get_peft_kwargs(rank, network_alpha_dict=None, peft_state_dict=state_dict)
        if "use_dora" in lora_config_kwargs:
            if lora_config_kwargs["use_dora"] and is_peft_version("<", "0.9.0"):
                raise ValueError(
                    "You need `peft` 0.9.0 at least to use DoRA-enabled LoRAs. Please upgrade your installation of `peft`."
                )
            else:
                lora_config_kwargs.pop("use_dora")
        lora_config = LoraConfig(**lora_config_kwargs)

        # adapter_name
        if adapter_name is None:
            adapter_name = get_adapter_name(transformer)

        inject_adapter_in_model(lora_config, transformer, adapter_name=adapter_name)
        incompatible_keys = set_peft_model_state_dict(transformer, state_dict, adapter_name)

        if incompatible_keys is not None:
            # check only for unexpected keys
            unexpected_keys = getattr(incompatible_keys, "unexpected_keys", None)
            if unexpected_keys:
                logger.warning(
                    f"Loading adapter weights from state_dict led to unexpected keys not found in the model: "
                    f" {unexpected_keys}. "
                )

mindone.diffusers.loaders.lora_pipeline.SD3LoraLoaderMixin.load_lora_weights(pretrained_model_name_or_path_or_dict, adapter_name=None, **kwargs)

Load LoRA weights specified in pretrained_model_name_or_path_or_dict into self.unet and self.text_encoder.

All kwargs are forwarded to self.lora_state_dict.

See [~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict] for more details on how the state dict is loaded.

See [~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_transformer] for more details on how the state dict is loaded into self.transformer.

PARAMETER	DESCRIPTION
pretrained_model_name_or_path_or_dict	See [~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict]. TYPE: `str` or `os.PathLike` or `dict`
kwargs	See [~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict]. TYPE: `dict`, *optional* DEFAULT: {}
adapter_name	Adapter name to be used for referencing the loaded adapter model. If not specified, it will use default_{i} where i is the total number of adapters being loaded. TYPE: `str`, *optional* DEFAULT: None

Source code in mindone/diffusers/loaders/lora_pipeline.py

def load_lora_weights(
    self, pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]], adapter_name=None, **kwargs
):
    """
    Load LoRA weights specified in `pretrained_model_name_or_path_or_dict` into `self.unet` and
    `self.text_encoder`.

    All kwargs are forwarded to `self.lora_state_dict`.

    See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`] for more details on how the state dict is
    loaded.

    See [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_into_transformer`] for more details on how the state
    dict is loaded into `self.transformer`.

    Parameters:
        pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
            See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
        kwargs (`dict`, *optional*):
            See [`~loaders.StableDiffusionLoraLoaderMixin.lora_state_dict`].
        adapter_name (`str`, *optional*):
            Adapter name to be used for referencing the loaded adapter model. If not specified, it will use
            `default_{i}` where i is the total number of adapters being loaded.
    """
    # if a dict is passed, copy it instead of modifying it inplace
    if isinstance(pretrained_model_name_or_path_or_dict, dict):
        pretrained_model_name_or_path_or_dict = pretrained_model_name_or_path_or_dict.copy()

    # First, ensure that the checkpoint is a compatible one and can be successfully loaded.
    state_dict = self.lora_state_dict(pretrained_model_name_or_path_or_dict, **kwargs)

    is_correct_format = all("lora" in key or "dora_scale" in key for key in state_dict.keys())
    if not is_correct_format:
        raise ValueError("Invalid LoRA checkpoint.")

    self.load_lora_into_transformer(
        state_dict,
        transformer=getattr(self, self.transformer_name) if not hasattr(self, "transformer") else self.transformer,
        adapter_name=adapter_name,
        _pipeline=self,
    )

    text_encoder_state_dict = {k: v for k, v in state_dict.items() if "text_encoder." in k}
    if len(text_encoder_state_dict) > 0:
        self.load_lora_into_text_encoder(
            text_encoder_state_dict,
            network_alphas=None,
            text_encoder=self.text_encoder,
            prefix="text_encoder",
            lora_scale=self.lora_scale,
            adapter_name=adapter_name,
            _pipeline=self,
        )

    text_encoder_2_state_dict = {k: v for k, v in state_dict.items() if "text_encoder_2." in k}
    if len(text_encoder_2_state_dict) > 0:
        self.load_lora_into_text_encoder(
            text_encoder_2_state_dict,
            network_alphas=None,
            text_encoder=self.text_encoder_2,
            prefix="text_encoder_2",
            lora_scale=self.lora_scale,
            adapter_name=adapter_name,
            _pipeline=self,
        )

mindone.diffusers.loaders.lora_pipeline.SD3LoraLoaderMixin.lora_state_dict(pretrained_model_name_or_path_or_dict, **kwargs) classmethod

Return state dict for lora weights and the network alphas.

We support loading A1111 formatted LoRA checkpoints in a limited capacity.

This function is experimental and might change in the future.

PARAMETER	DESCRIPTION
pretrained_model_name_or_path_or_dict	Can be either: - A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on the Hub. - A path to a *directory* (for example `./my_model_directory`) containing the model weights saved with [`ModelMixin.save_pretrained`]. - A MindSpore state dict. TYPE: `str` or `os.PathLike` or `dict`
cache_dir	Path to a directory where a downloaded pretrained model configuration is cached if the standard cache is not used. TYPE: `Union[str, os.PathLike]`, *optional*
force_download	Whether or not to force the (re-)download of the model weights and configuration files, overriding the cached versions if they exist. TYPE: `bool`, *optional*, defaults to `False`
proxies	A dictionary of proxy servers to use by protocol or endpoint, for example, {'http': 'foo.bar:3128', 'http://hostname': 'foo.bar:4012'}. The proxies are used on each request. TYPE: `Dict[str, str]`, *optional*
local_files_only	Whether to only load local model weights and configuration files or not. If set to True, the model won't be downloaded from the Hub. TYPE: `bool`, *optional*, defaults to `False`
token	The token to use as HTTP bearer authorization for remote files. If True, the token generated from diffusers-cli login (stored in ~/.huggingface) is used. TYPE: `str` or *bool*, *optional*
revision	The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier allowed by Git. TYPE: `str`, *optional*, defaults to `"main"`
subfolder	The subfolder location of a model file within a larger model repository on the Hub or locally. TYPE: `str`, *optional*, defaults to `""`

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
@validate_hf_hub_args
def lora_state_dict(
    cls,
    pretrained_model_name_or_path_or_dict: Union[str, Dict[str, ms.Tensor]],
    **kwargs,
):
    r"""
    Return state dict for lora weights and the network alphas.

    <Tip warning={true}>

    We support loading A1111 formatted LoRA checkpoints in a limited capacity.

    This function is experimental and might change in the future.

    </Tip>

    Parameters:
        pretrained_model_name_or_path_or_dict (`str` or `os.PathLike` or `dict`):
            Can be either:

                - A string, the *model id* (for example `google/ddpm-celebahq-256`) of a pretrained model hosted on
                  the Hub.
                - A path to a *directory* (for example `./my_model_directory`) containing the model weights saved
                  with [`ModelMixin.save_pretrained`].
                - A MindSpore state dict.

        cache_dir (`Union[str, os.PathLike]`, *optional*):
            Path to a directory where a downloaded pretrained model configuration is cached if the standard cache
            is not used.
        force_download (`bool`, *optional*, defaults to `False`):
            Whether or not to force the (re-)download of the model weights and configuration files, overriding the
            cached versions if they exist.

        proxies (`Dict[str, str]`, *optional*):
            A dictionary of proxy servers to use by protocol or endpoint, for example, `{'http': 'foo.bar:3128',
            'http://hostname': 'foo.bar:4012'}`. The proxies are used on each request.
        local_files_only (`bool`, *optional*, defaults to `False`):
            Whether to only load local model weights and configuration files or not. If set to `True`, the model
            won't be downloaded from the Hub.
        token (`str` or *bool*, *optional*):
            The token to use as HTTP bearer authorization for remote files. If `True`, the token generated from
            `diffusers-cli login` (stored in `~/.huggingface`) is used.
        revision (`str`, *optional*, defaults to `"main"`):
            The specific model version to use. It can be a branch name, a tag name, a commit id, or any identifier
            allowed by Git.
        subfolder (`str`, *optional*, defaults to `""`):
            The subfolder location of a model file within a larger model repository on the Hub or locally.

    """
    # Load the main state dict first which has the LoRA layers for either of
    # transformer and text encoder or both.
    cache_dir = kwargs.pop("cache_dir", None)
    force_download = kwargs.pop("force_download", False)
    proxies = kwargs.pop("proxies", None)
    local_files_only = kwargs.pop("local_files_only", None)
    token = kwargs.pop("token", None)
    revision = kwargs.pop("revision", None)
    subfolder = kwargs.pop("subfolder", None)
    weight_name = kwargs.pop("weight_name", None)
    use_safetensors = kwargs.pop("use_safetensors", None)

    allow_pickle = False
    if use_safetensors is None:
        use_safetensors = True
        allow_pickle = True

    user_agent = {
        "file_type": "attn_procs_weights",
        "framework": "pytorch",
    }

    state_dict = cls._fetch_state_dict(
        pretrained_model_name_or_path_or_dict=pretrained_model_name_or_path_or_dict,
        weight_name=weight_name,
        use_safetensors=use_safetensors,
        local_files_only=local_files_only,
        cache_dir=cache_dir,
        force_download=force_download,
        proxies=proxies,
        token=token,
        revision=revision,
        subfolder=subfolder,
        user_agent=user_agent,
        allow_pickle=allow_pickle,
    )

    return state_dict

mindone.diffusers.loaders.lora_pipeline.SD3LoraLoaderMixin.save_lora_weights(save_directory, transformer_lora_layers=None, text_encoder_lora_layers=None, text_encoder_2_lora_layers=None, is_main_process=True, weight_name=None, save_function=None, safe_serialization=True) classmethod

Save the LoRA parameters corresponding to the UNet and text encoder.

PARAMETER	DESCRIPTION
save_directory	Directory to save LoRA parameters to. Will be created if it doesn't exist. TYPE: `str` or `os.PathLike`
transformer_lora_layers	State dict of the LoRA layers corresponding to the transformer. TYPE: `Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]` DEFAULT: None
text_encoder_lora_layers	State dict of the LoRA layers corresponding to the text_encoder. Must explicitly pass the text encoder LoRA state dict because it comes from 🤗 Transformers. TYPE: `Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]` DEFAULT: None
text_encoder_2_lora_layers	State dict of the LoRA layers corresponding to the text_encoder_2. Must explicitly pass the text encoder LoRA state dict because it comes from 🤗 Transformers. TYPE: `Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]` DEFAULT: None
is_main_process	Whether the process calling this is the main process or not. Useful during distributed training and you need to call this function on all processes. In this case, set is_main_process=True only on the main process to avoid race conditions. TYPE: `bool`, *optional*, defaults to `True` DEFAULT: True
save_function	The function to use to save the state dictionary. Useful during distributed training when you need to replace mindspore.save_checkpoint with another method. Can be configured with the environment variable DIFFUSERS_SAVE_MODE. TYPE: `Callable` DEFAULT: None
safe_serialization	Whether to save the model using safetensors or the traditional MindSpore way. TYPE: `bool`, *optional*, defaults to `True` DEFAULT: True

Source code in mindone/diffusers/loaders/lora_pipeline.py

@classmethod
def save_lora_weights(
    cls,
    save_directory: Union[str, os.PathLike],
    transformer_lora_layers: Dict[str, nn.Cell] = None,
    text_encoder_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
    text_encoder_2_lora_layers: Dict[str, Union[nn.Cell, ms.Tensor]] = None,
    is_main_process: bool = True,
    weight_name: str = None,
    save_function: Callable = None,
    safe_serialization: bool = True,
):
    r"""
    Save the LoRA parameters corresponding to the UNet and text encoder.

    Arguments:
        save_directory (`str` or `os.PathLike`):
            Directory to save LoRA parameters to. Will be created if it doesn't exist.
        transformer_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
            State dict of the LoRA layers corresponding to the `transformer`.
        text_encoder_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
            State dict of the LoRA layers corresponding to the `text_encoder`. Must explicitly pass the text
            encoder LoRA state dict because it comes from 🤗 Transformers.
        text_encoder_2_lora_layers (`Dict[str, nn.Cell]` or `Dict[str, ms.Tensor]`):
            State dict of the LoRA layers corresponding to the `text_encoder_2`. Must explicitly pass the text
            encoder LoRA state dict because it comes from 🤗 Transformers.
        is_main_process (`bool`, *optional*, defaults to `True`):
            Whether the process calling this is the main process or not. Useful during distributed training and you
            need to call this function on all processes. In this case, set `is_main_process=True` only on the main
            process to avoid race conditions.
        save_function (`Callable`):
            The function to use to save the state dictionary. Useful during distributed training when you need to
            replace `mindspore.save_checkpoint` with another method. Can be configured with the environment variable
            `DIFFUSERS_SAVE_MODE`.
        safe_serialization (`bool`, *optional*, defaults to `True`):
            Whether to save the model using `safetensors` or the traditional MindSpore way.
    """
    state_dict = {}

    if not (transformer_lora_layers or text_encoder_lora_layers or text_encoder_2_lora_layers):
        raise ValueError(
            "You must pass at least one of `transformer_lora_layers`, `text_encoder_lora_layers`, `text_encoder_2_lora_layers`."
        )

    if transformer_lora_layers:
        state_dict.update(cls.pack_weights(transformer_lora_layers, cls.transformer_name))

    if text_encoder_lora_layers:
        state_dict.update(cls.pack_weights(text_encoder_lora_layers, "text_encoder"))

    if text_encoder_2_lora_layers:
        state_dict.update(cls.pack_weights(text_encoder_2_lora_layers, "text_encoder_2"))

    # Save the model
    cls.write_lora_layers(
        state_dict=state_dict,
        save_directory=save_directory,
        is_main_process=is_main_process,
        weight_name=weight_name,
        save_function=save_function,
        safe_serialization=safe_serialization,
    )

mindone.diffusers.loaders.lora_pipeline.SD3LoraLoaderMixin.unfuse_lora(components=['transformer', 'text_encoder', 'text_encoder_2'], **kwargs)

Reverses the effect of pipe.fuse_lora().

This is an experimental API.

PARAMETER	DESCRIPTION
components	List of LoRA-injectable components to unfuse LoRA from. TYPE: `List[str]` DEFAULT: ['transformer', 'text_encoder', 'text_encoder_2']
unfuse_unet	Whether to unfuse the UNet LoRA parameters. TYPE: `bool`, defaults to `True`
unfuse_text_encoder	Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the LoRA parameters then it won't have any effect. TYPE: `bool`, defaults to `True`

Source code in mindone/diffusers/loaders/lora_pipeline.py

def unfuse_lora(self, components: List[str] = ["transformer", "text_encoder", "text_encoder_2"], **kwargs):
    r"""
    Reverses the effect of
    [`pipe.fuse_lora()`](https://huggingface.co/docs/diffusers/main/en/api/loaders#diffusers.loaders.LoraBaseMixin.fuse_lora).

    <Tip warning={true}>

    This is an experimental API.

    </Tip>

    Args:
        components (`List[str]`): List of LoRA-injectable components to unfuse LoRA from.
        unfuse_unet (`bool`, defaults to `True`): Whether to unfuse the UNet LoRA parameters.
        unfuse_text_encoder (`bool`, defaults to `True`):
            Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the
            LoRA parameters then it won't have any effect.
    """
    super().unfuse_lora(components=components)

mindone.diffusers.loaders.lora_base.LoraBaseMixin

Utility class for handling LoRAs.

Source code in mindone/diffusers/loaders/lora_base.py

class LoraBaseMixin:
    """Utility class for handling LoRAs."""

    _lora_loadable_modules = []
    num_fused_loras = 0

    def load_lora_weights(self, **kwargs):
        raise NotImplementedError("`load_lora_weights()` is not implemented.")

    @classmethod
    def save_lora_weights(cls, **kwargs):
        raise NotImplementedError("`save_lora_weights()` not implemented.")

    @classmethod
    def lora_state_dict(cls, **kwargs):
        raise NotImplementedError("`lora_state_dict()` is not implemented.")

    @classmethod
    def _fetch_state_dict(
        cls,
        pretrained_model_name_or_path_or_dict,
        weight_name,
        use_safetensors,
        local_files_only,
        cache_dir,
        force_download,
        proxies,
        token,
        revision,
        subfolder,
        user_agent,
        allow_pickle,
    ):
        from .lora_pipeline import LORA_WEIGHT_NAME, LORA_WEIGHT_NAME_SAFE

        model_file = None
        if not isinstance(pretrained_model_name_or_path_or_dict, dict):
            # Let's first try to load .safetensors weights
            if (use_safetensors and weight_name is None) or (
                weight_name is not None and weight_name.endswith(".safetensors")
            ):
                try:
                    # Here we're relaxing the loading check to enable more Inference API
                    # friendliness where sometimes, it's not at all possible to automatically
                    # determine `weight_name`.
                    if weight_name is None:
                        weight_name = cls._best_guess_weight_name(
                            pretrained_model_name_or_path_or_dict,
                            file_extension=".safetensors",
                            local_files_only=local_files_only,
                        )
                    model_file = _get_model_file(
                        pretrained_model_name_or_path_or_dict,
                        weights_name=weight_name or LORA_WEIGHT_NAME_SAFE,
                        cache_dir=cache_dir,
                        force_download=force_download,
                        proxies=proxies,
                        local_files_only=local_files_only,
                        token=token,
                        revision=revision,
                        subfolder=subfolder,
                        user_agent=user_agent,
                    )
                    state_dict = load_file(model_file)
                except (IOError, safetensors.SafetensorError) as e:
                    if not allow_pickle:
                        raise e
                    # try loading non-safetensors weights
                    model_file = None
                    pass

            if model_file is None:
                if weight_name is None:
                    weight_name = cls._best_guess_weight_name(
                        pretrained_model_name_or_path_or_dict, file_extension=".bin", local_files_only=local_files_only
                    )
                model_file = _get_model_file(
                    pretrained_model_name_or_path_or_dict,
                    weights_name=weight_name or LORA_WEIGHT_NAME,
                    cache_dir=cache_dir,
                    force_download=force_download,
                    proxies=proxies,
                    local_files_only=local_files_only,
                    token=token,
                    revision=revision,
                    subfolder=subfolder,
                    user_agent=user_agent,
                )
                raise NotImplementedError(
                    f"Only supports deserialization of weights file in safetensors format, but got {model_file}"
                )
        else:
            state_dict = pretrained_model_name_or_path_or_dict

        return state_dict

    @classmethod
    def _best_guess_weight_name(
        cls, pretrained_model_name_or_path_or_dict, file_extension=".safetensors", local_files_only=False
    ):
        from .lora_pipeline import LORA_WEIGHT_NAME, LORA_WEIGHT_NAME_SAFE

        if local_files_only or HF_HUB_OFFLINE:
            raise ValueError("When using the offline mode, you must specify a `weight_name`.")

        targeted_files = []

        if os.path.isfile(pretrained_model_name_or_path_or_dict):
            return
        elif os.path.isdir(pretrained_model_name_or_path_or_dict):
            targeted_files = [
                f for f in os.listdir(pretrained_model_name_or_path_or_dict) if f.endswith(file_extension)
            ]
        else:
            files_in_repo = model_info(pretrained_model_name_or_path_or_dict).siblings
            targeted_files = [f.rfilename for f in files_in_repo if f.rfilename.endswith(file_extension)]
        if len(targeted_files) == 0:
            return

        # "scheduler" does not correspond to a LoRA checkpoint.
        # "optimizer" does not correspond to a LoRA checkpoint
        # only top-level checkpoints are considered and not the other ones, hence "checkpoint".
        unallowed_substrings = {"scheduler", "optimizer", "checkpoint"}
        targeted_files = list(
            filter(lambda x: all(substring not in x for substring in unallowed_substrings), targeted_files)
        )

        if any(f.endswith(LORA_WEIGHT_NAME) for f in targeted_files):
            targeted_files = list(filter(lambda x: x.endswith(LORA_WEIGHT_NAME), targeted_files))
        elif any(f.endswith(LORA_WEIGHT_NAME_SAFE) for f in targeted_files):
            targeted_files = list(filter(lambda x: x.endswith(LORA_WEIGHT_NAME_SAFE), targeted_files))

        if len(targeted_files) > 1:
            raise ValueError(
                f"Provided path contains more than one weights file in the {file_extension} format. Either specify `weight_name` in `load_lora_weights` or make sure there's only one  `.safetensors` or `.bin` file in  {pretrained_model_name_or_path_or_dict}."  # noqa: E501
            )
        weight_name = targeted_files[0]
        return weight_name

    def unload_lora_weights(self):
        """
        Unloads the LoRA parameters.

        Examples:

        ```python
        >>> # Assuming `pipeline` is already loaded with the LoRA parameters.
        >>> pipeline.unload_lora_weights()
        >>> ...
        ```
        """
        for component in self._lora_loadable_modules:
            model = getattr(self, component, None)
            if model is not None:
                if issubclass(model.__class__, ModelMixin):
                    model.unload_lora()
                elif issubclass(model.__class__, MSPreTrainedModel):
                    _remove_text_encoder_monkey_patch(model)

    def fuse_lora(
        self,
        components: List[str] = [],
        lora_scale: float = 1.0,
        safe_fusing: bool = False,
        adapter_names: Optional[List[str]] = None,
        **kwargs,
    ):
        r"""
        Fuses the LoRA parameters into the original parameters of the corresponding blocks.

        <Tip warning={true}>

        This is an experimental API.

        </Tip>

        Args:
            components: (`List[str]`): List of LoRA-injectable components to fuse the LoRAs into.
            lora_scale (`float`, defaults to 1.0):
                Controls how much to influence the outputs with the LoRA parameters.
            safe_fusing (`bool`, defaults to `False`):
                Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them.
            adapter_names (`List[str]`, *optional*):
                Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused.

        Example:

        ```py
        from mindone.diffusers import DiffusionPipeline
        import mindspore

        pipeline = DiffusionPipeline.from_pretrained(
            "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
        )
        pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
        pipeline.fuse_lora(lora_scale=0.7)
        ```
        """
        if "fuse_unet" in kwargs:
            depr_message = "Passing `fuse_unet` to `fuse_lora()` is deprecated and will be ignored. Please use the `components` argument and provide a list of the components whose LoRAs are to be fused. `fuse_unet` will be removed in a future version."  # noqa: E501
            deprecate(
                "fuse_unet",
                "1.0.0",
                depr_message,
            )
        if "fuse_transformer" in kwargs:
            depr_message = "Passing `fuse_transformer` to `fuse_lora()` is deprecated and will be ignored. Please use the `components` argument and provide a list of the components whose LoRAs are to be fused. `fuse_transformer` will be removed in a future version."  # noqa: E501
            deprecate(
                "fuse_transformer",
                "1.0.0",
                depr_message,
            )
        if "fuse_text_encoder" in kwargs:
            depr_message = "Passing `fuse_text_encoder` to `fuse_lora()` is deprecated and will be ignored. Please use the `components` argument and provide a list of the components whose LoRAs are to be fused. `fuse_text_encoder` will be removed in a future version."  # noqa: E501
            deprecate(
                "fuse_text_encoder",
                "1.0.0",
                depr_message,
            )

        if len(components) == 0:
            raise ValueError("`components` cannot be an empty list.")

        for fuse_component in components:
            if fuse_component not in self._lora_loadable_modules:
                raise ValueError(f"{fuse_component} is not found in {self._lora_loadable_modules=}.")

            model = getattr(self, fuse_component, None)
            if model is not None:
                # check if diffusers model
                if issubclass(model.__class__, ModelMixin):
                    model.fuse_lora(lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names)
                # handle transformers models.
                if issubclass(model.__class__, MSPreTrainedModel):
                    fuse_text_encoder_lora(
                        model, lora_scale=lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names
                    )

        self.num_fused_loras += 1

    def unfuse_lora(self, components: List[str] = [], **kwargs):
        r"""
        Reverses the effect of
        [`pipe.fuse_lora()`](https://huggingface.co/docs/diffusers/main/en/api/loaders#diffusers.loaders.LoraBaseMixin.fuse_lora).

        <Tip warning={true}>

        This is an experimental API.

        </Tip>

        Args:
            components (`List[str]`): List of LoRA-injectable components to unfuse LoRA from.
            unfuse_unet (`bool`, defaults to `True`): Whether to unfuse the UNet LoRA parameters.
            unfuse_text_encoder (`bool`, defaults to `True`):
                Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the
                LoRA parameters then it won't have any effect.
        """
        if "unfuse_unet" in kwargs:
            depr_message = "Passing `unfuse_unet` to `unfuse_lora()` is deprecated and will be ignored. Please use the `components` argument. `unfuse_unet` will be removed in a future version."  # noqa: E501
            deprecate(
                "unfuse_unet",
                "1.0.0",
                depr_message,
            )
        if "unfuse_transformer" in kwargs:
            depr_message = "Passing `unfuse_transformer` to `unfuse_lora()` is deprecated and will be ignored. Please use the `components` argument. `unfuse_transformer` will be removed in a future version."  # noqa: E501
            deprecate(
                "unfuse_transformer",
                "1.0.0",
                depr_message,
            )
        if "unfuse_text_encoder" in kwargs:
            depr_message = "Passing `unfuse_text_encoder` to `unfuse_lora()` is deprecated and will be ignored. Please use the `components` argument. `unfuse_text_encoder` will be removed in a future version."  # noqa: E501
            deprecate(
                "unfuse_text_encoder",
                "1.0.0",
                depr_message,
            )

        if len(components) == 0:
            raise ValueError("`components` cannot be an empty list.")

        for fuse_component in components:
            if fuse_component not in self._lora_loadable_modules:
                raise ValueError(f"{fuse_component} is not found in {self._lora_loadable_modules=}.")

            model = getattr(self, fuse_component, None)
            if model is not None:
                if issubclass(model.__class__, (ModelMixin, MSPreTrainedModel)):
                    for _, module in model.cells_and_names():
                        if isinstance(module, BaseTunerLayer):
                            module.unmerge()

        self.num_fused_loras -= 1

    def set_adapters(
        self,
        adapter_names: Union[List[str], str],
        adapter_weights: Optional[Union[float, Dict, List[float], List[Dict]]] = None,
    ):
        adapter_names = [adapter_names] if isinstance(adapter_names, str) else adapter_names

        adapter_weights = copy.deepcopy(adapter_weights)

        # Expand weights into a list, one entry per adapter
        if not isinstance(adapter_weights, list):
            adapter_weights = [adapter_weights] * len(adapter_names)

        if len(adapter_names) != len(adapter_weights):
            raise ValueError(
                f"Length of adapter names {len(adapter_names)} is not equal to the length of the weights {len(adapter_weights)}"
            )

        list_adapters = self.get_list_adapters()  # eg {"unet": ["adapter1", "adapter2"], "text_encoder": ["adapter2"]}
        all_adapters = {
            adapter for adapters in list_adapters.values() for adapter in adapters
        }  # eg ["adapter1", "adapter2"]
        invert_list_adapters = {
            adapter: [part for part, adapters in list_adapters.items() if adapter in adapters]
            for adapter in all_adapters
        }  # eg {"adapter1": ["unet"], "adapter2": ["unet", "text_encoder"]}

        # Decompose weights into weights for denoiser and text encoders.
        _component_adapter_weights = {}
        for component in self._lora_loadable_modules:
            model = getattr(self, component)

            for adapter_name, weights in zip(adapter_names, adapter_weights):
                if isinstance(weights, dict):
                    component_adapter_weights = weights.pop(component, None)

                    if component_adapter_weights is not None and not hasattr(self, component):
                        logger.warning(
                            f"Lora weight dict contains {component} weights but will be ignored because pipeline does not have {component}."
                        )

                    if component_adapter_weights is not None and component not in invert_list_adapters[adapter_name]:
                        logger.warning(
                            (
                                f"Lora weight dict for adapter '{adapter_name}' contains {component},"
                                f"but this will be ignored because {adapter_name} does not contain weights for {component}."
                                f"Valid parts for {adapter_name} are: {invert_list_adapters[adapter_name]}."
                            )
                        )

                else:
                    component_adapter_weights = weights

                _component_adapter_weights.setdefault(component, [])
                _component_adapter_weights[component].append(component_adapter_weights)

            if issubclass(model.__class__, ModelMixin):
                model.set_adapters(adapter_names, _component_adapter_weights[component])
            elif issubclass(model.__class__, MSPreTrainedModel):
                set_adapters_for_text_encoder(adapter_names, model, _component_adapter_weights[component])

    def disable_lora(self):
        for component in self._lora_loadable_modules:
            model = getattr(self, component, None)
            if model is not None:
                if issubclass(model.__class__, ModelMixin):
                    model.disable_lora()
                elif issubclass(model.__class__, MSPreTrainedModel):
                    disable_lora_for_text_encoder(model)

    def enable_lora(self):
        for component in self._lora_loadable_modules:
            model = getattr(self, component, None)
            if model is not None:
                if issubclass(model.__class__, ModelMixin):
                    model.enable_lora()
                elif issubclass(model.__class__, MSPreTrainedModel):
                    enable_lora_for_text_encoder(model)

    def delete_adapters(self, adapter_names: Union[List[str], str]):
        """
        Args:
        Deletes the LoRA layers of `adapter_name` for the unet and text-encoder(s).
            adapter_names (`Union[List[str], str]`):
                The names of the adapter to delete. Can be a single string or a list of strings
        """
        if isinstance(adapter_names, str):
            adapter_names = [adapter_names]

        for component in self._lora_loadable_modules:
            model = getattr(self, component, None)
            if model is not None:
                if issubclass(model.__class__, ModelMixin):
                    model.delete_adapters(adapter_names)
                elif issubclass(model.__class__, MSPreTrainedModel):
                    for adapter_name in adapter_names:
                        delete_adapter_layers(model, adapter_name)

    def get_active_adapters(self) -> List[str]:
        """
        Gets the list of the current active adapters.

        Example:

        ```python
        from mindone.diffusers import DiffusionPipeline

        pipeline = DiffusionPipeline.from_pretrained(
            "stabilityai/stable-diffusion-xl-base-1.0",
        )
        pipeline.load_lora_weights("CiroN2022/toy-face", weight_name="toy_face_sdxl.safetensors", adapter_name="toy")
        pipeline.get_active_adapters()
        ```
        """
        active_adapters = []

        for component in self._lora_loadable_modules:
            model = getattr(self, component, None)
            if model is not None and issubclass(model.__class__, ModelMixin):
                for _, module in model.cells_and_names():
                    if isinstance(module, BaseTunerLayer):
                        active_adapters = module.active_adapters
                        break

        return active_adapters

    def get_list_adapters(self) -> Dict[str, List[str]]:
        """
        Gets the current list of all available adapters in the pipeline.
        """
        set_adapters = {}

        for component in self._lora_loadable_modules:
            model = getattr(self, component, None)
            if (
                model is not None
                and issubclass(model.__class__, (ModelMixin, MSPreTrainedModel))
                and hasattr(model, "peft_config")
            ):
                set_adapters[component] = list(model.peft_config.keys())

        return set_adapters

    @staticmethod
    def pack_weights(layers, prefix):
        layers_weights = layers.parameters_dict() if isinstance(layers, nn.Cell) else layers
        layers_state_dict = {f"{prefix}.{module_name}": param for module_name, param in layers_weights.items()}
        return layers_state_dict

    @staticmethod
    def write_lora_layers(
        state_dict: Dict[str, ms.Tensor],
        save_directory: str,
        is_main_process: bool,
        weight_name: str,
        save_function: Callable,
        safe_serialization: bool,
    ):
        from .lora_pipeline import LORA_WEIGHT_NAME, LORA_WEIGHT_NAME_SAFE

        if os.path.isfile(save_directory):
            logger.error(f"Provided path ({save_directory}) should be a directory, not a file")
            return

        if save_function is None:
            if safe_serialization:

                def save_function(weights, filename):
                    return save_file(weights, filename, metadata={"format": "np"})

            else:
                save_function = ms.save_checkpoint

        os.makedirs(save_directory, exist_ok=True)

        if weight_name is None:
            if safe_serialization:
                weight_name = LORA_WEIGHT_NAME_SAFE
            else:
                weight_name = LORA_WEIGHT_NAME

        save_path = Path(save_directory, weight_name).as_posix()
        save_function(state_dict, save_path)
        logger.info(f"Model weights saved in {save_path}")

    @property
    def lora_scale(self) -> float:
        # property function that returns the lora scale which can be set at run time by the pipeline.
        # if _lora_scale has not been set, return 1
        return self._lora_scale if hasattr(self, "_lora_scale") else 1.0

mindone.diffusers.loaders.lora_base.LoraBaseMixin.delete_adapters(adapter_names)

Deletes the LoRA layers of adapter_name for the unet and text-encoder(s). adapter_names (Union[List[str], str]): The names of the adapter to delete. Can be a single string or a list of strings

Source code in mindone/diffusers/loaders/lora_base.py

def delete_adapters(self, adapter_names: Union[List[str], str]):
    """
    Args:
    Deletes the LoRA layers of `adapter_name` for the unet and text-encoder(s).
        adapter_names (`Union[List[str], str]`):
            The names of the adapter to delete. Can be a single string or a list of strings
    """
    if isinstance(adapter_names, str):
        adapter_names = [adapter_names]

    for component in self._lora_loadable_modules:
        model = getattr(self, component, None)
        if model is not None:
            if issubclass(model.__class__, ModelMixin):
                model.delete_adapters(adapter_names)
            elif issubclass(model.__class__, MSPreTrainedModel):
                for adapter_name in adapter_names:
                    delete_adapter_layers(model, adapter_name)

mindone.diffusers.loaders.lora_base.LoraBaseMixin.fuse_lora(components=[], lora_scale=1.0, safe_fusing=False, adapter_names=None, **kwargs)

Fuses the LoRA parameters into the original parameters of the corresponding blocks.

This is an experimental API.

PARAMETER	DESCRIPTION
components	(List[str]): List of LoRA-injectable components to fuse the LoRAs into. TYPE: List[str] DEFAULT: []
lora_scale	Controls how much to influence the outputs with the LoRA parameters. TYPE: `float`, defaults to 1.0 DEFAULT: 1.0
safe_fusing	Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them. TYPE: `bool`, defaults to `False` DEFAULT: False
adapter_names	Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused. TYPE: `List[str]`, *optional* DEFAULT: None

from mindone.diffusers import DiffusionPipeline
import mindspore

pipeline = DiffusionPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
)
pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
pipeline.fuse_lora(lora_scale=0.7)

Source code in mindone/diffusers/loaders/lora_base.py

def fuse_lora(
    self,
    components: List[str] = [],
    lora_scale: float = 1.0,
    safe_fusing: bool = False,
    adapter_names: Optional[List[str]] = None,
    **kwargs,
):
    r"""
    Fuses the LoRA parameters into the original parameters of the corresponding blocks.

    <Tip warning={true}>

    This is an experimental API.

    </Tip>

    Args:
        components: (`List[str]`): List of LoRA-injectable components to fuse the LoRAs into.
        lora_scale (`float`, defaults to 1.0):
            Controls how much to influence the outputs with the LoRA parameters.
        safe_fusing (`bool`, defaults to `False`):
            Whether to check fused weights for NaN values before fusing and if values are NaN not fusing them.
        adapter_names (`List[str]`, *optional*):
            Adapter names to be used for fusing. If nothing is passed, all active adapters will be fused.

    Example:

    ```py
    from mindone.diffusers import DiffusionPipeline
    import mindspore

    pipeline = DiffusionPipeline.from_pretrained(
        "stabilityai/stable-diffusion-xl-base-1.0", mindspore_dtype=mindspore.float16
    )
    pipeline.load_lora_weights("nerijs/pixel-art-xl", weight_name="pixel-art-xl.safetensors", adapter_name="pixel")
    pipeline.fuse_lora(lora_scale=0.7)
    ```
    """
    if "fuse_unet" in kwargs:
        depr_message = "Passing `fuse_unet` to `fuse_lora()` is deprecated and will be ignored. Please use the `components` argument and provide a list of the components whose LoRAs are to be fused. `fuse_unet` will be removed in a future version."  # noqa: E501
        deprecate(
            "fuse_unet",
            "1.0.0",
            depr_message,
        )
    if "fuse_transformer" in kwargs:
        depr_message = "Passing `fuse_transformer` to `fuse_lora()` is deprecated and will be ignored. Please use the `components` argument and provide a list of the components whose LoRAs are to be fused. `fuse_transformer` will be removed in a future version."  # noqa: E501
        deprecate(
            "fuse_transformer",
            "1.0.0",
            depr_message,
        )
    if "fuse_text_encoder" in kwargs:
        depr_message = "Passing `fuse_text_encoder` to `fuse_lora()` is deprecated and will be ignored. Please use the `components` argument and provide a list of the components whose LoRAs are to be fused. `fuse_text_encoder` will be removed in a future version."  # noqa: E501
        deprecate(
            "fuse_text_encoder",
            "1.0.0",
            depr_message,
        )

    if len(components) == 0:
        raise ValueError("`components` cannot be an empty list.")

    for fuse_component in components:
        if fuse_component not in self._lora_loadable_modules:
            raise ValueError(f"{fuse_component} is not found in {self._lora_loadable_modules=}.")

        model = getattr(self, fuse_component, None)
        if model is not None:
            # check if diffusers model
            if issubclass(model.__class__, ModelMixin):
                model.fuse_lora(lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names)
            # handle transformers models.
            if issubclass(model.__class__, MSPreTrainedModel):
                fuse_text_encoder_lora(
                    model, lora_scale=lora_scale, safe_fusing=safe_fusing, adapter_names=adapter_names
                )

    self.num_fused_loras += 1

mindone.diffusers.loaders.lora_base.LoraBaseMixin.get_active_adapters()

Gets the list of the current active adapters.

Example:

from mindone.diffusers import DiffusionPipeline

pipeline = DiffusionPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
)
pipeline.load_lora_weights("CiroN2022/toy-face", weight_name="toy_face_sdxl.safetensors", adapter_name="toy")
pipeline.get_active_adapters()

Source code in mindone/diffusers/loaders/lora_base.py

def get_active_adapters(self) -> List[str]:
    """
    Gets the list of the current active adapters.

    Example:

    ```python
    from mindone.diffusers import DiffusionPipeline

    pipeline = DiffusionPipeline.from_pretrained(
        "stabilityai/stable-diffusion-xl-base-1.0",
    )
    pipeline.load_lora_weights("CiroN2022/toy-face", weight_name="toy_face_sdxl.safetensors", adapter_name="toy")
    pipeline.get_active_adapters()
    ```
    """
    active_adapters = []

    for component in self._lora_loadable_modules:
        model = getattr(self, component, None)
        if model is not None and issubclass(model.__class__, ModelMixin):
            for _, module in model.cells_and_names():
                if isinstance(module, BaseTunerLayer):
                    active_adapters = module.active_adapters
                    break

    return active_adapters

mindone.diffusers.loaders.lora_base.LoraBaseMixin.get_list_adapters()

Gets the current list of all available adapters in the pipeline.

Source code in mindone/diffusers/loaders/lora_base.py

def get_list_adapters(self) -> Dict[str, List[str]]:
    """
    Gets the current list of all available adapters in the pipeline.
    """
    set_adapters = {}

    for component in self._lora_loadable_modules:
        model = getattr(self, component, None)
        if (
            model is not None
            and issubclass(model.__class__, (ModelMixin, MSPreTrainedModel))
            and hasattr(model, "peft_config")
        ):
            set_adapters[component] = list(model.peft_config.keys())

    return set_adapters

mindone.diffusers.loaders.lora_base.LoraBaseMixin.unfuse_lora(components=[], **kwargs)

Reverses the effect of pipe.fuse_lora().

This is an experimental API.

PARAMETER	DESCRIPTION
components	List of LoRA-injectable components to unfuse LoRA from. TYPE: `List[str]` DEFAULT: []
unfuse_unet	Whether to unfuse the UNet LoRA parameters. TYPE: `bool`, defaults to `True`
unfuse_text_encoder	Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the LoRA parameters then it won't have any effect. TYPE: `bool`, defaults to `True`

Source code in mindone/diffusers/loaders/lora_base.py

def unfuse_lora(self, components: List[str] = [], **kwargs):
    r"""
    Reverses the effect of
    [`pipe.fuse_lora()`](https://huggingface.co/docs/diffusers/main/en/api/loaders#diffusers.loaders.LoraBaseMixin.fuse_lora).

    <Tip warning={true}>

    This is an experimental API.

    </Tip>

    Args:
        components (`List[str]`): List of LoRA-injectable components to unfuse LoRA from.
        unfuse_unet (`bool`, defaults to `True`): Whether to unfuse the UNet LoRA parameters.
        unfuse_text_encoder (`bool`, defaults to `True`):
            Whether to unfuse the text encoder LoRA parameters. If the text encoder wasn't monkey-patched with the
            LoRA parameters then it won't have any effect.
    """
    if "unfuse_unet" in kwargs:
        depr_message = "Passing `unfuse_unet` to `unfuse_lora()` is deprecated and will be ignored. Please use the `components` argument. `unfuse_unet` will be removed in a future version."  # noqa: E501
        deprecate(
            "unfuse_unet",
            "1.0.0",
            depr_message,
        )
    if "unfuse_transformer" in kwargs:
        depr_message = "Passing `unfuse_transformer` to `unfuse_lora()` is deprecated and will be ignored. Please use the `components` argument. `unfuse_transformer` will be removed in a future version."  # noqa: E501
        deprecate(
            "unfuse_transformer",
            "1.0.0",
            depr_message,
        )
    if "unfuse_text_encoder" in kwargs:
        depr_message = "Passing `unfuse_text_encoder` to `unfuse_lora()` is deprecated and will be ignored. Please use the `components` argument. `unfuse_text_encoder` will be removed in a future version."  # noqa: E501
        deprecate(
            "unfuse_text_encoder",
            "1.0.0",
            depr_message,
        )

    if len(components) == 0:
        raise ValueError("`components` cannot be an empty list.")

    for fuse_component in components:
        if fuse_component not in self._lora_loadable_modules:
            raise ValueError(f"{fuse_component} is not found in {self._lora_loadable_modules=}.")

        model = getattr(self, fuse_component, None)
        if model is not None:
            if issubclass(model.__class__, (ModelMixin, MSPreTrainedModel)):
                for _, module in model.cells_and_names():
                    if isinstance(module, BaseTunerLayer):
                        module.unmerge()

    self.num_fused_loras -= 1

mindone.diffusers.loaders.lora_base.LoraBaseMixin.unload_lora_weights()

Unloads the LoRA parameters.

Examples:

>>> # Assuming `pipeline` is already loaded with the LoRA parameters.
>>> pipeline.unload_lora_weights()
>>> ...

Source code in mindone/diffusers/loaders/lora_base.py

def unload_lora_weights(self):
    """
    Unloads the LoRA parameters.

    Examples:

    ```python
    >>> # Assuming `pipeline` is already loaded with the LoRA parameters.
    >>> pipeline.unload_lora_weights()
    >>> ...
    ```
    """
    for component in self._lora_loadable_modules:
        model = getattr(self, component, None)
        if model is not None:
            if issubclass(model.__class__, ModelMixin):
                model.unload_lora()
            elif issubclass(model.__class__, MSPreTrainedModel):
                _remove_text_encoder_monkey_patch(model)