pipeline_stable_diffusion.py

# Copyright 2024 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.
importinspect
fromtypingimportAny, Callable, Dict, List, Optional, Union

importtorch
frompackagingimportversion
fromtransformersimportCLIPImageProcessor, CLIPTextModel, CLIPTokenizer, CLIPVisionModelWithProjection

from ...callbacksimportMultiPipelineCallbacks, PipelineCallback
from ...configuration_utilsimportFrozenDict
from ...image_processorimportPipelineImageInput, VaeImageProcessor
from ...loadersimportFromSingleFileMixin, IPAdapterMixin, StableDiffusionLoraLoaderMixin, TextualInversionLoaderMixin
from ...modelsimportAutoencoderKL, ImageProjection, UNet2DConditionModel
from ...models.loraimportadjust_lora_scale_text_encoder
from ...schedulersimportKarrasDiffusionSchedulers
from ...utilsimport (
USE_PEFT_BACKEND,
deprecate,
is_torch_xla_available,
logging,
replace_example_docstring,
scale_lora_layers,
unscale_lora_layers,
)
from ...utils.torch_utilsimportrandn_tensor
from ..pipeline_utilsimportDiffusionPipeline, StableDiffusionMixin
from .pipeline_outputimportStableDiffusionPipelineOutput
from .safety_checkerimportStableDiffusionSafetyChecker


ifis_torch_xla_available():
importtorch_xla.core.xla_modelasxm

XLA_AVAILABLE=True
else:
XLA_AVAILABLE=False

logger=logging.get_logger(__name__) # pylint: disable=invalid-name

EXAMPLE_DOC_STRING="""
 Examples:
 ```py
 >>> import torch
 >>> from diffusers import StableDiffusionPipeline

 >>> pipe = StableDiffusionPipeline.from_pretrained(
 ... "stable-diffusion-v1-5/stable-diffusion-v1-5", torch_dtype=torch.float16
 ... )
 >>> pipe = pipe.to("cuda")

 >>> prompt = "a photo of an astronaut riding a horse on mars"
 >>> image = pipe(prompt).images[0]
 ```
"""


defrescale_noise_cfg(noise_cfg, noise_pred_text, guidance_rescale=0.0):
r"""
 Rescales `noise_cfg` tensor based on `guidance_rescale` to improve image quality and fix overexposure. Based on
 Section 3.4 from [Common Diffusion Noise Schedules and Sample Steps are
 Flawed](https://arxiv.org/pdf/2305.08891.pdf).

 Args:
 noise_cfg (`torch.Tensor`):
 The predicted noise tensor for the guided diffusion process.
 noise_pred_text (`torch.Tensor`):
 The predicted noise tensor for the text-guided diffusion process.
 guidance_rescale (`float`, *optional*, defaults to 0.0):
 A rescale factor applied to the noise predictions.

 Returns:
 noise_cfg (`torch.Tensor`): The rescaled noise prediction tensor.
 """
std_text=noise_pred_text.std(dim=list(range(1, noise_pred_text.ndim)), keepdim=True)
std_cfg=noise_cfg.std(dim=list(range(1, noise_cfg.ndim)), keepdim=True)
# rescale the results from guidance (fixes overexposure)
noise_pred_rescaled=noise_cfg* (std_text/std_cfg)
# mix with the original results from guidance by factor guidance_rescale to avoid "plain looking" images
noise_cfg=guidance_rescale*noise_pred_rescaled+ (1-guidance_rescale) *noise_cfg
returnnoise_cfg


defretrieve_timesteps(
scheduler,
num_inference_steps: Optional[int] =None,
device: Optional[Union[str, torch.device]] =None,
timesteps: Optional[List[int]] =None,
sigmas: Optional[List[float]] =None,
**kwargs,
):
r"""
 Calls the scheduler's `set_timesteps` method and retrieves timesteps from the scheduler after the call. Handles
 custom timesteps. Any kwargs will be supplied to `scheduler.set_timesteps`.

 Args:
 scheduler (`SchedulerMixin`):
 The scheduler to get timesteps from.
 num_inference_steps (`int`):
 The number of diffusion steps used when generating samples with a pre-trained model. If used, `timesteps`
 must be `None`.
 device (`str` or `torch.device`, *optional*):
 The device to which the timesteps should be moved to. If `None`, the timesteps are not moved.
 timesteps (`List[int]`, *optional*):
 Custom timesteps used to override the timestep spacing strategy of the scheduler. If `timesteps` is passed,
 `num_inference_steps` and `sigmas` must be `None`.
 sigmas (`List[float]`, *optional*):
 Custom sigmas used to override the timestep spacing strategy of the scheduler. If `sigmas` is passed,
 `num_inference_steps` and `timesteps` must be `None`.

 Returns:
 `Tuple[torch.Tensor, int]`: A tuple where the first element is the timestep schedule from the scheduler and the
 second element is the number of inference steps.
 """
iftimestepsisnotNoneandsigmasisnotNone:
raiseValueError("Only one of `timesteps` or `sigmas` can be passed. Please choose one to set custom values")
iftimestepsisnotNone:
accepts_timesteps="timesteps"inset(inspect.signature(scheduler.set_timesteps).parameters.keys())
ifnotaccepts_timesteps:
raiseValueError(
f"The current scheduler class {scheduler.__class__}'s `set_timesteps` does not support custom"
f" timestep schedules. Please check whether you are using the correct scheduler."
 )
