src/pipeline.py

import os
from typing import TypeAlias

import torch
from PIL.Image import Image
from diffusers import FluxPipeline, FluxTransformer2DModel, AutoencoderKL, AutoencoderTiny
from huggingface_hub.constants import HF_HUB_CACHE
from pipelines.models import TextToImageRequest
from torch import Generator
from torchao.quantization import quantize_, int8_weight_only
from transformers import T5EncoderModel, CLIPTextModel, logging
from functools import partial
from para_attn.first_block_cache.diffusers_adapters import apply_cache_on_pipe

my_partial_compile = partial(torch.compile, mode="max-autotune")
Pipeline: TypeAlias = FluxPipeline
torch.backends.cudnn.benchmark = True
torch.backends.cudnn.benchmark = True
torch._inductor.config.conv_1x1_as_mm = True
torch._inductor.config.coordinate_descent_tuning = True
torch._inductor.config.epilogue_fusion = False
torch._inductor.config.coordinate_descent_check_all_directions = True

os.environ['PYTORCH_CUDA_ALLOC_CONF']="expandable_segments:True"

CHECKPOINT = "smartguy0505/flux.1-schnell"
REVISION = "f0fd35d9eef4536098c7e5e7c2fa60fe7c9fd1e2"

TinyVAE = "smartguy0505/tae"
TinyVAE_REV = "2ec20af4a1fc7b66b9e4823b8eaef9518687bb72"

def load_pipeline() -> Pipeline:
    path = os.path.join(HF_HUB_CACHE, "models--smartguy0505--flux.1-schnell/snapshots/f0fd35d9eef4536098c7e5e7c2fa60fe7c9fd1e2/transformer")
    transformer = FluxTransformer2DModel.from_pretrained(
            path,
            use_safetensors=False,
            local_files_only=True,
            torch_dtype=torch.bfloat16)
    vae = AutoencoderTiny.from_pretrained(
            TinyVAE,
            revision=TinyVAE_REV,
            local_files_only=True,
            torch_dtype=torch.bfloat16)
    pipeline = FluxPipeline.from_pretrained(
        CHECKPOINT,
        revision=REVISION,
        transformer=transformer,
        vae=vae,
        local_files_only=True,
        torch_dtype=torch.bfloat16,
    ).to("cuda")
    pipeline.to(memory_format=torch.channels_last)
    pipeline.vae = my_partial_compile(pipeline.vae)
    apply_cache_on_pipe(pipeline, residual_diff_threshold=0.64)
    with torch.inference_mode():
        for _ in range(2):
            pipeline("cats running on a road with a dog chasing", num_inference_steps=4)
    return pipeline

@torch.inference_mode()
def infer(request: TextToImageRequest, pipeline: Pipeline, generator: torch.Generator) -> Image:
    return pipeline(
        request.prompt,
        generator=generator,
        guidance_scale=0.0,
        num_inference_steps=4,
        max_sequence_length=256,
        height=request.height,
        width=request.width,
    ).images[0]