o ۷i!@sNdZddlmZddlmZddlZddlmZmZm Z GdddedZ dS) zI Base pipeline class for Diffusion models with shared CFG functionality. )ABCMeta)AnyN) get_cfg_group!get_classifier_free_guidance_rank'get_classifier_free_guidance_world_sizec@seZdZdZ  d!dededeeefdeeefdBded e dBd e j dBfd d Z d e j de j d e j fddZ d"d e j de j deded e j f ddZdeded e j fddZdeded efddZd e j de j de j d e j fddZd e j de j de j ded e j f dd ZdS)#CFGParallelMixinz Base Mixin class for Diffusion pipelines providing shared CFG methods. All pipelines should inherit from this class to reuse classifier-free guidance logic. TN do_true_cfgtrue_cfg_scalepositive_kwargsnegative_kwargs cfg_normalize output_slicereturncCs6|rtdk}|rQt}t} | dkr|jdi|} n|jdi|} |dur2| ddd|f} |j| dd} | dkrO| d} | d} || | ||} | SdS|jdi|}|jdi|}|dury|ddd|f}|ddd|f}|||||} | S|jdi|}|dur|ddd|f}|S)aF Predict noise with optional classifier-free guidance. Args: do_true_cfg: Whether to apply CFG true_cfg_scale: CFG scale factor positive_kwargs: Kwargs for positive/conditional prediction negative_kwargs: Kwargs for negative/unconditional prediction cfg_normalize: Whether to normalize CFG output (default: True) output_slice: If set, slice output to [:, :output_slice] for image editing Returns: Predicted noise tensor (only valid on rank 0 in CFG parallel mode) rNT)separate_tensors)rrr predict_noise all_gathercombine_cfg_noise)selfrr r r r r cfg_parallel_ready cfg_groupcfg_rank local_predgathered noise_predneg_noise_predpositive_noise_prednegative_noise_predpredrrb/home/ubuntu/vllm_env/lib/python3.10/site-packages/vllm_omni/diffusion/distributed/cfg_parallel.pypredict_noise_maybe_with_cfgs< z-CFGParallelMixin.predict_noise_maybe_with_cfgr comb_predcCs0tj|ddd}tj|ddd}|||}|S)z Normalize the combined noise prediction. Args: noise_pred: positive noise prediction comb_pred: combined noise prediction after CFG Returns: Normalized noise prediction tensor T)dimkeepdim)torchnorm)rrr" cond_norm noise_normrrr cfg_normalize_functioncs  z'CFGParallelMixin.cfg_normalize_functionFrcCs,||||}|r|||}|S|}|S)a Combine conditional and unconditional noise predictions with CFG. Args: noise_pred: Conditional noise prediction neg_noise_pred: Unconditional noise prediction true_cfg_scale: CFG scale factor cfg_normalize: Whether to normalize the combined prediction (default: False) Returns: Combined noise prediction tensor )r*)rrrr r r"rrr rss  z"CFGParallelMixin.combine_cfg_noiseargskwargscOs|j|i|dS)z Forward pass through transformer to predict noise. Subclasses should override this if they need custom behavior, but the default implementation calls self.transformer. r) transformerrr+r,rrr rszCFGParallelMixin.predict_noisecOstd)aU Diffusion loop with optional classifier-free guidance. Subclasses MUST implement this method to define the complete diffusion/denoising loop for their specific model. Typical implementation pattern: ```python def diffuse(self, latents, timesteps, prompt_embeds, negative_embeds, ...): for t in timesteps: # Prepare kwargs for positive and negative predictions positive_kwargs = {...} negative_kwargs = {...} # Predict noise with automatic CFG handling noise_pred = self.predict_noise_maybe_with_cfg( do_true_cfg=True, true_cfg_scale=self.guidance_scale, positive_kwargs=positive_kwargs, negative_kwargs=negative_kwargs, ) # Step scheduler with automatic CFG sync latents = self.scheduler_step_maybe_with_cfg( noise_pred, t, latents, do_true_cfg=True ) return latents ``` z!Subclasses must implement diffuse)NotImplementedErrorr.rrr diffuses#zCFGParallelMixin.diffusetlatentscCs|jj|||dddS)z Step the scheduler. Args: noise_pred: Predicted noise t: Current timestep latents: Current latents Returns: Updated latents after scheduler step F) return_dictr) schedulerstep)rrr1r2rrr scheduler_steps zCFGParallelMixin.scheduler_stepcCs`|otdk}|r't}t}|dkr||||}|}|j|dd|S||||}|S)a Step the scheduler with (maybe) automatic CFG parallel synchronization. In CFG parallel mode, only rank 0 computes the scheduler step, then broadcasts the result to other ranks. Args: noise_pred: Predicted noise (only valid on rank 0 in CFG parallel) t: Current timestep latents: Current latents do_true_cfg: Whether CFG is enabled Returns: Updated latents (synchronized across all CFG ranks) rr)src)rrrr6 contiguous broadcast)rrr1r2rrrrrrr scheduler_step_maybe_with_cfgsz.CFGParallelMixin.scheduler_step_maybe_with_cfg)TN)F)__name__ __module__ __qualname____doc__boolfloatdictstrrintr&Tensorr!r*rrr0r6r:rrrr rsh   G   "%r) metaclass) r>abcrtypingrr&.vllm_omni.diffusion.distributed.parallel_staterrrrrrrr s