Pipeline geri bildirimleri

Bir geri çağrı (callback), bir gürültü azaltma adımının sonunda çalışır ve sonraki adımlar için pipeline durumunu veya tensörleri değiştirebilir. Pipeline'ı yeniden yazmadan yeni davranışlar için öznitelikleri veya tensör değişkenlerini ayarlamak amacıyla kullanın.

Bu geri çağrılar, kutudan çıktığı gibi klasik DiffusionPipeline döngüleriyle çalışır. Gürültü azaltma döngüsüne bunları çalıştıran bir blok eklendiğinde Modular Diffusers boru hatlarıyla da çalışırlar; bkz. Modular boru hatları.

Diffusers, genel bakış bölümünde birkaç geri çağrı (callback) sunar.

Bir geri çağrının etkinleştirilmesi için, belirli bir gürültü azaltma adım sayısı sonrasında geri çağrının ne zaman çalıştırılacağını yapılandırmak için aşağıdaki argümanlardan birini kullanın.

• cutoff_step_ratio, toplam gürültü azaltma adımlarının yüzdesi olarak bir geri çağrının ne zaman etkinleştirileceğini belirtir. Kesme noktasının num_inference_steps ile ölçeklenmesi gerektiğinde kullanın (örneğin, çalışmanın %40'ından sonra CFG'yi devre dışı bırakın).
• cutoff_step_index, bir geri çağrının (callback) etkinleştirildiği kesin adım numarasını belirtir. Mutlak bir adım (örneğin, sabit 25 adımlık bir zaman çizelgesinde 10. adım) konusunda hassasiyet gösteriyorsanız kullanın.

Aşağıdaki örnek cutoff_step_ratio=0.4 kullanır; bu, geri çağrının (callback) gürültü azaltma (denoising) toplam çıkarım adımının %40'ına ulaştığında etkinleştirildiği anlamına gelir. SDXLCFGCutoffCallback, belirli bir adım sayısından sonra sınıflandırıcı-ücretsiz yönlendirmeyi (classifier-free guidance, CFG) devre dışı bırakır; bu, performansı önemli ölçüde etkilemeden hesaplama kaynağı tasarrufu sağlamaya yardımcı olabilir.

cutoff argümanlarından biriyle bir geri çağrı tanımlayın ve bunu pipeline'daki callback_on_step_end parametresine iletin.

import torch
from diffusers import DPMSolverMultistepScheduler, StableDiffusionXLPipeline
from diffusers.callbacks import SDXLCFGCutoffCallback

callback = SDXLCFGCutoffCallback(cutoff_step_ratio=0.4)
# if using cutoff_step_index
# callback = SDXLCFGCutoffCallback(cutoff_step_ratio=None, cutoff_step_index=10)

pipeline = StableDiffusionXLPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    dtype=torch.float16,
    device_map="cuda"  # or "mps", "xpu", "cpu"
)
pipeline.scheduler = DPMSolverMultistepScheduler.from_config(pipeline.scheduler.config, use_karras_sigmas=True)

prompt = "a sports car on the road, best quality, high quality, high detail, 8k resolution"
output = pipeline(
    prompt=prompt,
    negative_prompt="",
    guidance_scale=6.5,
    num_inference_steps=25,
    callback_on_step_end=callback,
)

Resmi geri çağrılar kendi girdi tensörlerini ayarlar. Özel bir işlev için, Ara görüntüleri görüntüleme bölümünde olduğu gibi callback_on_step_end_tensor_inputs parametresini iletin.

Yeni resmi bir geri bildirim (callback) eklemek istiyorsanız, bir özellik isteği açmaktan veya PR göndermekten çekinmeyin. Aksi takdirde, aşağıda gösterildiği gibi kendi geri bildiriminizi de oluşturabilirsiniz.

Erken durdurma

Erken durdurma, üretim sırasında ara sonuçlardan memnun değilseniz faydalıdır. Bu geri çağrı, _interrupt özniteliğini True olarak ayarlayarak sabitlenmiş bir durma noktası belirler; bu da gürültü azaltma döngüsünün kalan adımları atlamasına neden olur.

import torch
from diffusers import DiffusionPipeline

def interrupt_callback(pipeline, i, t, callback_kwargs):
    stop_idx = 10
    if i == stop_idx:
        pipeline._interrupt = True
    return callback_kwargs

pipeline = DiffusionPipeline.from_pretrained(
    "Qwen/Qwen-Image",
    dtype=torch.bfloat16,
    device_map="cuda",  # or "mps", "xpu", "cpu"
)
pipeline(
    prompt="A photo of a cat",
    num_inference_steps=50,
    callback_on_step_end=interrupt_callback,
)

Ara görüntüleri görüntüle

Ara görüntülerin görselleştirilmesi, ilerlemenin izlenmesi için faydalıdır. Aşağıdaki önizleme yalnızca SDXL içindir. SDXL latentlerini, gürültü azaltma sırasında hızlı bir bakış için doğrusal bir dönüşümle RGB'ye eşler. Bu ağırlıklar diğer modellere aktarılamaz. Qwen-Image ve benzeri checkpoint'ler için, bu yardımcı yerine modelin VAE'si ile çözümleme yapın.

Convert Stable Diffusion XL latenslerini (4 kanal) RGB tensörlere (3 kanal) dönüştürür.

import torch
from PIL import Image
from diffusers import AutoPipelineForText2Image

