Diffusers documentation
SequentialPipelineBlocks
SequentialPipelineBlocks
SequentialPipelineBlocks هي نوع متعدد الكتل يجمع بين كتل ModularPipelineBlocks أخرى في تسلسل. تتدفق البيانات خطياً من كتلة إلى التالية باستخدام inputs وintermediate_outputs. تمثل كل كتلة في SequentialPipelineBlocks عادةً خطوة في خط الأنابيب، ومن خلال دمجها، تبني خط أنابيب تدريجياً.
يُظهرك هذا الدليل على كيفية ربط كتلتين في SequentialPipelineBlocks.
أنشئ بلوكين من ModularPipelineBlocks. البلوك الأول، InputBlock، يُخرج قيمة batch_size، والبلوك الثاني، ImageEncoderBlock، يستخدم batch_size كـinputs.
from diffusers.modular_pipelines import ModularPipelineBlocks, InputParam, OutputParam
class InputBlock(ModularPipelineBlocks):
@property
def inputs(self):
return [
InputParam(name="prompt", type_hint=list, description="list of text prompts"),
InputParam(name="num_images_per_prompt", type_hint=int, description="number of images per prompt"),
]
@property
def intermediate_outputs(self):
return [
OutputParam(name="batch_size", description="calculated batch size"),
]
@property
def description(self):
return "A block that determines batch_size based on the number of prompts and num_images_per_prompt argument."
def __call__(self, components, state):
block_state = self.get_block_state(state)
batch_size = len(block_state.prompt)
block_state.batch_size = batch_size * block_state.num_images_per_prompt
self.set_block_state(state, block_state)
return components, stateimport torch
from diffusers.modular_pipelines import ModularPipelineBlocks, InputParam, OutputParam
class ImageEncoderBlock(ModularPipelineBlocks):
@property
def inputs(self):
return [
InputParam(name="image", type_hint="PIL.Image", description="raw input image to process"),
InputParam(name="batch_size", type_hint=int),
]
@property
def intermediate_outputs(self):
return [
OutputParam(name="image_latents", description="latents representing the image"),
]
@property
def description(self):
return "Encode raw image into its latent presentation"
def __call__(self, components, state):
block_state = self.get_block_state(state)
# Simulate processing the image
# This will change the state of the image from a PIL image to a tensor for all blocks
block_state.image = torch.randn(1, 3, 512, 512)
block_state.batch_size = block_state.batch_size * 2
block_state.image_latents = torch.randn(1, 4, 64, 64)
self.set_block_state(state, block_state)
return components, stateاربط الكتلتين عن طريق تعريف SequentialPipelineBlocks. اذكر أمثلة الكتل في block_classes وأسمائها المقابلة في block_names. تُنفَّذ الكتل بالترتيب الذي تظهر به في block_classes، ويتدفق البيانات من كتلة إلى التالية عبر PipelineState.
class ImageProcessingStep(SequentialPipelineBlocks):
"""
# auto_docstring
"""
model_name = "my_model"
block_classes = [InputBlock(), ImageEncoderBlock()]
block_names = ["input", "image_encoder"]
@property
def description(self):
return (
"Process text prompts and images for the pipeline. It:\n"
" - Determines the batch size from the prompts.\n"
" - Encodes the image into latent space."
)عند إنشاء SequentialPipelineBlocks، تُجمَّع خصائص مثل inputs وintermediate_outputs وexpected_components تلقائيًا من الكتل الفرعية، لذا لا حاجة إلى تعريفها مرة أخرى.
هناك بعض الخصائص التي يجب عليك ضبطها:
description: ننصح بإضافة وصف للكتلة المجمّعة لتوضيح ما يفعله الإجراء المدمج.model_name: يُشتق هذا تلقائيًا من الكتل الفرعية، لكنه ليس دائمًا صحيحًا، لذا قد تحتاج إلى تجاوز قيمته.outputs: افتراضيًا، تكون هذه القيمة هي نفسهاintermediate_outputs، لكن يمكنك ضبطها يدويًا للتحكم في القيم التي تظهر في المستند. وهذا مفيد لعرض المخرجات النهائية فقط بدلاً من جميع القيم الوسيطة.
تُستخدم هذه الخصائص، إلى جانب inputs وintermediate_outputs وexpected_components المجمّعة، لتوليد خاصية doc تلقائيًا.
اطبع كتلة ImageProcessingStep لفحص كتلها الفرعية، واستخدم doc للحصول على ملخص كامل لمُدخلات ومُخرجات ومكوّنات الكتلة.
blocks = ImageProcessingStep()
print(blocks)
print(blocks.doc)