"""FLUX.3 image-small — layout-control canvas (private ZeroGPU demo, diffusers modular). A custom HTML/JS canvas on top of the diffusers-native FLUX.3 modular pipeline (PR #29, vendored `diffusers/`), running the **step-distilled** flow-map model (`Flux3DistilledAutoBlocks`, 16 steps, no guider / negative prompt), with its transformer block stacks **AOT-compiled via aokit** for ZeroGPU (`aokit.load` + a per-block eager fallback; set `F3_AOTI=0` to run eager). You drag bounding boxes on an aspect-ratio artboard, give each box an id + a description, write a one-line scene prompt, and FLUX.3 renders the scene with every element placed inside its box. Layout control uses FLUX.3's region format (see `integration-notes/flux3_examples.py`): the final prompt is the scene caption followed by a JSON list of regions, each `{"id", "bbox": [y0, x0, y1, x1] (normalized 0-1000), "desc"}`. """ import importlib.util import os os.environ.setdefault("TORCH_ALLOW_TF32_CUBLAS_OVERRIDE", "1") spaces = None if importlib.util.find_spec("spaces"): import spaces # before torch so ZeroGPU can patch torch import base64 import hashlib import io import json import random from pathlib import Path import gradio as gr import torch from PIL import Image from huggingface_hub import hf_hub_download, InferenceClient from diffusers import ModularPipeline # Step-distilled FLUX.3 image model + its private AOTI artifact. REPO = os.environ.get("F3_REPO", "diffusers-internal-dev/f3-small-image-distilled") AOTI_REPO = os.environ.get("F3_AOTI_REPO", "diffusers-modular/f3-small-image-distilled-aoti-sm120") AOTI_ENABLED = os.environ.get("F3_AOTI", "1") != "0" HF_TOKEN = os.environ.get("HF_TOKEN") # Header "button" links (klein-style pills). Point at the model cards; override via env # when switching to the public klein repos. LINK_DISTILLED = os.environ.get("F3_LINK_DISTILLED", "https://huggingface.co/black-forest-labs/flux-3-image-klein") LINK_BASE = os.environ.get("F3_LINK_BASE", "https://huggingface.co/black-forest-labs/flux-3-image-klein-base") LINK_CODE = os.environ.get("F3_LINK_CODE", "https://github.com/black-forest-labs/flux-3-image-klein") LINK_BOXES = os.environ.get("F3_LINK_BOXES", "https://docs.bfl.ai/flux_3/flux3_image_bounding_boxes") MAX_SEED = 2**63 - 1 DEFAULT_STEPS = int(os.environ.get("F3_STEPS", "16")) # distilled default torch.backends.cuda.matmul.allow_tf32 = True torch.backends.cudnn.allow_tf32 = True # Aspect ratio -> pixel dims (multiples of 16, ~1 MP, the model's default area). RATIOS = { "1:1": (1024, 1024), "3:2": (1216, 816), "2:3": (816, 1216), "4:3": (1152, 864), "3:4": (864, 1152), "16:9": (1344, 768), "9:16": (768, 1344), } def _eager_fallback(compiled, eager): """Call the AOTI graph; fall back to the eager forward on any shape it can't handle (keeps odd resolutions / layouts working instead of crashing).""" def forward(*args, **kwargs): try: return compiled(*args, **kwargs) except Exception: return eager(*args, **kwargs) return forward def _load_aoti(transformer, repo, token): """aokit.load the compiled stacks, then wrap each block with an eager fallback. The artifact's torch version is asserted against the runtime.""" import aokit cfg = json.loads(Path(hf_hub_download(repo, "config.json", token=token)).read_text()) if cfg.get("torch") != torch.__version__: raise RuntimeError( f"AOTI artifact built for torch {cfg.get('torch')}, runtime has {torch.__version__}" ) stack_names = list(cfg.get("stacks", {})) or [ "mode_blocks.image", "mode_blocks.image_cond", "txt_mode_blocks", "single_transformer_blocks", ] def _blocks(name): try: stack = transformer.get_submodule(name) except AttributeError: return [] return list(stack) if isinstance(stack, torch.nn.ModuleList) else [stack] eager = {name: [blk.forward for blk in _blocks(name)] for name in stack_names} aokit.load(transformer, repo) # patches block.forward -> compiled (weights stay live) wrapped = 0 for name in stack_names: for blk, eager_fwd in zip(_blocks(name), eager[name]): blk.forward = _eager_fallback(blk.forward, eager_fwd) wrapped += 1 print(f"[app] AOTI loaded from {repo}: {len(stack_names)} stacks, {wrapped} blocks wrapped " f"(torch {cfg.get('torch')}, {cfg.get('arch')})", flush=True) return cfg # ---- load the pipeline on CPU at import; place on cuda + AOTI under ZeroGPU ---- print(f"[app] loading distilled modular pipeline from {REPO} ...", flush=True) PIPE = ModularPipeline.from_pretrained(REPO, token=HF_TOKEN) PIPE.load_components(dtype={"vae": torch.float32, "default": torch.bfloat16}, token=HF_TOKEN) print("[app] components loaded on CPU.", flush=True) AOTI_LOADED = False if spaces is not None: PIPE.to("cuda") if AOTI_ENABLED: try: _load_aoti(PIPE.transformer, AOTI_REPO, HF_TOKEN) AOTI_LOADED = True except Exception as exc: # fall back to eager if the artifact is unavailable print(f"[app] AOTI unavailable, running eager: {exc!r}", flush=True) print(f"[app] pipeline on cuda; ready (aoti={AOTI_LOADED}).", flush=True) def _build_prompt(scene: str, elements: list) -> str: """Scene caption + JSON region list, FLUX.3's layout-control format.""" scene = (scene or "").strip() regions = [] for e in elements: desc = (e.get("desc") or "").strip() if not desc: continue bbox = e.get("bbox") if not (isinstance(bbox, list) and len(bbox) == 4): continue regions.append({ "id": e.get("id") or "element", "bbox": [int(round(v)) for v in bbox], "desc": desc, }) if not regions: return scene return (scene + " " if scene else "") + json.dumps(regions) REF_MAX_SIDE = 1024 # cap the reference's long side so the VAE encode stays bounded def _decode_ref(data_uri): """Decode a `data:image/...;base64,...` reference from the canvas into a capped RGB PIL image, or None.""" if not data_uri or not isinstance(data_uri, str) or "," not in data_uri: return None try: raw = base64.b64decode(data_uri.split(",", 1)[1]) img = Image.open(io.BytesIO(raw)).convert("RGB") except Exception: return None w, h = img.size s = REF_MAX_SIDE / max(w, h) if s < 1.0: img = img.resize((max(16, round(w * s)), max(16, round(h * s))), Image.LANCZOS) return img # ---- optional VLM prompt upsampling (klein-style) --------------------------- VLM_MODEL = os.environ.get("F3_VLM_MODEL", "Qwen/Qwen3-VL-235B-A22B-Instruct") _hf_client = InferenceClient(api_key=os.environ.get("INFERENCE_TOKEN") or HF_TOKEN) SYSTEM_PROMPT_TEXT_ONLY = """You are an expert prompt engineer for FLUX.3 by Black Forest Labs. Rewrite user prompts into a full scene caption while strictly preserving their core subject and intent. Guidelines: 1. Structure: Keep structured inputs structured (enhance within fields). Convert natural language to detailed paragraphs. 2. Details: Describe the subject, composition, lighting (quality, direction, color), atmosphere, textures, materials, spatial relationships, and environmental context. 3. Text in Images: Put ALL text in quotation marks, matching the prompt's language. Always provide explicit quoted text for objects that would contain text in reality (signs, labels, screens, etc.) - without it, the model generates gibberish. Output only the revised prompt and nothing else.""" SYSTEM_PROMPT_WITH_IMAGES = """You are FLUX.3 by Black Forest Labs, an image-editing expert. You turn an editing request into a complete caption of the target image: describe the reference image(s) in full with the requested edit folded in (60-120 words). Rules: - One caption only, no commentary - Use clear, analytical language (avoid "whimsical," "cascading," etc.) - Keep everything the request does not change (faces, lighting, composition) and describe it as it is - Refer to the references as , , ... in the order given ("the person from ") - Make abstractions concrete ("futuristic" -> "glowing cyan neon, metallic panels") - Keep content PG-13 Output only the final caption in plain text and nothing else.""" def _img_data_uri(img): buf = io.BytesIO() img.save(buf, format="PNG") return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode() def _upsample_prompt(prompt, image_list=None): """Rewrite the prompt into a full scene caption via a VLM; returns the original prompt on any failure so generation always proceeds.""" try: if image_list: content = [{"type": "text", "text": prompt}] for img in image_list: content.append({"type": "image_url", "image_url": {"url": _img_data_uri(img)}}) messages = [{"role": "system", "content": SYSTEM_PROMPT_WITH_IMAGES}, {"role": "user", "content": content}] else: messages = [{"role": "system", "content": SYSTEM_PROMPT_TEXT_ONLY}, {"role": "user", "content": prompt}] r = _hf_client.chat.completions.create(model=VLM_MODEL, messages=messages, max_tokens=1024) out = (r.choices[0].message.content or "").strip() return out or prompt except Exception as e: print(f"[app] prompt upsampling failed, using raw prompt: {e!r}", flush=True) return prompt def _duration(state_json, *_): try: st = json.loads(state_json) if state_json else {} w, h = RATIOS.get(st.get("aspect", "1:1"), (1024, 1024)) ref = 12 if st.get("ref_image") else 0 # i2i adds a VAE encode except Exception: w, h, ref = 1024, 1024, 0 px = w * h # Distilled flow-map model: single forward per step (no CFG), so cheaper/step. return int(min(300, 40 + ref + DEFAULT_STEPS * 1.6 * max(1.0, px / (1024 * 1024)))) def _generate(state_json): try: st = json.loads(state_json) if state_json else {} except Exception as e: raise gr.Error(f"Bad canvas state: {e}") elements = st.get("elements") or [] scene = st.get("scene") or "" aspect = st.get("aspect") or "1:1" # `generate` may have pre-upsampled the prompt (off-GPU); use it if present. full_prompt = st.get("_final_prompt") or _build_prompt(scene, elements) if not full_prompt.strip(): raise gr.Error("Add at least a scene prompt or one described element.") seed = st.get("seed") seed = int(seed) if isinstance(seed, (int, float)) and seed >= 0 else random.randint(0, MAX_SEED) ref = _decode_ref(st.get("ref_image")) # Distilled flow-map model has no guider / guidance. kwargs = dict( prompt=full_prompt, num_inference_steps=DEFAULT_STEPS, generator=torch.Generator().manual_seed(seed), output="images", ) if ref is not None: # i2i: pass the reference; the model keeps its aspect (like the PR examples), # so don't force height/width. kwargs["image"] = ref w, h = ref.size else: w, h = RATIOS.get(aspect, (1024, 1024)) kwargs["height"], kwargs["width"] = h, w image = PIPE(**kwargs)[0] if torch.cuda.is_available(): torch.cuda.synchronize() w, h = image.size buf = io.BytesIO() image.save(buf, format="WEBP", quality=92) data_uri = "data:image/webp;base64," + base64.b64encode(buf.getvalue()).decode() print(f"[app] generate {w}x{h} steps={DEFAULT_STEPS} seed={seed} i2i={ref is not None} " f"elements={len(elements)} prompt_len={len(full_prompt)}", flush=True) return json.dumps({ "image": data_uri, "aspect": aspect, "w": w, "h": h, "seed": seed, "i2i": ref is not None, "elements": [{"id": e.get("id"), "bbox": e.get("bbox"), "color": e.get("color")} for e in elements if e.get("bbox")], "prompt": full_prompt, }) _generate_gpu = spaces.GPU(_generate, duration=_duration) if spaces is not None else _generate # ---- lazy example cache (gradio cache_examples="lazy" style) ----------------- # A deterministic request (fixed seed, no reference image) is cached to disk on # the first generation, so later identical requests - i.e. any user opening the # same example - are served from the cache with no GPU. Custom/edited generations # use a random seed (seed < 0) and are never cached. CACHE_DIR = os.environ.get( "F3_EXAMPLE_CACHE", os.path.join(os.path.dirname(os.path.abspath(__file__)), "example_cache")) os.makedirs(CACHE_DIR, exist_ok=True) _CACHE_MEM = {} def _has_boxes(st): return any((e.get("desc") or "").strip() for e in (st.get("elements") or [])) def _cache_key(st): if st.get("ref_image"): return None # i2i is not a fixed example if st.get("prompt_upsampling") and not _has_boxes(st): return None # VLM upsampling is non-deterministic -> don't cache seed = st.get("seed") if not (isinstance(seed, (int, float)) and seed >= 0): return None # random seed -> not a reproducible example full_prompt = _build_prompt(st.get("scene") or "", st.get("elements") or []) if not full_prompt.strip(): return None sig = f"{REPO}|{DEFAULT_STEPS}|{st.get('aspect') or '1:1'}|{int(seed)}|{full_prompt}" return hashlib.sha256(sig.encode("utf-8")).hexdigest()[:24] def _cache_get(key): if key in _CACHE_MEM: return _CACHE_MEM[key] fp = os.path.join(CACHE_DIR, key + ".json") if os.path.exists(fp): try: data = Path(fp).read_text() _CACHE_MEM[key] = data return data except Exception: return None return None def _cache_put(key, result): _CACHE_MEM[key] = result try: tmp = os.path.join(CACHE_DIR, key + ".json.tmp") Path(tmp).write_text(result) os.replace(tmp, os.path.join(CACHE_DIR, key + ".json")) except Exception as e: print(f"[app] example cache write failed: {e!r}", flush=True) def generate(state_json): try: st = json.loads(state_json) if state_json else {} except Exception: st = {} # Optional VLM prompt upsampling, done OFF-GPU before allocation. A prompt that # carries a box table (layout) is sent as-is, matching the klein space. if st.get("prompt_upsampling") and not _has_boxes(st): base = _build_prompt(st.get("scene") or "", st.get("elements") or []) if base.strip(): ref = _decode_ref(st.get("ref_image")) up = _upsample_prompt(base, [ref] if ref is not None else None) if up and up.strip(): st["_final_prompt"] = up state_json = json.dumps(st) print(f"[app] upsampled prompt -> {up[:90]!r}", flush=True) key = _cache_key(st) if key: cached = _cache_get(key) if cached is not None: print(f"[app] example cache HIT {key}", flush=True) return cached result = _generate_gpu(state_json) if key: _cache_put(key, result) print(f"[app] example cache STORE {key}", flush=True) return result # --------------------------------------------------------------------------- # Preset scenes (the top tabs), reconstructed from BFL's FLUX.3 layout examples. # bbox is [y0, x0, y1, x1] normalized 0-1000. # --------------------------------------------------------------------------- PRESETS = { "Le Festival du Soleil": { "aspect": "3:4", "scene": "A glowing concrete dome rises from a twilight bay, a crowd on the beach before it.", "elements": [ {"id": "Fr_Text_1", "bbox": [90, 230, 230, 780], "desc": "\"LE FESTIVAL DU SOLEIL\" written in a thin, elegant, serif typeface in a light cream color."}, {"id": "dome_1", "bbox": [250, 150, 650, 850], "desc": "a massive, smooth parabolic dome of pale concrete"}, {"id": "town_1", "bbox": [250, 640, 430, 900], "desc": "faint lights and small buildings at the foot of the hills"}, {"id": "swimmers_1", "bbox": [580, 200, 720, 800], "desc": "dozens of small, silhouetted figures scattered in the dark water, wading"}, {"id": "crowd_1", "bbox": [740, 0, 1000, 1000], "desc": "a large crowd of people seated on the beach in casual, light-colored summer attire"}, ], }, "Five frames of winter": { "aspect": "16:9", "scene": "Five vertical frames of a quiet winter day, left to right from dawn to dusk.", "elements": [ {"id": "frame_1", "bbox": [0, 0, 1000, 190], "desc": "a bare birch forest at pale blue dawn, frost on every branch"}, {"id": "frame_2", "bbox": [0, 200, 1000, 390], "desc": "a frozen lake under flat grey light, a lone figure skating"}, {"id": "frame_3", "bbox": [0, 400, 1000, 590], "desc": "heavy snowfall over a red wooden cabin with a lit window"}, {"id": "frame_4", "bbox": [0, 600, 1000, 790], "desc": "a snowy pine ridge at golden hour, long blue shadows"}, {"id": "frame_5", "bbox": [0, 800, 1000, 1000], "desc": "the same valley at dusk, deep indigo sky and the first stars"}, ], }, "Knitted iceberg village": { "aspect": "4:3", "scene": "A knitted diorama: three cable-knit icebergs with tiny houses, a banner and a magenta whale tail.", "elements": [ {"id": "En_Text_1", "bbox": [130, 220, 220, 780], "desc": "The exact words \"KNITTED ICEBERG VILLAGE\" embroidered in thick, raised navy-blue yarn across a banner."}, {"id": "iceberg_3", "bbox": [200, 300, 650, 700], "desc": "The largest central iceberg, a bulky mound of white and pale grey ribbed knitting, dominating the midground with thick, vertical columns of stitches."}, {"id": "iceberg_1", "bbox": [230, 60, 650, 320], "desc": "A conical soft sculpture made of stark white, chunky roving yarn in a thick cable-knit pattern. The left side is brightly lit, while the right side falls into deep, fuzzy shadow."}, {"id": "iceberg_2", "bbox": [230, 680, 640, 960], "desc": "A tall, asymmetrical peak of white cable-knit yarn, featuring a few visibly dropped stitches near the base and slightly uneven tension in the cables."}, {"id": "whale_tail_1", "bbox": [500, 380, 850, 620], "desc": "A giant whale tail knitted from bright magenta and coral-pink yarn, bursting vertically from the teal sea."}, ], }, "Up from the subway": { "aspect": "3:2", "scene": "Looking up a subway staircase at dusk, a blurred figure climbing toward the skyline.", "elements": [ {"id": "building_1", "bbox": [50, 120, 630, 470], "desc": "a modern, red-brick high-rise skyscraper stretching into the sky"}, {"id": "building_2", "bbox": [0, 450, 650, 850], "desc": "a tall, classical white stone skyscraper with a tiered, set-back top"}, {"id": "person_1", "bbox": [300, 380, 820, 640], "desc": "a heavily blurred, dark shape of a person seen from behind"}, {"id": "stairs_1", "bbox": [600, 200, 1000, 800], "desc": "a flight of concrete subway stairs filling the bottom center of the frame"}, {"id": "En_Text_1", "bbox": [300, 760, 640, 1000], "desc": "a partially visible, glowing green and white sign on the building"}, ], }, } # A fixed seed per example, so its lazily-cached render is deterministic and the # cache key is stable across restarts. Order is stable -> seeds are stable. for _i, _name in enumerate(PRESETS): PRESETS[_name].setdefault("seed", 1000 + _i) RATIO_LIST = list(RATIOS.keys()) CONFIG_JS = ( "" ) # Styles + behavior live in ; the component mounts into #f3-root (gr.HTML). STYLE = """ """ # Selection accent (green) + small extras, appended to the head styles. STYLE2 = """ """ SCRIPT = r""" """ HEAD = STYLE + STYLE2 + CONFIG_JS + SCRIPT with gr.Blocks(head=HEAD, title="FLUX.3 layout canvas", css="footer{display:none!important} .gradio-container{max-width:1240px!important}" " .f3-bridge{position:absolute!important;left:-9999px!important;top:0;" "width:1px!important;height:1px!important;overflow:hidden!important;" "opacity:0!important;pointer-events:none!important;}") as demo: # Bridge components: kept MOUNTED (not visible=False, which unmounts them from the # DOM in gradio 5.x) but hidden off-screen via CSS, so the canvas JS can find them. state_box = gr.Textbox(elem_id="f3-state", elem_classes=["f3-bridge"]) result_box = gr.Textbox(elem_id="f3-result", elem_classes=["f3-bridge"]) gen_btn = gr.Button("gen", elem_id="f3-gen", elem_classes=["f3-bridge"]) gr.HTML('
') ev = gen_btn.click(generate, [state_box], [result_box], api_name="generate", show_progress="hidden") ev.then(None, [result_box], None, js="(r)=>{ if(window.f3_render) window.f3_render(r); }") if __name__ == "__main__": demo.queue(max_size=12).launch(show_error=True)