Buckets:

hf-doc-build
/

doc

Files

xet

hf-doc-build/doc / diffusers /main /ko /using-diffusers /callback.md

mishig

about 20 hours ago

preview code

download

raw

10.8 kB

# Pipeline 콜백

Callback은 denoising 단계의 끝에서 실행되며, 이후 단계를 위해 pipeline 상태나 텐서를 변경할 수 있습니다. pipeline을 다시 작성하지 않고 새로운 동작을 위해 속성이나 텐서 변수를 조정하는 데 사용하세요.

이 콜백은 기본적으로 클래식 [DiffusionPipeline](/docs/diffusers/main/ko/api/pipelines/overview#diffusers.DiffusionPipeline) 루프와 함께 작동합니다. 디노이징 루프에 이를 실행하는 블록이 삽입되면 [Modular Diffusers](../modular_diffusers/overview) 파이프라인에서도 작동하며, [Modular pipelines](#modular-pipelines)를 참고하세요.

Diffusers는 파이프라인 [개요](../api/pipelines/overview#diffusers.callbacks.PipelineCallback)에서 여러 콜백을 제공합니다.

콜백을 활성화하려면, … 중 하나의 인수를 사용하여 특정 수의 디노이징 단계 이후에 콜백이 실행되는 시점을 구성합니다.

- `cutoff_step_ratio`는 전체 디노이징 스텝의 백분율로 콜백이 활성화되는 시점을 지정합니다. 컷오프가 `num_inference_steps`에 따라 비례적으로 조정되어야 하는 경우(예: 실행의 40% 이후에 CFG 제거)에 사용합니다.
- `cutoff_step_index`는 콜백이 활성화되는 정확한 스텝 번호를 지정합니다. 절대 스텝(예: 고정된 25스텝 스케줄의 `10`번째 스텝)에 관심이 있을 때 사용합니다.

아래 예시는 `cutoff_step_ratio=0.4`를 사용하여, 디노이징이 전체 추론 단계의 40%에 도달하면 콜백이 활성화됩니다. [SDXLCFGCutoffCallback](/docs/diffusers/main/ko/api/pipelines/overview#diffusers.callbacks.SDXLCFGCutoffCallback)은 특정 단계 이후에 classifier-free guidance(CFG)를 비활성화하여, 성능에 큰 영향을 주지 않으면서도 계산 비용을 절감하는 데 도움이 될 수 있습니다.

`cutoff` 인자 중 하나를 사용하여 콜백을 정의하고, 파이프라인의 `callback_on_step_end` 매개변수에 전달합니다.

```py
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,
)
```

공식 콜백은 자체적으로 텐서 입력을 설정합니다. 사용자 정의 함수의 경우, [중간 이미지 표시](#display-intermediate-images)에서와 같이 `callback_on_step_end_tensor_inputs`를 전달합니다.

새로운 공식 callback을 추가하려면 [기능 요청](https://github.com/huggingface/diffusers/issues/new/choose)을 열거나 [PR을 제출](https://huggingface.co/docs/diffusers/main/en/conceptual/contribution#how-to-open-a-pr)할 수 있습니다. 그렇지 않으면 아래에 나와 있는 대로 직접 callback을 만들 수도 있습니다.

## 조기 종료

Early stopping은 생성 과정 중 중간 결과에 만족하지 못할 때 유용합니다. 이 콜백은 `_interrupt` 속성을 `True`로 설정하여 하드코딩된 중단 지점을 설정하며, 이를 통해 denoising 루프가 나머지 스텝을 건너뛰게 됩니다.

```py
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,
)
```

## 중간 이미지 표시

중간 이미지를 시각화하는 것은 진행 상황 모니터링에 유용합니다. 아래 미리보기는 SDXL 전용입니다. 디노이징(denoising) 중에 빠르게 확인하기 위해 SDXL의 latents를 선형 변환(linear transform)으로 RGB로 매핑합니다. 해당 가중치는 다른 모델에 적용되지 않습니다. Qwen-Image 및 유사한 체크포인트의 경우, 이 헬퍼 대신 모델의 VAE로 디코딩하세요.

[변환](https://huggingface.co/blog/TimothyAlexisVass/explaining-the-sdxl-latent-space) Stable Diffusion XL latents(4채널)를 RGB 텐서(3채널)로 변환합니다.

```py
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)
```

latents를 추출하고 배치의 첫 번째 이미지를 RGB로 변환합니다. 이미지를 단계 번호를 포함하여 PNG 파일로 저장합니다.

```py
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_on_step_end_tensor_inputs`를 사용하여 콜백이 수신할 텐서를 선택합니다. 이 경우 해당 텐서는 latents입니다.

```py
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]
```

## 모듈형 파이프라인

모듈형 파이프라인은 `callback_on_step_end` 및 `interrupt` 메커니즘을 지원하지 않습니다. 대신 일급 스트리밍 지원이 제공됩니다. 아직 병합되지 않은 [huggingface/diffusers#14159](https://github.com/huggingface/diffusers/pull/14159)는 `pipe.stream()`을 추가하며, 이는 디노이징 단계마다 파이프라인 상태를 반환하고 반복을 중단하면 실행을 종료합니다. 현재는 기존 콜백 함수를 재사용하려면 디노이징 루프에 블록을 삽입해야 합니다. 이 블록은 단계 인덱스와 타임스텝을 포함한 루프의 `BlockState`를 수신하며, `latents`를 읽거나 수정할 수 있습니다.

> [!TIP]
> `pipe.blocks`는 복사본을 반환하므로, 파이프라인이 생성된 이후에 수정해도 효과가 없습니다. 따라서 [init_pipeline()](/docs/diffusers/main/ko/api/modular_diffusers/pipeline_blocks#diffusers.ModularPipelineBlocks.init_pipeline) *이전*에 블록을 수정해야 합니다.

아래 블록은 콜백을 파이프라인 입력으로 받아 위에서 설명한 것과 동일한 시그니처(파이프라인, 단계 인덱스, 타임스텝, 텐서 딕셔너리)로 호출합니다. 반환된 값은 다시 해당 딕셔너리에 기록됩니다. `walk_denoise_loops`는 해당 블록이 어떤 워크플로에 속해 있든 모든 디노이징 루프를 찾아내므로, 동일한 스니펫이 모든 모듈형 파이프라인에서 작동합니다.

```py
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`는 이제 파이프라인의 입력이며, `DiffusionPipeline`용으로 작성된 `decode_tensors`와 같은 콜백은 그대로 작동합니다:

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

[Early stopping](#early-stopping)은 동작 방식이 다릅니다. 블록은 실행 중인 루프를 건너뛸 수 없으므로 `_interrupt`를 설정해도 효과가 없습니다. 대신 콜백에서 예외를 발생시키세요. 예외는 `pipe(...)`에서 전파되어 나옵니다(이미지가 디코딩되지 않음) 그리고 파이프라인은 이후에도 계속 사용할 수 있습니다.

이 방법은 Helios와 Wan-Animate-2와 같은 자기회귀 비디오 모델처럼 외부 루프의 블록 내부에서 각 청크를 디노이즈하는 중첩 루프의 내부 루프를 제외한 모든 모듈형 파이프라인에 적용됩니다. 해당 경우, 블록을 외부 루프에 삽입하십시오. 이 경우 블록은 청크마다 한 번씩 실행되며, `i`와 `t` 대신 청크 인덱스 `k`를 수신합니다.

Xet Storage Details

Size:: 10.8 kB
Xet hash:: 33dc786f0eaf2dca832a5fb18db07529049eba3aad73c0885d6cca5e1b12b5c2

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	# Pipeline 콜백

	Callback은 denoising 단계의 끝에서 실행되며, 이후 단계를 위해 pipeline 상태나 텐서를 변경할 수 있습니다. pipeline을 다시 작성하지 않고 새로운 동작을 위해 속성이나 텐서 변수를 조정하는 데 사용하세요.

	이 콜백은 기본적으로 클래식 [DiffusionPipeline](/docs/diffusers/main/ko/api/pipelines/overview#diffusers.DiffusionPipeline) 루프와 함께 작동합니다. 디노이징 루프에 이를 실행하는 블록이 삽입되면 [Modular Diffusers](../modular_diffusers/overview) 파이프라인에서도 작동하며, [Modular pipelines](#modular-pipelines)를 참고하세요.

	Diffusers는 파이프라인 [개요](../api/pipelines/overview#diffusers.callbacks.PipelineCallback)에서 여러 콜백을 제공합니다.

	콜백을 활성화하려면, … 중 하나의 인수를 사용하여 특정 수의 디노이징 단계 이후에 콜백이 실행되는 시점을 구성합니다.

	- `cutoff_step_ratio`는 전체 디노이징 스텝의 백분율로 콜백이 활성화되는 시점을 지정합니다. 컷오프가 `num_inference_steps`에 따라 비례적으로 조정되어야 하는 경우(예: 실행의 40% 이후에 CFG 제거)에 사용합니다.
	- `cutoff_step_index`는 콜백이 활성화되는 정확한 스텝 번호를 지정합니다. 절대 스텝(예: 고정된 25스텝 스케줄의 `10`번째 스텝)에 관심이 있을 때 사용합니다.

	아래 예시는 `cutoff_step_ratio=0.4`를 사용하여, 디노이징이 전체 추론 단계의 40%에 도달하면 콜백이 활성화됩니다. [SDXLCFGCutoffCallback](/docs/diffusers/main/ko/api/pipelines/overview#diffusers.callbacks.SDXLCFGCutoffCallback)은 특정 단계 이후에 classifier-free guidance(CFG)를 비활성화하여, 성능에 큰 영향을 주지 않으면서도 계산 비용을 절감하는 데 도움이 될 수 있습니다.

	`cutoff` 인자 중 하나를 사용하여 콜백을 정의하고, 파이프라인의 `callback_on_step_end` 매개변수에 전달합니다.

	```py
	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,
	)
	```

	공식 콜백은 자체적으로 텐서 입력을 설정합니다. 사용자 정의 함수의 경우, [중간 이미지 표시](#display-intermediate-images)에서와 같이 `callback_on_step_end_tensor_inputs`를 전달합니다.

	새로운 공식 callback을 추가하려면 [기능 요청](https://github.com/huggingface/diffusers/issues/new/choose)을 열거나 [PR을 제출](https://huggingface.co/docs/diffusers/main/en/conceptual/contribution#how-to-open-a-pr)할 수 있습니다. 그렇지 않으면 아래에 나와 있는 대로 직접 callback을 만들 수도 있습니다.

	## 조기 종료

	Early stopping은 생성 과정 중 중간 결과에 만족하지 못할 때 유용합니다. 이 콜백은 `_interrupt` 속성을 `True`로 설정하여 하드코딩된 중단 지점을 설정하며, 이를 통해 denoising 루프가 나머지 스텝을 건너뛰게 됩니다.

	```py
	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,
	)
	```

	## 중간 이미지 표시

	중간 이미지를 시각화하는 것은 진행 상황 모니터링에 유용합니다. 아래 미리보기는 SDXL 전용입니다. 디노이징(denoising) 중에 빠르게 확인하기 위해 SDXL의 latents를 선형 변환(linear transform)으로 RGB로 매핑합니다. 해당 가중치는 다른 모델에 적용되지 않습니다. Qwen-Image 및 유사한 체크포인트의 경우, 이 헬퍼 대신 모델의 VAE로 디코딩하세요.

	[변환](https://huggingface.co/blog/TimothyAlexisVass/explaining-the-sdxl-latent-space) Stable Diffusion XL latents(4채널)를 RGB 텐서(3채널)로 변환합니다.

	```py
	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)
	```

	latents를 추출하고 배치의 첫 번째 이미지를 RGB로 변환합니다. 이미지를 단계 번호를 포함하여 PNG 파일로 저장합니다.

	```py
	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_on_step_end_tensor_inputs`를 사용하여 콜백이 수신할 텐서를 선택합니다. 이 경우 해당 텐서는 latents입니다.

	```py
	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]
	```

	## 모듈형 파이프라인

	모듈형 파이프라인은 `callback_on_step_end` 및 `interrupt` 메커니즘을 지원하지 않습니다. 대신 일급 스트리밍 지원이 제공됩니다. 아직 병합되지 않은 [huggingface/diffusers#14159](https://github.com/huggingface/diffusers/pull/14159)는 `pipe.stream()`을 추가하며, 이는 디노이징 단계마다 파이프라인 상태를 반환하고 반복을 중단하면 실행을 종료합니다. 현재는 기존 콜백 함수를 재사용하려면 디노이징 루프에 블록을 삽입해야 합니다. 이 블록은 단계 인덱스와 타임스텝을 포함한 루프의 `BlockState`를 수신하며, `latents`를 읽거나 수정할 수 있습니다.

	> [!TIP]
	> `pipe.blocks`는 복사본을 반환하므로, 파이프라인이 생성된 이후에 수정해도 효과가 없습니다. 따라서 [init_pipeline()](/docs/diffusers/main/ko/api/modular_diffusers/pipeline_blocks#diffusers.ModularPipelineBlocks.init_pipeline) 이전에 블록을 수정해야 합니다.

	아래 블록은 콜백을 파이프라인 입력으로 받아 위에서 설명한 것과 동일한 시그니처(파이프라인, 단계 인덱스, 타임스텝, 텐서 딕셔너리)로 호출합니다. 반환된 값은 다시 해당 딕셔너리에 기록됩니다. `walk_denoise_loops`는 해당 블록이 어떤 워크플로에 속해 있든 모든 디노이징 루프를 찾아내므로, 동일한 스니펫이 모든 모듈형 파이프라인에서 작동합니다.

	```py
	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`는 이제 파이프라인의 입력이며, `DiffusionPipeline`용으로 작성된 `decode_tensors`와 같은 콜백은 그대로 작동합니다:

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

	[Early stopping](#early-stopping)은 동작 방식이 다릅니다. 블록은 실행 중인 루프를 건너뛸 수 없으므로 `_interrupt`를 설정해도 효과가 없습니다. 대신 콜백에서 예외를 발생시키세요. 예외는 `pipe(...)`에서 전파되어 나옵니다(이미지가 디코딩되지 않음) 그리고 파이프라인은 이후에도 계속 사용할 수 있습니다.

	이 방법은 Helios와 Wan-Animate-2와 같은 자기회귀 비디오 모델처럼 외부 루프의 블록 내부에서 각 청크를 디노이즈하는 중첩 루프의 내부 루프를 제외한 모든 모듈형 파이프라인에 적용됩니다. 해당 경우, 블록을 외부 루프에 삽입하십시오. 이 경우 블록은 청크마다 한 번씩 실행되며, `i`와 `t` 대신 청크 인덱스 `k`를 수신합니다.