def latents_to_rgb(latents):
    weights = (
        (60, -60, 25, -70),
        (60,  -5, 15, -50),
        (60,  10, -5, -35),
    )

    weights_tensor = torch.t(torch.tensor(weights, dtype=latents.dtype).to(latents.device))
    biases_tensor = torch.tensor((150, 140, 130), dtype=latents.dtype).to(latents.device)
    rgb_tensor = torch.einsum("...lxy,lr -> ...rxy", latents, weights_tensor) + biases_tensor.unsqueeze(-1).unsqueeze(-1)
    image_array = rgb_tensor.clamp(0, 255).byte().cpu().numpy().transpose(1, 2, 0)

    return Image.fromarray(image_array)

Latentleri çıkarın ve batch'teki ilk görüntüyü RGB'ye dönüştürün. Görüntüyü adım numarasıyla birlikte PNG dosyası olarak kaydedin.

def decode_tensors(pipe, step, timestep, callback_kwargs):
    latents = callback_kwargs["latents"]

    image = latents_to_rgb(latents[0])
    image.save(f"{step}.png")

    return callback_kwargs

Callback'ın hangi tensörleri alacağını seçmek için callback_on_step_end_tensor_inputs kullanılır; bu durumda bunlar latentlerdir.

pipeline = AutoPipelineForText2Image.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    dtype=torch.float16,
    device_map="cuda"  # or "mps", "xpu", "cpu"
)

image = pipeline(
    prompt="A croissant shaped like a cute bear.",
    negative_prompt="Deformed, ugly, bad anatomy",
    callback_on_step_end=decode_tensors,
    callback_on_step_end_tensor_inputs=["latents"],
).images[0]

Modüler hatlar

Modüler pipeline'lar callback_on_step_end ve interrupt mekanizmasını taşımaz. Bunun yerine birinci sınıf akış desteği gelecektir: henüz birleştirilmemiş olan huggingface/diffusers#14159, her gürültü azaltma adımında pipeline durumunu veren ve iterasyonu durdurduğunuzda çalışmayı sonlandıran pipe.stream() işlevini ekler. Şimdilik mevcut geri çağırma işlevlerinizi yeniden kullanmak için gürültü azaltma döngüsüne bir blok ekleyin: bu blok, adım dizini ve zaman adımını içeren döngünün BlockState nesnesini alır ve latents değerlerini okuyabilir veya düzenleyebilir.

init_pipeline() öncesinde blokları düzenleyin: pipe.blocks bir kopya döndürür, bu nedenle pipeline oluşturulduktan sonra değiştirilmesi hiçbir etki yaratmaz.

Aşağıdaki blok, geri çağrıyı (callback) bir pipeline girdisi olarak alır ve yukarıdakiyle aynı imza ile çağırır: pipeline, adım indeksi, zaman adımı (timestep) ve geri dönen değerleri geri yazılan tensörlerden oluşan bir sözlük. walk_denoise_loops, ait olduğu iş akışından bağımsız olarak bloklardaki tüm gürültü azaltma (denoising) döngülerini bulur, böylece aynı kod parçası herhangi bir modüler pipeline için çalışır.

from typing import Callable

import torch
from diffusers import ModularPipeline
from diffusers.modular_pipelines import InputParam, LoopSequentialPipelineBlocks, ModularPipelineBlocks


class CallbackOnStepEnd(ModularPipelineBlocks):
    @property
    def inputs(self):
        return [
            InputParam(
                "callback_on_step_end",
                type_hint=Callable,
                description="Called after each scheduler step as `callback(pipe, i, t, {'latents': ...})`; the dict it returns is written back.",
            ),
            InputParam("latents", required=True),
        ]

    def __call__(self, components, block_state, i, t):
        if block_state.callback_on_step_end is not None:
            outputs = block_state.callback_on_step_end(components, i, t, {"latents": block_state.latents})
            block_state.latents = outputs.get("latents", block_state.latents)
        return components, block_state


def walk_denoise_loops(block):
    """Every denoising loop in the blocks, whichever workflow it belongs to."""
    if isinstance(block, LoopSequentialPipelineBlocks) and "denoiser" in block.sub_blocks:
        yield block
    for sub in getattr(block, "sub_blocks", {}).values():
        yield from walk_denoise_loops(sub)


repo_id = "stabilityai/stable-diffusion-xl-base-1.0"
blocks = ModularPipeline.from_pretrained(repo_id).blocks
for loop in walk_denoise_loops(blocks):
    loop.sub_blocks["callback"] = CallbackOnStepEnd()  # appended after the last step of the loop
pipe = blocks.init_pipeline(repo_id)
pipe.load_components(dtype=torch.float16)
pipe.to("cuda")

callback_on_step_end artık pipeline'ın bir girdisidir ve DiffusionPipeline için yazılmış bir geri çağrı, yukarıdaki decode_tensors gibi, olduğu gibi çalışır:

image = pipe(prompt="A croissant shaped like a cute bear.", callback_on_step_end=decode_tensors, output="images")[0]

Erken durdurma farklı çalışır: bir blok, içinde çalıştığı döngüyü atlayamaz, bu nedenle _interrupt ayarlamak bir etki yaratmaz. Bunun yerine geri çağrıdan (callback) bir istisna fırlatın: istisna pipe(...) dışına yayılır (görsel çözümlenmez) ve boru hattı sonrasında kullanılabilir durumda kalır.

Bu, iç içe döngülerin iç döngüsü hariç her modüler pipeline için geçerlidir (Helios ve Wan-Animate-2 gibi otoregresif video modelleri, dış döngünün bir bloğu içinde her parçayı gürültüden arındırır). Burada bloğu dış döngüye ekleyin: her parça için bir kez çalışır ve i ile t yerine parça indeksini k alır.