Upload 2 files

handler.py

@@ -0,0 +1,79 @@
+import os
+from typing import Any, Dict
+
+from diffusers import FluxPipeline, FluxTransformer2DModel, AutoencoderKL, TorchAoConfig
+from diffusers.utils.remote_utils import remote_decode
+from PIL import Image
+import torch
+
+IS_COMPILE = True
+
+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.fuse_qkv_projections()
+    pipe.transformer.to(memory_format=torch.channels_last)
+    #pipe.transformer = torch.compile(pipe.transformer, mode="reduce-overhead", fullgraph=False, dynamic=False, backend="inductor")
+    pipe.transformer = torch.compile(pipe.transformer, mode="max-autotune", fullgraph=True, 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("int8wo")
+        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)
+        if IS_COMPILE: self.pipeline = compile_pipeline(self.pipeline)
+        self.pipeline.to("cuda")
+
+    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)
+
+        latent = self.pipeline(  # type: ignore
+            prompt,
+            height=height,
+            width=width,
+            guidance_scale=guidance_scale,
+            num_inference_steps=num_inference_steps,
+            generator=generator,
+            output_type="latent",
+        ).images
+    
+        image = remote_decode(
+            endpoint="https://whhx50ex1aryqvw6.us-east-1.aws.endpoints.huggingface.cloud/",
+            tensor=latent,
+            height=height,
+            width=width,
+            scaling_factor=0.3611,
+            shift_factor=0.1159,
+        )
+
+        return image

requirements.txt

@@ -0,0 +1,14 @@
+huggingface_hub
+torch
+torchvision
+torchao
+git+https://github.com/huggingface/diffusers
+peft
+accelerate
+transformers
+numpy
+scipy
+Pillow
+sentencepiece
+protobuf
+pytorch-lightning