Add support for FLUX ControlNet models (XLabs and InstantX)

invokeai/app/invocations/fields.py

@@ -192,6 +192,7 @@ class FieldDescriptions:
     freeu_s2 = 'Scaling factor for stage 2 to attenuate the contributions of the skip features. This is done to mitigate the "oversmoothing effect" in the enhanced denoising process.'
     freeu_b1 = "Scaling factor for stage 1 to amplify the contributions of backbone features."
     freeu_b2 = "Scaling factor for stage 2 to amplify the contributions of backbone features."
+    instantx_control_mode = "The control mode for InstantX ControlNet union models. Ignored for other ControlNet models. The standard mapping is: canny (0), tile (1), depth (2), blur (3), pose (4), gray (5), low quality (6). Negative values will be treated as 'None'."
 
 
 class ImageField(BaseModel):

invokeai/app/invocations/flux_controlnet.py

@@ -0,0 +1,99 @@
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+from invokeai.app.invocations.baseinvocation import (
+    BaseInvocation,
+    BaseInvocationOutput,
+    Classification,
+    invocation,
+    invocation_output,
+)
+from invokeai.app.invocations.fields import FieldDescriptions, ImageField, InputField, OutputField, UIType
+from invokeai.app.invocations.model import ModelIdentifierField
+from invokeai.app.invocations.util import validate_begin_end_step, validate_weights
+from invokeai.app.services.shared.invocation_context import InvocationContext
+from invokeai.app.util.controlnet_utils import CONTROLNET_RESIZE_VALUES
+
+
+class FluxControlNetField(BaseModel):
+    image: ImageField = Field(description="The control image")
+    control_model: ModelIdentifierField = Field(description="The ControlNet model to use")
+    control_weight: float | list[float] = Field(default=1, description="The weight given to the ControlNet")
+    begin_step_percent: float = Field(
+        default=0, ge=0, le=1, description="When the ControlNet is first applied (% of total steps)"
+    )
+    end_step_percent: float = Field(
+        default=1, ge=0, le=1, description="When the ControlNet is last applied (% of total steps)"
+    )
+    resize_mode: CONTROLNET_RESIZE_VALUES = Field(default="just_resize", description="The resize mode to use")
+    instantx_control_mode: int | None = Field(default=-1, description=FieldDescriptions.instantx_control_mode)
+
+    @field_validator("control_weight")
+    @classmethod
+    def validate_control_weight(cls, v: float | list[float]) -> float | list[float]:
+        validate_weights(v)
+        return v
+
+    @model_validator(mode="after")
+    def validate_begin_end_step_percent(self):
+        validate_begin_end_step(self.begin_step_percent, self.end_step_percent)
+        return self
+
+
+@invocation_output("flux_controlnet_output")
+class FluxControlNetOutput(BaseInvocationOutput):
+    """FLUX ControlNet info"""
+
+    control: FluxControlNetField = OutputField(description=FieldDescriptions.control)
+
+
+@invocation(
+    "flux_controlnet",
+    title="FLUX ControlNet",
+    tags=["controlnet", "flux"],
+    category="controlnet",
+    version="1.0.0",
+    classification=Classification.Prototype,
+)
+class FluxControlNetInvocation(BaseInvocation):
+    """Collect FLUX ControlNet info to pass to other nodes."""
+
+    image: ImageField = InputField(description="The control image")
+    control_model: ModelIdentifierField = InputField(
+        description=FieldDescriptions.controlnet_model, ui_type=UIType.ControlNetModel
+    )
+    control_weight: float | list[float] = InputField(
+        default=1.0, ge=-1, le=2, description="The weight given to the ControlNet"
+    )
+    begin_step_percent: float = InputField(
+        default=0, ge=0, le=1, description="When the ControlNet is first applied (% of total steps)"
+    )
+    end_step_percent: float = InputField(
+        default=1, ge=0, le=1, description="When the ControlNet is last applied (% of total steps)"
+    )
+    resize_mode: CONTROLNET_RESIZE_VALUES = InputField(default="just_resize", description="The resize mode used")
+    # Note: We default to -1 instead of None, because in the workflow editor UI None is not currently supported.
+    instantx_control_mode: int | None = InputField(default=-1, description=FieldDescriptions.instantx_control_mode)
+
+    @field_validator("control_weight")
+    @classmethod
+    def validate_control_weight(cls, v: float | list[float]) -> float | list[float]:
+        validate_weights(v)
+        return v
+
+    @model_validator(mode="after")
+    def validate_begin_end_step_percent(self):
+        validate_begin_end_step(self.begin_step_percent, self.end_step_percent)
+        return self
+
+    def invoke(self, context: InvocationContext) -> FluxControlNetOutput:
+        return FluxControlNetOutput(
+            control=FluxControlNetField(
+                image=self.image,
+                control_model=self.control_model,
+                control_weight=self.control_weight,
+                begin_step_percent=self.begin_step_percent,
+                end_step_percent=self.end_step_percent,
+                resize_mode=self.resize_mode,
+                instantx_control_mode=self.instantx_control_mode,
+            ),
+        )

