Split View: 생성형 AI & 디퓨전 모델 완전 정복: Stable Diffusion, ControlNet, 비디오 생성까지

# Basic GAN architecture
import torch
import torch.nn as nn

class Generator(nn.Module):
    def __init__(self, latent_dim=100, img_size=64):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(latent_dim, 256),
            nn.ReLU(),
            nn.Linear(256, 512),
            nn.ReLU(),
            nn.Linear(512, img_size * img_size * 3),
            nn.Tanh()
        )

    def forward(self, z):
        return self.net(z).view(-1, 3, 64, 64)

import torch
import torch.nn.functional as F

class DDPMScheduler:
    def __init__(self, num_timesteps=1000, beta_start=1e-4, beta_end=0.02):
        self.T = num_timesteps
        # Linear noise schedule
        self.betas = torch.linspace(beta_start, beta_end, num_timesteps)
        self.alphas = 1.0 - self.betas
        self.alpha_bar = torch.cumprod(self.alphas, dim=0)

    def add_noise(self, x0, noise, t):
        """Add t steps of noise to x0 via reparameterization trick"""
        sqrt_alpha_bar = self.alpha_bar[t] ** 0.5
        sqrt_one_minus = (1 - self.alpha_bar[t]) ** 0.5
        # Reshape for broadcasting
        sqrt_alpha_bar = sqrt_alpha_bar.view(-1, 1, 1, 1)
        sqrt_one_minus = sqrt_one_minus.view(-1, 1, 1, 1)
        return sqrt_alpha_bar * x0 + sqrt_one_minus * noise

def ddpm_training_step(model, scheduler, x0, optimizer):
    batch_size = x0.shape[0]
    # Sample random timesteps
    t = torch.randint(0, scheduler.T, (batch_size,))
    # Sample Gaussian noise
    noise = torch.randn_like(x0)
    # Add noise (forward process)
    xt = scheduler.add_noise(x0, noise, t)
    # Predict noise
    predicted_noise = model(xt, t)
    # MSE loss
    loss = F.mse_loss(predicted_noise, noise)
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()
    return loss.item()

from diffusers import StableDiffusionPipeline
import torch

# Basic pipeline usage
pipe = StableDiffusionPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5",
    torch_dtype=torch.float16
).to("cuda")

# Generate image
image = pipe(
    prompt="a serene mountain landscape at sunset, photorealistic",
    negative_prompt="blurry, low quality, distorted",
    num_inference_steps=30,
    guidance_scale=7.5,
    width=512,
    height=512
).images[0]

image.save("output.png")

# Visualizing the VAE latent space
from diffusers import AutoencoderKL
from PIL import Image
import torchvision.transforms as T

vae = AutoencoderKL.from_pretrained("stabilityai/sd-vae-ft-mse")
vae = vae.to("cuda").eval()

transform = T.Compose([T.Resize((512, 512)), T.ToTensor(),
                        T.Normalize([0.5], [0.5])])

img = transform(Image.open("input.png")).unsqueeze(0).to("cuda")
with torch.no_grad():
    # Pixel → Latent (encoding)
    latent = vae.encode(img).latent_dist.sample()
    latent = latent * vae.config.scaling_factor
    print(f"Latent space shape: {latent.shape}")  # [1, 4, 64, 64]

from transformers import CLIPTextModel, CLIPTokenizer

tokenizer = CLIPTokenizer.from_pretrained("openai/clip-vit-large-patch14")
text_encoder = CLIPTextModel.from_pretrained("openai/clip-vit-large-patch14")

prompt = "a fantasy castle in the clouds"
tokens = tokenizer(prompt, padding="max_length", max_length=77,
                   return_tensors="pt")
with torch.no_grad():
    text_emb = text_encoder(tokens.input_ids)[0]
print(f"Text embedding shape: {text_emb.shape}")  # [1, 77, 768]

from diffusers import StableDiffusionPipeline
from peft import LoraConfig, get_peft_model
import torch

# LoRA configuration
lora_config = LoraConfig(
    r=16,                          # Rank
    lora_alpha=32,                 # Scaling parameter
    target_modules=["to_q", "to_v", "to_k", "to_out.0"],
    lora_dropout=0.05,
    bias="none",
)

# Apply LoRA to model
pipe = StableDiffusionPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5"
)
unet_lora = get_peft_model(pipe.unet, lora_config)
unet_lora.print_trainable_parameters()
# Trainable params: ~3M / Total: ~860M (about 0.3%)

from diffusers import DiffusionPipeline
import torch

# Load DreamBooth-trained model
pipe = DiffusionPipeline.from_pretrained(
    "./dreambooth-sks-dog",  # Trained checkpoint
    torch_dtype=torch.float16
).to("cuda")

