src/pipeline.py

import gc
import os
from typing import TypeAlias

import torch
from PIL.Image import Image
from diffusers import (
    FluxPipeline,
    FluxTransformer2DModel,
    AutoencoderKL,
    AutoencoderTiny,
    DiffusionPipeline,
)
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
import torch._dynamo

torch._dynamo.config.suppress_errors = True

Pipeline: TypeAlias = FluxPipeline

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"
os.environ["TOKENIZERS_PARALLELISM"] = "True"
CHECKPOINT = "jokerbit/flux.1-schnell-Robert-int8wo"
REVISION = "5ef0012f11a863e5111ec56540302a023bc8587b"


def load_pipeline() -> Pipeline:
    path = os.path.join(
        HF_HUB_CACHE,
        "models--jokerbit--flux.1-schnell-Robert-int8wo/snapshots/5ef0012f11a863e5111ec56540302a023bc8587b/transformer",
    )
    transformer = FluxTransformer2DModel.from_pretrained(
        path, use_safetensors=False, local_files_only=True, torch_dtype=torch.bfloat16
    )

    pipeline = FluxPipeline.from_pretrained(
        CHECKPOINT,
        revision=REVISION,
        transformer=transformer,
        local_files_only=True,
        torch_dtype=torch.bfloat16,
    ).to("cuda")

    pipeline.to(memory_format=torch.channels_last)
    pipeline.transformer = torch.compile(pipeline.transformer, mode="max-autotune")
    quantize_(pipeline.vae, int8_weight_only())
    pipeline.vae = torch.compile(pipeline.vae, mode="max-autotune")

    with torch.no_grad():
        for _ in range(5):
            pipeline(
                prompt="onomancy, aftergo, spirantic, Platyhelmia, modificator, drupaceous, jobbernowl, hereness",
                width=1024,
                height=1024,
                guidance_scale=0.0,
                num_inference_steps=4,
                max_sequence_length=256,
            )
    torch.cuda.empty_cache()
    return pipeline


@torch.no_grad()
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]