10 - FLUX与新一代架构¶

import torch
import torch.nn as nn
import torch.nn.functional as F
from einops import rearrange
import math

class MMDiTBlock(nn.Module):  # 继承nn.Module定义网络层
    """MM-DiT双流注意力块

    双流设计：img/txt各自独立Norm/MLP参数，
    但注意力时拼接QKV实现双向信息交互。
    """
    def __init__(self, dim=3072, heads=24, mlp_ratio=4.0):
        super().__init__()  # super()调用父类方法
        self.heads = heads
        self.head_dim = dim // heads

        # ====== 图像流参数 ======
        self.img_norm1 = nn.LayerNorm(dim, elementwise_affine=False)
        self.img_qkv = nn.Linear(dim, dim * 3)
        self.img_out = nn.Linear(dim, dim)
        self.img_norm2 = nn.LayerNorm(dim, elementwise_affine=False)
        self.img_mlp = nn.Sequential(
            nn.Linear(dim, int(dim * mlp_ratio)),
            nn.GELU(approximate="tanh"),
            nn.Linear(int(dim * mlp_ratio), dim))

        # ====== 文本流参数（结构相同，权重独立）======
        self.txt_norm1 = nn.LayerNorm(dim, elementwise_affine=False)
        self.txt_qkv = nn.Linear(dim, dim * 3)
        self.txt_out = nn.Linear(dim, dim)
        self.txt_norm2 = nn.LayerNorm(dim, elementwise_affine=False)
        self.txt_mlp = nn.Sequential(
            nn.Linear(dim, int(dim * mlp_ratio)),
            nn.GELU(approximate="tanh"),
            nn.Linear(int(dim * mlp_ratio), dim))

        # AdaLN-Zero: 12个调制参数
        # img: (shift_a, scale_a, gate_a, shift_m, scale_m, gate_m) × 1
        # txt: 同上 × 1
        self.adaLN = nn.Sequential(nn.SiLU(), nn.Linear(dim, dim * 12))

    def _modulate(self, x, shift, scale):
        return x * (1 + scale.unsqueeze(1)) + shift.unsqueeze(1)  # unsqueeze增加一个维度

    def forward(self, img, txt, cond):
        """
        img: [B, N_img, D]  图像patch序列
        txt: [B, N_txt, D]  文本token序列
        cond: [B, D]        时间步+pooled文本条件
        """
        # chunk(12)将条件向量切分为12个调制参数：每对(shift,scale,gate)控制norm/attn/mlp，图像和文本各6个
        m = self.adaLN(cond).chunk(12, dim=-1)

        # ---- Joint Attention ----
        img_n = self._modulate(self.img_norm1(img), m[0], m[1])
        txt_n = self._modulate(self.txt_norm1(txt), m[6], m[7])
        iq, ik, iv = self.img_qkv(img_n).chunk(3, dim=-1)
        tq, tk, tv = self.txt_qkv(txt_n).chunk(3, dim=-1)

        # 拼接Q/K/V做联合注意力
        q = torch.cat([iq, tq], dim=1)  # torch.cat沿已有维度拼接张量
        k = torch.cat([ik, tk], dim=1)
        v = torch.cat([iv, tv], dim=1)

        q = rearrange(q, 'b n (h d) -> b h n d', h=self.heads)
        k = rearrange(k, 'b n (h d) -> b h n d', h=self.heads)
        v = rearrange(v, 'b n (h d) -> b h n d', h=self.heads)
        attn = F.scaled_dot_product_attention(q, k, v)  # Flash Attention  # F.xxx PyTorch函数式API
        attn = rearrange(attn, 'b h n d -> b n (h d)')

        # 拆分回img和txt，各自output projection + gate
        N = img.shape[1]
        img = img + m[2].unsqueeze(1) * self.img_out(attn[:, :N])
        txt = txt + m[8].unsqueeze(1) * self.txt_out(attn[:, N:])

        # ---- MLP (各自独立) ----
        img = img + m[5].unsqueeze(1) * self.img_mlp(
            self._modulate(self.img_norm2(img), m[3], m[4]))
        txt = txt + m[11].unsqueeze(1) * self.txt_mlp(
            self._modulate(self.txt_norm2(txt), m[9], m[10]))
        return img, txt

import torch

