diffusers/v0.43.0/ko/using-diffusers/write_own_pipeline.md

파이프라인, 모델 및 스케줄러 이해하기

🧨 Diffusers는 사용자의 사용 사례에 맞게 확산 시스템을 구축할 수 있도록 사용자 친화적이고 유연한 도구 모음으로 설계되었습니다. 도구 모음의 핵심은 모델과 스케줄러입니다. DiffusionPipeline은 편의를 위해 이러한 구성 요소를 하나로 묶어 제공하지만, 파이프라인을 분리하여 모델과 스케줄러를 개별적으로 사용해 새로운 확산 시스템을 만들 수도 있습니다.

이 튜토리얼에서는 기본적인 파이프라인을 시작으로 Stable Diffusion 파이프라인까지, 모델과 스케줄러를 사용하여 추론용 diffusion 시스템을 구성하는 방법을 학습합니다.

기본 파이프라인 분해하기

pipeline는 추론을 위해 모델을 실행하는 빠르고 쉬운 방법이며, 이미지를 생성하는 데 필요한 코드는 4줄에 불과합니다:

>>> from diffusers import DDPMPipeline

>>> ddpm = DDPMPipeline.from_pretrained("google/ddpm-cat-256", use_safetensors=True).to("cuda")  # or "mps", "xpu", "cpu"
>>> image = ddpm(num_inference_steps=25).images[0]
>>> image

그것은 매우 쉬웠지만, 파이프라인은 어떻게 그렇게 했을까요? 파이프라인을 분해하여 내부에서 일어나는 일을 살펴봅시다.

위 예시에서 파이프라인에는 UNet2DModel 모델과 DDPMScheduler가 포함되어 있습니다. 파이프라인은 원하는 출력 크기의 랜덤 노이즈를 받아 모델을 여러 번 통과시킴으로써 이미지를 디노이즈합니다. 각 타임스텝에서 모델은 noise residual을 예측하고, 스케줄러는 이를 사용하여 노이즈가 적은 이미지를 예측합니다. 파이프라인은 지정된 추론 스텝 수의 끝까지 도달할 때까지 이 과정을 반복합니다.

모델과 스케줄러를 별도로 사용하여 파이프라인을 재현하려면, 자체적인 디노이징 프로세스를 작성해 보겠습니다.

1. 모델과 스케줄러를 로드합니다:

>>> from diffusers import DDPMScheduler, UNet2DModel

>>> scheduler = DDPMScheduler.from_pretrained("google/ddpm-cat-256")
>>> model = UNet2DModel.from_pretrained("google/ddpm-cat-256", use_safetensors=True).to("cuda")  # or "mps", "xpu", "cpu"

2. 프로세스를 실행할 수를 설정합니다:

>>> scheduler.set_timesteps(50)

3. 스케줄러의 timesteps을 설정하면, 이 예시에서는 50개의 요소가 균등한 간격으로 배치된 텐서가 생성됩니다. 각 요소는 모델이 이미지를 디노이즈하는 타임스텝에 해당합니다. 나중에 디노이즈 루프를 만들 때, 이 텐서를 반복하여 이미지를 디노이즈합니다:

>>> scheduler.timesteps
tensor([980, 960, 940, 920, 900, 880, 860, 840, 820, 800, 780, 760, 740, 720,
    700, 680, 660, 640, 620, 600, 580, 560, 540, 520, 500, 480, 460, 440,
    420, 400, 380, 360, 340, 320, 300, 280, 260, 240, 220, 200, 180, 160,
    140, 120, 100,  80,  60,  40,  20,   0])

4. 원하는 출력과 동일한 모양의 무작위 노이즈를 생성합니다:

>>> import torch

>>> sample_size = model.config.sample_size
>>> noise = torch.randn((1, 3, sample_size, sample_size), device="cuda")  # or "mps", "xpu", "cpu"

5. 이제 timesteps를 반복하는 루프를 작성합니다. 각 timestep마다 모델은 UNet2DModel.forward() 패스를 수행하고 노이즈 잔차를 반환합니다. 스케줄러의 step() 메서드는 노이즈 잔차, timestep, 입력을 받아 이전 timestep의 이미지를 예측합니다. 이 출력은 디노이징 루프에서 모델의 다음 입력이 되며, timesteps 배열의 끝까지 도달할 때까지 반복됩니다.

>>> input = noise

>>> for t in scheduler.timesteps:
...     with torch.no_grad():
...         noisy_residual = model(input, t).sample
...     previous_noisy_sample = scheduler.step(noisy_residual, t, input).prev_sample
...     input = previous_noisy_sample

이것이 전체 노이즈 제거 과정이며, 동일한 패턴을 사용하여 어떤 diffusion 시스템이든 작성할 수 있습니다.

6. 마지막 단계는 노이즈가 제거된 출력을 이미지로 변환하는 것입니다:

>>> from PIL import Image
>>> import numpy as np

>>> image = (input / 2 + 0.5).clamp(0, 1).squeeze()
>>> image = (image.permute(1, 2, 0) * 255).round().to(torch.uint8).cpu().numpy()
>>> image = Image.fromarray(image)
>>> image