scheduler.set_timesteps(timesteps=timesteps, device=device, **kwargs)
timesteps=scheduler.timesteps
num_inference_steps=len(timesteps)
elifsigmasisnotNone:
accept_sigmas="sigmas"inset(inspect.signature(scheduler.set_timesteps).parameters.keys())
ifnotaccept_sigmas:
raiseValueError(
f"The current scheduler class {scheduler.__class__}'s `set_timesteps` does not support custom"
f" sigmas schedules. Please check whether you are using the correct scheduler."
 )
scheduler.set_timesteps(sigmas=sigmas, device=device, **kwargs)
timesteps=scheduler.timesteps
num_inference_steps=len(timesteps)
else:
scheduler.set_timesteps(num_inference_steps, device=device, **kwargs)
timesteps=scheduler.timesteps
returntimesteps, num_inference_steps


classStableDiffusionPipeline(
DiffusionPipeline,
StableDiffusionMixin,
TextualInversionLoaderMixin,
StableDiffusionLoraLoaderMixin,
IPAdapterMixin,
FromSingleFileMixin,
):
"""
 Pipeline for text-to-image generation using Stable Diffusion.

 This model inherits from [`DiffusionPipeline`]. Check the superclass documentation for the generic methods
 implemented for all pipelines (downloading, saving, running on a particular device, etc.).

 The pipeline also inherits the following loading methods:
 - [`~loaders.TextualInversionLoaderMixin.load_textual_inversion`] for loading textual inversion embeddings
 - [`~loaders.StableDiffusionLoraLoaderMixin.load_lora_weights`] for loading LoRA weights
 - [`~loaders.StableDiffusionLoraLoaderMixin.save_lora_weights`] for saving LoRA weights
 - [`~loaders.FromSingleFileMixin.from_single_file`] for loading `.ckpt` files
 - [`~loaders.IPAdapterMixin.load_ip_adapter`] for loading IP Adapters

 Args:
 vae ([`AutoencoderKL`]):
 Variational Auto-Encoder (VAE) model to encode and decode images to and from latent representations.
 text_encoder ([`~transformers.CLIPTextModel`]):
 Frozen text-encoder ([clip-vit-large-patch14](https://huggingface.co/openai/clip-vit-large-patch14)).
 tokenizer ([`~transformers.CLIPTokenizer`]):
 A `CLIPTokenizer` to tokenize text.
 unet ([`UNet2DConditionModel`]):
 A `UNet2DConditionModel` to denoise the encoded image latents.
 scheduler ([`SchedulerMixin`]):
 A scheduler to be used in combination with `unet` to denoise the encoded image latents. Can be one of
 [`DDIMScheduler`], [`LMSDiscreteScheduler`], or [`PNDMScheduler`].
 safety_checker ([`StableDiffusionSafetyChecker`]):
 Classification module that estimates whether generated images could be considered offensive or harmful.
 Please refer to the [model card](https://huggingface.co/stable-diffusion-v1-5/stable-diffusion-v1-5) for
 more details about a model's potential harms.
 feature_extractor ([`~transformers.CLIPImageProcessor`]):
 A `CLIPImageProcessor` to extract features from generated images; used as inputs to the `safety_checker`.
 """

model_cpu_offload_seq="text_encoder->image_encoder->unet->vae"
_optional_components= ["safety_checker", "feature_extractor", "image_encoder"]
_exclude_from_cpu_offload= ["safety_checker"]
_callback_tensor_inputs= ["latents", "prompt_embeds", "negative_prompt_embeds"]

def__init__(
self,
vae: AutoencoderKL,
text_encoder: CLIPTextModel,
tokenizer: CLIPTokenizer,
unet: UNet2DConditionModel,
scheduler: KarrasDiffusionSchedulers,
safety_checker: StableDiffusionSafetyChecker,
feature_extractor: CLIPImageProcessor,
image_encoder: CLIPVisionModelWithProjection=None,
requires_safety_checker: bool=True,
 ):
super().__init__()

ifschedulerisnotNoneandgetattr(scheduler.config, "steps_offset", 1) !=1:
deprecation_message= (
f"The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"
f" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure "
"to update the config accordingly as leaving `steps_offset` might led to incorrect results"
" in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,"
" it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`"
" file"
 )
deprecate("steps_offset!=1", "1.0.0", deprecation_message, standard_warn=False)
new_config=dict(scheduler.config)
new_config["steps_offset"] =1
scheduler._internal_dict=FrozenDict(new_config)

ifschedulerisnotNoneandgetattr(scheduler.config, "clip_sample", False) isTrue:
deprecation_message= (
f"The configuration file of this scheduler: {scheduler} has not set the configuration `clip_sample`."
" `clip_sample` should be set to False in the configuration file. Please make sure to update the"
" config accordingly as not setting `clip_sample` in the config might lead to incorrect results in"
" future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it would be very"
" nice if you could open a Pull request for the `scheduler/scheduler_config.json` file"
 )
deprecate("clip_sample not set", "1.0.0", deprecation_message, standard_warn=False)
new_config=dict(scheduler.config)
new_config["clip_sample"] =False
scheduler._internal_dict=FrozenDict(new_config)