def get_2d_rope_freqs(height, width, dim, theta=10000.0):
    """生成2D RoPE复数频率矩阵

    Returns: [H*W, dim//2] 复数频率
    """
    quarter_dim = dim // 4
    freqs = 1.0 / (theta ** (torch.arange(0, quarter_dim).float() / quarter_dim))

    # 行频率 → 前半维度
    h_freqs = torch.outer(torch.arange(height).float(), freqs)  # [H, qd]
    h_freqs = h_freqs.unsqueeze(1).expand(-1, width, -1).reshape(-1, quarter_dim)  # 重塑张量形状

    # 列频率 → 后半维度
    w_freqs = torch.outer(torch.arange(width).float(), freqs)  # [W, qd]
    w_freqs = w_freqs.unsqueeze(0).expand(height, -1, -1).reshape(-1, quarter_dim)

    freqs_2d = torch.cat([h_freqs, w_freqs], dim=-1)  # [H*W, dim//2]
    return torch.polar(torch.ones_like(freqs_2d), freqs_2d)

def apply_2d_rope(x, freqs):
    """x: [B, heads, N, head_dim] → 旋转后同形状"""
    # reshape将最后一维两两配对为(实部,虚部)，view_as_complex转为复数，乘以频率复数实现旋转位置编码
    xc = torch.view_as_complex(x.float().reshape(*x.shape[:-1], -1, 2))
    rotated = xc * freqs.unsqueeze(0).unsqueeze(0)
    return torch.view_as_real(rotated).reshape_as(x).to(x.dtype)

# 示例：1024×1024图像, patch=2, VAE下采样8x → 64×64 patches
freqs = get_2d_rope_freqs(height=64, width=64, dim=128)
print(f"2D RoPE freqs: {freqs.shape}")  # [4096, 64]

class SingleDiTBlock(nn.Module):
    """单流DiT块（第二阶段，img+txt共享全部参数）"""
    def __init__(self, dim=3072, heads=24, mlp_ratio=4.0):
        super().__init__()
        self.heads = heads
        self.norm1 = nn.LayerNorm(dim, elementwise_affine=False)
        self.qkv = nn.Linear(dim, dim * 3)
        self.proj = nn.Linear(dim, dim)
        self.norm2 = nn.LayerNorm(dim, elementwise_affine=False)
        self.mlp = nn.Sequential(
            nn.Linear(dim, int(dim * mlp_ratio)),
            nn.GELU(approximate="tanh"),
            nn.Linear(int(dim * mlp_ratio), dim))
        self.adaLN = nn.Sequential(nn.SiLU(), nn.Linear(dim, dim * 6))

    def forward(self, x, cond):
        """x: [B, N_img+N_txt, D] 拼接后的统一序列"""
        s_a, sc_a, g_a, s_m, sc_m, g_m = self.adaLN(cond).chunk(6, dim=-1)
        xn = self.norm1(x) * (1 + sc_a.unsqueeze(1)) + s_a.unsqueeze(1)
        # 列表推导：qkv线性层输出chunk(3)拆为Q/K/V，每个用rearrange从(B,N,heads*d)重排为(B,heads,N,d)多头格式
        q, k, v = [rearrange(t, 'b n (h d) -> b h n d', h=self.heads)
                    for t in self.qkv(xn).chunk(3, dim=-1)]
        attn = rearrange(F.scaled_dot_product_attention(q, k, v), 'b h n d -> b n (h d)')
        x = x + g_a.unsqueeze(1) * self.proj(attn)
        xn2 = self.norm2(x) * (1 + sc_m.unsqueeze(1)) + s_m.unsqueeze(1)
        x = x + g_m.unsqueeze(1) * self.mlp(xn2)
        return x

def timestep_embedding(t, dim, max_period=10000):
    """正弦时间步嵌入"""
    half = dim // 2
    freqs = torch.exp(-math.log(max_period) * torch.arange(half, device=t.device) / half)
    args = t[:, None].float() * freqs[None]
    return torch.cat([torch.cos(args), torch.sin(args)], dim=-1)