# Generate the specific dog using "sks dog"
images = pipe(
    "a photo of sks dog in front of the Eiffel Tower",
    num_inference_steps=50,
    guidance_scale=7.5
).images

from diffusers import StableDiffusionControlNetPipeline, ControlNetModel
from diffusers.utils import load_image
import torch
import cv2
import numpy as np

# Load ControlNet model (Canny edge)
controlnet = ControlNetModel.from_pretrained(
    "lllyasviel/sd-controlnet-canny",
    torch_dtype=torch.float16
)
pipe = StableDiffusionControlNetPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5",
    controlnet=controlnet,
    torch_dtype=torch.float16
).to("cuda")

# Extract Canny edges
image = load_image("input.png")
image_np = np.array(image)
low_threshold, high_threshold = 100, 200
canny_image = cv2.Canny(image_np, low_threshold, high_threshold)
canny_image = canny_image[:, :, None]
canny_image = np.concatenate([canny_image] * 3, axis=2)

# ControlNet inference
result = pipe(
    prompt="a beautiful landscape, detailed, 8k",
    image=canny_image,
    num_inference_steps=30,
    controlnet_conditioning_scale=1.0,
).images[0]

from diffusers import StableDiffusionInstructPix2PixPipeline
import torch
from diffusers.utils import load_image

pipe = StableDiffusionInstructPix2PixPipeline.from_pretrained(
    "timbrooks/instruct-pix2pix",
    torch_dtype=torch.float16,
    safety_checker=None
).to("cuda")

image = load_image("horse.png")
result = pipe(
    "turn the horse into a zebra",
    image=image,
    num_inference_steps=30,
    image_guidance_scale=1.5,  # Fidelity to original image
    guidance_scale=7.5          # Text instruction strength
).images[0]

from diffusers import CogVideoXPipeline
import torch

pipe = CogVideoXPipeline.from_pretrained(
    "THUDM/CogVideoX-5b",
    torch_dtype=torch.bfloat16
).to("cuda")

video = pipe(
    prompt="A serene lake with rippling water, birds flying overhead",
    num_videos_per_prompt=1,
    num_inference_steps=50,
    num_frames=49,
    guidance_scale=6,
).frames[0]

from audiocraft.models import MusicGen
import torchaudio

model = MusicGen.get_pretrained("facebook/musicgen-large")
model.set_generation_params(duration=30)  # Generate 30 seconds

descriptions = ["happy jazz piano with upbeat rhythm"]
wav = model.generate(descriptions)
torchaudio.save("music.wav", wav[0].cpu(), sample_rate=32000)

from diffusers import StableDiffusionPipeline
import torch

pipe = StableDiffusionPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5",
    torch_dtype=torch.float16
)

# Memory optimizations
pipe.enable_attention_slicing()           # Attention slicing
pipe.enable_vae_slicing()                 # VAE slicing
pipe.enable_model_cpu_offload()           # CPU offload

# xformers acceleration (if installed)
try:
    pipe.enable_xformers_memory_efficient_attention()
    print("xformers enabled")
except:
    print("xformers not found, using default attention")

import json
import urllib.request

def queue_prompt(prompt_workflow, server_address="127.0.0.1:8188"):
    """Execute workflow via ComfyUI API"""
    p = {"prompt": prompt_workflow}
    data = json.dumps(p).encode("utf-8")
    req = urllib.request.Request(
        f"http://{server_address}/prompt",
        data=data,
        headers={"Content-Type": "application/json"}
    )
    with urllib.request.urlopen(req) as response:
        return json.loads(response.read())

# ComfyUI workflow (JSON format)
workflow = {
    "1": {
        "class_type": "CheckpointLoaderSimple",
        "inputs": {"ckpt_name": "v1-5-pruned-emaonly.ckpt"}
    },
    "2": {
        "class_type": "CLIPTextEncode",
        "inputs": {
            "text": "a beautiful sunset over mountains",
            "clip": ["1", 1]
        }
    },
    "3": {
        "class_type": "KSampler",
        "inputs": {
            "model": ["1", 0],
            "positive": ["2", 0],
            "negative": ["4", 0],
            "latent_image": ["5", 0],
            "seed": 42,
            "steps": 30,
            "cfg": 7.5,
            "sampler_name": "euler",
            "scheduler": "karras",
            "denoise": 1.0
        }
    }
}

result = queue_prompt(workflow)
print(f"Prompt ID: {result['prompt_id']}")

from diffusers import OnnxStableDiffusionPipeline

# ONNX Runtime inference (works on CPU and GPU)
pipe = OnnxStableDiffusionPipeline.from_pretrained(
    "runwayml/stable-diffusion-v1-5",
    revision="onnx",
    provider="CUDAExecutionProvider",
)

image = pipe("a mountain lake at dawn").images[0]