Port LoRA to new classification API (#7849)

## Summary

- Port LoRA to new classification API
- Add 2 additional tests cases (ControlLora and Flux Diffusers LoRA)
- Moved `ModelOnDisk` to its own module

## Related Issues / Discussions

<!--WHEN APPLICABLE: List any related issues or discussions on github or
discord. If this PR closes an issue, please use the "Closes #1234"
format, so that the issue will be automatically closed when the PR
merges.-->

## QA Instructions

<!--WHEN APPLICABLE: Describe how you have tested the changes in this
PR. Provide enough detail that a reviewer can reproduce your tests.-->

## Merge Plan

<!--WHEN APPLICABLE: Large PRs, or PRs that touch sensitive things like
DB schemas, may need some care when merging. For example, a careful
rebase by the change author, timing to not interfere with a pending
release, or a message to contributors on discord after merging.-->

## Checklist

- [ ] _The PR has a short but descriptive title, suitable for a
changelog_
- [ ] _Tests added / updated (if applicable)_
- [ ] _Documentation added / updated (if applicable)_
- [ ] _Updated `What's New` copy (if doing a release after this PR)_

invokeai/backend/model_manager/config.py

@@ -30,19 +30,18 @@ from inspect import isabstract
 from pathlib import Path
 from typing import ClassVar, Literal, Optional, TypeAlias, Union
 
-import safetensors.torch
-import torch
-from picklescan.scanner import scan_file_path
 from pydantic import BaseModel, ConfigDict, Discriminator, Field, Tag, TypeAdapter
 from typing_extensions import Annotated, Any, Dict
 
 from invokeai.app.util.misc import uuid_string
 from invokeai.backend.model_hash.hash_validator import validate_hash
-from invokeai.backend.model_hash.model_hash import HASHING_ALGORITHMS, ModelHash
+from invokeai.backend.model_hash.model_hash import HASHING_ALGORITHMS
+from invokeai.backend.model_manager.model_on_disk import ModelOnDisk
 from invokeai.backend.model_manager.taxonomy import (
     AnyVariant,
     BaseModelType,
     ClipVariantType,
+    FluxLoRAFormat,
     ModelFormat,
     ModelRepoVariant,
     ModelSourceType,
@@ -51,9 +50,8 @@ from invokeai.backend.model_manager.taxonomy import (
     SchedulerPredictionType,
     SubModelType,
 )
-from invokeai.backend.quantization.gguf.loaders import gguf_sd_loader
+from invokeai.backend.model_manager.util.model_util import lora_token_vector_length
 from invokeai.backend.stable_diffusion.schedulers.schedulers import SCHEDULER_NAME_VALUES
-from invokeai.backend.util.silence_warnings import SilenceWarnings
 
 logger = logging.getLogger(__name__)
 
@@ -67,11 +65,6 @@ class InvalidModelConfigException(Exception):
 DEFAULTS_PRECISION = Literal["fp16", "fp32"]
 
 
-class FSLayout(Enum):
-    FILE = "file"
-    DIRECTORY = "directory"
-
-
 class SubmodelDefinition(BaseModel):
     path_or_prefix: str
     model_type: ModelType
@@ -102,87 +95,6 @@ class ControlAdapterDefaultSettings(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
 
-class ModelOnDisk:
-    """A utility class representing a model stored on disk."""
-
-    def __init__(self, path: Path, hash_algo: HASHING_ALGORITHMS = "blake3_single"):
-        self.path = path
-        # TODO: Revisit checkpoint vs diffusers terminology
-        self.layout = FSLayout.DIRECTORY if path.is_dir() else FSLayout.FILE
-        if self.path.suffix in {".safetensors", ".bin", ".pt", ".ckpt"}:
-            self.name = path.stem
-        else:
-            self.name = path.name
-        self.hash_algo = hash_algo
-        self._state_dict_cache = {}
-
-    def hash(self) -> str:
-        return ModelHash(algorithm=self.hash_algo).hash(self.path)
-
-    def size(self) -> int:
-        if self.layout == FSLayout.FILE:
-            return self.path.stat().st_size
-        return sum(file.stat().st_size for file in self.path.rglob("*"))
-
-    def component_paths(self) -> set[Path]:
-        if self.layout == FSLayout.FILE:
-            return {self.path}
-        extensions = {".safetensors", ".pt", ".pth", ".ckpt", ".bin", ".gguf"}
-        return {f for f in self.path.rglob("*") if f.suffix in extensions}
-
-    def repo_variant(self) -> Optional[ModelRepoVariant]:
-        if self.layout == FSLayout.FILE:
-            return None
-
-        weight_files = list(self.path.glob("**/*.safetensors"))
-        weight_files.extend(list(self.path.glob("**/*.bin")))
-        for x in weight_files:
-            if ".fp16" in x.suffixes:
-                return ModelRepoVariant.FP16
-            if "openvino_model" in x.name:
-                return ModelRepoVariant.OpenVINO
-            if "flax_model" in x.name:
-                return ModelRepoVariant.Flax
-            if x.suffix == ".onnx":
-                return ModelRepoVariant.ONNX
-        return ModelRepoVariant.Default
-
-    def load_state_dict(self, path: Optional[Path] = None) -> Dict[str | int, Any]:
-        if path in self._state_dict_cache:
-            return self._state_dict_cache[path]
-
-        if not path:
-            components = list(self.component_paths())
-            match components:
-                case []:
-                    raise ValueError("No weight files found for this model")
-                case [p]:
-                    path = p
-                case ps if len(ps) >= 2:
-                    raise ValueError(
-                        f"Multiple weight files found for this model: {ps}. "
-                        f"Please specify the intended file using the 'path' argument"
-                    )
-
-        with SilenceWarnings():
-            if path.suffix.endswith((".ckpt", ".pt", ".pth", ".bin")):
-                scan_result = scan_file_path(path)
-                if scan_result.infected_files != 0 or scan_result.scan_err:
-                    raise RuntimeError(f"The model {path.stem} is potentially infected by malware. Aborting import.")
-                checkpoint = torch.load(path, map_location="cpu")
-                assert isinstance(checkpoint, dict)
-            elif path.suffix.endswith(".gguf"):
-                checkpoint = gguf_sd_loader(path, compute_dtype=torch.float32)
-            elif path.suffix.endswith(".safetensors"):
-                checkpoint = safetensors.torch.load_file(path)
-            else:
-                raise ValueError(f"Unrecognized model extension: {path.suffix}")
-
-        state_dict = checkpoint.get("state_dict", checkpoint)
-        self._state_dict_cache[path] = state_dict
-        return state_dict
-
-
 class MatchSpeed(int, Enum):
     """Represents the estimated runtime speed of a config's 'matches' method."""
 
@@ -257,7 +169,7 @@ class ModelConfigBase(ABC, BaseModel):
         Created to deprecate ModelProbe.probe
         """
         candidates = ModelConfigBase._USING_CLASSIFY_API
-        sorted_by_match_speed = sorted(candidates, key=lambda cls: cls._MATCH_SPEED)
+        sorted_by_match_speed = sorted(candidates, key=lambda cls: (cls._MATCH_SPEED, cls.__name__))
         mod = ModelOnDisk(model_path, hash_algo)
 
         for config_cls in sorted_by_match_speed:
@@ -308,6 +220,9 @@ class ModelConfigBase(ABC, BaseModel):
         if "source_type" in overrides:
             overrides["source_type"] = ModelSourceType(overrides["source_type"])
 
+        if "variant" in overrides:
+            overrides["variant"] = ModelVariantType(overrides["variant"])
+
     @classmethod
     def from_model_on_disk(cls, mod: ModelOnDisk, **overrides):
         """Creates an instance of this config or raises InvalidModelConfigException."""
@@ -367,6 +282,38 @@ class LoRAConfigBase(ABC, BaseModel):
     type: Literal[ModelType.LoRA] = ModelType.LoRA
     trigger_phrases: Optional[set[str]] = Field(description="Set of trigger phrases for this model", default=None)
 
+    @classmethod
+    def flux_lora_format(cls, mod: ModelOnDisk):
+        key = "FLUX_LORA_FORMAT"
+        if key in mod.cache:
+            return mod.cache[key]
+
+        from invokeai.backend.patches.lora_conversions.formats import flux_format_from_state_dict
+
+        sd = mod.load_state_dict(mod.path)
+        value = flux_format_from_state_dict(sd)
+        mod.cache[key] = value
+        return value
+
+    @classmethod
+    def base_model(cls, mod: ModelOnDisk) -> BaseModelType:
+        if cls.flux_lora_format(mod):
+            return BaseModelType.Flux
+
+        state_dict = mod.load_state_dict()
+        # If we've gotten here, we assume that the model is a Stable Diffusion model
+        token_vector_length = lora_token_vector_length(state_dict)
+        if token_vector_length == 768:
+            return BaseModelType.StableDiffusion1
+        elif token_vector_length == 1024:
+            return BaseModelType.StableDiffusion2
+        elif token_vector_length == 1280:
+            return BaseModelType.StableDiffusionXL  # recognizes format at https://civitai.com/models/224641
+        elif token_vector_length == 2048:
+            return BaseModelType.StableDiffusionXL
+        else:
+            raise InvalidModelConfigException("Unknown LoRA type")
+
 
 class T5EncoderConfigBase(ABC, BaseModel):
     """Base class for diffusers-style models."""
@@ -382,11 +329,40 @@ class T5EncoderBnbQuantizedLlmInt8bConfig(T5EncoderConfigBase, LegacyProbeMixin,
     format: Literal[ModelFormat.BnbQuantizedLlmInt8b] = ModelFormat.BnbQuantizedLlmInt8b
 
 
-class LoRALyCORISConfig(LoRAConfigBase, LegacyProbeMixin, ModelConfigBase):
+class LoRALyCORISConfig(LoRAConfigBase, ModelConfigBase):
     """Model config for LoRA/Lycoris models."""
 
     format: Literal[ModelFormat.LyCORIS] = ModelFormat.LyCORIS
 
+    @classmethod
+    def matches(cls, mod: ModelOnDisk) -> bool:
+        if mod.path.is_dir():
+            return False
+
+        # Avoid false positive match against ControlLoRA and Diffusers
+        if cls.flux_lora_format(mod) in [FluxLoRAFormat.Control, FluxLoRAFormat.Diffusers]:
+            return False
+
+        state_dict = mod.load_state_dict()
+        for key in state_dict.keys():
+            if type(key) is int:
+                continue
+
+            if key.startswith(("lora_te_", "lora_unet_", "lora_te1_", "lora_te2_", "lora_transformer_")):
+                return True
+            # "lora_A.weight" and "lora_B.weight" are associated with models in PEFT format. We don't support all PEFT
+            # LoRA models, but as of the time of writing, we support Diffusers FLUX PEFT LoRA models.
+            if key.endswith(("to_k_lora.up.weight", "to_q_lora.down.weight", "lora_A.weight", "lora_B.weight")):
+                return True
+
+        return False
+
+    @classmethod
+    def parse(cls, mod: ModelOnDisk) -> dict[str, Any]:
+        return {
+            "base": cls.base_model(mod),
+        }
+
 
 class ControlAdapterConfigBase(ABC, BaseModel):
     default_settings: Optional[ControlAdapterDefaultSettings] = Field(
@@ -410,11 +386,26 @@ class ControlLoRADiffusersConfig(ControlAdapterConfigBase, LegacyProbeMixin, Mod
     format: Literal[ModelFormat.Diffusers] = ModelFormat.Diffusers
 
 
-class LoRADiffusersConfig(LoRAConfigBase, LegacyProbeMixin, ModelConfigBase):
+class LoRADiffusersConfig(LoRAConfigBase, ModelConfigBase):
     """Model config for LoRA/Diffusers models."""
 
     format: Literal[ModelFormat.Diffusers] = ModelFormat.Diffusers
 
+    @classmethod
+    def matches(cls, mod: ModelOnDisk) -> bool:
+        if mod.path.is_file():
+            return cls.flux_lora_format(mod) == FluxLoRAFormat.Diffusers
+
+        suffixes = ["bin", "safetensors"]
+        weight_files = [mod.path / f"pytorch_lora_weights.{sfx}" for sfx in suffixes]
+        return any(wf.exists() for wf in weight_files)
+
+    @classmethod
+    def parse(cls, mod: ModelOnDisk) -> dict[str, Any]:
+        return {
+            "base": cls.base_model(mod),
+        }
+
 
 class VAECheckpointConfig(CheckpointConfigBase, LegacyProbeMixin, ModelConfigBase):
     """Model config for standalone VAE models."""
@@ -586,7 +577,7 @@ class LlavaOnevisionConfig(DiffusersConfigBase, ModelConfigBase):
 
     @classmethod
     def matches(cls, mod: ModelOnDisk) -> bool:
-        if mod.layout == FSLayout.FILE:
+        if mod.path.is_file():
             return False
 
         config_path = mod.path / "config.json"

invokeai/backend/model_manager/model_on_disk.py

@@ -0,0 +1,96 @@
+from pathlib import Path
+from typing import Any, Optional, TypeAlias
+
+import safetensors.torch
+import torch
+from picklescan.scanner import scan_file_path
+
+from invokeai.backend.model_hash.model_hash import HASHING_ALGORITHMS, ModelHash
+from invokeai.backend.model_manager.taxonomy import ModelRepoVariant
+from invokeai.backend.quantization.gguf.loaders import gguf_sd_loader
+from invokeai.backend.util.silence_warnings import SilenceWarnings
+
+StateDict: TypeAlias = dict[str | int, Any]  # When are the keys int?
+
+
+class ModelOnDisk:
+    """A utility class representing a model stored on disk."""
+
+    def __init__(self, path: Path, hash_algo: HASHING_ALGORITHMS = "blake3_single"):
+        self.path = path
+        if self.path.suffix in {".safetensors", ".bin", ".pt", ".ckpt"}:
+            self.name = path.stem
+        else:
+            self.name = path.name
+        self.hash_algo = hash_algo
+        # Having a cache helps users of ModelOnDisk (i.e. configs) to save state
+        # This prevents redundant computations during matching and parsing
+        self.cache = {"_CACHED_STATE_DICTS": {}}
+
+    def hash(self) -> str:
+        return ModelHash(algorithm=self.hash_algo).hash(self.path)
+
+    def size(self) -> int:
+        if self.path.is_file():
+            return self.path.stat().st_size
+        return sum(file.stat().st_size for file in self.path.rglob("*"))
+
+    def component_paths(self) -> set[Path]:
+        if self.path.is_file():
+            return {self.path}
+        extensions = {".safetensors", ".pt", ".pth", ".ckpt", ".bin", ".gguf"}
+        return {f for f in self.path.rglob("*") if f.suffix in extensions}
+
+    def repo_variant(self) -> Optional[ModelRepoVariant]:
+        if self.path.is_file():
+            return None
+
+        weight_files = list(self.path.glob("**/*.safetensors"))
+        weight_files.extend(list(self.path.glob("**/*.bin")))
+        for x in weight_files:
+            if ".fp16" in x.suffixes:
+                return ModelRepoVariant.FP16
+            if "openvino_model" in x.name:
+                return ModelRepoVariant.OpenVINO
+            if "flax_model" in x.name:
+                return ModelRepoVariant.Flax
+            if x.suffix == ".onnx":
+                return ModelRepoVariant.ONNX
+        return ModelRepoVariant.Default
+
+    def load_state_dict(self, path: Optional[Path] = None) -> StateDict:
+        sd_cache = self.cache["_CACHED_STATE_DICTS"]
+
+        if path in sd_cache:
+            return sd_cache[path]
+
+        if not path:
+            components = list(self.component_paths())
+            match components:
+                case []:
+                    raise ValueError("No weight files found for this model")
+                case [p]:
+                    path = p
+                case ps if len(ps) >= 2:
+                    raise ValueError(
+                        f"Multiple weight files found for this model: {ps}. "
+                        f"Please specify the intended file using the 'path' argument"
+                    )
+
+        with SilenceWarnings():
+            if path.suffix.endswith((".ckpt", ".pt", ".pth", ".bin")):
+                scan_result = scan_file_path(path)
+                if scan_result.infected_files != 0 or scan_result.scan_err:
+                    raise RuntimeError(f"The model {path.stem} is potentially infected by malware. Aborting import.")
+                checkpoint = torch.load(path, map_location="cpu")
+                assert isinstance(checkpoint, dict)
+            elif path.suffix.endswith(".gguf"):
+                checkpoint = gguf_sd_loader(path, compute_dtype=torch.float32)
+            elif path.suffix.endswith(".safetensors"):
+                checkpoint = safetensors.torch.load_file(path)
+            else:
+                raise ValueError(f"Unrecognized model extension: {path.suffix}")
+
+        state_dict = checkpoint.get("state_dict", checkpoint)
+        sd_cache[path] = state_dict
+        return state_dict

invokeai/backend/model_manager/taxonomy.py

@@ -126,4 +126,13 @@ class ModelSourceType(str, Enum):
     HFRepoID = "hf_repo_id"
 
 
+class FluxLoRAFormat(str, Enum):
+    """Flux LoRA formats."""
+
+    Diffusers = "flux.diffusers"
+    Kohya = "flux.kohya"
+    OneTrainer = "flux.onetrainer"
+    Control = "flux.control"
+
+
 AnyVariant: TypeAlias = Union[ModelVariantType, ClipVariantType, None]

invokeai/backend/patches/lora_conversions/formats.py

@@ -0,0 +1,24 @@
+from invokeai.backend.model_manager.taxonomy import FluxLoRAFormat
+from invokeai.backend.patches.lora_conversions.flux_control_lora_utils import is_state_dict_likely_flux_control
+from invokeai.backend.patches.lora_conversions.flux_diffusers_lora_conversion_utils import (
+    is_state_dict_likely_in_flux_diffusers_format,
+)
+from invokeai.backend.patches.lora_conversions.flux_kohya_lora_conversion_utils import (
+    is_state_dict_likely_in_flux_kohya_format,
+)
+from invokeai.backend.patches.lora_conversions.flux_onetrainer_lora_conversion_utils import (
+    is_state_dict_likely_in_flux_onetrainer_format,
+)
+
+
+def flux_format_from_state_dict(state_dict):
+    if is_state_dict_likely_in_flux_kohya_format(state_dict):
+        return FluxLoRAFormat.Kohya
+    elif is_state_dict_likely_in_flux_onetrainer_format(state_dict):
+        return FluxLoRAFormat.OneTrainer
+    elif is_state_dict_likely_in_flux_diffusers_format(state_dict):
+        return FluxLoRAFormat.Diffusers
+    elif is_state_dict_likely_flux_control(state_dict):
+        return FluxLoRAFormat.Control
+    else:
+        return None

scripts/strip_models.py

@@ -22,9 +22,8 @@ from pathlib import Path
 import humanize
 import torch
 
-from invokeai.backend.model_manager.config import ModelOnDisk
+from invokeai.backend.model_manager.model_on_disk import ModelOnDisk
 from invokeai.backend.model_manager.search import ModelSearch
-from invokeai.backend.model_manager.taxonomy import ModelFormat
 
 
 def strip(v):
@@ -63,7 +62,7 @@ def load_stripped_model(path: Path, *args, **kwargs):
 
 def create_stripped_model(original_model_path: Path, stripped_model_path: Path) -> ModelOnDisk:
     original = ModelOnDisk(original_model_path)
-    if original.format_type == ModelFormat.Checkpoint:
+    if original.path.is_file():
         shutil.copy2(original.path, stripped_model_path)
     else:
         shutil.copytree(original.path, stripped_model_path, dirs_exist_ok=True)

tests/test_model_probe.py

@@ -17,7 +17,6 @@ from invokeai.backend.model_manager.config import (
     MainDiffusersConfig,
     ModelConfigBase,
     ModelConfigFactory,
-    ModelOnDisk,
     get_model_discriminator_value,
 )
 from invokeai.backend.model_manager.legacy_probe import (
@@ -27,6 +26,7 @@ from invokeai.backend.model_manager.legacy_probe import (
     get_default_settings_control_adapters,
     get_default_settings_main,
 )
+from invokeai.backend.model_manager.model_on_disk import ModelOnDisk
 from invokeai.backend.model_manager.search import ModelSearch
 from invokeai.backend.util.logging import InvokeAILogger
 

tests/test_model_probe/stripped_models/Hard Edge Detection (Canny).safetensors

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:5acefb3658338a4126736e2da02cfef5a9ce6e2469564a6c7994ae34e8ef2e8a
+size 192447

tests/test_model_probe/stripped_models/flat_colour_anime_style_schnell_v3.4.safetensors

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:55aafd0f9b4ac2863361573b070320e13b800b2359a81a73878008bdffc3edfa
+size 201040