configs = {
    "SDXL":         {"steps": 25, "cfg": 7.0,  "scheduler": "DPM++ 2M Karras"},
    "SD3-Medium":   {"steps": 28, "cfg": 7.0,  "scheduler": "Euler + logit-normal"},
    "FLUX.1-dev":   {"steps": 28, "cfg": 3.5,  "scheduler": "Euler"},
    "FLUX.1-schnell": {"steps": 4, "cfg": 0.0, "scheduler": "Euler (无CFG)"},
}

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader
from torchvision import datasets, transforms

class VelocityNet(nn.Module):
    """简化的速度场网络（用于小图像验证）"""
    def __init__(self, ch=1, size=28, hidden=256):
        super().__init__()
        d = ch * size * size
        self.net = nn.Sequential(
            nn.Linear(d + 1, hidden), nn.SiLU(),
            nn.Linear(hidden, hidden), nn.SiLU(),
            nn.Linear(hidden, hidden), nn.SiLU(),
            nn.Linear(hidden, d))
        self.shape = (ch, size, size)

    def forward(self, x_t, t):
        B = x_t.shape[0]
        x_flat = x_t.reshape(B, -1)
        return self.net(torch.cat([x_flat, t.reshape(B,1)], -1)).reshape(B, *self.shape)

class RectifiedFlowTrainer:
    """Rectified Flow 训练与采样"""
    def __init__(self, model, lr=2e-4, device="cuda"):
        self.model = model.to(device)  # 移至GPU/CPU
        self.opt = torch.optim.AdamW(model.parameters(), lr=lr)
        self.device = device

    def train_step(self, x_1):
        """x_1: [B,C,H,W] 真实数据（对应t=1）"""
        B = x_1.shape[0]
        x_0 = torch.randn_like(x_1)                          # 噪声
        t = torch.rand(B, 1, device=self.device)
        t_e = t.unsqueeze(-1).unsqueeze(-1)                   # [B,1,1,1]
        x_t = (1 - t_e) * x_0 + t_e * x_1                   # 线性插值
        target = x_1 - x_0                                     # 目标速度
        loss = F.mse_loss(self.model(x_t, t), target)
        self.opt.zero_grad(); loss.backward(); self.opt.step()  # 清零梯度  # 反向传播计算梯度
        return loss.item()  # 将单元素张量转为Python数值

    @torch.no_grad()  # 禁用梯度计算，节省内存
    def sample_ode(self, batch_size, num_steps=20):
        """ODE Euler采样（确定性）"""
        self.model.eval()  # eval()评估模式
        x = torch.randn(batch_size, *self.model.shape, device=self.device)
        dt = 1.0 / num_steps
        for i in range(num_steps):
            t = torch.full((batch_size, 1), i * dt, device=self.device)
            x = x + dt * self.model(x, t)
        self.model.train()
        return x.clamp(-1, 1)

    @torch.no_grad()
    def sample_sde(self, batch_size, num_steps=20, sigma=0.01):
        """SDE采样（带随机噪声修正）"""
        self.model.eval()
        x = torch.randn(batch_size, *self.model.shape, device=self.device)
        dt = 1.0 / num_steps
        for i in range(num_steps):
            t = torch.full((batch_size, 1), i * dt, device=self.device)
            noise = torch.randn_like(x) * sigma * dt**0.5 if i < num_steps - 1 else 0
            x = x + dt * self.model(x, t) + noise
        self.model.train()
        return x.clamp(-1, 1)

def train_rectified_flow():
    device = "cuda" if torch.cuda.is_available() else "cpu"
    transform = transforms.Compose([
        transforms.ToTensor(), transforms.Normalize([0.5], [0.5])])
    dl = DataLoader(datasets.MNIST("./data", train=True, download=True,  # DataLoader批量加载数据
        transform=transform), batch_size=128, shuffle=True)

    trainer = RectifiedFlowTrainer(VelocityNet(), device=device)
    for epoch in range(10):
        avg_loss = sum(trainer.train_step(x.to(device)) for x,_ in dl) / len(dl)
        print(f"Epoch {epoch}: loss={avg_loss:.4f}")
        if epoch % 5 == 0:
            samples = trainer.sample_ode(16, num_steps=10)
            print(f"  ODE 10步: range [{samples.min():.2f}, {samples.max():.2f}]")
            samples_sde = trainer.sample_sde(16, num_steps=10, sigma=0.05)
            print(f"  SDE 10步: range [{samples_sde.min():.2f}, {samples_sde.max():.2f}]")

