Upload handler.py

handler.py

@@ -0,0 +1,61 @@
+import os
+from typing import Any, Dict
+
+from diffusers import DiffusionPipeline  # type: ignore
+from PIL.Image import Image
+import torch
+
+from huggingface_inference_toolkit.logging import logger
+
+
+class EndpointHandler:
+    def __init__(self, model_dir: str, **kwargs: Any) -> None:  # type: ignore
+        """The current `EndpointHandler` works with any FLUX.1-dev LoRA Adapter."""
+        if os.getenv("HF_TOKEN") is None:
+            raise ValueError(
+                "Since `black-forest-labs/FLUX.1-dev` is a gated model, you will need to provide a valid "
+                "`HF_TOKEN` as an environment variable for the handler to work properly."
+            )
+
+        self.pipeline = DiffusionPipeline.from_pretrained(
+            "black-forest-labs/FLUX.1-dev",
+            torch_dtype=torch.bfloat16,
+            token=os.getenv("HF_TOKEN"),
+        )
+        self.pipeline.load_lora_weights(model_dir)
+        self.pipeline.to("cuda")
+
+    def __call__(self, data: Dict[str, Any]) -> 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", 30)
+        width = parameters.get("width", 1024)
+        height = parameters.get("height", 768)
+        guidance_scale = parameters.get("guidance_scale", 3.5)
+        lora_scale = parameters.get("lora_scale", 1.0)
+
+        # 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,
+            lora_scale=lora_scale,
+            generator=generator,
+        ).images[0]