ifsafety_checkerisNoneandrequires_safety_checker:
logger.warning(
f"You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"
" that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered"
" results in services or applications open to the public. Both the diffusers team and Hugging Face"
" strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling"
" it only for use-cases that involve analyzing network behavior or auditing its results. For more"
" information, please have a look at https://github.com/huggingface/diffusers/pull/254 ."
 )

ifsafety_checkerisnotNoneandfeature_extractorisNone:
raiseValueError(
"Make sure to define a feature extractor when loading {self.__class__} if you want to use the safety"
" checker. If you do not want to use the safety checker, you can pass `'safety_checker=None'` instead."
 )

is_unet_version_less_0_9_0= (
unetisnotNone
andhasattr(unet.config, "_diffusers_version")
andversion.parse(version.parse(unet.config._diffusers_version).base_version) <version.parse("0.9.0.dev0")
 )
self._is_unet_config_sample_size_int=unetisnotNoneandisinstance(unet.config.sample_size, int)
is_unet_sample_size_less_64= (
unetisnotNone
andhasattr(unet.config, "sample_size")
andself._is_unet_config_sample_size_int
andunet.config.sample_size<64
 )
ifis_unet_version_less_0_9_0andis_unet_sample_size_less_64:
deprecation_message= (
"The configuration file of the unet has set the default `sample_size` to smaller than"
" 64 which seems highly unlikely. If your checkpoint is a fine-tuned version of any of the"
" following: \n- CompVis/stable-diffusion-v1-4 \n- CompVis/stable-diffusion-v1-3 \n-"
" CompVis/stable-diffusion-v1-2 \n- CompVis/stable-diffusion-v1-1 \n- stable-diffusion-v1-5/stable-diffusion-v1-5"
" \n- stable-diffusion-v1-5/stable-diffusion-inpainting \n you should change 'sample_size' to 64 in the"
" configuration file. Please make sure to update the config accordingly as leaving `sample_size=32`"
" in the config might lead to incorrect results in future versions. If you have downloaded this"
" checkpoint from the Hugging Face Hub, it would be very nice if you could open a Pull request for"
" the `unet/config.json` file"
 )
deprecate("sample_size<64", "1.0.0", deprecation_message, standard_warn=False)
new_config=dict(unet.config)
new_config["sample_size"] =64
unet._internal_dict=FrozenDict(new_config)

self.register_modules(
vae=vae,
text_encoder=text_encoder,
tokenizer=tokenizer,
unet=unet,
scheduler=scheduler,
safety_checker=safety_checker,
feature_extractor=feature_extractor,
image_encoder=image_encoder,
 )
self.vae_scale_factor=2** (len(self.vae.config.block_out_channels) -1) ifgetattr(self, "vae", None) else8
self.image_processor=VaeImageProcessor(vae_scale_factor=self.vae_scale_factor)
self.register_to_config(requires_safety_checker=requires_safety_checker)

def_encode_prompt(
self,
prompt,
device,
num_images_per_prompt,
do_classifier_free_guidance,
negative_prompt=None,
prompt_embeds: Optional[torch.Tensor] =None,
negative_prompt_embeds: Optional[torch.Tensor] =None,
lora_scale: Optional[float] =None,
**kwargs,
 ):
deprecation_message="`_encode_prompt()` is deprecated and it will be removed in a future version. Use `encode_prompt()` instead. Also, be aware that the output format changed from a concatenated tensor to a tuple."
deprecate("_encode_prompt()", "1.0.0", deprecation_message, standard_warn=False)

prompt_embeds_tuple=self.encode_prompt(
prompt=prompt,
device=device,
num_images_per_prompt=num_images_per_prompt,
do_classifier_free_guidance=do_classifier_free_guidance,
negative_prompt=negative_prompt,
prompt_embeds=prompt_embeds,
negative_prompt_embeds=negative_prompt_embeds,
lora_scale=lora_scale,
**kwargs,
 )

# concatenate for backwards comp
prompt_embeds=torch.cat([prompt_embeds_tuple[1], prompt_embeds_tuple[0]])

returnprompt_embeds

defencode_prompt(
self,
prompt,
device,
num_images_per_prompt,
do_classifier_free_guidance,
negative_prompt=None,
prompt_embeds: Optional[torch.Tensor] =None,
negative_prompt_embeds: Optional[torch.Tensor] =None,
lora_scale: Optional[float] =None,
clip_skip: Optional[int] =None,
 ):
r"""
 Encodes the prompt into text encoder hidden states.

 Args:
 prompt (`str` or `List[str]`, *optional*):
 prompt to be encoded
 device: (`torch.device`):
 torch device
 num_images_per_prompt (`int`):
 number of images that should be generated per prompt
 do_classifier_free_guidance (`bool`):
 whether to use classifier free guidance or not
 negative_prompt (`str` or `List[str]`, *optional*):
 The prompt or prompts not to guide the image generation. If not defined, one has to pass
 `negative_prompt_embeds` instead. Ignored when not using guidance (i.e., ignored if `guidance_scale` is
 less than `1`).
 prompt_embeds (`torch.Tensor`, *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.
 negative_prompt_embeds (`torch.Tensor`, *optional*):
 Pre-generated negative text embeddings. Can be used to easily tweak text inputs, *e.g.* prompt
 weighting. If not provided, negative_prompt_embeds will be generated from `negative_prompt` input
 argument.
 lora_scale (`float`, *optional*):
 A LoRA scale that will be applied to all LoRA layers of the text encoder if LoRA layers are loaded.
 clip_skip (`int`, *optional*):
 Number of layers to be skipped from CLIP while computing the prompt embeddings. A value of 1 means that
 the output of the pre-final layer will be used for computing the prompt embeddings.
 """