import torch, torch.nn as nn, copy

class ZeroConv(nn.Module):
    """零初始化1×1卷积"""
    def __init__(self, ch):
        super().__init__()
        self.conv = nn.Conv2d(ch, ch, kernel_size=1)
        nn.init.zeros_(self.conv.weight)
        nn.init.zeros_(self.conv.bias)
    def forward(self, x):
        return self.conv(x)

class ControlNet(nn.Module):
    """ControlNet for U-Net based models (SD1.5/SDXL, 简化示意)"""
    def __init__(self, base_unet, cond_channels=3):
        super().__init__()
        # 条件编码器：将条件图（如Canny边缘）编码为latent尺寸特征
        self.cond_encoder = nn.Sequential(
            nn.Conv2d(cond_channels, 64, 3, stride=2, padding=1), nn.SiLU(),
            nn.Conv2d(64, 128, 3, stride=2, padding=1), nn.SiLU(),
            nn.Conv2d(128, 256, 3, stride=2, padding=1), nn.SiLU(),
            nn.Conv2d(256, 320, 3, stride=1, padding=1))

        # 复制encoder各层 + zero-conv连接
        encoder_channels = [320, 640, 1280, 1280]
        self.ctrl_blocks = nn.ModuleList()
        self.zero_convs = nn.ModuleList()
        for ch in encoder_channels:
            self.ctrl_blocks.append(nn.Sequential(
                nn.Conv2d(ch, ch, 3, padding=1),
                nn.GroupNorm(32, ch), nn.SiLU()))
            self.zero_convs.append(ZeroConv(ch))

        # 冻结原始U-Net全部参数
        for p in base_unet.parameters():
            p.requires_grad = False

    def forward(self, cond_image):
        """返回各层控制信号（加到decoder对应层）"""
        h = self.cond_encoder(cond_image)
        signals = []
        for block, zc in zip(self.ctrl_blocks, self.zero_convs):  # zip按位置配对
            h = block(h)
            signals.append(zc(h))
        return signals

# 关键训练细节：
# 1. 冻结原始U-Net全部参数
# 2. 只训练：条件编码器 + ControlNet副本 + zero-convolution
# 3. 学习率1e-5（较小，从预训练权重开始）
# 4. 数据：成对的(图像, 条件图)，如(照片, Canny边缘)
# 5. 数据量仅需原始模型训练量的5-10%

