File size: 2,237 Bytes
253e509 87c555a |
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 |
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)
# 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,
).images[0] |