# set lora scale so that monkey patched LoRA
# function of text encoder can correctly access it
iflora_scaleisnotNoneandisinstance(self, StableDiffusionLoraLoaderMixin):
self._lora_scale=lora_scale

# dynamically adjust the LoRA scale
ifnotUSE_PEFT_BACKEND:
adjust_lora_scale_text_encoder(self.text_encoder, lora_scale)
else:
scale_lora_layers(self.text_encoder, lora_scale)

ifpromptisnotNoneandisinstance(prompt, str):
batch_size=1
elifpromptisnotNoneandisinstance(prompt, list):
batch_size=len(prompt)
else:
batch_size=prompt_embeds.shape[0]

ifprompt_embedsisNone:
# textual inversion: process multi-vector tokens if necessary
ifisinstance(self, TextualInversionLoaderMixin):
prompt=self.maybe_convert_prompt(prompt, self.tokenizer)

text_inputs=self.tokenizer(
prompt,
padding="max_length",
max_length=self.tokenizer.model_max_length,
truncation=True,
return_tensors="pt",
 )
text_input_ids=text_inputs.input_ids
untruncated_ids=self.tokenizer(prompt, padding="longest", return_tensors="pt").input_ids

ifuntruncated_ids.shape[-1] >=text_input_ids.shape[-1] andnottorch.equal(
text_input_ids, untruncated_ids
 ):
removed_text=self.tokenizer.batch_decode(
untruncated_ids[:, self.tokenizer.model_max_length-1 : -1]
 )
logger.warning(
"The following part of your input was truncated because CLIP can only handle sequences up to"
f" {self.tokenizer.model_max_length} tokens: {removed_text}"
 )

ifhasattr(self.text_encoder.config, "use_attention_mask") andself.text_encoder.config.use_attention_mask:
attention_mask=text_inputs.attention_mask.to(device)
else:
attention_mask=None

ifclip_skipisNone:
prompt_embeds=self.text_encoder(text_input_ids.to(device), attention_mask=attention_mask)
prompt_embeds=prompt_embeds[0]
else:
prompt_embeds=self.text_encoder(
text_input_ids.to(device), attention_mask=attention_mask, output_hidden_states=True
 )
# Access the `hidden_states` first, that contains a tuple of
# all the hidden states from the encoder layers. Then index into
# the tuple to access the hidden states from the desired layer.
prompt_embeds=prompt_embeds[-1][-(clip_skip+1)]
# We also need to apply the final LayerNorm here to not mess with the
# representations. The `last_hidden_states` that we typically use for
# obtaining the final prompt representations passes through the LayerNorm
# layer.
prompt_embeds=self.text_encoder.text_model.final_layer_norm(prompt_embeds)

ifself.text_encoderisnotNone:
prompt_embeds_dtype=self.text_encoder.dtype
elifself.unetisnotNone:
prompt_embeds_dtype=self.unet.dtype
else:
prompt_embeds_dtype=prompt_embeds.dtype

prompt_embeds=prompt_embeds.to(dtype=prompt_embeds_dtype, device=device)

bs_embed, seq_len, _=prompt_embeds.shape
# duplicate text embeddings for each generation per prompt, using mps friendly method
prompt_embeds=prompt_embeds.repeat(1, num_images_per_prompt, 1)
prompt_embeds=prompt_embeds.view(bs_embed*num_images_per_prompt, seq_len, -1)

# get unconditional embeddings for classifier free guidance
ifdo_classifier_free_guidanceandnegative_prompt_embedsisNone:
uncond_tokens: List[str]
ifnegative_promptisNone:
uncond_tokens= [""] *batch_size
elifpromptisnotNoneandtype(prompt) isnottype(negative_prompt):
raiseTypeError(
f"`negative_prompt` should be the same type to `prompt`, but got {type(negative_prompt)} !="
f" {type(prompt)}."
 )
elifisinstance(negative_prompt, str):
uncond_tokens= [negative_prompt]
elifbatch_size!=len(negative_prompt):
raiseValueError(
f"`negative_prompt`: {negative_prompt} has batch size {len(negative_prompt)}, but `prompt`:"
f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"
" the batch size of `prompt`."
 )
else:
uncond_tokens=negative_prompt

# textual inversion: process multi-vector tokens if necessary
ifisinstance(self, TextualInversionLoaderMixin):
uncond_tokens=self.maybe_convert_prompt(uncond_tokens, self.tokenizer)

max_length=prompt_embeds.shape[1]
uncond_input=self.tokenizer(
uncond_tokens,
padding="max_length",
max_length=max_length,
truncation=True,
return_tensors="pt",
 )

ifhasattr(self.text_encoder.config, "use_attention_mask") andself.text_encoder.config.use_attention_mask:
attention_mask=uncond_input.attention_mask.to(device)
else:
attention_mask=None