class FLUXControlNet(nn.Module):
    """FLUX ControlNet适配（简化版）"""
    def __init__(self, flux_model, n_ctrl=8, dim=3072, cond_ch=3):
        super().__init__()
        # 条件编码器：图像条件 → token序列
        self.cond_enc = nn.Sequential(
            nn.Conv2d(cond_ch, 64, 3, 2, 1), nn.SiLU(),
            nn.Conv2d(64, 128, 3, 2, 1), nn.SiLU(),
            nn.Conv2d(128, 256, 3, 2, 1), nn.SiLU(),
            nn.Conv2d(256, dim // 4, 1))
        self.proj = nn.Linear(dim // 4 * 4, dim)  # patchify后投影

        # 复制前N个双流MM-DiT块
        self.ctrl_blocks = nn.ModuleList([
            copy.deepcopy(flux_model.double_blocks[i]) for i in range(n_ctrl)])
        self.zero_lins = nn.ModuleList([
            ZeroLinear(dim) for _ in range(n_ctrl)])

        for p in flux_model.parameters():
            p.requires_grad = False

    def forward(self, img_tok, txt_tok, cond, cond_img, flux):
        cond_tok = self.proj(rearrange(
            self.cond_enc(cond_img),
            'b c (h p1) (w p2) -> b (h w) (c p1 p2)', p1=2, p2=2))
        ci, ct = img_tok.clone(), txt_tok.clone()
        for i, main_blk in enumerate(flux.double_blocks):  # enumerate同时获取索引和元素
            img_tok, txt_tok = main_blk(img_tok, txt_tok, cond)
            if i < len(self.ctrl_blocks):
                ci, ct = self.ctrl_blocks[i](ci + cond_tok, ct, cond)
                img_tok = img_tok + self.zero_lins[i](ci)  # zero-linear注入
        return img_tok, txt_tok

class ZeroLinear(nn.Module):
    """零初始化线性投影"""
    def __init__(self, d):
        super().__init__()
        self.lin = nn.Linear(d, d)
        nn.init.zeros_(self.lin.weight); nn.init.zeros_(self.lin.bias)
    def forward(self, x): return self.lin(x)

class DecoupledCrossAttn(nn.Module):
    """解耦交叉注意力：分别处理文本和图像条件"""
    def __init__(self, dim=768, heads=12):
        super().__init__()
        self.heads, self.hd = heads, dim // heads
        # 原始文本cross-attention (冻结)
        self.q = nn.Linear(dim, dim)
        self.k_text = nn.Linear(dim, dim)
        self.v_text = nn.Linear(dim, dim)
        self.out_text = nn.Linear(dim, dim)
        # 新增图像cross-attention (可训练)
        self.k_img = nn.Linear(dim, dim)
        self.v_img = nn.Linear(dim, dim)
        self.out_img = nn.Linear(dim, dim)
        self.ip_scale = nn.Parameter(torch.tensor(1.0))

    def forward(self, x, text_emb, img_tokens=None):
        B, N, D = x.shape
        def to_heads(t, s):
            return t.view(B, s, self.heads, self.hd).transpose(1, 2)
        q = to_heads(self.q(x), N)
        # 文本cross-attention
        kt = to_heads(self.k_text(text_emb), text_emb.shape[1])
        vt = to_heads(self.v_text(text_emb), text_emb.shape[1])
        out = self.out_text(
            F.scaled_dot_product_attention(q, kt, vt).transpose(1,2).reshape(B,N,D))
        # 图像cross-attention（解耦路径）
        if img_tokens is not None:
            ki = to_heads(self.k_img(img_tokens), img_tokens.shape[1])
            vi = to_heads(self.v_img(img_tokens), img_tokens.shape[1])
            img_out = self.out_img(
                F.scaled_dot_product_attention(q, ki, vi).transpose(1,2).reshape(B,N,D))
            out = out + self.ip_scale * img_out
        return out

import torch
from diffusers import FluxPipeline, BitsAndBytesConfig

# 方式1: NF4量化加载 (~8GB VRAM)
pipe = FluxPipeline.from_pretrained(
    "black-forest-labs/FLUX.1-dev",
    quantization_config=BitsAndBytesConfig(
        load_in_4bit=True, bnb_4bit_quant_type="nf4",
        bnb_4bit_compute_dtype=torch.bfloat16),
    torch_dtype=torch.bfloat16)

# 方式2: CPU offload (更通用)
# pipe = FluxPipeline.from_pretrained("black-forest-labs/FLUX.1-dev",
#     torch_dtype=torch.bfloat16)
# pipe.enable_model_cpu_offload()

image = pipe(
    "A serene mountain landscape at sunset, crystal-clear lake, photorealistic, 8K",
    height=1024, width=1024,
    guidance_scale=3.5,       # FLUX推荐低CFG
    num_inference_steps=28,
    max_sequence_length=512,  # T5最大token长度
    generator=torch.Generator("cuda").manual_seed(42)
).images[0]
image.save("flux_dev.png")

pipe = FluxPipeline.from_pretrained(
    "black-forest-labs/FLUX.1-schnell", torch_dtype=torch.bfloat16)
pipe.enable_model_cpu_offload()

image = pipe(
    "A cyberpunk cityscape, neon lights, rain reflections",
    height=1024, width=1024,
    guidance_scale=0.0,    # schnell不需CFG（蒸馏已内化）
    num_inference_steps=4,  # 仅4步！
    generator=torch.Generator("cuda").manual_seed(42)
).images[0]
image.save("flux_schnell.png")

# 2D RoPE支持任意分辨率
for name, (h, w) in [("横屏16:9", (576, 1024)), ("竖屏9:16", (1024, 576))]:
    img = pipe("cinematic landscape", height=h, width=w,
               guidance_scale=3.5, num_inference_steps=28).images[0]
    img.save(f"flux_{name}.png")

# 加载LoRA微调权重
pipe.load_lora_weights("your-user/flux-lora", weight_name="lora.safetensors")
pipe.fuse_lora(lora_scale=0.8)  # 可调节LoRA强度
# 卸载: pipe.unfuse_lora(); pipe.unload_lora_weights()

import torch, cv2, numpy as np
from PIL import Image
from diffusers import StableDiffusionXLControlNetPipeline, ControlNetModel
from diffusers.utils import load_image

def get_canny(path, lo=100, hi=200):
    """提取Canny边缘图"""
    edges = cv2.Canny(np.array(load_image(path)), lo, hi)  # np.array创建NumPy数组
    return Image.fromarray(np.stack([edges]*3, axis=-1))

controlnet = ControlNetModel.from_pretrained(
    "diffusers/controlnet-canny-sdxl-1.0",
    torch_dtype=torch.float16, variant="fp16")
pipe = StableDiffusionXLControlNetPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    controlnet=controlnet, torch_dtype=torch.float16)
pipe.enable_model_cpu_offload()

image = pipe(
    "A beautiful garden, oil painting style, masterpiece",
    negative_prompt="blurry, low quality",
    image=get_canny("reference.jpg"),
    controlnet_conditioning_scale=0.7,  # 条件强度
    guidance_scale=7.5, num_inference_steps=30,
    generator=torch.Generator("cuda").manual_seed(42)
).images[0]
image.save("controlnet_canny.png")

from transformers import DPTForDepthEstimation, DPTImageProcessor

def get_depth(path):
    """使用DPT估计深度图"""
    proc = DPTImageProcessor.from_pretrained("Intel/dpt-large")
    model = DPTForDepthEstimation.from_pretrained(
        "Intel/dpt-large", torch_dtype=torch.float16).to("cuda")
    with torch.no_grad():
        d = model(**proc(images=load_image(path), return_tensors="pt").to("cuda")
            ).predicted_depth.squeeze().cpu().numpy()  # squeeze压缩维度
    d = ((d - d.min()) / (d.max() - d.min()) * 255).astype(np.uint8)
    return Image.fromarray(np.stack([d]*3, -1)).resize((1024, 1024))

controlnet = ControlNetModel.from_pretrained(
    "diffusers/controlnet-depth-sdxl-1.0", torch_dtype=torch.float16, variant="fp16")
pipe = StableDiffusionXLControlNetPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    controlnet=controlnet, torch_dtype=torch.float16)
pipe.enable_model_cpu_offload()

image = pipe("Futuristic city, cyberpunk, neon", image=get_depth("scene.jpg"),
    controlnet_conditioning_scale=0.6, guidance_scale=7.5,
    num_inference_steps=30).images[0]
image.save("controlnet_depth.png")

from controlnet_aux import OpenposeDetector

# 提取人体姿态骨架
pose_img = OpenposeDetector.from_pretrained("lllyasviel/ControlNet")(
    load_image("person.jpg"), detect_resolution=1024, image_resolution=1024)

# MultiControlNet: 多条件组合
controlnets = [
    ControlNetModel.from_pretrained("diffusers/controlnet-canny-sdxl-1.0",
        torch_dtype=torch.float16, variant="fp16"),
    ControlNetModel.from_pretrained("diffusers/controlnet-depth-sdxl-1.0",
        torch_dtype=torch.float16, variant="fp16")]

pipe = StableDiffusionXLControlNetPipeline.from_pretrained(
    "stabilityai/stable-diffusion-xl-base-1.0",
    controlnet=controlnets, torch_dtype=torch.float16)
pipe.enable_model_cpu_offload()

image = pipe("Beautiful forest scene, photorealistic",
    image=[canny_image, depth_image],          # 多个条件图
    controlnet_conditioning_scale=[0.6, 0.4],  # 各自权重
    guidance_scale=7.5, num_inference_steps=30).images[0]

from diffusers import FluxControlNetPipeline, FluxControlNetModel

controlnet = FluxControlNetModel.from_pretrained(
    "InstantX/FLUX.1-dev-Controlnet-Canny", torch_dtype=torch.bfloat16)
pipe = FluxControlNetPipeline.from_pretrained(
    "black-forest-labs/FLUX.1-dev",
    controlnet=controlnet, torch_dtype=torch.bfloat16)
pipe.enable_model_cpu_offload()

image = pipe(
    "A detailed watercolor painting of an ancient castle",
    control_image=canny_image,
    controlnet_conditioning_scale=0.6,
    guidance_scale=3.5, num_inference_steps=28,
    height=1024, width=1024).images[0]
image.save("flux_controlnet.png")