Diffusers documentation
AutoPipelineBlocks
AutoPipelineBlocks
تُعد AutoPipelineBlocks نوعًا متعدد الكتل يحتوي على كتل تدعم سير عمل مختلفة. يقوم تلقائيًا باختيار الكتل الفرعية التي يجب تشغيلها بناءً على المدخلات المُقدمة وقت التشغيل. يُستخدم هذا عادةً لتجميع عدة سير عمل - من النص إلى الصورة، من الصورة إلى الصورة، وإعادة الرسم - في خط أنابيب واحد لتسهيل الاستخدام.
يُظهر هذا الدليل كيفية إنشاء AutoPipelineBlocks.
أنشئ ثلاثة ModularPipelineBlocks لتحويل النص إلى صورة، وتحويل الصورة إلى صورة، والتعبئة (Inpainting). تمثل هذه الكتل سير العمل المختلفة المتاحة في خط الأنابيب.
import torch
from diffusers.modular_pipelines import ModularPipelineBlocks, InputParam, OutputParam
class TextToImageBlock(ModularPipelineBlocks):
model_name = "text2img"
@property
def inputs(self):
return [InputParam(name="prompt")]
@property
def intermediate_outputs(self):
return []
@property
def description(self):
return "I'm a text-to-image workflow!"
def __call__(self, components, state):
block_state = self.get_block_state(state)
print("running the text-to-image workflow")
# Add your text-to-image logic here
# For example: generate image from prompt
self.set_block_state(state, block_state)
return components, stateclass ImageToImageBlock(ModularPipelineBlocks):
model_name = "img2img"
@property
def inputs(self):
return [InputParam(name="prompt"), InputParam(name="image")]
@property
def intermediate_outputs(self):
return []
@property
def description(self):
return "I'm an image-to-image workflow!"
def __call__(self, components, state):
block_state = self.get_block_state(state)
print("running the image-to-image workflow")
# Add your image-to-image logic here
# For example: transform input image based on prompt
self.set_block_state(state, block_state)
return components, stateclass InpaintBlock(ModularPipelineBlocks):
model_name = "inpaint"
@property
def inputs(self):
return [InputParam(name="prompt"), InputParam(name="image"), InputParam(name="mask")]
@property
def intermediate_outputs(self):
return []
@property
def description(self):
return "I'm an inpaint workflow!"
def __call__(self, components, state):
block_state = self.get_block_state(state)
print("running the inpaint workflow")
# Add your inpainting logic here
# For example: fill masked areas based on prompt
self.set_block_state(state, block_state)
return components, stateأنشئ فئة AutoPipelineBlocks تتضمن قائمة بفئات الأجزاء الفرعية وأسماء الأجزاء المقابلة لها.
يجب أيضًا تضمين block_trigger_inputs، وهو قائمة بأسماء المدخلات التي تُطلق الكتلة المقابلة. إذا تم توفير مدخل محفّز وقت التشغيل، فسيتم اختيار تلك الكتلة للتشغيل. استخدم None لتحديد الكتلة الافتراضية للتشغيل إذا لم يتم اكتشاف أي مدخلات محفّزة.
أخيرًا، من المهم تضمين description يوضح بوضوح أي المدخلات تُطلق أي مسار عمل. يساعد ذلك المستخدمين على فهم كيفية تشغيل مسارات عمل محددة.
from diffusers.modular_pipelines import AutoPipelineBlocks
class AutoImageBlocks(AutoPipelineBlocks):
# List of sub-block classes to choose from
block_classes = [InpaintBlock, ImageToImageBlock, TextToImageBlock]
# Names for each block in the same order
block_names = ["inpaint", "img2img", "text2img"]
# Trigger inputs that determine which block to run
# - "mask" triggers inpaint workflow
# - "image" triggers img2img workflow (but only if mask is not provided)
# - if none of above, runs the text2img workflow (default)
block_trigger_inputs = ["mask", "image", None]
@property
def description(self):
return (
"Pipeline generates images given different types of conditions!\n"
+ "This is an auto pipeline block that works for text2img, img2img and inpainting tasks.\n"
+ " - inpaint workflow is run when `mask` is provided.\n"
+ " - img2img workflow is run when `image` is provided (but only when `mask` is not provided).\n"
+ " - text2img workflow is run when neither `image` nor `mask` is provided.\n"
)من الهام للغاية تضمين description لتجنب أي لبس حول كيفية تشغيل الكتلة وما هي المدخلات المطلوبة. وعلى الرغم من أن AutoPipelineBlocks مريحة، إلا أن منطقها الشرطي قد يكون صعب الفهم إذا لم يتم شرحه بشكل صحيح.
أنشئ مثيلًا من AutoImageBlocks.
auto_blocks = AutoImageBlocks()
في حالات التركيبات الأكثر تعقيدًا، مثل كتل AutoPipelineBlocks المتداخلة عندما تُستخدم ككتل فرعية في خطوط أنابيب أكبر، استخدم طريقة get_execution_blocks() لاستخراج الكتلة التي يتم تشغيلها فعليًا بناءً على مدخلاتك.
auto_blocks.get_execution_blocks(mask=True)ConditionalPipelineBlocks
يُعدّ AutoPipelineBlocks حالة خاصة من ConditionalPipelineBlocks. بينما يختار AutoPipelineBlocks الكتل بناءً على ما إذا كان مدخل المحفّز مُقدَّمًا أم لا، فإن ConditionalPipelineBlocks قادر على اختيار كتلة بناءً على منطق اختيار مخصص مُقدَّم في طريقة select_block.
إليك المثال نفسه مكتوبًا باستخدام ConditionalPipelineBlocks مباشرةً:
from diffusers.modular_pipelines import ConditionalPipelineBlocks
class AutoImageBlocks(ConditionalPipelineBlocks):
block_classes = [InpaintBlock, ImageToImageBlock, TextToImageBlock]
block_names = ["inpaint", "img2img", "text2img"]
block_trigger_inputs = ["mask", "image"]
default_block_name = "text2img"
@property
def description(self):
return (
"Pipeline generates images given different types of conditions!\n"
+ "This is an auto pipeline block that works for text2img, img2img and inpainting tasks.\n"
+ " - inpaint workflow is run when `mask` is provided.\n"
+ " - img2img workflow is run when `image` is provided (but only when `mask` is not provided).\n"
+ " - text2img workflow is run when neither `image` nor `mask` is provided.\n"
)
def select_block(self, mask=None, image=None) -> str | None:
if mask is not None:
return "inpaint"
if image is not None:
return "img2img"
return None # falls back to default_block_name ("text2img")تُمرَّر المدخلات المدرجة في block_trigger_inputs كحجج اسمية إلى select_block(). عند إرجاع select_block القيمة None، يتم الرجوع إلى default_block_name. وإذا كانت default_block_name هي أيضاً None، يتم تجاوز الكتلة الشرطية بأكملها — وهذا مفيد لخطوات المعالجة الاختيارية التي يجب أن تعمل فقط عند توفير مدخلات محددة.
سير العمل
يمكن أن تدعم خطوط الأنابيب التي تحتوي على كتل شرطية (AutoPipelineBlocks أو ConditionalPipelineBlocks) سير عمل متعددة — على سبيل المثال، يدعم خط أنابيب SDXL المعياري لدينا اثني عشر سير عمل في خط أنابيب واحد. لكن هذا يعني أيضًا أنه قد يكون من الصعب على المستخدمين معرفة سلاسل العمل المدعومة وكيفية تشغيلها. وبالنسبة لمطوري خطوط الأنابيب، فمن المفيد القدرة على استخراج الكتل ذات الصلة بسير عمل محدد فقط.
نوصي بتعريف _workflow_map لمنح كل مسار عمل اسمًا وقائمةً صريحةً بالمداخل التي يتطلبها.
from diffusers.modular_pipelines import SequentialPipelineBlocks
class MyPipelineBlocks(SequentialPipelineBlocks):
block_classes = [TextEncoderBlock, AutoImageBlocks, DecodeBlock]
block_names = ["text_encoder", "auto_image", "decode"]
_workflow_map = {
"text2image": {"prompt": True},
"image2image": {"image": True, "prompt": True},
"inpaint": {"mask": True, "image": True, "prompt": True},
}جميع خطوط الأنابيب المعيارية المدمجة لدينا تأتي مع مسارات عمل محددة مسبقًا. يحدد خاصية available_workflows جميع مسارات العمل المدعومة:
pipeline_blocks = MyPipelineBlocks()
pipeline_blocks.available_workflows
# ['text2image', 'image2image', 'inpaint']استرجع مسار عمل محددًا باستخدام get_workflow لفحص وتصحيح كتلة محددة تنفّذ مسار العمل.
pipeline_blocks.get_workflow("inpaint")