File size: 2,939 Bytes
b748322 fb6d062 b748322 acf2d5f fb6d062 acf2d5f b748322 2393f58 9b4d4c0 b748322 9b4d4c0 fecd4a4 b748322 31523d1 b748322 f28bd15 b748322 9b4d4c0 b748322 9b4d4c0 b748322 2393f58 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
import os
from typing import Any, Dict
from diffusers import FluxPipeline, FluxTransformer2DModel, AutoencoderKL, TorchAoConfig
from PIL import Image
import torch
IS_COMPILE = False
if IS_COMPILE:
import torch._dynamo
torch._dynamo.config.suppress_errors = True
#from huggingface_inference_toolkit.logging import logger
def compile_pipeline(pipe) -> Any:
pipe.transformer.to(memory_format=torch.channels_last)
pipe.transformer = torch.compile(pipe.transformer, mode="default", fullgraph=False, dynamic=False, backend="inductor")
#pipe.vae.to(memory_format=torch.channels_last)
#pipe.vae = torch.compile(pipe.vae, mode="reduce-overhead", fullgraph=False, dynamic=False, backend="inductor")
return pipe
class EndpointHandler:
def __init__(self, path=""):
repo_id = "camenduru/FLUX.1-dev-diffusers"
#repo_id = "NoMoreCopyright/FLUX.1-dev-test"
dtype = torch.bfloat16
quantization_config = TorchAoConfig("int8dq")
vae = AutoencoderKL.from_pretrained(repo_id, subfolder="vae", torch_dtype=dtype)
#transformer = FluxTransformer2DModel.from_pretrained(repo_id, subfolder="transformer", torch_dtype=dtype, quantization_config=quantization_config).to("cuda")
self.pipeline = FluxPipeline.from_pretrained(repo_id, vae=vae, torch_dtype=dtype, quantization_config=quantization_config)
self.pipeline.transformer.fuse_qkv_projections()
self.pipeline.vae.fuse_qkv_projections()
if IS_COMPILE: self.pipeline = compile_pipeline(self.pipeline)
self.pipeline.to("cuda")
@torch.inference_mode()
def __call__(self, data: Dict[str, Any]) -> Image.Image:
#logger.info(f"Received incoming request with {data=}")
if "inputs" in data and isinstance(data["inputs"], str):
prompt = data.pop("inputs")
elif "prompt" in data and isinstance(data["prompt"], str):
prompt = data.pop("prompt")
else:
raise ValueError(
"Provided input body must contain either the key `inputs` or `prompt` with the"
" prompt to use for the image generation, and it needs to be a non-empty string."
)
parameters = data.pop("parameters", {})
num_inference_steps = parameters.get("num_inference_steps", 28)
width = parameters.get("width", 1024)
height = parameters.get("height", 1024)
guidance_scale = parameters.get("guidance_scale", 3.5)
# seed generator (seed cannot be provided as is but via a generator)
seed = parameters.get("seed", 0)
generator = torch.manual_seed(seed)
return self.pipeline( # type: ignore
prompt,
height=height,
width=width,
guidance_scale=guidance_scale,
num_inference_steps=num_inference_steps,
generator=generator,
output_type="pil",
).images[0]
|