다음 섹션에서는 실력을 시험해 보며 더 복잡한 Stable Diffusion 파이프라인을 분석해 보겠습니다. 단계는 대체로 동일합니다. 필요한 구성 요소를 초기화하고, timestep 배열을 생성하기 위해 타임스텝 수를 설정합니다. timestep 배열은 노이즈 제거 루프에서 사용되며, 이 배열의 각 요소에 대해 모델은 노이즈가 덜한 이미지를 예측합니다. 노이즈 제거 루프는 timestep을 반복하며, 각 타임스텝에서 노이즈 잔차를 출력하고 스케줄러는 이를 사용해 이전 타임스텝에서 노이즈가 덜한 이미지를 예측합니다. 이 과정은 timestep 배열의 끝까지 도달할 때까지 반복됩니다.

지금 바로 시도해 보세요!

Stable Diffusion 파이프라인 분해하기

Stable Diffusion은 텍스트-이미지 latent diffusion 모델입니다. 실제 픽셀 공간 대신 이미지의 저차원 표현을 사용하므로 메모리 효율이 더 높아 latent diffusion 모델이라고 불립니다. 인코더는 이미지를 더 작은 표현으로 압축하고, 디코더는 압축된 표현을 다시 이미지로 변환합니다. 텍스트-이미지 모델의 경우 텍스트 임베딩을 생성하려면 토크나이저와 인코더가 필요합니다. 이전 예제에서 이미 UNet 모델과 스케줄러가 필요하다는 것을 알고 있습니다.

알 수 있듯이, 이는 UNet 모델만 포함하는 DDPM 파이프라인보다 이미 더 복잡합니다. Stable Diffusion 모델은 세 개의 독립적인 사전 학습된 모델을 가지고 있습니다.

💡 VAE, UNet, 텍스트 인코더 모델의 작동 방식에 대한 자세한 내용은 Stable Diffusion은 어떻게 작동하나요? 블로그를 참고하세요.

Stable Diffusion 파이프라인에 필요한 구성 요소를 이제 파악했으므로, from_pretrained() 메서드로 이 모든 구성 요소를 로드하세요. 사전 학습된 stable-diffusion-v1-5/stable-diffusion-v1-5 체크포인트에서 이들을 찾을 수 있으며, 각 구성 요소는 별도의 하위 폴더에 저장되어 있습니다:

>>> from PIL import Image
>>> import torch
>>> from transformers import CLIPTextModel, CLIPTokenizer
>>> from diffusers import AutoencoderKL, UNet2DConditionModel, PNDMScheduler

>>> vae = AutoencoderKL.from_pretrained("CompVis/stable-diffusion-v1-4", subfolder="vae", use_safetensors=True)
>>> tokenizer = CLIPTokenizer.from_pretrained("CompVis/stable-diffusion-v1-4", subfolder="tokenizer")
>>> text_encoder = CLIPTextModel.from_pretrained(
...     "CompVis/stable-diffusion-v1-4", subfolder="text_encoder", use_safetensors=True
... )
>>> unet = UNet2DConditionModel.from_pretrained(
...     "CompVis/stable-diffusion-v1-4", subfolder="unet", use_safetensors=True
... )

기본값인 PNDMScheduler 대신 UniPCMultistepScheduler로 교체하여 다른 스케줄러를 얼마나 쉽게 적용할 수 있는지 확인해 보세요:

>>> from diffusers import UniPCMultistepScheduler

>>> scheduler = UniPCMultistepScheduler.from_pretrained("CompVis/stable-diffusion-v1-4", subfolder="scheduler")

추론 속도를 높이기 위해, 스케줄러와 달리 학습 가능한 가중치를 가지고 있으므로 모델을 GPU로 이동하세요:

>>> torch_device = "cuda"  # or "mps", "xpu", "cpu"
>>> vae.to(torch_device)
>>> text_encoder.to(torch_device)
>>> unet.to(torch_device)

텍스트 임베딩 생성

다음 단계는 텍스트를 토큰화하여 임베딩을 생성하는 것입니다. 이 텍스트는 UNet 모델을 조건화하고, 확산 과정을 입력 프롬프트와 유사한 방향으로 유도하는 데 사용됩니다.

💡 guidance_scale 매개변수는 이미지 생성 시 프롬프트에 부여할 가중치를 결정합니다.

다른 것을 생성하려면 원하는 프롬프트를 자유롭게 선택하세요!

>>> prompt = ["a photograph of an astronaut riding a horse"]
>>> height = 512  # default height of Stable Diffusion
>>> width = 512  # default width of Stable Diffusion
>>> num_inference_steps = 25  # Number of denoising steps
>>> guidance_scale = 7.5  # Scale for classifier-free guidance
>>> generator = torch.manual_seed(0)  # Seed generator to create the initial latent noise
>>> batch_size = len(prompt)

텍스트를 토큰화하고 프롬프트에서 임베딩을 생성합니다:

>>> text_input = tokenizer(
...     prompt, padding="max_length", max_length=tokenizer.model_max_length, truncation=True, return_tensors="pt"
... )

