|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import inspect |
|
from typing import Any, Callable, Dict, List, Optional, Union |
|
|
|
import numpy as np |
|
import torch |
|
from transformers import CLIPTextModel, CLIPTokenizer, T5EncoderModel, T5TokenizerFast |
|
|
|
from diffusers.image_processor import VaeImageProcessor, PipelineImageInput |
|
from diffusers.models.autoencoders import AutoencoderKL |
|
from diffusers.models.transformers import FluxTransformer2DModel |
|
from diffusers.schedulers import FlowMatchEulerDiscreteScheduler |
|
from diffusers.utils import ( |
|
is_torch_xla_available, |
|
logging, |
|
replace_example_docstring, |
|
) |
|
from diffusers.utils.torch_utils import randn_tensor |
|
from diffusers.pipelines.flux.pipeline_output import FluxPipelineOutput |
|
|
|
from .pipeline_flux import FluxPipeline, calculate_shift, retrieve_timesteps |
|
|
|
if is_torch_xla_available(): |
|
import torch_xla.core.xla_model as xm |
|
|
|
XLA_AVAILABLE = True |
|
else: |
|
XLA_AVAILABLE = False |
|
|
|
|
|
logger = logging.get_logger(__name__) |
|
|
|
EXAMPLE_DOC_STRING = """ |
|
Examples: |
|
```py |
|
>>> import torch |
|
>>> from diffusers import FluxPipeline |
|
|
|
>>> pipe = FluxPipeline.from_pretrained("black-forest-labs/FLUX.1-schnell", torch_dtype=torch.bfloat16) |
|
>>> pipe.to("cuda") |
|
>>> prompt = "A cat holding a sign that says hello world" |
|
>>> # Depending on the variant being used, the pipeline call will slightly vary. |
|
>>> # Refer to the pipeline documentation for more details. |
|
>>> image = pipe(prompt, num_inference_steps=4, guidance_scale=0.0).images[0] |
|
>>> image.save("flux.png") |
|
``` |
|
""" |
|
|
|
|
|
def retrieve_latents( |
|
encoder_output: torch.Tensor, generator: Optional[torch.Generator] = None, sample_mode: str = "sample" |
|
): |
|
if hasattr(encoder_output, "latent_dist") and sample_mode == "sample": |
|
return encoder_output.latent_dist.sample(generator) |
|
elif hasattr(encoder_output, "latent_dist") and sample_mode == "argmax": |
|
return encoder_output.latent_dist.mode() |
|
elif hasattr(encoder_output, "latents"): |
|
return encoder_output.latents |
|
else: |
|
raise AttributeError("Could not access latents of provided encoder_output") |
|
|
|
|
|
class FluxFillPipeline(FluxPipeline): |
|
r""" |
|
The Flux pipeline for text-to-image generation. |
|
|
|
Reference: https://blackforestlabs.ai/announcing-black-forest-labs/ |
|
|
|
Args: |
|
transformer ([`FluxTransformer2DModel`]): |
|
Conditional Transformer (MMDiT) architecture to denoise the encoded image latents. |
|
scheduler ([`FlowMatchEulerDiscreteScheduler`]): |
|
A scheduler to be used in combination with `transformer` to denoise the encoded image latents. |
|
vae ([`AutoencoderKL`]): |
|
Variational Auto-Encoder (VAE) Model to encode and decode images to and from latent representations. |
|
text_encoder ([`CLIPTextModel`]): |
|
[CLIP](https://huggingface.co/docs/transformers/model_doc/clip#transformers.CLIPTextModel), specifically |
|
the [clip-vit-large-patch14](https://huggingface.co/openai/clip-vit-large-patch14) variant. |
|
text_encoder_2 ([`T5EncoderModel`]): |
|
[T5](https://huggingface.co/docs/transformers/en/model_doc/t5#transformers.T5EncoderModel), specifically |
|
the [google/t5-v1_1-xxl](https://huggingface.co/google/t5-v1_1-xxl) variant. |
|
tokenizer (`CLIPTokenizer`): |
|
Tokenizer of class |
|
[CLIPTokenizer](https://huggingface.co/docs/transformers/en/model_doc/clip#transformers.CLIPTokenizer). |
|
tokenizer_2 (`T5TokenizerFast`): |
|
Second Tokenizer of class |
|
[T5TokenizerFast](https://huggingface.co/docs/transformers/en/model_doc/t5#transformers.T5TokenizerFast). |
|
""" |
|
|
|
model_cpu_offload_seq = "text_encoder->text_encoder_2->transformer->vae" |
|
_optional_components = [] |
|
_callback_tensor_inputs = ["latents", "prompt_embeds"] |
|
|
|
def __init__( |
|
self, |
|
scheduler: FlowMatchEulerDiscreteScheduler, |
|
vae: AutoencoderKL, |
|
text_encoder: CLIPTextModel, |
|
tokenizer: CLIPTokenizer, |
|
text_encoder_2: T5EncoderModel, |
|
tokenizer_2: T5TokenizerFast, |
|
transformer: FluxTransformer2DModel, |
|
): |
|
super().__init__( |
|
scheduler=scheduler, |
|
vae=vae, |
|
text_encoder=text_encoder, |
|
tokenizer=tokenizer, |
|
text_encoder_2=text_encoder_2, |
|
tokenizer_2=tokenizer_2, |
|
transformer=transformer |
|
) |
|
self.mask_processor = VaeImageProcessor( |
|
vae_scale_factor=self.vae_scale_factor, |
|
vae_latent_channels=self.vae.config.latent_channels, |
|
do_normalize=False, |
|
do_binarize=True, |
|
do_convert_grayscale=True, |
|
) |
|
|
|
def prepare_mask_latents( |
|
self, |
|
mask, |
|
masked_image, |
|
batch_size, |
|
num_channels_latents, |
|
num_images_per_prompt, |
|
height, |
|
width, |
|
dtype, |
|
device, |
|
generator, |
|
): |
|
|
|
|
|
|
|
height = 2 * (int(height) // self.vae_scale_factor) |
|
width = 2 * (int(width) // self.vae_scale_factor) |
|
|
|
|
|
if masked_image.shape[1] == num_channels_latents: |
|
masked_image_latents = masked_image |
|
else: |
|
masked_image_latents = retrieve_latents(self.vae.encode(masked_image), generator=generator) |
|
|
|
masked_image_latents = (masked_image_latents - self.vae.config.shift_factor) * self.vae.config.scaling_factor |
|
masked_image_latents = masked_image_latents.to(device=device, dtype=dtype) |
|
|
|
|
|
batch_size = batch_size * num_images_per_prompt |
|
if mask.shape[0] < batch_size: |
|
if not batch_size % mask.shape[0] == 0: |
|
raise ValueError( |
|
"The passed mask and the required batch size don't match. Masks are supposed to be duplicated to" |
|
f" a total batch size of {batch_size}, but {mask.shape[0]} masks were passed. Make sure the number" |
|
" of masks that you pass is divisible by the total requested batch size." |
|
) |
|
mask = mask.repeat(batch_size // mask.shape[0], 1, 1, 1) |
|
if masked_image_latents.shape[0] < batch_size: |
|
if not batch_size % masked_image_latents.shape[0] == 0: |
|
raise ValueError( |
|
"The passed images and the required batch size don't match. Images are supposed to be duplicated" |
|
f" to a total batch size of {batch_size}, but {masked_image_latents.shape[0]} images were passed." |
|
" Make sure the number of images that you pass is divisible by the total requested batch size." |
|
) |
|
masked_image_latents = masked_image_latents.repeat(batch_size // masked_image_latents.shape[0], 1, 1, 1) |
|
|
|
|
|
|
|
masked_image_latents = self._pack_latents( |
|
masked_image_latents, |
|
batch_size, |
|
num_channels_latents, |
|
height, |
|
width, |
|
) |
|
|
|
|
|
mask = mask[:, 0, :, :] |
|
mask = mask.view( |
|
batch_size, height, self.vae_scale_factor // 2, width, self.vae_scale_factor // 2 |
|
) |
|
mask = mask.permute(0, 2, 4, 1, 3) |
|
mask = mask.reshape( |
|
batch_size, (self.vae_scale_factor // 2) * (self.vae_scale_factor // 2), height, width |
|
) |
|
|
|
|
|
|
|
mask = self._pack_latents( |
|
mask, |
|
batch_size, |
|
(self.vae_scale_factor // 2) * (self.vae_scale_factor // 2), |
|
height, |
|
width, |
|
) |
|
mask = mask.to(device=device, dtype=dtype) |
|
|
|
return mask, masked_image_latents |
|
|
|
|
|
def get_timesteps(self, num_inference_steps, strength, device): |
|
|
|
init_timestep = min(num_inference_steps * strength, num_inference_steps) |
|
|
|
t_start = int(max(num_inference_steps - init_timestep, 0)) |
|
timesteps = self.scheduler.timesteps[t_start * self.scheduler.order :] |
|
if hasattr(self.scheduler, "set_begin_index"): |
|
self.scheduler.set_begin_index(t_start * self.scheduler.order) |
|
|
|
return timesteps, num_inference_steps - t_start |
|
|
|
def get_latents_with_image(self, image, latent_timestep, batch_size, num_channels_latents, height, width, generator, device, dtype): |
|
image = image.to(device=device, dtype=dtype) |
|
image_latents = self.vae.encode(image).latent_dist.sample(generator) |
|
image_latents = (image_latents - self.vae.config.shift_factor) * self.vae.config.scaling_factor |
|
batch_size, num_channels_latents, height, width = image_latents.size() |
|
noise = randn_tensor(image_latents.size(), generator=generator, device=device, dtype=dtype) |
|
latents = self.scheduler.scale_noise(image_latents, latent_timestep, noise) |
|
latents = self._pack_latents(latents, batch_size, num_channels_latents, height, width) |
|
|
|
return latents |
|
|
|
@torch.no_grad() |
|
@replace_example_docstring(EXAMPLE_DOC_STRING) |
|
def __call__( |
|
self, |
|
prompt: Union[str, List[str]] = None, |
|
prompt_2: Optional[Union[str, List[str]]] = None, |
|
img_cond: torch.FloatTensor = None, |
|
mask: torch.FloatTensor = None, |
|
height: Optional[int] = None, |
|
width: Optional[int] = None, |
|
strength: float = 1.0, |
|
image: PipelineImageInput = None, |
|
num_inference_steps: int = 28, |
|
timesteps: List[int] = None, |
|
guidance_scale: float = 3.5, |
|
num_images_per_prompt: Optional[int] = 1, |
|
generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None, |
|
latents: Optional[torch.FloatTensor] = None, |
|
prompt_embeds: Optional[torch.FloatTensor] = None, |
|
pooled_prompt_embeds: Optional[torch.FloatTensor] = None, |
|
output_type: Optional[str] = "pil", |
|
return_dict: bool = True, |
|
joint_attention_kwargs: Optional[Dict[str, Any]] = None, |
|
callback_on_step_end: Optional[Callable[[int, int, Dict], None]] = None, |
|
callback_on_step_end_tensor_inputs: List[str] = ["latents"], |
|
max_sequence_length: int = 512, |
|
): |
|
r""" |
|
Function invoked when calling the pipeline for generation. |
|
|
|
Args: |
|
prompt (`str` or `List[str]`, *optional*): |
|
The prompt or prompts to guide the image generation. If not defined, one has to pass `prompt_embeds`. |
|
instead. |
|
prompt_2 (`str` or `List[str]`, *optional*): |
|
The prompt or prompts to be sent to `tokenizer_2` and `text_encoder_2`. If not defined, `prompt` is |
|
will be used instead |
|
height (`int`, *optional*, defaults to self.unet.config.sample_size * self.vae_scale_factor): |
|
The height in pixels of the generated image. This is set to 1024 by default for the best results. |
|
width (`int`, *optional*, defaults to self.unet.config.sample_size * self.vae_scale_factor): |
|
The width in pixels of the generated image. This is set to 1024 by default for the best results. |
|
num_inference_steps (`int`, *optional*, defaults to 50): |
|
The number of denoising steps. More denoising steps usually lead to a higher quality image at the |
|
expense of slower inference. |
|
timesteps (`List[int]`, *optional*): |
|
Custom timesteps to use for the denoising process with schedulers which support a `timesteps` argument |
|
in their `set_timesteps` method. If not defined, the default behavior when `num_inference_steps` is |
|
passed will be used. Must be in descending order. |
|
guidance_scale (`float`, *optional*, defaults to 7.0): |
|
Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598). |
|
`guidance_scale` is defined as `w` of equation 2. of [Imagen |
|
Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale > |
|
1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`, |
|
usually at the expense of lower image quality. |
|
num_images_per_prompt (`int`, *optional*, defaults to 1): |
|
The number of images to generate per prompt. |
|
generator (`torch.Generator` or `List[torch.Generator]`, *optional*): |
|
One or a list of [torch generator(s)](https://pytorch.org/docs/stable/generated/torch.Generator.html) |
|
to make generation deterministic. |
|
latents (`torch.FloatTensor`, *optional*): |
|
Pre-generated noisy latents, sampled from a Gaussian distribution, to be used as inputs for image |
|
generation. Can be used to tweak the same generation with different prompts. If not provided, a latents |
|
tensor will ge generated by sampling using the supplied random `generator`. |
|
prompt_embeds (`torch.FloatTensor`, *optional*): |
|
Pre-generated text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt weighting. If not |
|
provided, text embeddings will be generated from `prompt` input argument. |
|
pooled_prompt_embeds (`torch.FloatTensor`, *optional*): |
|
Pre-generated pooled text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt weighting. |
|
If not provided, pooled text embeddings will be generated from `prompt` input argument. |
|
output_type (`str`, *optional*, defaults to `"pil"`): |
|
The output format of the generate image. Choose between |
|
[PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`. |
|
return_dict (`bool`, *optional*, defaults to `True`): |
|
Whether or not to return a [`~pipelines.flux.FluxPipelineOutput`] instead of a plain tuple. |
|
joint_attention_kwargs (`dict`, *optional*): |
|
A kwargs dictionary that if specified is passed along to the `AttentionProcessor` as defined under |
|
`self.processor` in |
|
[diffusers.models.attention_processor](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/attention_processor.py). |
|
callback_on_step_end (`Callable`, *optional*): |
|
A function that calls at the end of each denoising steps during the inference. The function is called |
|
with the following arguments: `callback_on_step_end(self: DiffusionPipeline, step: int, timestep: int, |
|
callback_kwargs: Dict)`. `callback_kwargs` will include a list of all tensors as specified by |
|
`callback_on_step_end_tensor_inputs`. |
|
callback_on_step_end_tensor_inputs (`List`, *optional*): |
|
The list of tensor inputs for the `callback_on_step_end` function. The tensors specified in the list |
|
will be passed as `callback_kwargs` argument. You will only be able to include variables listed in the |
|
`._callback_tensor_inputs` attribute of your pipeline class. |
|
max_sequence_length (`int` defaults to 512): Maximum sequence length to use with the `prompt`. |
|
|
|
Examples: |
|
|
|
Returns: |
|
[`~pipelines.flux.FluxPipelineOutput`] or `tuple`: [`~pipelines.flux.FluxPipelineOutput`] if `return_dict` |
|
is True, otherwise a `tuple`. When returning a tuple, the first element is a list with the generated |
|
images. |
|
""" |
|
|
|
height = height or self.default_sample_size * self.vae_scale_factor |
|
width = width or self.default_sample_size * self.vae_scale_factor |
|
|
|
|
|
self.check_inputs( |
|
prompt, |
|
prompt_2, |
|
height, |
|
width, |
|
prompt_embeds=prompt_embeds, |
|
pooled_prompt_embeds=pooled_prompt_embeds, |
|
callback_on_step_end_tensor_inputs=callback_on_step_end_tensor_inputs, |
|
max_sequence_length=max_sequence_length, |
|
) |
|
|
|
self._guidance_scale = guidance_scale |
|
self._joint_attention_kwargs = joint_attention_kwargs |
|
self._interrupt = False |
|
|
|
|
|
if prompt is not None and isinstance(prompt, str): |
|
batch_size = 1 |
|
elif prompt is not None and isinstance(prompt, list): |
|
batch_size = len(prompt) |
|
else: |
|
batch_size = prompt_embeds.shape[0] |
|
|
|
device = self._execution_device |
|
|
|
lora_scale = ( |
|
self.joint_attention_kwargs.get("scale", None) if self.joint_attention_kwargs is not None else None |
|
) |
|
( |
|
prompt_embeds, |
|
pooled_prompt_embeds, |
|
text_ids, |
|
) = self.encode_prompt( |
|
prompt=prompt, |
|
prompt_2=prompt_2, |
|
prompt_embeds=prompt_embeds, |
|
pooled_prompt_embeds=pooled_prompt_embeds, |
|
device=device, |
|
num_images_per_prompt=num_images_per_prompt, |
|
max_sequence_length=max_sequence_length, |
|
lora_scale=lora_scale, |
|
) |
|
|
|
|
|
num_channels_latents = self.vae.config.latent_channels |
|
latents, latent_image_ids = self.prepare_latents( |
|
batch_size * num_images_per_prompt, |
|
num_channels_latents, |
|
height, |
|
width, |
|
prompt_embeds.dtype, |
|
device, |
|
generator, |
|
latents, |
|
) |
|
|
|
|
|
img_cond = self.image_processor.preprocess(img_cond, height=height, width=width) |
|
mask = self.mask_processor.preprocess(mask, height=height, width=width) |
|
masked_image = img_cond * (1 - mask) |
|
masked_image = masked_image.to(device=device, dtype=prompt_embeds.dtype) |
|
|
|
height, width = masked_image.shape[-2:] |
|
mask, masked_image_latents = self.prepare_mask_latents( |
|
mask, |
|
masked_image, |
|
batch_size, |
|
num_channels_latents, |
|
num_images_per_prompt, |
|
height, |
|
width, |
|
prompt_embeds.dtype, |
|
device, |
|
generator, |
|
) |
|
masked_image_latents = torch.cat((masked_image_latents, mask), dim=-1) |
|
|
|
|
|
sigmas = np.linspace(1.0, 1 / num_inference_steps, num_inference_steps) |
|
image_seq_len = latents.shape[1] |
|
mu = calculate_shift( |
|
image_seq_len, |
|
self.scheduler.config.base_image_seq_len, |
|
self.scheduler.config.max_image_seq_len, |
|
self.scheduler.config.base_shift, |
|
self.scheduler.config.max_shift, |
|
) |
|
timesteps, num_inference_steps = retrieve_timesteps( |
|
self.scheduler, |
|
num_inference_steps, |
|
device, |
|
timesteps, |
|
sigmas, |
|
mu=mu, |
|
) |
|
|
|
if strength != 1.0 and image is not None: |
|
timesteps, num_inference_steps = self.get_timesteps(num_inference_steps, strength, device) |
|
latent_timestep = timesteps[:1].repeat(batch_size * num_images_per_prompt) |
|
latents = self.get_latents_with_image(image, latent_timestep, batch_size, num_channels_latents, height, width, generator, device, latents.dtype) |
|
|
|
num_warmup_steps = max(len(timesteps) - num_inference_steps * self.scheduler.order, 0) |
|
self._num_timesteps = len(timesteps) |
|
|
|
|
|
if self.transformer.config.guidance_embeds: |
|
guidance = torch.full([1], guidance_scale, device=device, dtype=torch.float32) |
|
guidance = guidance.expand(latents.shape[0]) |
|
else: |
|
guidance = None |
|
|
|
|
|
with self.progress_bar(total=num_inference_steps) as progress_bar: |
|
for i, t in enumerate(timesteps): |
|
if self.interrupt: |
|
continue |
|
|
|
|
|
timestep = t.expand(latents.shape[0]).to(latents.dtype) |
|
noise_pred = self.transformer( |
|
hidden_states=torch.cat((latents, masked_image_latents), dim=-1), |
|
timestep=timestep / 1000, |
|
guidance=guidance, |
|
pooled_projections=pooled_prompt_embeds, |
|
encoder_hidden_states=prompt_embeds, |
|
txt_ids=text_ids, |
|
img_ids=latent_image_ids, |
|
joint_attention_kwargs=self.joint_attention_kwargs, |
|
return_dict=False, |
|
)[0] |
|
|
|
|
|
latents_dtype = latents.dtype |
|
latents = self.scheduler.step(noise_pred, t, latents, return_dict=False)[0] |
|
|
|
if latents.dtype != latents_dtype: |
|
if torch.backends.mps.is_available(): |
|
|
|
latents = latents.to(latents_dtype) |
|
|
|
if callback_on_step_end is not None: |
|
callback_kwargs = {} |
|
for k in callback_on_step_end_tensor_inputs: |
|
callback_kwargs[k] = locals()[k] |
|
callback_outputs = callback_on_step_end(self, i, t, callback_kwargs) |
|
|
|
latents = callback_outputs.pop("latents", latents) |
|
prompt_embeds = callback_outputs.pop("prompt_embeds", prompt_embeds) |
|
|
|
|
|
if i == len(timesteps) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0): |
|
progress_bar.update() |
|
|
|
if XLA_AVAILABLE: |
|
xm.mark_step() |
|
|
|
if output_type == "latent": |
|
image = latents |
|
|
|
else: |
|
latents = self._unpack_latents(latents, height, width, self.vae_scale_factor) |
|
latents = (latents / self.vae.config.scaling_factor) + self.vae.config.shift_factor |
|
image = self.vae.decode(latents, return_dict=False)[0] |
|
image = self.image_processor.postprocess(image, output_type=output_type) |
|
|
|
|
|
self.maybe_free_model_hooks() |
|
|
|
if not return_dict: |
|
return (image,) |
|
|
|
return FluxPipelineOutput(images=image) |
|
|
|
|