negative_prompt_embeds=self.text_encoder(
uncond_input.input_ids.to(device),
attention_mask=attention_mask,
 )
negative_prompt_embeds=negative_prompt_embeds[0]

ifdo_classifier_free_guidance:
# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
seq_len=negative_prompt_embeds.shape[1]

negative_prompt_embeds=negative_prompt_embeds.to(dtype=prompt_embeds_dtype, device=device)

negative_prompt_embeds=negative_prompt_embeds.repeat(1, num_images_per_prompt, 1)
negative_prompt_embeds=negative_prompt_embeds.view(batch_size*num_images_per_prompt, seq_len, -1)

ifself.text_encoderisnotNone:
ifisinstance(self, StableDiffusionLoraLoaderMixin) andUSE_PEFT_BACKEND:
# Retrieve the original scale by scaling back the LoRA layers
unscale_lora_layers(self.text_encoder, lora_scale)

returnprompt_embeds, negative_prompt_embeds

defencode_image(self, image, device, num_images_per_prompt, output_hidden_states=None):
dtype=next(self.image_encoder.parameters()).dtype

ifnotisinstance(image, torch.Tensor):
image=self.feature_extractor(image, return_tensors="pt").pixel_values

image=image.to(device=device, dtype=dtype)
ifoutput_hidden_states:
image_enc_hidden_states=self.image_encoder(image, output_hidden_states=True).hidden_states[-2]
image_enc_hidden_states=image_enc_hidden_states.repeat_interleave(num_images_per_prompt, dim=0)
uncond_image_enc_hidden_states=self.image_encoder(
torch.zeros_like(image), output_hidden_states=True
 ).hidden_states[-2]
uncond_image_enc_hidden_states=uncond_image_enc_hidden_states.repeat_interleave(
num_images_per_prompt, dim=0
 )
returnimage_enc_hidden_states, uncond_image_enc_hidden_states
else:
image_embeds=self.image_encoder(image).image_embeds
image_embeds=image_embeds.repeat_interleave(num_images_per_prompt, dim=0)
uncond_image_embeds=torch.zeros_like(image_embeds)

returnimage_embeds, uncond_image_embeds

defprepare_ip_adapter_image_embeds(
self, ip_adapter_image, ip_adapter_image_embeds, device, num_images_per_prompt, do_classifier_free_guidance
 ):
image_embeds= []
ifdo_classifier_free_guidance:
negative_image_embeds= []
ifip_adapter_image_embedsisNone:
ifnotisinstance(ip_adapter_image, list):
ip_adapter_image= [ip_adapter_image]

iflen(ip_adapter_image) !=len(self.unet.encoder_hid_proj.image_projection_layers):
raiseValueError(
f"`ip_adapter_image` must have same length as the number of IP Adapters. Got {len(ip_adapter_image)} images and {len(self.unet.encoder_hid_proj.image_projection_layers)} IP Adapters."
 )

forsingle_ip_adapter_image, image_proj_layerinzip(
ip_adapter_image, self.unet.encoder_hid_proj.image_projection_layers
 ):
output_hidden_state=notisinstance(image_proj_layer, ImageProjection)
single_image_embeds, single_negative_image_embeds=self.encode_image(
single_ip_adapter_image, device, 1, output_hidden_state
 )

image_embeds.append(single_image_embeds[None, :])
ifdo_classifier_free_guidance:
negative_image_embeds.append(single_negative_image_embeds[None, :])
else:
forsingle_image_embedsinip_adapter_image_embeds:
ifdo_classifier_free_guidance:
single_negative_image_embeds, single_image_embeds=single_image_embeds.chunk(2)
negative_image_embeds.append(single_negative_image_embeds)
image_embeds.append(single_image_embeds)

ip_adapter_image_embeds= []
fori, single_image_embedsinenumerate(image_embeds):
single_image_embeds=torch.cat([single_image_embeds] *num_images_per_prompt, dim=0)
ifdo_classifier_free_guidance:
single_negative_image_embeds=torch.cat([negative_image_embeds[i]] *num_images_per_prompt, dim=0)
single_image_embeds=torch.cat([single_negative_image_embeds, single_image_embeds], dim=0)

single_image_embeds=single_image_embeds.to(device=device)
ip_adapter_image_embeds.append(single_image_embeds)

returnip_adapter_image_embeds

defrun_safety_checker(self, image, device, dtype):
ifself.safety_checkerisNone:
has_nsfw_concept=None
else:
iftorch.is_tensor(image):
feature_extractor_input=self.image_processor.postprocess(image, output_type="pil")
else:
feature_extractor_input=self.image_processor.numpy_to_pil(image)
safety_checker_input=self.feature_extractor(feature_extractor_input, return_tensors="pt").to(device)
image, has_nsfw_concept=self.safety_checker(
images=image, clip_input=safety_checker_input.pixel_values.to(dtype)
 )
returnimage, has_nsfw_concept

defdecode_latents(self, latents):
deprecation_message="The decode_latents method is deprecated and will be removed in 1.0.0. Please use VaeImageProcessor.postprocess(...) instead"
deprecate("decode_latents", "1.0.0", deprecation_message, standard_warn=False)

latents=1/self.vae.config.scaling_factor*latents
image=self.vae.decode(latents, return_dict=False)[0]
image= (image/2+0.5).clamp(0, 1)
# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
image=image.cpu().permute(0, 2, 3, 1).float().numpy()
returnimage