invokeai/app/invocations/flux_denoise.py

@@ -16,11 +16,16 @@
     WithBoard,
     WithMetadata,
 )
-from invokeai.app.invocations.model import TransformerField
+from invokeai.app.invocations.flux_controlnet import FluxControlNetField
+from invokeai.app.invocations.model import TransformerField, VAEField
 from invokeai.app.invocations.primitives import LatentsOutput
 from invokeai.app.services.shared.invocation_context import InvocationContext
+from invokeai.backend.flux.controlnet.instantx_controlnet_flux import InstantXControlNetFlux
+from invokeai.backend.flux.controlnet.xlabs_controlnet_flux import XLabsControlNetFlux
 from invokeai.backend.flux.denoise import denoise
-from invokeai.backend.flux.inpaint_extension import InpaintExtension
+from invokeai.backend.flux.extensions.inpaint_extension import InpaintExtension
+from invokeai.backend.flux.extensions.instantx_controlnet_extension import InstantXControlNetExtension
+from invokeai.backend.flux.extensions.xlabs_controlnet_extension import XLabsControlNetExtension
 from invokeai.backend.flux.model import Flux
 from invokeai.backend.flux.sampling_utils import (
     clip_timestep_schedule_fractional,
@@ -44,7 +49,7 @@
     title="FLUX Denoise",
     tags=["image", "flux"],
     category="image",
-    version="3.0.0",
+    version="3.1.0",
     classification=Classification.Prototype,
 )
 class FluxDenoiseInvocation(BaseInvocation, WithMetadata, WithBoard):
@@ -87,6 +92,13 @@ class FluxDenoiseInvocation(BaseInvocation, WithMetadata, WithBoard):
         description="The guidance strength. Higher values adhere more strictly to the prompt, and will produce less diverse images. FLUX dev only, ignored for schnell.",
     )
     seed: int = InputField(default=0, description="Randomness seed for reproducibility.")
+    control: FluxControlNetField | list[FluxControlNetField] | None = InputField(
+        default=None, input=Input.Connection, description="ControlNet models."
+    )
+    controlnet_vae: VAEField | None = InputField(
+        description=FieldDescriptions.vae,
+        input=Input.Connection,
+    )
 
     @torch.no_grad()
     def invoke(self, context: InvocationContext) -> LatentsOutput:
@@ -167,8 +179,8 @@ def _run_diffusion(
 
         inpaint_mask = self._prep_inpaint_mask(context, x)
 
-        b, _c, h, w = x.shape
-        img_ids = generate_img_ids(h=h, w=w, batch_size=b, device=x.device, dtype=x.dtype)
+        b, _c, latent_h, latent_w = x.shape
+        img_ids = generate_img_ids(h=latent_h, w=latent_w, batch_size=b, device=x.device, dtype=x.dtype)
 
         bs, t5_seq_len, _ = t5_embeddings.shape
         txt_ids = torch.zeros(bs, t5_seq_len, 3, dtype=inference_dtype, device=TorchDevice.choose_torch_device())
@@ -192,12 +204,21 @@ def _run_diffusion(
                 noise=noise,
             )
 
-        with (
-            transformer_info.model_on_device() as (cached_weights, transformer),
-            ExitStack() as exit_stack,
-        ):
-            assert isinstance(transformer, Flux)
+        with ExitStack() as exit_stack:
+            # Prepare ControlNet extensions.
+            # Note: We do this before loading the transformer model to minimize peak memory (see implementation).
+            controlnet_extensions = self._prep_controlnet_extensions(
+                context=context,
+                exit_stack=exit_stack,
+                latent_height=latent_h,
+                latent_width=latent_w,
+                dtype=inference_dtype,
+                device=x.device,
+            )
 
+            # Load the transformer model.
+            (cached_weights, transformer) = exit_stack.enter_context(transformer_info.model_on_device())
+            assert isinstance(transformer, Flux)
             config = transformer_info.config
             assert config is not None
 
@@ -242,6 +263,7 @@ def _run_diffusion(
                 step_callback=self._build_step_callback(context),
                 guidance=self.guidance,
                 inpaint_extension=inpaint_extension,
+                controlnet_extensions=controlnet_extensions,
             )
 
         x = unpack(x.float(), self.height, self.width)
@@ -288,6 +310,104 @@ def _prep_inpaint_mask(self, context: InvocationContext, latents: torch.Tensor)
         # `latents`.
         return mask.expand_as(latents)
 
+    def _prep_controlnet_extensions(
+        self,
+        context: InvocationContext,
+        exit_stack: ExitStack,
+        latent_height: int,
+        latent_width: int,
+        dtype: torch.dtype,
+        device: torch.device,
+    ) -> list[XLabsControlNetExtension | InstantXControlNetExtension]:
+        # Normalize the controlnet input to list[ControlField].
+        controlnets: list[FluxControlNetField]
+        if self.control is None:
+            controlnets = []
+        elif isinstance(self.control, FluxControlNetField):
+            controlnets = [self.control]
+        elif isinstance(self.control, list):
+            controlnets = self.control
+        else:
+            raise ValueError(f"Unsupported controlnet type: {type(self.control)}")
+
+        # TODO(ryand): Add a field to the model config so that we can distinguish between XLabs and InstantX ControlNets
+        # before loading the models. Then make sure that all VAE encoding is done before loading the ControlNets to
+        # minimize peak memory.
+
+        # First, load the ControlNet models so that we can determine the ControlNet types.
+        controlnet_models = [context.models.load(controlnet.control_model) for controlnet in controlnets]
+
+        # Calculate the controlnet conditioning tensors.
+        # We do this before loading the ControlNet models because it may require running the VAE, and we are trying to
+        # keep peak memory down.
+        controlnet_conds: list[torch.Tensor] = []
+        for controlnet, controlnet_model in zip(controlnets, controlnet_models, strict=True):
+            image = context.images.get_pil(controlnet.image.image_name)
+            if isinstance(controlnet_model.model, InstantXControlNetFlux):
+                if self.controlnet_vae is None:
+                    raise ValueError("A ControlNet VAE is required when using an InstantX FLUX ControlNet.")
+                vae_info = context.models.load(self.controlnet_vae.vae)
+                controlnet_conds.append(
+                    InstantXControlNetExtension.prepare_controlnet_cond(
+                        controlnet_image=image,
+                        vae_info=vae_info,
+                        latent_height=latent_height,
+                        latent_width=latent_width,
+                        dtype=dtype,
+                        device=device,
+                        resize_mode=controlnet.resize_mode,
+                    )
+                )
+            elif isinstance(controlnet_model.model, XLabsControlNetFlux):
+                controlnet_conds.append(
+                    XLabsControlNetExtension.prepare_controlnet_cond(
+                        controlnet_image=image,
+                        latent_height=latent_height,
+                        latent_width=latent_width,
+                        dtype=dtype,
+                        device=device,
+                        resize_mode=controlnet.resize_mode,
+                    )
+                )
+
+        # Finally, load the ControlNet models and initialize the ControlNet extensions.
+        controlnet_extensions: list[XLabsControlNetExtension | InstantXControlNetExtension] = []
+        for controlnet, controlnet_cond, controlnet_model in zip(
+            controlnets, controlnet_conds, controlnet_models, strict=True
+        ):
+            model = exit_stack.enter_context(controlnet_model)
+
+            if isinstance(model, XLabsControlNetFlux):
+                controlnet_extensions.append(
+                    XLabsControlNetExtension(
+                        model=model,
+                        controlnet_cond=controlnet_cond,
+                        weight=controlnet.control_weight,
+                        begin_step_percent=controlnet.begin_step_percent,
+                        end_step_percent=controlnet.end_step_percent,
+                    )
+                )
+            elif isinstance(model, InstantXControlNetFlux):
+                instantx_control_mode: torch.Tensor | None = None
+                if controlnet.instantx_control_mode is not None and controlnet.instantx_control_mode >= 0:
+                    instantx_control_mode = torch.tensor(controlnet.instantx_control_mode, dtype=torch.long)
+                    instantx_control_mode = instantx_control_mode.reshape([-1, 1])
+
+                controlnet_extensions.append(
+                    InstantXControlNetExtension(
+                        model=model,
+                        controlnet_cond=controlnet_cond,
+                        instantx_control_mode=instantx_control_mode,
+                        weight=controlnet.control_weight,
+                        begin_step_percent=controlnet.begin_step_percent,
+                        end_step_percent=controlnet.end_step_percent,
+                    )
+                )
+            else:
+                raise ValueError(f"Unsupported ControlNet model type: {type(model)}")
+
+        return controlnet_extensions
+
     def _lora_iterator(self, context: InvocationContext) -> Iterator[Tuple[LoRAModelRaw, float]]:
         for lora in self.transformer.loras:
             lora_info = context.models.load(lora.lora)

invokeai/backend/flux/controlnet/controlnet_flux_output.py

@@ -0,0 +1,58 @@
+from dataclasses import dataclass
+
+import torch
+
+
+@dataclass
+class ControlNetFluxOutput:
+    single_block_residuals: list[torch.Tensor] | None
+    double_block_residuals: list[torch.Tensor] | None
+
+    def apply_weight(self, weight: float):
+        if self.single_block_residuals is not None:
+            for i in range(len(self.single_block_residuals)):
+                self.single_block_residuals[i] = self.single_block_residuals[i] * weight
+        if self.double_block_residuals is not None:
+            for i in range(len(self.double_block_residuals)):
+                self.double_block_residuals[i] = self.double_block_residuals[i] * weight
+
+
+def add_tensor_lists_elementwise(
+    list1: list[torch.Tensor] | None, list2: list[torch.Tensor] | None
+) -> list[torch.Tensor] | None:
+    """Add two tensor lists elementwise that could be None."""
+    if list1 is None and list2 is None:
+        return None
+    if list1 is None:
+        return list2
+    if list2 is None:
+        return list1
+
+    new_list: list[torch.Tensor] = []
+    for list1_tensor, list2_tensor in zip(list1, list2, strict=True):
+        new_list.append(list1_tensor + list2_tensor)
+    return new_list
+
+
+def add_controlnet_flux_outputs(
+    controlnet_output_1: ControlNetFluxOutput, controlnet_output_2: ControlNetFluxOutput
+) -> ControlNetFluxOutput:
+    return ControlNetFluxOutput(
+        single_block_residuals=add_tensor_lists_elementwise(
+            controlnet_output_1.single_block_residuals, controlnet_output_2.single_block_residuals
+        ),
+        double_block_residuals=add_tensor_lists_elementwise(
+            controlnet_output_1.double_block_residuals, controlnet_output_2.double_block_residuals
+        ),
+    )
+
+
+def sum_controlnet_flux_outputs(
+    controlnet_outputs: list[ControlNetFluxOutput],
+) -> ControlNetFluxOutput:
+    controlnet_output_sum = ControlNetFluxOutput(single_block_residuals=None, double_block_residuals=None)
+
+    for controlnet_output in controlnet_outputs:
+        controlnet_output_sum = add_controlnet_flux_outputs(controlnet_output_sum, controlnet_output)
+
+    return controlnet_output_sum