Spaces:
Running on Zero
Running on Zero
Download diffusers/modular_pipelines/flux3/modular_blocks_flux3.py from diffusers-modular/flux3-image-klein-bbox: direct link, hf CLI and curl.
- Browser
- Download file 16.7 kB
-
https://huggingface.co/spaces/diffusers-modular/flux3-image-klein-bbox/resolve/main/diffusers/modular_pipelines/flux3/modular_blocks_flux3.py
- Command line
-
hf download hf://spaces/diffusers-modular/flux3-image-klein-bbox/diffusers/modular_pipelines/flux3/modular_blocks_flux3.py
-
curl -L -o modular_blocks_flux3.py https://huggingface.co/spaces/diffusers-modular/flux3-image-klein-bbox/resolve/main/diffusers/modular_pipelines/flux3/modular_blocks_flux3.py
16.7 kB
| # Copyright 2026 Black Forest Labs and The HuggingFace Team. All rights reserved. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| from ...utils import logging | |
| from ..modular_pipeline import AutoPipelineBlocks, SequentialPipelineBlocks | |
| from ..modular_pipeline_utils import InsertableDict, OutputParam | |
| from .before_denoise import ( | |
| Flux3PrepareImageCondLatentsStep, | |
| Flux3PrepareLatentsStep, | |
| Flux3SetTimestepsStep, | |
| Flux3TextRoPEInputsStep, | |
| ) | |
| from .decoders import Flux3DecodeStep, Flux3UnpackLatentsStep | |
| from .denoise import Flux3DenoiseStep | |
| from .encoders import Flux3ProcessImagesInputStep, Flux3TextEncoderStep, Flux3VaeEncoderStep | |
| from .inputs import Flux3ImageCondInputStep, Flux3TextInputStep | |
| logger = logging.get_logger(__name__) # pylint: disable=invalid-name | |
| ### | |
| ### VAE encoder (reference images) | |
| ### | |
| # auto_docstring | |
| class Flux3VaeEncoderSequentialStep(SequentialPipelineBlocks): | |
| """ | |
| VAE encoder step that preprocesses and encodes the reference images into their latent representations. | |
| Components: | |
| image_processor (`Flux3ImageProcessor`) vae (`AutoencoderKLFlux2`) | |
| Inputs: | |
| image (`Image | list`): | |
| The reference image or images to condition the generation on; each keeps its own resolution | |
| (center-cropped to the 16-pixel grid) | |
| Outputs: | |
| condition_images (`list`): | |
| The preprocessed reference images, one tensor of shape (1, 3, height_i, width_i) in [-1, 1] per image | |
| image_cond_latents (`list`): | |
| Normalized packed latents of the reference images, one tensor of shape (1, 128, height_i//16, | |
| width_i//16) per image | |
| """ | |
| model_name = "flux3" | |
| block_classes = [Flux3ProcessImagesInputStep(), Flux3VaeEncoderStep()] | |
| block_names = ["preprocess", "encode"] | |
| def description(self) -> str: | |
| return "VAE encoder step that preprocesses and encodes the reference images into their latent representations." | |
| # auto_docstring | |
| class Flux3AutoVaeEncoderStep(AutoPipelineBlocks): | |
| """ | |
| VAE encoder step that encodes the reference images into their latent representations. | |
| - `Flux3VaeEncoderSequentialStep` is used when `image` is provided (image-to-image). | |
| - If `image` is not provided, step will be skipped. | |
| Components: | |
| image_processor (`Flux3ImageProcessor`) vae (`AutoencoderKLFlux2`) | |
| Inputs: | |
| image (`Image | list`, *optional*): | |
| The reference image or images to condition the generation on; each keeps its own resolution | |
| (center-cropped to the 16-pixel grid) | |
| Outputs: | |
| condition_images (`list`): | |
| The preprocessed reference images, one tensor of shape (1, 3, height_i, width_i) in [-1, 1] per image | |
| image_cond_latents (`list`): | |
| Normalized packed latents of the reference images, one tensor of shape (1, 128, height_i//16, | |
| width_i//16) per image | |
| """ | |
| model_name = "flux3" | |
| block_classes = [Flux3VaeEncoderSequentialStep] | |
| block_names = ["image_conditioning"] | |
| block_trigger_inputs = ["image"] | |
| def description(self): | |
| return ( | |
| "VAE encoder step that encodes the reference images into their latent representations.\n" | |
| " - `Flux3VaeEncoderSequentialStep` is used when `image` is provided (image-to-image).\n" | |
| " - If `image` is not provided, step will be skipped." | |
| ) | |
| ### | |
| ### Core denoise | |
| ### | |
| Flux3CoreDenoiseBlocks = InsertableDict( | |
| [ | |
| ("input", Flux3TextInputStep()), | |
| ("prepare_latents", Flux3PrepareLatentsStep()), | |
| ("set_timesteps", Flux3SetTimestepsStep()), | |
| ("prepare_rope_inputs", Flux3TextRoPEInputsStep()), | |
| ("denoise", Flux3DenoiseStep()), | |
| ("after_denoise", Flux3UnpackLatentsStep()), | |
| ] | |
| ) | |
| # auto_docstring | |
| class Flux3CoreDenoiseStep(SequentialPipelineBlocks): | |
| """ | |
| Core denoise step that performs the denoising process for FLUX.3, for text-to-image generation. | |
| Components: | |
| transformer (`Flux3Transformer2DModel`) scheduler (`FlowMatchEulerDiscreteScheduler`) guider | |
| (`ClassifierFreeGuidance`) | |
| Inputs: | |
| num_images_per_prompt (`int`, *optional*, defaults to 1): | |
| The number of images to generate per prompt. | |
| prompt_embeds (`Tensor`): | |
| Pre-generated text embeddings. Can be generated from text_encoder step. | |
| negative_prompt_embeds (`Tensor`, *optional*): | |
| Pre-generated negative text embeddings. Can be generated from text_encoder step. | |
| height (`int`, *optional*, defaults to 1024): | |
| The height in pixels of the generated image. | |
| width (`int`, *optional*, defaults to 1024): | |
| The width in pixels of the generated image. | |
| latents (`Tensor`, *optional*): | |
| Pre-generated noisy latents for image generation. | |
| generator (`Generator`, *optional*): | |
| Torch generator for deterministic generation. | |
| num_inference_steps (`int`, *optional*, defaults to 50): | |
| The number of denoising steps. | |
| timesteps (`Tensor`, *optional*): | |
| Timesteps for the denoising process. (overrides `num_inference_steps` when provided) | |
| sigmas (`list`, *optional*): | |
| Custom sigmas for the denoising process. (overrides `num_inference_steps` when provided) | |
| **denoiser_input_fields (`None`, *optional*): | |
| conditional model inputs for the denoiser: e.g. prompt_embeds, negative_prompt_embeds, etc. | |
| Outputs: | |
| latents (`Tensor`): | |
| Denoised latents. | |
| """ | |
| model_name = "flux3" | |
| block_classes = Flux3CoreDenoiseBlocks.values() | |
| block_names = Flux3CoreDenoiseBlocks.keys() | |
| def description(self): | |
| return "Core denoise step that performs the denoising process for FLUX.3, for text-to-image generation." | |
| def outputs(self): | |
| return [ | |
| OutputParam.template("latents"), | |
| ] | |
| Flux3ImageConditionedCoreDenoiseBlocks = InsertableDict( | |
| [ | |
| ("input", Flux3TextInputStep()), | |
| ("cond_input", Flux3ImageCondInputStep()), | |
| ("prepare_latents", Flux3PrepareLatentsStep()), | |
| ("prepare_image_cond_latents", Flux3PrepareImageCondLatentsStep()), | |
| ("set_timesteps", Flux3SetTimestepsStep()), | |
| ("prepare_rope_inputs", Flux3TextRoPEInputsStep()), | |
| ("denoise", Flux3DenoiseStep()), | |
| ("after_denoise", Flux3UnpackLatentsStep()), | |
| ] | |
| ) | |
| # auto_docstring | |
| class Flux3ImageConditionedCoreDenoiseStep(SequentialPipelineBlocks): | |
| """ | |
| Core denoise step that performs the denoising process for FLUX.3 with reference images (image-to-image). | |
| Components: | |
| transformer (`Flux3Transformer2DModel`) scheduler (`FlowMatchEulerDiscreteScheduler`) guider | |
| (`ClassifierFreeGuidance`) | |
| Inputs: | |
| num_images_per_prompt (`int`, *optional*, defaults to 1): | |
| The number of images to generate per prompt. | |
| prompt_embeds (`Tensor`): | |
| Pre-generated text embeddings. Can be generated from text_encoder step. | |
| negative_prompt_embeds (`Tensor`, *optional*): | |
| Pre-generated negative text embeddings. Can be generated from text_encoder step. | |
| image_cond_latents (`list`): | |
| Normalized packed latents of the reference images, one tensor of shape (B, C, height_i//16, width_i//16) | |
| per image | |
| height (`int`, *optional*): | |
| The height in pixels of the generated image; defaults to the first reference image's aspect ratio at the | |
| default area | |
| width (`int`, *optional*): | |
| The width in pixels of the generated image; defaults to the first reference image's aspect ratio at the | |
| default area | |
| latents (`Tensor`, *optional*): | |
| Pre-generated noisy latents for image generation. | |
| generator (`Generator`, *optional*): | |
| Torch generator for deterministic generation. | |
| num_inference_steps (`int`, *optional*, defaults to 50): | |
| The number of denoising steps. | |
| timesteps (`Tensor`, *optional*): | |
| Timesteps for the denoising process. (overrides `num_inference_steps` when provided) | |
| sigmas (`list`, *optional*): | |
| Custom sigmas for the denoising process. (overrides `num_inference_steps` when provided) | |
| **denoiser_input_fields (`None`, *optional*): | |
| conditional model inputs for the denoiser: e.g. prompt_embeds, negative_prompt_embeds, etc. | |
| Outputs: | |
| latents (`Tensor`): | |
| Denoised latents. | |
| """ | |
| model_name = "flux3" | |
| block_classes = Flux3ImageConditionedCoreDenoiseBlocks.values() | |
| block_names = Flux3ImageConditionedCoreDenoiseBlocks.keys() | |
| def description(self): | |
| return ( | |
| "Core denoise step that performs the denoising process for FLUX.3 with reference images (image-to-image)." | |
| ) | |
| def outputs(self): | |
| return [ | |
| OutputParam.template("latents"), | |
| ] | |
| # auto_docstring | |
| class Flux3AutoCoreDenoiseStep(AutoPipelineBlocks): | |
| """ | |
| Core denoise step that selects the FLUX.3 denoising workflow by the conditioning inputs. | |
| - `Flux3ImageConditionedCoreDenoiseStep` is used when `image_cond_latents` is provided (image-to-image). | |
| - `Flux3CoreDenoiseStep` is used otherwise (text-to-image). | |
| Components: | |
| transformer (`Flux3Transformer2DModel`) scheduler (`FlowMatchEulerDiscreteScheduler`) guider | |
| (`ClassifierFreeGuidance`) | |
| Inputs: | |
| num_images_per_prompt (`int`, *optional*, defaults to 1): | |
| The number of images to generate per prompt. | |
| prompt_embeds (`Tensor`): | |
| Pre-generated text embeddings. Can be generated from text_encoder step. | |
| negative_prompt_embeds (`Tensor`, *optional*): | |
| Pre-generated negative text embeddings. Can be generated from text_encoder step. | |
| image_cond_latents (`list`, *optional*): | |
| Normalized packed latents of the reference images, one tensor of shape (B, C, height_i//16, width_i//16) | |
| per image | |
| height (`int`, *optional*, defaults to None or 1024, depending on the workflow): | |
| The height in pixels of the generated image; defaults to the first reference image's aspect ratio at the | |
| default area | |
| width (`int`, *optional*, defaults to None or 1024, depending on the workflow): | |
| The width in pixels of the generated image; defaults to the first reference image's aspect ratio at the | |
| default area | |
| latents (`Tensor`): | |
| Pre-generated noisy latents for image generation. | |
| generator (`Generator`, *optional*): | |
| Torch generator for deterministic generation. | |
| num_inference_steps (`int`): | |
| The number of denoising steps. | |
| timesteps (`Tensor`): | |
| Timesteps for the denoising process. (overrides `num_inference_steps` when provided) | |
| sigmas (`list`, *optional*): | |
| Custom sigmas for the denoising process. (overrides `num_inference_steps` when provided) | |
| **denoiser_input_fields (`None`, *optional*): | |
| conditional model inputs for the denoiser: e.g. prompt_embeds, negative_prompt_embeds, etc. | |
| Outputs: | |
| latents (`Tensor`): | |
| Denoised latents. | |
| """ | |
| model_name = "flux3" | |
| block_classes = [Flux3ImageConditionedCoreDenoiseStep, Flux3CoreDenoiseStep] | |
| block_names = ["image_conditioned", "text2image"] | |
| block_trigger_inputs = ["image_cond_latents", None] | |
| def description(self): | |
| return ( | |
| "Core denoise step that selects the FLUX.3 denoising workflow by the conditioning inputs.\n" | |
| " - `Flux3ImageConditionedCoreDenoiseStep` is used when `image_cond_latents` is provided (image-to-image).\n" | |
| " - `Flux3CoreDenoiseStep` is used otherwise (text-to-image)." | |
| ) | |
| ### | |
| ### Auto blocks | |
| ### | |
| AUTO_BLOCKS = InsertableDict( | |
| [ | |
| ("text_encoder", Flux3TextEncoderStep()), | |
| ("vae_encoder", Flux3AutoVaeEncoderStep()), | |
| ("denoise", Flux3AutoCoreDenoiseStep()), | |
| ("decode", Flux3DecodeStep()), | |
| ] | |
| ) | |
| # auto_docstring | |
| class Flux3AutoBlocks(SequentialPipelineBlocks): | |
| """ | |
| Auto blocks that perform text-to-image and image-to-image generation using FLUX.3. | |
| Supported workflows: | |
| - `text2image`: requires `prompt` | |
| - `image_conditioned`: requires `prompt`, `image` | |
| Components: | |
| text_encoder (`Qwen3VLForConditionalGeneration`) tokenizer (`AutoProcessor`) image_processor | |
| (`Flux3ImageProcessor`) vae (`AutoencoderKLFlux2`) transformer (`Flux3Transformer2DModel`) scheduler | |
| (`FlowMatchEulerDiscreteScheduler`) guider (`ClassifierFreeGuidance`) | |
| Inputs: | |
| prompt (`str`): | |
| The prompt to guide the generation (one prompt per call) | |
| negative_prompt (`str`, *optional*): | |
| The prompt not to guide the generation | |
| max_sequence_length (`int`, *optional*, defaults to 8192): | |
| Maximum number of prompt tokens; longer prompts are truncated. | |
| text_encoder_out_layers (`tuple`, *optional*, defaults to (4, 8, 12, 16, 20, 24, 28, 32)): | |
| Indices of the text-encoder hidden layers whose states are stacked to form the text embeddings | |
| image (`Image | list`, *optional*): | |
| The reference image or images to condition the generation on; each keeps its own resolution | |
| (center-cropped to the 16-pixel grid) | |
| num_images_per_prompt (`int`, *optional*, defaults to 1): | |
| The number of images to generate per prompt. | |
| image_cond_latents (`list`, *optional*): | |
| Normalized packed latents of the reference images, one tensor of shape (B, C, height_i//16, width_i//16) | |
| per image | |
| height (`int`, *optional*, defaults to None or 1024, depending on the workflow): | |
| The height in pixels of the generated image; defaults to the first reference image's aspect ratio at the | |
| default area | |
| width (`int`, *optional*, defaults to None or 1024, depending on the workflow): | |
| The width in pixels of the generated image; defaults to the first reference image's aspect ratio at the | |
| default area | |
| latents (`Tensor`): | |
| Pre-generated noisy latents for image generation. | |
| generator (`Generator`, *optional*): | |
| Torch generator for deterministic generation. | |
| num_inference_steps (`int`): | |
| The number of denoising steps. | |
| timesteps (`Tensor`): | |
| Timesteps for the denoising process. (overrides `num_inference_steps` when provided) | |
| sigmas (`list`, *optional*): | |
| Custom sigmas for the denoising process. (overrides `num_inference_steps` when provided) | |
| **denoiser_input_fields (`None`, *optional*): | |
| conditional model inputs for the denoiser: e.g. prompt_embeds, negative_prompt_embeds, etc. | |
| output_type (`str`, *optional*, defaults to pil): | |
| Output format: 'pil', 'np', 'pt'. | |
| Outputs: | |
| images (`list`): | |
| Generated images. | |
| latents (`Tensor`): | |
| Denoised latents. | |
| """ | |
| model_name = "flux3" | |
| block_classes = AUTO_BLOCKS.values() | |
| block_names = AUTO_BLOCKS.keys() | |
| _workflow_map = { | |
| "text2image": {"prompt": True}, | |
| "image_conditioned": {"prompt": True, "image": True}, | |
| } | |
| def description(self): | |
| return "Auto blocks that perform text-to-image and image-to-image generation using FLUX.3." | |
| def outputs(self): | |
| return [ | |
| OutputParam.template("images"), | |
| OutputParam.template("latents"), | |
| ] | |