defprepare_extra_step_kwargs(self, generator, eta):
# prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
# eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
# eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
# and should be between [0, 1]

accepts_eta="eta"inset(inspect.signature(self.scheduler.step).parameters.keys())
extra_step_kwargs= {}
ifaccepts_eta:
extra_step_kwargs["eta"] =eta

# check if the scheduler accepts generator
accepts_generator="generator"inset(inspect.signature(self.scheduler.step).parameters.keys())
ifaccepts_generator:
extra_step_kwargs["generator"] =generator
returnextra_step_kwargs

defcheck_inputs(
self,
prompt,
height,
width,
callback_steps,
negative_prompt=None,
prompt_embeds=None,
negative_prompt_embeds=None,
ip_adapter_image=None,
ip_adapter_image_embeds=None,
callback_on_step_end_tensor_inputs=None,
 ):
ifheight%8!=0orwidth%8!=0:
raiseValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}.")

ifcallback_stepsisnotNoneand (notisinstance(callback_steps, int) orcallback_steps<=0):
raiseValueError(
f"`callback_steps` has to be a positive integer but is {callback_steps} of type"
f" {type(callback_steps)}."
 )
ifcallback_on_step_end_tensor_inputsisnotNoneandnotall(
kinself._callback_tensor_inputsforkincallback_on_step_end_tensor_inputs
 ):
raiseValueError(
f"`callback_on_step_end_tensor_inputs` has to be in {self._callback_tensor_inputs}, but found {[kforkincallback_on_step_end_tensor_inputsifknotinself._callback_tensor_inputs]}"
 )

ifpromptisnotNoneandprompt_embedsisnotNone:
raiseValueError(
f"Cannot forward both `prompt`: {prompt} and `prompt_embeds`: {prompt_embeds}. Please make sure to"
" only forward one of the two."
 )
elifpromptisNoneandprompt_embedsisNone:
raiseValueError(
"Provide either `prompt` or `prompt_embeds`. Cannot leave both `prompt` and `prompt_embeds` undefined."
 )
elifpromptisnotNoneand (notisinstance(prompt, str) andnotisinstance(prompt, list)):
raiseValueError(f"`prompt` has to be of type `str` or `list` but is {type(prompt)}")

ifnegative_promptisnotNoneandnegative_prompt_embedsisnotNone:
raiseValueError(
f"Cannot forward both `negative_prompt`: {negative_prompt} and `negative_prompt_embeds`:"
f" {negative_prompt_embeds}. Please make sure to only forward one of the two."
 )

ifprompt_embedsisnotNoneandnegative_prompt_embedsisnotNone:
ifprompt_embeds.shape!=negative_prompt_embeds.shape:
raiseValueError(
"`prompt_embeds` and `negative_prompt_embeds` must have the same shape when passed directly, but"
f" got: `prompt_embeds` {prompt_embeds.shape} != `negative_prompt_embeds`"
f" {negative_prompt_embeds.shape}."
 )

ifip_adapter_imageisnotNoneandip_adapter_image_embedsisnotNone:
raiseValueError(
"Provide either `ip_adapter_image` or `ip_adapter_image_embeds`. Cannot leave both `ip_adapter_image` and `ip_adapter_image_embeds` defined."
 )

ifip_adapter_image_embedsisnotNone:
ifnotisinstance(ip_adapter_image_embeds, list):
raiseValueError(
f"`ip_adapter_image_embeds` has to be of type `list` but is {type(ip_adapter_image_embeds)}"
 )
elifip_adapter_image_embeds[0].ndimnotin [3, 4]:
raiseValueError(
f"`ip_adapter_image_embeds` has to be a list of 3D or 4D tensors but is {ip_adapter_image_embeds[0].ndim}D"
 )

defprepare_latents(self, batch_size, num_channels_latents, height, width, dtype, device, generator, latents=None):
shape= (
batch_size,
num_channels_latents,
int(height) //self.vae_scale_factor,
int(width) //self.vae_scale_factor,
 )
ifisinstance(generator, list) andlen(generator) !=batch_size:
raiseValueError(
f"You have passed a list of generators of length {len(generator)}, but requested an effective batch"
f" size of {batch_size}. Make sure the batch size matches the length of the generators."
 )

iflatentsisNone:
latents=randn_tensor(shape, generator=generator, device=device, dtype=dtype)
else:
latents=latents.to(device)

# scale the initial noise by the standard deviation required by the scheduler
latents=latents*self.scheduler.init_noise_sigma
returnlatents

# Copied from diffusers.pipelines.latent_consistency_models.pipeline_latent_consistency_text2img.LatentConsistencyModelPipeline.get_guidance_scale_embedding
defget_guidance_scale_embedding(
self, w: torch.Tensor, embedding_dim: int=512, dtype: torch.dtype=torch.float32
 ) ->torch.Tensor:
"""
 See https://github.com/google-research/vdm/blob/dc27b98a554f65cdc654b800da5aa1846545d41b/model_vdm.py#L298

 Args:
 w (`torch.Tensor`):
 Generate embedding vectors with a specified guidance scale to subsequently enrich timestep embeddings.
 embedding_dim (`int`, *optional*, defaults to 512):
 Dimension of the embeddings to generate.
 dtype (`torch.dtype`, *optional*, defaults to `torch.float32`):
 Data type of the generated embeddings.

 Returns:
 `torch.Tensor`: Embedding vectors with shape `(len(w), embedding_dim)`.
 """