>>> with torch.no_grad():
...     text_embeddings = text_encoder(text_input.input_ids.to(torch_device))[0]

패딩 토큰의 임베딩인 *무조건 텍스트 임베딩(unconditional text embeddings)*도 생성해야 합니다. 이는 조건부 text_embeddings와 동일한 형태(batch_size 및 seq_length)여야 합니다:

>>> max_length = text_input.input_ids.shape[-1]
>>> uncond_input = tokenizer([""] * batch_size, padding="max_length", max_length=max_length, return_tensors="pt")
>>> uncond_embeddings = text_encoder(uncond_input.input_ids.to(torch_device))[0]

forward 패스를 두 번 수행하지 않도록 조건부 임베딩과 무조건부 임베딩을 배치로 연결합시다:

>>> text_embeddings = torch.cat([uncond_embeddings, text_embeddings])

random noise 생성

다음으로, 확산 과정의 시작점으로 사용할 초기 랜덤 노이즈를 생성합니다. 이는 이미지의 잠재 표현(latent representation)이며, 점차 노이즈가 제거됩니다. 이 시점에서 latent 이미지는 최종 이미지 크기보다 작지만, 모델이 나중에 이를 최종 512x512 이미지 차원으로 변환하므로 문제가 되지 않습니다.

💡 vae 모델에 다운샘플링 레이어가 3개 있으므로 높이와 너비는 8로 나뉩니다. 다음을 실행하여 확인할 수 있습니다:

2 ** (len(vae.config.block_out_channels) - 1) == 8

>>> latents = torch.randn(
...     (batch_size, unet.config.in_channels, height // 8, width // 8),
...     generator=generator,
...     device=torch_device,
... )

이미지를 디노이즈합니다

초기 노이즈 분포인 sigma, 즉 UniPCMultistepScheduler와 같은 개선된 스케줄러에 필요한 노이즈 스케일 값으로 입력을 스케일링하는 것부터 시작합니다:

>>> latents = latents * scheduler.init_noise_sigma

마지막 단계는 latents의 순수한 노이즈를 프롬프트로 설명된 이미지로 점진적으로 변환하는 디노이징 루프를 만드는 것입니다. 디노이징 루프는 세 가지를 수행해야 합니다.

1. 디노이징 과정에서 사용할 스케줄러의 timesteps를 설정합니다.
2. timesteps를 반복합니다.
3. 각 timestep마다 UNet 모델을 호출하여 노이즈 잔차를 예측한 후, 이를 스케줄러에 전달하여 이전의 노이즈 샘플을 계산합니다.

>>> from tqdm.auto import tqdm

>>> scheduler.set_timesteps(num_inference_steps)

>>> for t in tqdm(scheduler.timesteps):
...     # expand the latents if we are doing classifier-free guidance to avoid doing two forward passes.
...     latent_model_input = torch.cat([latents] * 2)

...     latent_model_input = scheduler.scale_model_input(latent_model_input, timestep=t)

...     # predict the noise residual
...     with torch.no_grad():
...         noise_pred = unet(latent_model_input, t, encoder_hidden_states=text_embeddings).sample

...     # perform guidance
...     noise_pred_uncond, noise_pred_text = noise_pred.chunk(2)
...     noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

...     # compute the previous noisy sample x_t -> x_t-1
...     latents = scheduler.step(noise_pred, t, latents).prev_sample

이미지 디코딩

마지막 단계는 vae를 사용하여 잠재 표현을 이미지로 디코딩하고 sample로 디코딩된 출력을 가져오는 것입니다:

# scale and decode the image latents with vae
latents = 1 / 0.18215 * latents
with torch.no_grad():
    image = vae.decode(latents).sample

마지막으로, 생성된 이미지를 확인하기 위해 이미지를 PIL.Image로 변환하세요!

>>> image = (image / 2 + 0.5).clamp(0, 1).squeeze()
>>> image = (image.permute(1, 2, 0) * 255).to(torch.uint8).cpu().numpy()
>>> image = Image.fromarray(image)
>>> image

다음 단계

기본적인 파이프라인에서 복잡한 파이프라인에 이르기까지, 자체 diffusion 시스템을 작성하는 데 실제로 필요한 것은 denoising loop뿐임을 확인했습니다. 이 loop는 scheduler의 timesteps를 설정하고, 이를 순회하며, UNet 모델을 호출하여 noise residual을 예측한 후 이를 scheduler에 전달하여 이전의 noisy sample을 계산하는 과정을 번갈아 수행해야 합니다.

이것이 바로 🧨 Diffusers의 설계 목적입니다. 모델과 스케줄러를 사용하여 자체 확산 시스템을 직관적이고 쉽게 작성할 수 있도록 하는 것입니다.

다음 단계로, 자유롭게 다음을 수행해 보세요:

• 🧨 Diffusers에 파이프라인을 구축하고 기여하는 방법을 배워보세요. 여러분이 어떤 것을 만들어낼지 기대가 됩니다!
• 라이브러리의 기존 파이프라인을 탐색하고, 모델과 스케줄러를 각각 사용하여 파이프라인을 분해하고 처음부터 구축해 볼 수 있는지 확인해 보세요.