# -*- coding: utf-8 -*- # @Time : 2024/5/31 # @Author : White Jiang from diffusers.schedulers import ( DDIMScheduler, DPMSolverMultistepScheduler, EulerAncestralDiscreteScheduler, EulerDiscreteScheduler, LMSDiscreteScheduler, PNDMScheduler, ) from diffusers.utils import is_accelerate_available from diffusers.pipelines.controlnet.pipeline_controlnet import * import os import sys from safetensors import safe_open BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(BASE_DIR) from adapter.resampler import ProjPlusModel from adapter.attention_processor import RefSAttnProcessor2_0, RefLoraSAttnProcessor2_0, IPAttnProcessor2_0, LoRAIPAttnProcessor2_0 class PipIpaControlNet(StableDiffusionControlNetPipeline): _optional_components = [] def __init__( self, vae, reference_unet, unet, tokenizer, text_encoder, controlnet, image_encoder, ImgProj, ip_ckpt, scheduler: Union[ DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler, EulerDiscreteScheduler, EulerAncestralDiscreteScheduler, DPMSolverMultistepScheduler, ], safety_checker: StableDiffusionSafetyChecker, feature_extractor: CLIPImageProcessor, ): super().__init__(vae, text_encoder, tokenizer, unet, controlnet, scheduler, safety_checker, feature_extractor) self.register_modules( vae=vae, reference_unet=reference_unet, unet=unet, controlnet=controlnet, scheduler=scheduler, tokenizer=tokenizer, text_encoder=text_encoder, image_encoder=image_encoder, ImgProj=ImgProj, safety_checker=safety_checker, feature_extractor=feature_extractor ) self.vae_scale_factor = 2 ** (len(self.vae.config.block_out_channels) - 1) self.clip_image_processor = CLIPImageProcessor() self.image_processor = VaeImageProcessor(vae_scale_factor=self.vae_scale_factor) self.ref_image_processor = VaeImageProcessor( vae_scale_factor=self.vae_scale_factor, do_convert_rgb=True, do_normalize=False, ) self.cond_image_processor = VaeImageProcessor( vae_scale_factor=self.vae_scale_factor, do_convert_rgb=True, do_normalize=False, ) self.ip_ckpt = ip_ckpt self.num_tokens = 4 # image proj model self.image_proj_model = self.init_proj() self.load_ip_adapter() def init_proj(self): image_proj_model = ProjPlusModel( cross_attention_dim=self.unet.config.cross_attention_dim, id_embeddings_dim=512, clip_embeddings_dim=self.image_encoder.config.hidden_size, num_tokens=self.num_tokens, ).to(self.unet.device, dtype=torch.float16) return image_proj_model def load_ip_adapter(self): if os.path.splitext(self.ip_ckpt)[-1] == ".safetensors": state_dict = {"image_proj": {}, "ip_adapter": {}} with safe_open(self.ip_ckpt, framework="pt", device="cpu") as f: for key in f.keys(): if key.startswith("image_proj."): state_dict["image_proj"][key.replace("image_proj.", "")] = f.get_tensor(key) elif key.startswith("ip_adapter."): state_dict["ip_adapter"][key.replace("ip_adapter.", "")] = f.get_tensor(key) else: state_dict = torch.load(self.ip_ckpt, map_location="cpu") self.image_proj_model.load_state_dict(state_dict["image_proj"]) ip_layers = torch.nn.ModuleList(self.unet.attn_processors.values()) ip_layers.load_state_dict(state_dict["ip_adapter"], strict=False) @property def cross_attention_kwargs(self): return self._cross_attention_kwargs def enable_vae_slicing(self): self.vae.enable_slicing() def disable_vae_slicing(self): self.vae.disable_slicing() def enable_sequential_cpu_offload(self, gpu_id=0): if is_accelerate_available(): from accelerate import cpu_offload else: raise ImportError("Please install accelerate via `pip install accelerate`") device = torch.device(f"cuda:{gpu_id}") for cpu_offloaded_model in [self.unet, self.text_encoder, self.vae]: if cpu_offloaded_model is not None: cpu_offload(cpu_offloaded_model, device) @property def _execution_device(self): if self.device != torch.device("meta") or not hasattr(self.unet, "_hf_hook"): return self.device for module in self.unet.modules(): if ( hasattr(module, "_hf_hook") and hasattr(module._hf_hook, "execution_device") and module._hf_hook.execution_device is not None ): return torch.device(module._hf_hook.execution_device) return self.device def prepare_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" in set( inspect.signature(self.scheduler.step).parameters.keys() ) extra_step_kwargs = {} if accepts_eta: extra_step_kwargs["eta"] = eta # check if the scheduler accepts generator accepts_generator = "generator" in set( inspect.signature(self.scheduler.step).parameters.keys() ) if accepts_generator: extra_step_kwargs["generator"] = generator return extra_step_kwargs # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline.encode_prompt def encode_prompt( self, prompt, device, num_images_per_prompt, do_classifier_free_guidance, negative_prompt=None, prompt_embeds: Optional[torch.FloatTensor] = None, negative_prompt_embeds: Optional[torch.FloatTensor] = None, lora_scale: Optional[float] = None, clip_skip: Optional[int] = None, ): if lora_scale is not None and isinstance(self, LoraLoaderMixin): self._lora_scale = lora_scale # dynamically adjust the LoRA scale if not USE_PEFT_BACKEND: adjust_lora_scale_text_encoder(self.text_encoder, lora_scale) else: scale_lora_layers(self.text_encoder, lora_scale) 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] if prompt_embeds is None: # textual inversion: procecss multi-vector tokens if necessary if isinstance(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 if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.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}" ) if hasattr(self.text_encoder.config, "use_attention_mask") and self.text_encoder.config.use_attention_mask: attention_mask = text_inputs.attention_mask.to(device) else: attention_mask = None if clip_skip is None: 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) if self.text_encoder is not None: prompt_embeds_dtype = self.text_encoder.dtype elif self.unet is not None: 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 if do_classifier_free_guidance and negative_prompt_embeds is None: uncond_tokens: List[str] if negative_prompt is None: uncond_tokens = [""] * batch_size elif prompt is not None and type(prompt) is not type(negative_prompt): raise TypeError( f"`negative_prompt` should be the same type to `prompt`, but got {type(negative_prompt)} !=" f" {type(prompt)}." ) elif isinstance(negative_prompt, str): uncond_tokens = [negative_prompt] elif batch_size != len(negative_prompt): raise ValueError( 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: procecss multi-vector tokens if necessary if isinstance(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", ) if hasattr(self.text_encoder.config, "use_attention_mask") and self.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] if do_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) if isinstance(self, LoraLoaderMixin) and USE_PEFT_BACKEND: # Retrieve the original scale by scaling back the LoRA layers unscale_lora_layers(self.text_encoder, lora_scale) return prompt_embeds, negative_prompt_embeds def prepare_latents( self, batch_size, num_channels_latents, width, height, dtype, device, generator, latents=None, ): shape = ( batch_size, num_channels_latents, height // self.vae_scale_factor, width // self.vae_scale_factor, ) if isinstance(generator, list) and len(generator) != batch_size: raise ValueError( 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." ) if latents is None: 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 return latents def prepare_condition( self, cond_image, width, height, device, dtype, do_classififer_free_guidance=False, ): image = self.cond_image_processor.preprocess( cond_image, height=height, width=width ).to(dtype=torch.float32) image = image.to(device=device, dtype=dtype) if do_classififer_free_guidance: image = torch.cat([image] * 2) return image def get_image_embeds(self, clip_image=None, faceid_embeds=None): with torch.no_grad(): clip_image_embeds = self.image_encoder(clip_image.to(self.device, dtype=torch.float16), output_hidden_states=True).hidden_states[-2] uncond_clip_image_embeds = self.image_encoder( torch.zeros_like(clip_image).to(self.device, dtype=torch.float16), output_hidden_states=True ).hidden_states[-2] faceid_embeds = faceid_embeds.to(self.device, dtype=torch.float16) image_prompt_embeds = self.image_proj_model(faceid_embeds, clip_image_embeds) uncond_image_prompt_embeds = self.image_proj_model(torch.zeros_like(faceid_embeds),uncond_clip_image_embeds) return image_prompt_embeds, uncond_image_prompt_embeds def set_scale(self, scale, lora_scale): for attn_processor in self.unet.attn_processors.values(): if isinstance(attn_processor, RefLoraSAttnProcessor2_0): attn_processor.scale = scale attn_processor.lora_scale = lora_scale # elif isinstance(attn_processor, RefCAttnProcessor2_0): # attn_processor.scale = scale def set_ipa_scale(self, ipa_scale, lora_scale): for attn_processor in self.unet.attn_processors.values(): if isinstance(attn_processor, LoRAIPAttnProcessor2_0): attn_processor.scale = ipa_scale attn_processor.lora_scale = lora_scale elif isinstance(attn_processor, IPAttnProcessor2_0): attn_processor.scale = ipa_scale attn_processor.lora_scale = lora_scale @torch.no_grad() def __call__( self, prompt, null_prompt, negative_prompt, ref_image, width, height, num_inference_steps, guidance_scale, pose_image=None, ref_clip_image=None, face_clip_image=None, faceid_embeds=None, num_images_per_prompt=1, image_scale=1.0, ipa_scale=0.0, s_lora_scale=0.0, c_lora_scale=0.0, num_samples=1, eta: float = 0.0, generator: Optional[Union[torch.Generator, List[torch.Generator]]] = None, output_type: Optional[str] = "pil", return_dict: bool = True, clip_skip: Optional[int] = None, callback: Optional[Callable[[int, int, torch.FloatTensor], None]] = None, callback_steps: Optional[int] = 1, prompt_embeds: Optional[torch.FloatTensor] = None, negative_prompt_embeds: Optional[torch.FloatTensor] = None, cross_attention_kwargs: Optional[Dict[str, Any]] = None, controlnet_conditioning_scale: Union[float, List[float]] = 1.0, guess_mode: bool = False, control_guidance_start: Union[float, List[float]] = 0.0, control_guidance_end: Union[float, List[float]] = 1.0, **kwargs, ): if face_clip_image is None: self.set_scale(image_scale, lora_scale=0.0) self.set_ipa_scale(ipa_scale=0.0, lora_scale=0.0) else: self.set_scale(image_scale, lora_scale=s_lora_scale) self.set_ipa_scale(ipa_scale, lora_scale=c_lora_scale) # controlnet controlnet = self.controlnet._orig_mod if is_compiled_module(self.controlnet) else self.controlnet # align format for control guidance if not isinstance(control_guidance_start, list) and isinstance(control_guidance_end, list): control_guidance_start = len(control_guidance_end) * [control_guidance_start] elif not isinstance(control_guidance_end, list) and isinstance(control_guidance_start, list): control_guidance_end = len(control_guidance_start) * [control_guidance_end] elif not isinstance(control_guidance_start, list) and not isinstance(control_guidance_end, list): mult = len(controlnet.nets) if isinstance(controlnet, MultiControlNetModel) else 1 control_guidance_start, control_guidance_end = ( mult * [control_guidance_start], mult * [control_guidance_end], ) if isinstance(controlnet, MultiControlNetModel) and isinstance(controlnet_conditioning_scale, float): controlnet_conditioning_scale = [controlnet_conditioning_scale] * len(controlnet.nets) global_pool_conditions = ( controlnet.config.global_pool_conditions if isinstance(controlnet, ControlNetModel) else controlnet.nets[0].config.global_pool_conditions ) guess_mode = guess_mode or global_pool_conditions # Default height and width to unet height = height or self.unet.config.sample_size * self.vae_scale_factor width = width or self.unet.config.sample_size * self.vae_scale_factor device = self._execution_device self._cross_attention_kwargs = cross_attention_kwargs self._clip_skip = clip_skip do_classifier_free_guidance = guidance_scale > 1.0 # Prepare timesteps self.scheduler.set_timesteps(num_inference_steps, device=device) timesteps = self.scheduler.timesteps batch_size = 1 if pose_image is not None: # Prepare control image if isinstance(controlnet, ControlNetModel): image = self.prepare_image( image=pose_image, width=width, height=height, batch_size=batch_size * num_images_per_prompt, num_images_per_prompt=num_images_per_prompt, device=device, dtype=controlnet.dtype, do_classifier_free_guidance=do_classifier_free_guidance, guess_mode=guess_mode, ) if do_classifier_free_guidance and not guess_mode: image = image.chunk(2)[0] height, width = image.shape[-2:] else: assert False # print(image.shape) # 3. Encode input prompt text_encoder_lora_scale = ( self.cross_attention_kwargs.get("scale", None) if self.cross_attention_kwargs is not None else None ) prompt_embeds, negative_prompt_embeds = self.encode_prompt( prompt, device, num_images_per_prompt, do_classifier_free_guidance, negative_prompt, prompt_embeds=prompt_embeds, negative_prompt_embeds=negative_prompt_embeds, lora_scale=text_encoder_lora_scale, clip_skip=self.clip_skip, ) if face_clip_image is not None: # for face image condition image_prompt_embeds, uncond_image_prompt_embeds = self.get_image_embeds(face_clip_image, faceid_embeds) bs_embed, seq_len, _ = image_prompt_embeds.shape image_prompt_embeds = image_prompt_embeds.repeat(1, num_samples, 1) image_prompt_embeds = image_prompt_embeds.view(bs_embed * num_samples, seq_len, -1) uncond_image_prompt_embeds = uncond_image_prompt_embeds.repeat(1, num_samples, 1) uncond_image_prompt_embeds = uncond_image_prompt_embeds.view(bs_embed * num_samples, seq_len, -1) if ref_clip_image is not None: with torch.no_grad(): image_embeds = self.image_encoder(ref_clip_image.to(device, dtype=prompt_embeds.dtype), output_hidden_states=True).hidden_states[-2] image_null_embeds = \ self.image_encoder(torch.zeros_like(ref_clip_image).to(device, dtype=prompt_embeds.dtype), output_hidden_states=True).hidden_states[-2] cloth_proj_embed = self.ImgProj(image_embeds) cloth_null_embeds = self.ImgProj(image_null_embeds) # cloth_null_embeds = self.ImgProj(torch.zeros_like(image_embeds)) else: null_prompt_embeds, _ = self.encode_prompt( null_prompt, device, num_images_per_prompt, do_classifier_free_guidance, negative_prompt, prompt_embeds=prompt_embeds, negative_prompt_embeds=negative_prompt_embeds, lora_scale=text_encoder_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 if do_classifier_free_guidance: prompt_embeds_control = torch.cat([negative_prompt_embeds, prompt_embeds]) if ref_clip_image is not None: null_prompt_embeds = torch.cat([cloth_null_embeds, cloth_proj_embed]) else: null_prompt_embeds = torch.cat([negative_prompt_embeds, null_prompt_embeds]) if face_clip_image is not None: prompt_embeds = torch.cat([prompt_embeds, image_prompt_embeds], dim=1) negative_prompt_embeds = torch.cat([negative_prompt_embeds, uncond_image_prompt_embeds], dim=1) else: prompt_embeds = prompt_embeds negative_prompt_embeds = negative_prompt_embeds num_channels_latents = self.unet.in_channels latents = self.prepare_latents( batch_size * num_images_per_prompt, num_channels_latents, width, height, prompt_embeds.dtype, device, generator, ) # Prepare extra step kwargs. extra_step_kwargs = self.prepare_extra_step_kwargs(generator, eta) # Prepare ref image latents ref_image_tensor = ref_image.to( dtype=self.vae.dtype, device=self.vae.device ) ref_image_latents = self.vae.encode(ref_image_tensor).latent_dist.mean ref_image_latents = ref_image_latents * 0.18215 # (b, 4, h, w) if pose_image is not None: # Create tensor stating which controlnets to keep controlnet_keep = [] for i in range(len(timesteps)): keeps = [ 1.0 - float(i / len(timesteps) < s or (i + 1) / len(timesteps) > e) for s, e in zip(control_guidance_start, control_guidance_end) ] controlnet_keep.append(keeps[0] if isinstance(controlnet, ControlNetModel) else keeps) # denoising loop num_warmup_steps = len(timesteps) - num_inference_steps * self.scheduler.order with self.progress_bar(total=num_inference_steps) as progress_bar: for i, t in enumerate(timesteps): # 1. Forward reference image if i == 0: _ = self.reference_unet( ref_image_latents.repeat( (2 if do_classifier_free_guidance else 1), 1, 1, 1 ), torch.zeros_like(t), encoder_hidden_states=null_prompt_embeds, return_dict=False, ) # get cache tensors sa_hidden_states = {} for name in self.reference_unet.attn_processors.keys(): sa_hidden_states[name] = self.reference_unet.attn_processors[name].cache["hidden_states"][ 1].unsqueeze(0) # sa_hidden_states[name][0, :, :] = 0 # 3.1 expand the latents if we are doing classifier free guidance latent_model_input = ( torch.cat([latents] * 2) if do_classifier_free_guidance else latents ) latent_model_input = self.scheduler.scale_model_input( latent_model_input, t ) # Optionally get Guidance Scale Embedding timestep_cond = None if self.unet.config.time_cond_proj_dim is not None: guidance_scale_tensor = torch.tensor(self.guidance_scale - 1).repeat( batch_size * num_images_per_prompt) timestep_cond = self.get_guidance_scale_embedding( guidance_scale_tensor, embedding_dim=self.unet.config.time_cond_proj_dim ).to(device=device, dtype=latents.dtype) # for control if pose_image is not None: # controlnet(s) inference if guess_mode and self.do_classifier_free_guidance: # Infer ControlNet only for the conditional batch. control_model_input = latents control_model_input = self.scheduler.scale_model_input(control_model_input, t) controlnet_prompt_embeds = prompt_embeds_control.chunk(2)[1] # controlnet_prompt_embeds = prompt_embeds else: control_model_input = latent_model_input controlnet_prompt_embeds = prompt_embeds_control if isinstance(controlnet_keep[i], list): cond_scale = [c * s for c, s in zip(controlnet_conditioning_scale, controlnet_keep[i])] else: controlnet_cond_scale = controlnet_conditioning_scale if isinstance(controlnet_cond_scale, list): controlnet_cond_scale = controlnet_cond_scale[0] cond_scale = controlnet_cond_scale * controlnet_keep[i] down_block_res_samples, mid_block_res_sample = self.controlnet( control_model_input, t, encoder_hidden_states=controlnet_prompt_embeds, controlnet_cond=image, conditioning_scale=cond_scale, guess_mode=guess_mode, return_dict=False, ) # if do_classifier_free_guidance: down_block_res_samples_con = [] down_block_res_samples_uncon = [] for down_block in down_block_res_samples: down_block_res_samples_con.append(down_block[1]) down_block_res_samples_uncon.append(down_block[0]) # for prompt_embeds ref + text noise_pred = self.unet( latent_model_input[0].unsqueeze(0), t, encoder_hidden_states=prompt_embeds, cross_attention_kwargs={ "sa_hidden_states": sa_hidden_states, }, timestep_cond=timestep_cond, down_block_additional_residuals=down_block_res_samples_con, mid_block_additional_residual=mid_block_res_sample[1], added_cond_kwargs=None, return_dict=False, )[0] # for negative_prompt_embeds non text unc_noise_pred = self.unet( latent_model_input[1].unsqueeze(0), t, encoder_hidden_states=negative_prompt_embeds, timestep_cond=timestep_cond, down_block_additional_residuals=down_block_res_samples_uncon, mid_block_additional_residual=mid_block_res_sample[0], added_cond_kwargs=None, return_dict=False, )[0] # for no control else: noise_pred = self.unet( latent_model_input[1].unsqueeze(0), t, encoder_hidden_states=prompt_embeds, cross_attention_kwargs={ "sa_hidden_states": sa_hidden_states, }, timestep_cond=timestep_cond, added_cond_kwargs=None, return_dict=False, )[0] # for negative_prompt_embeds non text unc_noise_pred = self.unet( latent_model_input[0].unsqueeze(0), t, encoder_hidden_states=negative_prompt_embeds, timestep_cond=timestep_cond, added_cond_kwargs=None, return_dict=False, )[0] # perform guidance if do_classifier_free_guidance: # noise_pred_uncond, noise_pred_text = noise_pred.chunk(2) noise_pred_uncond, noise_pred_text = unc_noise_pred, noise_pred noise_pred = noise_pred_uncond + guidance_scale * ( noise_pred_text - noise_pred_uncond ) # compute the previous noisy sample x_t -> x_t-1 latents = self.scheduler.step( noise_pred, t, latents, **extra_step_kwargs, return_dict=False )[0] # call the callback, if provided if i == len(timesteps) - 1 or ( (i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0 ): progress_bar.update() if callback is not None and i % callback_steps == 0: step_idx = i // getattr(self.scheduler, "order", 1) callback(step_idx, t, latents) # Post-processing image = self.vae.decode(latents / self.vae.config.scaling_factor, return_dict=False, generator=generator)[0] do_denormalize = [True] * image.shape[0] image = self.image_processor.postprocess(image, output_type=output_type, do_denormalize=do_denormalize) return StableDiffusionPipelineOutput(images=image, nsfw_content_detected=None)