assertlen(w.shape) ==1
w=w*1000.0

half_dim=embedding_dim//2
emb=torch.log(torch.tensor(10000.0)) / (half_dim-1)
emb=torch.exp(torch.arange(half_dim, dtype=dtype) *-emb)
emb=w.to(dtype)[:, None] *emb[None, :]
emb=torch.cat([torch.sin(emb), torch.cos(emb)], dim=1)
ifembedding_dim%2==1: # zero pad
emb=torch.nn.functional.pad(emb, (0, 1))
assertemb.shape== (w.shape[0], embedding_dim)
returnemb

@property
defguidance_scale(self):
returnself._guidance_scale

@property
defguidance_rescale(self):
returnself._guidance_rescale

@property
defclip_skip(self):
returnself._clip_skip

# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
# corresponds to doing no classifier free guidance.
@property
defdo_classifier_free_guidance(self):
returnself._guidance_scale>1andself.unet.config.time_cond_proj_dimisNone

@property
defcross_attention_kwargs(self):
returnself._cross_attention_kwargs

@property
defnum_timesteps(self):
returnself._num_timesteps

@property
definterrupt(self):
returnself._interrupt

@torch.no_grad()
@replace_example_docstring(EXAMPLE_DOC_STRING)
def__call__(
self,
prompt: Union[str, List[str]] =None,
height: Optional[int] =None,
width: Optional[int] =None,
num_inference_steps: int=50,
timesteps: List[int] =None,
sigmas: List[float] =None,
guidance_scale: float=7.5,
negative_prompt: Optional[Union[str, List[str]]] =None,
num_images_per_prompt: Optional[int] =1,
eta: float=0.0,
generator: Optional[Union[torch.Generator, List[torch.Generator]]] =None,
latents: Optional[torch.Tensor] =None,
prompt_embeds: Optional[torch.Tensor] =None,
negative_prompt_embeds: Optional[torch.Tensor] =None,
ip_adapter_image: Optional[PipelineImageInput] =None,
ip_adapter_image_embeds: Optional[List[torch.Tensor]] =None,
output_type: Optional[str] ="pil",
return_dict: bool=True,
cross_attention_kwargs: Optional[Dict[str, Any]] =None,
guidance_rescale: float=0.0,
clip_skip: Optional[int] =None,
callback_on_step_end: Optional[
Union[Callable[[int, int, Dict], None], PipelineCallback, MultiPipelineCallbacks]
 ] =None,
callback_on_step_end_tensor_inputs: List[str] = ["latents"],
**kwargs,
 ):
r"""
 The call function to the pipeline for generation.

 Args:
 prompt (`str` or `List[str]`, *optional*):
 The prompt or prompts to guide image generation. If not defined, you need to pass `prompt_embeds`.
 height (`int`, *optional*, defaults to `self.unet.config.sample_size * self.vae_scale_factor`):
 The height in pixels of the generated image.
 width (`int`, *optional*, defaults to `self.unet.config.sample_size * self.vae_scale_factor`):
 The width in pixels of the generated image.
 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.
 sigmas (`List[float]`, *optional*):
 Custom sigmas to use for the denoising process with schedulers which support a `sigmas` argument in
 their `set_timesteps` method. If not defined, the default behavior when `num_inference_steps` is passed
 will be used.
 guidance_scale (`float`, *optional*, defaults to 7.5):
 A higher guidance scale value encourages the model to generate images closely linked to the text
 `prompt` at the expense of lower image quality. Guidance scale is enabled when `guidance_scale > 1`.
 negative_prompt (`str` or `List[str]`, *optional*):
 The prompt or prompts to guide what to not include in image generation. If not defined, you need to
 pass `negative_prompt_embeds` instead. Ignored when not using guidance (`guidance_scale < 1`).
 num_images_per_prompt (`int`, *optional*, defaults to 1):
 The number of images to generate per prompt.
 eta (`float`, *optional*, defaults to 0.0):
 Corresponds to parameter eta (η) from the [DDIM](https://arxiv.org/abs/2010.02502) paper. Only applies
 to the [`~schedulers.DDIMScheduler`], and is ignored in other schedulers.
 generator (`torch.Generator` or `List[torch.Generator]`, *optional*):
 A [`torch.Generator`](https://pytorch.org/docs/stable/generated/torch.Generator.html) to make
 generation deterministic.
 latents (`torch.Tensor`, *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 is generated by sampling using the supplied random `generator`.
 prompt_embeds (`torch.Tensor`, *optional*):
 Pre-generated text embeddings. Can be used to easily tweak text inputs (prompt weighting). If not
 provided, text embeddings are generated from the `prompt` input argument.
 negative_prompt_embeds (`torch.Tensor`, *optional*):
 Pre-generated negative text embeddings. Can be used to easily tweak text inputs (prompt weighting). If
 not provided, `negative_prompt_embeds` are generated from the `negative_prompt` input argument.
 ip_adapter_image: (`PipelineImageInput`, *optional*): Optional image input to work with IP Adapters.
 ip_adapter_image_embeds (`List[torch.Tensor]`, *optional*):
 Pre-generated image embeddings for IP-Adapter. It should be a list of length same as number of
 IP-adapters. Each element should be a tensor of shape `(batch_size, num_images, emb_dim)`. It should
 contain the negative image embedding if `do_classifier_free_guidance` is set to `True`. If not
 provided, embeddings are computed from the `ip_adapter_image` input argument.
 output_type (`str`, *optional*, defaults to `"pil"`):
 The output format of the generated image. Choose between `PIL.Image` or `np.array`.
 return_dict (`bool`, *optional*, defaults to `True`):
 Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
 plain tuple.
 cross_attention_kwargs (`dict`, *optional*):
 A kwargs dictionary that if specified is passed along to the [`AttentionProcessor`] as defined in
 [`self.processor`](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/attention_processor.py).
 guidance_rescale (`float`, *optional*, defaults to 0.0):
 Guidance rescale factor from [Common Diffusion Noise Schedules and Sample Steps are
 Flawed](https://arxiv.org/pdf/2305.08891.pdf). Guidance rescale factor should fix overexposure when
 using zero terminal SNR.
 clip_skip (`int`, *optional*):
 Number of layers to be skipped from CLIP while computing the prompt embeddings. A value of 1 means that
 the output of the pre-final layer will be used for computing the prompt embeddings.
 callback_on_step_end (`Callable`, `PipelineCallback`, `MultiPipelineCallbacks`, *optional*):
 A function or a subclass of `PipelineCallback` or `MultiPipelineCallbacks` that is called at the end of
 each denoising step during the inference. 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.

 Examples:

 Returns:
 [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`:
 If `return_dict` is `True`, [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] is returned,
 otherwise a `tuple` is returned where the first element is a list with the generated images and the
 second element is a list of `bool`s indicating whether the corresponding generated image contains
 "not-safe-for-work" (nsfw) content.
 """

