"""MiniMax-H3 as an image model, split deployment — the denoising half. MiniMax-H3 is a joint video-and-audio model with no image mode. The method this Space implements is the one described by `iamkaikai/MiniMax-H3-Single-Frame-VAE-500K`: run H3's ordinary `t2va` generation, then decode **one temporal latent slice** of the result as a still, using that repo's decoder-only retrain of H3's video autoencoder. The decode is that card's reference recipe, unmodified — see `h3_single_frame_vae.py`. The one deliberate divergence is **generation length**: the card generates a full-length clip, this defaults to the shortest H3 will produce (5 frames), because measurement says the floor is both far cheaper and no worse. Length is a user control, so the divergence is exposed rather than hidden. The 62.14 GiB Qwen3-VL conditioner cannot live here — MiniMax-H3 is 195.9 GiB in bfloat16 and a ZeroGPU Space is evicted at 150 GB of storage — so it runs in `multimodalart/qwen3vl-conditioner` and this Space calls it over the gradio API for every request, exactly as its two video siblings do. """ from __future__ import annotations import os import tempfile import time import traceback from functools import cache # Before anything that could initialize CUDA: `import spaces` patches `torch.cuda` so the transformer load can happen # at startup rather than on GPU time. import spaces import gradio as gr MODEL_REPO = os.environ.get("H3_MODEL_REPO", "multimodalart/MiniMax-H3-Pruned") CONDITIONER_SPACE = os.environ.get("H3_CONDITIONER", "multimodalart/qwen3vl-conditioner") # `pack` places the transformer at startup, `lazy` moves it on the first GPU call. Scoped to the transformer either # way: `spaces` packs every startup-resident CUDA tensor into a second on-disk copy, so the 9.03 GiB float32 decoder # moves on the first GPU call rather than being written to disk twice. PLACEMENT = os.environ.get("H3_PLACEMENT", "pack").lower() # cuDNN's fused attention is 10-20% faster than the SDPA default on this pool and needs nothing installed. ATTENTION = os.environ.get("H3_ATTENTION", "_native_cudnn").lower() GPU_SIZE = os.environ.get("H3_GPU_SIZE", "xlarge") # The turbo LoRA. `larryvrh/MiniMax-H3-Turbo-Lora` v4 is distilled for *video*, and the surprise measured on a single # frame is that it wins outright: at 6 NFE it is sharper and more legible than the 27-NFE base schedule, not merely # close to it. Chart axis labels that are illegible in the base leg read as clock times in the turbo one, so this is # resolved detail rather than acutance. # # `num_inference_steps` counts sigma grid points **including the terminal one, which has no model evaluation**, so a # run makes `steps - 1` forward passes (NFE). 6 NFE is 7 steps. That is the convention throughout MiniMax-H3 and is # not a typo here. # # What a correct attach looks like: 363 modules — 51 AdaLN + 312 attention/feed-forward — and 51 AdaLN projection # offsets. The offsets are there because this LoRA is *released*-trained: on the pruned checkpoint served here its # AdaLN factors go through the shipped full->pruned projection, whose constant term cannot live in the projection # itself. A file that only partly claims the model loads without complaint and costs quality silently, so both counts # are checked rather than trusted. TURBO_REPO = "larryvrh/MiniMax-H3-Turbo-Lora" TURBO_WEIGHT = "minimax_h3_turbo_v4_step600_ema.safetensors" TURBO_ADAPTER = "larryvrh_v4" TURBO_MODULES, TURBO_OFFSETS = 363, 51 TURBO_STEPS, BASE_STEPS = 7, 28 # The demo ships the **base** schedule only, so the adapter is not attached by default and costs nothing. It stays # reachable with `H3_TURBO=1` because the measurement above is worth keeping runnable, but it is env-only: there is # no request-level control and no UI for it. TURBO = os.environ.get("H3_TURBO", "0") == "1" TURBO_LABEL = f"Turbo · {TURBO_STEPS - 1} steps" BASE_LABEL = f"Base · {BASE_STEPS - 1} steps" # The video VAE's temporal geometry is fixed: `17n + 5` pixel frames map to `5n + 2` latent frames. Only those counts # can be encoded or decoded — so the length slider steps by 17 from 5, which lands on exactly those values and no # others (5, 22, 39, ... 124) without a user ever needing to know the rule. `snap_frames` covers API callers, who can # pass anything. FPS, FRAMES_PER_CHUNK, LATENTS_PER_CHUNK = 24, 17, 5 MIN_FRAMES, MAX_FRAMES = 5, 124 DEFAULT_FRAMES = MIN_FRAMES # What `/encode` is told. The conditioner's blocks never see `num_frames` — its `_encode` takes only the prompt, the # keyframes and the canvas, so `prompt_embeds` is frame-count independent — but its request validator holds a 2 s # floor, which 5 frames (0.21 s) would trip. 56 frames is the smallest `17n + 5` that clears it. Verified by # generating from embeddings encoded this way at every offered length: prompt adherence is intact. CONDITIONER_NUM_FRAMES = 56 # The *wire* label goes to `/encode` and must be one the conditioner's own table knows, so it is the value here and # the pretty name is the key. A canvas the conditioner does not know is rejected there and surfaces as a failure here. CANVASES = { "1024 × 1024 · square": ("1024x1024 · 1:1 max", (1024, 1024)), "1344 × 768 · landscape": ("1344x768 · 16:9 full", (768, 1344)), "768 × 1344 · portrait": ("768x1344 · 9:16 full", (1344, 768)), "1024 × 768 · 4:3": ("1024x768 · 4:3 full", (768, 1024)), "768 × 1024 · 3:4": ("768x1024 · 3:4 full", (1024, 768)), "1536 × 672 · panorama": ("1536x672 · 21:9 full", (672, 1536)), "768 × 768 · small square": ("768x768 · 1:1 full", (768, 768)), "960 × 544 · fastest": ("960x544 · 16:9 fast", (544, 960)), } DEFAULT_CANVAS = "1024 × 1024 · square" # How many slices the contact strip shows alongside the chosen one. The whole sequence is generated either way, so # each extra slice costs one decoder pass and nothing else. STRIP_MAX = 5 PIPE = None VAE = None MANAGER = None LOAD_ERROR: str | None = None LOADED_IN: float | None = None TURBO_STATUS = "off (base schedule; `H3_TURBO=1` to attach)" if not TURBO else "not loaded" TURBO_ON = False def snap_frames(num_frames: int) -> int: """The nearest valid `17n + 5` count at or above the request. The slider is already on the grid; API callers are not, and an off-grid count fails deep inside the layout step rather than here.""" n = max(MIN_FRAMES, min(int(num_frames), MAX_FRAMES)) while n % FRAMES_PER_CHUNK != LATENTS_PER_CHUNK: n += 1 return n def latent_frames(num_frames: int) -> int: """`17n + 5` pixel frames -> `5n + 2` latent frames.""" return (num_frames - LATENTS_PER_CHUNK) // FRAMES_PER_CHUNK * LATENTS_PER_CHUNK + 2 def lower_duration_floor(seconds: float = 5.0 / FPS) -> None: """Let the pipeline generate its own 5-frame floor. `MiniMaxH3ModularPipeline.min_duration` is 5 *seconds*, which refuses 5 *frames*. Patched here as a property override, the way both video siblings do it — never by editing the installed package. """ from diffusers.modular_pipelines.minimax_h3.modular_pipeline import MiniMaxH3ModularPipeline MiniMaxH3ModularPipeline.min_duration = property(lambda self: float(seconds)) def default_steps() -> int: return TURBO_STEPS if TURBO_ON else BASE_STEPS def status() -> str: if LOAD_ERROR: return LOAD_ERROR if PIPE is None: return f"Loading `{MODEL_REPO}` (transformer) and the single-frame decoder. Watch the Space logs." return ( f"Ready · transformer **bfloat16, unquantized**, decoder **float32** · placement `{PLACEMENT}` · " f"attention `{ATTENTION}` · turbo {TURBO_STATUS} · loaded in {LOADED_IN:.0f}s · " f"conditioner `{CONDITIONER_SPACE}`" ) def detach_turbo(transformer) -> None: """Delete the adapter *and* the AdaLN constant terms that ride alongside it. `delete_adapters` is PEFT's, and PEFT knows nothing about the pruned checkpoint's `_lora_adaln_offsets`: those are plain non-persistent buffers this checkpoint's own loader registers for a released-trained LoRA. Dropping them here keeps "the adapter was removed" true of the whole adapter, including the half of it that does not live in a `lora_A`/`lora_B` pair. """ if TURBO_ADAPTER in (getattr(transformer, "peft_config", None) or {}): transformer.delete_adapters(TURBO_ADAPTER) for _, module in transformer.named_modules(): registry = getattr(module, "_lora_adaln_offsets", None) if isinstance(registry, dict) and TURBO_ADAPTER in registry: module._buffers.pop(registry.pop(TURBO_ADAPTER), None) def attach_turbo(pipe) -> str: """Inject the turbo LoRA once, at startup, and count what actually attached. Runs in the parent process before `pipe.transformer.to("cuda")`, so `spaces` packs the adapter's parameters along with everything else and every forked ZeroGPU worker inherits it. Per-request switching is then `enable_adapters()` / `disable_adapters()`, which was verified **bitwise** in both directions against a transformer that never saw the adapter — the 51 AdaLN offsets follow PEFT's `active_adapters` and `disable_adapters`, so nothing leaks into a base-quality request. """ started = time.time() pipe.load_lora_weights(TURBO_REPO, weight_name=TURBO_WEIGHT, adapter_name=TURBO_ADAPTER) modules, offsets, scalings = [], [], set() for path, module in pipe.transformer.named_modules(): scaling = getattr(module, "scaling", None) if isinstance(scaling, dict) and TURBO_ADAPTER in scaling: modules.append(path) scalings.add(round(float(scaling[TURBO_ADAPTER]), 6)) if TURBO_ADAPTER in getattr(module, "_lora_adaln_offsets", {}): offsets.append(path) adaln = sum(path.endswith(("adaln_proj.linear", "norm_out.linear")) for path in modules) census = ( f"{len(modules)} modules ({adaln} AdaLN + {len(modules) - adaln} attention/FF), " f"{len(offsets)} AdaLN projection offsets, scale {sorted(scalings)}" ) print(f"[turbo] {TURBO_REPO}/{TURBO_WEIGHT}: {census}, in {time.time() - started:.1f}s", flush=True) if len(modules) != TURBO_MODULES or len(offsets) != TURBO_OFFSETS: # Loud rather than quiet: the Space keeps serving, without the adapter and at the base step count, and the # "Turbo" choice disappears from the UI rather than silently meaning nothing. detach_turbo(pipe.transformer) print( f"[turbo] expected {TURBO_MODULES} modules and {TURBO_OFFSETS} AdaLN offsets, got {len(modules)} and " f"{len(offsets)} — adapter deleted", flush=True, ) return ( f"**under-attached** ({len(modules)} of {TURBO_MODULES} modules, {len(offsets)} of {TURBO_OFFSETS} " "AdaLN offsets), removed — running the base model" ) return f"`{TURBO_REPO}` · {census}" def load_models() -> str | None: """Load the denoising half plus the single-frame decoder at startup. `MiniMaxH3ImageBlocks` declares `image_processor`, the two schedulers and `transformer` and nothing else, so `load_components` fetches exactly the transformer — `text_encoder/`, `transformer_ref/`, `vae/` and `audio_vae/` are never touched. The decoder is built separately from `vae/config.json` and the single-frame checkpoint. """ global PIPE, VAE, MANAGER, LOAD_ERROR, LOADED_IN, TURBO_STATUS, TURBO_ON if PIPE is not None or LOAD_ERROR is not None: return LOAD_ERROR started = time.time() try: import torch from diffusers import ComponentsManager from h3_image_blocks import MiniMaxH3ImageBlocks from h3_single_frame_vae import load_single_frame_vae lower_duration_floor() manager = ComponentsManager() blocks = MiniMaxH3ImageBlocks() names = [component.name for component in blocks.expected_components] print(f"[img] loading {names} from {MODEL_REPO} ...", flush=True) if "vae" in names or "audio_vae" in names: raise RuntimeError(f"the image blockset must declare no autoencoder, got {names}") pipe = blocks.init_pipeline(MODEL_REPO, components_manager=manager, collection="h3img") # The pruned DiT is served as remote code (`transformer/modeling_minimax_h3_pruned.py`, reached through the # `AutoModel` type hint in `modular_model_index.json`). pipe.load_components(dtype=torch.bfloat16, trust_remote_code=True) pipe.transformer.set_attention_backend(ATTENTION) if TURBO: try: TURBO_STATUS = attach_turbo(pipe) TURBO_ON = TURBO_STATUS.startswith("`") except Exception as error: # noqa: BLE001 traceback.print_exc() TURBO_STATUS = f"**failed to load**: `{type(error).__name__}: {error}` — running the base model" TURBO_ON = False print("[img] building the single-frame decoder ...", flush=True) vae = load_single_frame_vae() if PLACEMENT == "pack": # Scoped to the transformer, mirroring the video siblings: `spaces` writes a second on-disk copy of every # startup-resident CUDA tensor, and the 9.03 GiB decoder does not need to be on disk twice for the sake # of a one-off host-to-device copy on a cold worker. pipe.transformer.to("cuda") PIPE, VAE, MANAGER = pipe, vae, manager LOADED_IN = time.time() - started print(f"[img] ready in {LOADED_IN:.0f}s", flush=True) except Exception as error: # noqa: BLE001 traceback.print_exc() LOAD_ERROR = ( f"**Loading `{MODEL_REPO}` failed** after {time.time() - started:.0f}s: " f"`{type(error).__name__}: {error}`" ) return LOAD_ERROR @cache def conditioner(): """The other half, over the gradio API. `gradio_client` attaches the caller's own ZeroGPU token per call, so the conditioner's booking is billed to whoever asked for the image.""" from gradio_client import Client return Client(CONDITIONER_SPACE) def encode_remote(prompt, canvas_wire, rewrite_prompt=False): """`/encode` on the conditioner Space: a safetensors file holding `prompt_embeds` + `text_token_tags`, with the resolved `height` / `width` in its metadata, plus the plan.""" from safetensors import safe_open path, plan = conditioner().predict( prompt=prompt, image_path=None, last_image_path=None, canvas=canvas_wire, num_frames=CONDITIONER_NUM_FRAMES, rewrite_prompt=bool(rewrite_prompt), api_name="/encode", ) with safe_open(path, framework="pt") as handle: metadata = handle.metadata() return handle.get_tensor("prompt_embeds"), handle.get_tensor("text_token_tags"), metadata, plan # Seconds of GPU one request needs, from the packed video rows it is about to denoise: linear in the rows for the # matmuls, quadratic for the attention. Carried over from the video sibling, which fitted them on this pool. _DUR_B, _DUR_C = 1.1745e-4, 3.8396e-9 # Each decoded slice, which scales with the canvas rather than with the step count. _DECODE_BASE, _DECODE_PER_MEGAPIXEL = 3, 3 # `pack` mode: the 9.03 GiB float32 decoder moves on a cold worker. _PLACEMENT_ALLOWANCE, _PAD = 14, 8 def get_duration(prompt_embeds, text_token_tags, height, width, num_frames, steps, seed, turbo, strip, *a, **k): """Sized from the *actual* requested length, so the short default reserves little and a long clip reserves more.""" height, width, steps = int(height), int(width), int(steps) rows = latent_frames(int(num_frames)) * (height // 32) * (width // 32) denoise = steps * (_DUR_B * rows + _DUR_C * rows**2) decodes = 1 + (min(STRIP_MAX, latent_frames(int(num_frames))) if strip else 0) decode = _DECODE_BASE + decodes * _DECODE_PER_MEGAPIXEL * (height * width) / 1e6 return max(30, int(denoise + decode) + _PLACEMENT_ALLOWANCE + _PAD) @spaces.GPU(duration=get_duration, size=GPU_SIZE) def _generate(prompt_embeds, text_token_tags, height, width, num_frames, steps, seed, turbo, strip, latent_index): """The only thing on GPU time: the denoise loop and the slice decodes. Only the finished images come back — a `@spaces.GPU` return crosses a process boundary by pickling, and the full `PipelineState` still holds the packed latents, the rotary grid and the row indices on the card. """ import torch from h3_single_frame_vae import decode_slice if PLACEMENT == "lazy": PIPE.transformer.to("cuda") VAE.to("cuda") if TURBO_ON: # Verified bitwise in both directions, AdaLN offsets included, so this is a real switch and not an # approximation of one. PIPE.transformer.enable_adapters() if turbo else PIPE.transformer.disable_adapters() started = time.time() state = PIPE( prompt_embeds=prompt_embeds.to("cuda"), text_token_tags=text_token_tags, height=int(height), width=int(width), num_frames=int(num_frames), num_inference_steps=int(steps), generator=torch.Generator("cpu").manual_seed(int(seed)), ) denoise_seconds = time.time() - started # The audio rows were denoised along with the video rows and are dropped here as latents: MiniMax-H3 is jointly # trained and the packed row geometry is fixed, so removing the rows would change what every other row attends # to. They are 16 rows out of 2 064 at 1024x1024 and 5 frames, and no audio autoencoder is loaded to decode them. latents = state.get("latents") vlat = latents.shape[2] started = time.time() index = max(0, min(int(latent_index), vlat - 1)) chosen = decode_slice(VAE, latents, index) others = [] if strip and vlat > 1: # Evenly spaced across the sequence, so the strip shows the transition the card describes rather than a # cluster. The generation is paid for either way; each extra slice is one decoder pass. count = min(STRIP_MAX, vlat) picks = sorted({round(i * (vlat - 1) / (count - 1)) for i in range(count)}) if count > 1 else [0] others = [(p, decode_slice(VAE, latents, p)) for p in picks] return chosen, others, index, vlat, denoise_seconds, time.time() - started def generate( prompt, canvas=DEFAULT_CANVAS, num_frames=DEFAULT_FRAMES, latent_index=0, strip=True, steps=0, seed=42, upsample=False, progress=gr.Progress(track_tqdm=True), ): """One request.""" if LOAD_ERROR: raise gr.Error(LOAD_ERROR) if PIPE is None: raise gr.Error("The denoiser is still loading.") if not prompt or not prompt.strip(): raise gr.Error("MiniMax-H3 always takes a prompt.") canvas_wire, _ = CANVASES[canvas] num_frames = snap_frames(num_frames) # The demo runs the base schedule only; the turbo adapter is never attached (see `TURBO`). turbo = False # 0 means "use the default". It has to be a real number rather than `None`, because gradio validates a # Slider's value *before* the handler runs and rejects `None` outright — which an API caller hits immediately. steps = max(4, int(steps)) if steps and int(steps) > 0 else BASE_STEPS progress(0.0, desc=f"Upsampling the prompt on {CONDITIONER_SPACE} ..." if upsample else f"Conditioning on {CONDITIONER_SPACE} ...") conditioned = time.time() prompt_embeds, text_token_tags, metadata, plan = encode_remote(prompt, canvas_wire, rewrite_prompt=upsample) condition_seconds = time.time() - conditioned height, width = int(metadata["height"]), int(metadata["width"]) refined = plan.get("refined_prompt") or "" progress(0.25, desc=f"Denoising {steps - 1} steps of {num_frames} frames at {width}x{height} ...") chosen, others, index, vlat, denoise_seconds, decode_seconds = _generate( prompt_embeds, text_token_tags, height, width, num_frames, steps, seed, turbo, bool(strip), int(latent_index) ) directory = os.path.join(tempfile.gettempdir(), "h3-images") os.makedirs(directory, exist_ok=True) stamp = int(time.time() * 1000) path = os.path.join(directory, f"h3-{stamp}.png") chosen.save(path) gallery = [] for slice_index, image in others: strip_path = os.path.join(directory, f"h3-{stamp}-s{slice_index}.png") image.save(strip_path) gallery.append((strip_path, f"latent_index {slice_index}" + (" · shown" if slice_index == index else ""))) report = ( f"`{width}x{height}` · {num_frames} frames → {vlat} latent slices, decoded index **{index}** · " f"{'turbo' if turbo else 'base'}, {steps - 1} NFE · " f"conditioner {condition_seconds:.1f}s ({plan['num_text_tokens']} tokens" f"{', upsampled' if refined else ''}) · " f"denoise {denoise_seconds:.1f}s ({denoise_seconds / (steps - 1):.2f} s/NFE) · " f"decode {decode_seconds:.1f}s · seed {int(seed)}" ) print(f"[img] {report}", flush=True) return path, gallery, report, refined, gr.update(visible=bool(refined)) def _follow_length(num_frames): """`latent_index` must clamp to the slices that exist, or index 9 on a 5-frame generation would error.""" vlat = latent_frames(snap_frames(num_frames)) return gr.update(maximum=vlat - 1, value=0, info=f"{vlat} slices at {snap_frames(num_frames)} frames") load_models() INTRO = """# MiniMax-H3 · images
[ model ]   [ single-frame decoder ]   [ blog ]   [ video ]
**MiniMax-H3** is a 33B parameter video model with no image mode. This runs an ordinary video generation and decodes one temporal latent slice of it as a still. Longer clips cost more and are not sharper, so it defaults to the shortest clip H3 will produce. """ CSS = """ .main.fillable {max-width: 1250px !important} .dark .gradio-container { color: var(--body-text-color); } """ with gr.Blocks(title="MiniMax-H3 images") as demo: gr.Markdown(INTRO) with gr.Row(): with gr.Column(): prompt = gr.Textbox( label="Prompt", lines=3, value="Studio product photograph of a compact futuristic electric espresso machine, brushed aluminium and matte black, precise industrial design, centered three-quarter view, soft gray seamless background, crisp edges", ) canvas = gr.Dropdown(label="Canvas", choices=list(CANVASES), value=DEFAULT_CANVAS) run = gr.Button("Generate", variant="primary") with gr.Accordion("Advanced options", open=False): with gr.Row(): frames = gr.Slider( label="Frames generated", minimum=MIN_FRAMES, maximum=MAX_FRAMES, step=FRAMES_PER_CHUNK, value=DEFAULT_FRAMES, info="Longer costs more and measured no sharper.", ) # `maximum` is the widest any length allows, not the default length's. gradio validates a Slider # against the *initial* component config, and an API call carries no session state, so a maximum # narrowed by `_follow_length` would reject a legitimate index from a client. The UI still gets # the narrowed range via that handler, and `_generate` clamps to the slices that exist. latent_index = gr.Slider( label="Latent slice", minimum=0, maximum=latent_frames(MAX_FRAMES) - 1, step=1, value=0, info=f"{latent_frames(DEFAULT_FRAMES)} slices at {DEFAULT_FRAMES} frames", ) steps = gr.Slider( label="Steps", minimum=0, maximum=40, step=1, value=default_steps(), info=f"0 uses the default {BASE_STEPS}.", ) seed = gr.Number(label="Seed", value=42, precision=0) strip = gr.Checkbox(label="Show other slices", value=True) upsample = gr.Checkbox(label="Upsample prompt", value=False) with gr.Column(): image = gr.Image(label="Image", type="filepath", format="png", height=520) slices = gr.Gallery(label="Other latent slices", columns=5, height=150, object_fit="cover") report = gr.Markdown(visible=False) with gr.Accordion("Upsampled prompt", open=False, visible=False) as upsampled_panel: upsampled = gr.Textbox(show_label=False, lines=8, interactive=False) frames.change(_follow_length, frames, latent_index) gr.Examples( examples=[ ["Studio product photograph of a compact futuristic electric espresso machine, brushed aluminium and matte black, precise industrial design, centered three-quarter view, soft gray seamless background, crisp edges", "1024 × 1024 · square"], ["A technical exploded-view diagram of a mechanical wristwatch on a white background, thin black line art, labelled callout lines, engineering illustration style", "1024 × 1024 · square"], ["A red fox standing in a snowy pine forest at dawn, soft golden light through the trees, shallow depth of field, photorealistic wildlife photograph", "1344 × 768 · landscape"], ["Close-up portrait photograph of an elderly fisherman with a weathered face and a wool cap, overcast daylight, sharp detail in the skin and beard", "768 × 1024 · 3:4"], ["A clean analytics dashboard UI on a dark background, sidebar navigation, three KPI cards, one line chart and one bar chart, crisp vector rendering, flat design", "1024 × 1024 · square"], ], inputs=[prompt, canvas], outputs=[image, slices, report, upsampled, upsampled_panel], fn=generate, cache_examples=True, cache_mode="lazy", ) run.click( generate, [prompt, canvas, frames, latent_index, strip, steps, seed, upsample], [image, slices, report, upsampled, upsampled_panel], api_name="generate", ) if __name__ == "__main__": demo.launch(show_error=True, theme=gr.themes.Citrus(), css=CSS, max_threads=1000)