深入 Diffusers 模塊化管線之 LoopSequentialPipelineBlocks構(gòu)建可迭代去噪循環(huán)的循環(huán)包裝器【免費(fèi)下載鏈接】diffusers Diffusers: State-of-the-art diffusion models for image, video, and audio generation in PyTorch.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/di/diffusersLoopSequentialPipelineBlocks是 Diffusers 模塊化管線Modular Pipelines體系中的核心多塊類型之一它把一組 [~modular_pipelines.ModularPipelineBlocks] 組織成循環(huán)結(jié)構(gòu)讓數(shù)據(jù)通過(guò)intermediate_inputs/intermediate_outputs在每次迭代中循環(huán)流動(dòng)是構(gòu)建默認(rèn)就是迭代式去噪循環(huán)這類管線的標(biāo)準(zhǔn)手段。本指南將帶你從零創(chuàng)建一個(gè)循環(huán)包裝器與循環(huán)塊并用from_blocks_dict組裝出可運(yùn)行的LoopSequentialPipelineBlocks最后結(jié)合倉(cāng)庫(kù)中 Wan 視頻模型的真實(shí)去噪循環(huán)源碼理解循環(huán)結(jié)構(gòu)在生成管線中的實(shí)際落地方式。從多塊類型說(shuō)起順序 vs 循環(huán)在模塊化管線的世界觀里管道是由一個(gè)個(gè) [~modular_pipelines.ModularPipelineBlocks] 拼裝而成的而多塊類型multi-block types負(fù)責(zé)定義這些塊之間的編排方式SequentialPipelineBlocks將多個(gè)塊按順序串接數(shù)據(jù)線性地從上一個(gè)塊流向下一個(gè)塊LoopSequentialPipelineBlocks將多個(gè)塊以循環(huán)方式組合數(shù)據(jù)在塊之間循環(huán)流動(dòng)每個(gè)塊被迭代執(zhí)行多次此外還有ConditionalPipelineBlocks/AutoPipelineBlocks這類根據(jù)輸入條件選擇執(zhí)行分支的多塊類型。LoopSequentialPipelineBlocks之所以重要是因?yàn)閿U(kuò)散模型的去噪過(guò)程本身就是一次迭代同一組預(yù)測(cè)噪聲 → 更新 latent的操作要在num_inference_steps個(gè)時(shí)間步上反復(fù)執(zhí)行。如果只用順序塊你就不得不把去噪循環(huán)手寫進(jìn)某個(gè)單一塊內(nèi)部而循環(huán)塊則把循環(huán)體拆成多個(gè)可獨(dú)立替換的子塊循環(huán)邏輯與每輪具體做什么被徹底解耦。本文配套的模塊化管線總覽與快速上手可以幫你先建立整體認(rèn)知本文聚焦LoopSequentialPipelineBlocks這一個(gè)點(diǎn)。循環(huán)包裝器定義循環(huán)結(jié)構(gòu)與迭代邏輯[~modular_pipelines.LoopSequentialPipelineBlocks] 也被稱為循環(huán)包裝器loop wrapper因?yàn)樗?fù)責(zé)定義三件事循環(huán)結(jié)構(gòu)、迭代變量和配置。你不需要像寫普通塊那樣覆寫inputs/intermediate_outputs屬性而是覆寫一套以loop_前綴命名的等價(jià)屬性。循環(huán)包裝器內(nèi)需要定義的變量如下表所示變量作用對(duì)應(yīng)普通塊的屬性loop_inputs用戶提供的值例如迭代步數(shù)num_steps[~modular_pipelines.ModularPipelineBlocks.inputs]loop_intermediate_inputs來(lái)自 [~modular_pipelines.PipelineState] 的中間變量供循環(huán)體消費(fèi)[~modular_pipelines.ModularPipelineBlocks.intermediate_inputs]loop_intermediate_outputs由循環(huán)內(nèi)塊創(chuàng)建、最終寫回 [~modular_pipelines.PipelineState] 的新中間變量[~modular_pipelines.ModularPipelineBlocks.intermediate_outputs]__call__定義循環(huán)結(jié)構(gòu)和迭代邏輯是整個(gè)包裝器的核心——下面的示例定義了一個(gè)名為L(zhǎng)oopWrapper的最小循環(huán)包裝器它聲明了唯一的循環(huán)輸入num_steps并在__call__中用一個(gè)for循環(huán)迭代每次迭代調(diào)用self.loop_step(...)按順序執(zhí)行所有注冊(cè)的子塊import torch from diffusers.modular_pipelines import LoopSequentialPipelineBlocks, ModularPipelineBlocks, InputParam, OutputParam class LoopWrapper(LoopSequentialPipelineBlocks): model_name test property def description(self): return Im a loop!! property def loop_inputs(self): return [InputParam(namenum_steps)] torch.no_grad() def __call__(self, components, state): block_state self.get_block_state(state) # 循環(huán)結(jié)構(gòu) - 可以根據(jù)您的需求定制 for i in range(block_state.num_steps): # loop_step 按順序執(zhí)行所有注冊(cè)的塊 components, block_state self.loop_step(components, block_state, ii) self.set_block_state(state, block_state) return components, state注意幾個(gè)關(guān)鍵細(xì)節(jié)loop_step與set_block_state的分工loop_step內(nèi)部已經(jīng)遍歷并執(zhí)行了sub_blocks中的全部子塊但整個(gè)循環(huán)結(jié)束后包裝器還要顯式調(diào)用self.set_block_state(state, block_state)把循環(huán)體內(nèi)累積的BlockState寫回全局的PipelineState這是循環(huán)包裝器與普通塊__call__結(jié)構(gòu)上最大的不同。迭代參數(shù)可以透?jìng)餮h(huán)包裝器可以把額外的參數(shù)如當(dāng)前迭代索引i通過(guò)loop_step(components, block_state, ii)傳給循環(huán)塊。從源碼可以看到loop_step的實(shí)現(xiàn)是依次調(diào)用每個(gè)子塊并把**kwargs原樣透?jìng)鱠ef loop_step(self, components, state: PipelineState, **kwargs): for block_name, block in self.sub_blocks.items(): try: components, state block(components, state, **kwargs) except Exception as e: ... return components, state因此循環(huán)塊只要在__call__簽名里聲明i: int甚至t: torch.Tensor就能拿到每次迭代的上下文。循環(huán)塊每次迭代執(zhí)行的葉子單元循環(huán)塊本質(zhì)上仍然是一個(gè) [~modular_pipelines.ModularPipelineBlocks]但與普通塊相比它的__call__方法行為有三個(gè)顯著區(qū)別它從循環(huán)包裝器接收調(diào)用而不是從順序容器接收調(diào)用它直接與 [~modular_pipelines.BlockState] 協(xié)作而不是與 [~modular_pipelines.PipelineState] 協(xié)作它不需要自行檢索或更新 [~modular_pipelines.BlockState] —— 該狀態(tài)由循環(huán)包裝器統(tǒng)一管理。循環(huán)塊共享同一個(gè) [~modular_pipelines.BlockState]這正是值可以在循環(huán)的每次迭代中累積和變化的機(jī)制來(lái)源WanLoopBeforeDenoiser準(zhǔn)備本輪輸入、WanLoopDenoiser算出noise_pred、WanLoopAfterDenoiser更新latents同一個(gè)BlockState在迭代間被反復(fù)讀寫。class LoopBlock(ModularPipelineBlocks): model_name test property def inputs(self): return [InputParam(namex)] property def intermediate_outputs(self): # 這個(gè)塊產(chǎn)生的輸出 return [OutputParam(namex)] property def description(self): return 我是一個(gè)在LoopWrapper類內(nèi)部使用的塊 def __call__(self, components, block_state, i: int): block_state.x 1 return components, block_state上面的LoopBlock每被調(diào)用一次就把block_state.x加一。注意它的__call__簽名是(self, components, block_state, i: int)接收的是block_state而不是state且直接返回(components, block_state)——這正是它與普通塊的__call__結(jié)構(gòu)檢索BlockState→ 計(jì)算 → 寫回PipelineState的差異所在。組裝用 from_blocks_dict 注冊(cè)循環(huán)塊要得到一個(gè)可運(yùn)行的 [~modular_pipelines.LoopSequentialPipelineBlocks]使用類方法 [~modular_pipelines.LoopSequentialPipelineBlocks.from_blocks_dict] 把循環(huán)塊注冊(cè)進(jìn)循環(huán)包裝器即可。blocks_dict的鍵是塊名值可以是塊類會(huì)自動(dòng)實(shí)例化也可以是塊實(shí)例loop LoopWrapper.from_blocks_dict({block1: LoopBlock})如果想要在每次迭代中運(yùn)行多個(gè)塊往字典里繼續(xù)添加即可。這允許你在完全不改動(dòng)循環(huán)邏輯本身的前提下自由增刪、替換循環(huán)體內(nèi)的步驟loop LoopWrapper.from_blocks_dict({block1: LoopBlock(), block2: LoopBlock})從源碼實(shí)現(xiàn)看from_blocks_dict會(huì)先實(shí)例化傳入的類再統(tǒng)一設(shè)置block_classes、block_names與sub_blocksclassmethod def from_blocks_dict(cls, blocks_dict: dict[str, Any]) - LoopSequentialPipelineBlocks: instance cls() sub_blocks InsertableDict() for name, block in blocks_dict.items(): if inspect.isclass(block): sub_blocks[name] block() else: sub_blocks[name] block instance.block_classes [block.__class__ for block in blocks_dict.values()] instance.block_names list(blocks_dict.keys()) instance.sub_blocks blocks_dict return instance一個(gè)值得注意的約束是LoopSequentialPipelineBlocks.__init__會(huì)校驗(yàn)所有子塊必須是葉子塊leaf blocks即不允許子塊自身再嵌套sub_blocks否則直接拋出ValueError。這是因?yàn)檠h(huán)體內(nèi)的每一步應(yīng)當(dāng)是原子操作嵌套容器會(huì)讓循環(huán)語(yǔ)義變得不可控。因此循環(huán)塊只能是普通的ModularPipelineBlocks不能是SequentialPipelineBlocks或另一個(gè)ConditionalPipelineBlocks。源碼視角LoopSequentialPipelineBlocks 的完整契約回到src/diffusers/modular_pipelines/modular_pipeline.py中LoopSequentialPipelineBlocks類的實(shí)現(xiàn)可以提煉出它區(qū)別于SequentialPipelineBlocks的完整行為契約輸入合并_get_inputs()先并入loop_inputs再按順序并入各子塊的輸入跳過(guò)那些已被中間輸出覆蓋的變量required_inputs則取首個(gè)子塊必填 ∪ 循環(huán)必填 ∪ 其余子塊必填的并集。中間輸出合并intermediate_outputs先合并所有子塊的輸出再補(bǔ)充loop_intermediate_outputs中未出現(xiàn)的新變量outputs直接取最后一個(gè)子塊的intermediate_outputs。組件與配置透?jìng)鱡xpected_components/expected_configs在聚合全部子塊的聲明之外還會(huì)追加循環(huán)包裝器自己聲明的loop_expected_components/loop_expected_configs——這使得調(diào)度器、guider 等循環(huán)級(jí)組件可以掛在包裝器上而不是重復(fù)聲明在每個(gè)子塊里。__call__必須由子類實(shí)現(xiàn)基類只拋出NotImplementedError__call__method needs to be implemented by the subclass也就是說(shuō)循環(huán)的具體迭代方式如何遍歷 timesteps、是否顯示進(jìn)度條、warmup 步數(shù)如何計(jì)算完全留給你的包裝器類定制。進(jìn)度條工具基類提供了progress_bar(iterableNone, totalNone)與set_progress_bar_config(**kwargs)方便在長(zhǎng)循環(huán)中渲染 tqdm 進(jìn)度條并已用torch.compiler.disable標(biāo)記避免與 torch.compile 沖突。真實(shí)案例Wan 視頻模型的迭代去噪循環(huán)倉(cāng)庫(kù)中最能說(shuō)明LoopSequentialPipelineBlocks實(shí)戰(zhàn)價(jià)值的案例是 Wan 系列視頻生成模型的去噪步驟位于src/diffusers/modular_pipelines/wan/denoise.py。首先定義一個(gè)循環(huán)包裝器WanDenoiseLoopWrapper它聲明了循環(huán)輸入timesteps與num_inference_steps并定制了循環(huán)邏輯遍歷每個(gè) timestept把i和t同時(shí)透?jìng)鹘o循環(huán)體同時(shí)維護(hù)進(jìn)度條與 warmup 步數(shù)class WanDenoiseLoopWrapper(LoopSequentialPipelineBlocks): model_name wan property def loop_inputs(self): return [ InputParam(timesteps, requiredTrue, type_hinttorch.Tensor, description...), InputParam(num_inference_steps, requiredTrue, type_hintint, description...), ] torch.no_grad() def __call__(self, components, state): block_state self.get_block_state(state) block_state.num_warmup_steps max( len(block_state.timesteps) - block_state.num_inference_steps * components.scheduler.order, 0 ) with self.progress_bar(totalblock_state.num_inference_steps) as progress_bar: for i, t in enumerate(block_state.timesteps): components, block_state self.loop_step(components, block_state, ii, tt) if i len(block_state.timesteps) - 1 or ( (i 1) block_state.num_warmup_steps and (i 1) % components.scheduler.order 0 ): progress_bar.update() self.set_block_state(state, block_state) return components, state然后在包裝器的子類里通過(guò)block_classesblock_names聲明循環(huán)體的三個(gè)步驟把每輪迭代做什么配置化class WanDenoiseStep(WanDenoiseLoopWrapper): block_classes [ WanLoopBeforeDenoiser, # 準(zhǔn)備 latent_model_input WanLoopDenoiser( # 帶 guidance 調(diào)用 transformer 預(yù)測(cè)噪聲 guider_input_fields{ encoder_hidden_states: (prompt_embeds, negative_prompt_embeds), } ), WanLoopAfterDenoiser, # 用 scheduler.step 更新 latents ] block_names [before_denoiser, denoiser, after_denoiser]三個(gè)子塊正是葉子塊 共享BlockState的典范WanLoopBeforeDenoiser.__call__(self, components, block_state, i, t)計(jì)算latent_model_inputWanLoopDenoiser通過(guò) guider 拆分條件/無(wú)條件 batch 后調(diào)用transformer得到noise_predWanLoopAfterDenoiser調(diào)用components.scheduler.step(...)更新latents。同一個(gè)block_state.latents在數(shù)百次迭代中被反復(fù)讀出、更新、寫回最終得到去噪完成的 latent。Wan2.2 版本W(wǎng)an22DenoiseStep只需替換其中的 denoiser 子塊就能在同一循環(huán)邏輯下切換到雙 transformer / 雙 guider 的高噪聲與低噪聲兩階段去噪loop_expected_configs中的boundary_ratio即用于切分這兩個(gè)階段。除 Wan 之外倉(cāng)庫(kù)中 FLUX、FLUX2、SDXL、SD3、LTX、HunyuanVideo1.5、QwenImage、Krea2、Ideogram4、MiniMax 等多個(gè)模型的denoise.py也都采用了同樣的循環(huán)包裝器模式如src/diffusers/modular_pipelines/flux/denoise.py印證了循環(huán)包裝器定義迭代結(jié)構(gòu)、循環(huán)塊定義每步操作這一設(shè)計(jì)在圖像與視頻生成管線中的普適性。與順序塊的分工總結(jié)維度SequentialPipelineBlocksLoopSequentialPipelineBlocks數(shù)據(jù)流向線性一次執(zhí)行循環(huán)多次迭代執(zhí)行子塊狀態(tài)管理每個(gè)子塊自行g(shù)et/set_block_state包裝器統(tǒng)一管理共享BlockState子塊直接使用子塊__call__簽名(components, state)(components, block_state, i, ...)子塊約束無(wú)特殊約束必須是葉子塊不能再嵌套sub_blocks典型用途編碼 → 去噪 → 解碼 的單遍步驟逐 timestep 迭代的去噪循環(huán)對(duì)于大多數(shù)一次性通過(guò)的步驟如文本編碼、latent 準(zhǔn)備、VAE 解碼選擇SequentialPipelineBlocks對(duì)于天然迭代的計(jì)算去噪、擴(kuò)散反向過(guò)程則應(yīng)該使用LoopSequentialPipelineBlocks。兩者的組合使用方式——順序塊把循環(huán)包裝器當(dāng)作其中一步嵌入整條管線——正是模塊化 Diffusers 用統(tǒng)一而可組合的積木搭建完整生成工作流的核心理念。相關(guān)測(cè)試可參考tests/modular_pipelines/test_modular_pipelines_custom_blocks.py中對(duì)from_blocks_dict與循環(huán)塊的驗(yàn)證用例。【免費(fèi)下載鏈接】diffusers Diffusers: State-of-the-art diffusion models for image, video, and audio generation in PyTorch.項(xiàng)目地址: https://gitcode.com/GitHub_Trending/di/diffusers創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考