callback=kwargs.pop("callback", None)
callback_steps=kwargs.pop("callback_steps", None)

ifcallbackisnotNone:
deprecate(
"callback",
"1.0.0",
"Passing `callback` as an input argument to `__call__` is deprecated, consider using `callback_on_step_end`",
 )
ifcallback_stepsisnotNone:
deprecate(
"callback_steps",
"1.0.0",
"Passing `callback_steps` as an input argument to `__call__` is deprecated, consider using `callback_on_step_end`",
 )

ifisinstance(callback_on_step_end, (PipelineCallback, MultiPipelineCallbacks)):
callback_on_step_end_tensor_inputs=callback_on_step_end.tensor_inputs

# 0. Default height and width to unet
ifnotheightornotwidth:
height= (
self.unet.config.sample_size
ifself._is_unet_config_sample_size_int
elseself.unet.config.sample_size[0]
 )
width= (
self.unet.config.sample_size
ifself._is_unet_config_sample_size_int
elseself.unet.config.sample_size[1]
 )
height, width=height*self.vae_scale_factor, width*self.vae_scale_factor
# to deal with lora scaling and other possible forward hooks

# 1. Check inputs. Raise error if not correct
self.check_inputs(
prompt,
height,
width,
callback_steps,
negative_prompt,
prompt_embeds,
negative_prompt_embeds,
ip_adapter_image,
ip_adapter_image_embeds,
callback_on_step_end_tensor_inputs,
 )

self._guidance_scale=guidance_scale
self._guidance_rescale=guidance_rescale
self._clip_skip=clip_skip
self._cross_attention_kwargs=cross_attention_kwargs
self._interrupt=False

# 2. Define call parameters
ifpromptisnotNoneandisinstance(prompt, str):
batch_size=1
elifpromptisnotNoneandisinstance(prompt, list):
batch_size=len(prompt)
else:
batch_size=prompt_embeds.shape[0]

device=self._execution_device

# 3. Encode input prompt
lora_scale= (
self.cross_attention_kwargs.get("scale", None) ifself.cross_attention_kwargsisnotNoneelseNone
 )

prompt_embeds, negative_prompt_embeds=self.encode_prompt(
prompt,
device,
num_images_per_prompt,
self.do_classifier_free_guidance,
negative_prompt,
prompt_embeds=prompt_embeds,
negative_prompt_embeds=negative_prompt_embeds,
lora_scale=lora_scale,
clip_skip=self.clip_skip,
 )

# For classifier free guidance, we need to do two forward passes.
# Here we concatenate the unconditional and text embeddings into a single batch
# to avoid doing two forward passes
ifself.do_classifier_free_guidance:
prompt_embeds=torch.cat([negative_prompt_embeds, prompt_embeds])

ifip_adapter_imageisnotNoneorip_adapter_image_embedsisnotNone:
image_embeds=self.prepare_ip_adapter_image_embeds(
ip_adapter_image,
ip_adapter_image_embeds,
device,
batch_size*num_images_per_prompt,
self.do_classifier_free_guidance,
 )

# 4. Prepare timesteps
timesteps, num_inference_steps=retrieve_timesteps(
self.scheduler, num_inference_steps, device, timesteps, sigmas
 )

# 5. Prepare latent variables
num_channels_latents=self.unet.config.in_channels
latents=self.prepare_latents(
batch_size*num_images_per_prompt,
num_channels_latents,