|

Implementing a MiniMax-H3 Multimodal Video and Audio Generation Pipeline with ComfyUI APIs

In this tutorial, we implement an end-to-end MiniMax-H3 video technology workflow utilizing ComfyUI as a headless inference backend. We configure the atmosphere round GPU reminiscence, disk capability, mannequin precision, decision, length, sampling technique, and a number of technology modes, whereas dynamically deciding on an applicable weight profile primarily based on the accessible {hardware}. We set up and launch ComfyUI programmatically, obtain the required diffusion, text-encoder, video-VAE, and audio-VAE weights from Hugging Face, and talk with the working server by way of its HTTP and WebSocket APIs. We additionally assemble the ComfyUI execution graph immediately in Python, validate node schemas towards the stay /object_info endpoint, and help text-to-video, first- and last-frame-conditioned technology, and reference-image-conditioned technology. By combining automated mannequin setup, schema-aware graph building, joint video-audio decoding, progress monitoring, and output assortment, we create a reproducible pipeline for experimenting with MiniMax-H3 with out counting on the graphical ComfyUI interface.

import json, os, re, shutil, subprocess, sys, time, uuid, urllib.request, urllib.error
from pathlib import Path
CFG = {
  
   "MODE": "t2v",                
   "PROMPT": (
       "Realistic live-action cinematic look. A lone lighthouse keeper on a storm-lashed "
       "cliff at nightfall, anamorphic lens, shallow depth of area, movie grain, volumetric sea spray.n"
       "[0s-2s] Wide shot: waves detonate towards black rock, the lighthouse beam sweeps the body.n"
       "[2s-4s] Medium shot: the keeper braces towards the wind, coat snapping, rain on his face.n"
       "[4s-5s] Close up: he squints into the darkish and says "She's holding."n"
       "Camera: onerous cuts between photographs, slight handheld jitter, no dissolves.n"
       "Audio: roaring surf and howling wind all through, low cello drone beneath, "
       "a heavy wave affect on every lower, the road delivered clearly over the storm.n"
       "No textual content, subtitles, logos or watermarks."
   ),
   "ASPECT": (16, 9),            
   "MEGAPIXELS": 0.4,            
   "SECONDS": 5.0,               
   "SEED": 556589502035082,
   "STEPS": 20,                  
   "SAMPLER": "res_multistep",   
   "SCHEDULER": "easy",
  
   "FIRST_FRAME": None,          
   "LAST_FRAME": None,           
   "REF_IMAGES": [],             
   "REF_IMAGE_SIZE": "match",    
  
   "SIGMA_SHIFT": None,          
   "TURBO_LORA": False,          
   "TURBO_STEPS": 8,
   "TURBO_SAMPLER": "euler",
   "TURBO_SCHEDULER": "beta",
  
   "COMFY_DIR": "/content material/ComfyUI",
   "OUT_DIR": "/content material/outputs",
   "MODELS_ROOT": "/content material/fashions",  
   "PORT": 8188,
   "HF_TOKEN": os.environ.get("HF_TOKEN", ""), 
   "SKIP_INSTALL": False,             
}
REPO = "Comfy-Org/MiniMax-H3"
API = f"http://127.0.0.1:{CFG['PORT']}"
PROFILES = [
   dict(name="quality", min_vram=70,
        unet_fl="minimax_h3_fl2va_bf16.safetensors",           
        unet_ref="minimax_h3_ref2va_bf16.safetensors",
        te="qwen3vl_32b_minimax_h3_int8_convrot.safetensors",  
        flags=["--normalvram"]),
   dict(title="balanced", min_vram=38,
        unet_fl="minimax_h3_fl2va_pruned_int8_convrot.safetensors",  
        unet_ref="minimax_h3_ref2va_pruned_int8_convrot.safetensors",
        te="qwen3vl_32b_minimax_h3_nvfp4_awq.safetensors",           
        flags=["--normalvram", "--cache-none"]),
   dict(title="squeeze", min_vram=20,
        unet_fl="minimax_h3_fl2va_pruned_fp8_scaled.safetensors",    
        unet_ref="minimax_h3_ref2va_pruned_fp8_scaled.safetensors",
        te="qwen3vl_32b_minimax_h3_nvfp4_awq.safetensors",
        flags=["--lowvram", "--cache-none", "--disable-smart-memory"]),
]
VAE_VIDEO = "minimax_h3_video_vae_fp16.safetensors"
VAE_AUDIO = "minimax_h3_audio_vae_fp32.safetensors"
def sh(cmd, cwd=None, examine=True, quiet=False):
   """Run a shell command, streaming output."""
   print(f"$ {cmd}")
   p = subprocess.run(cmd, shell=True, cwd=cwd,
                      stdout=subprocess.DEVNULL if quiet else None,
                      stderr=subprocess.STDOUT if quiet else None)
   if examine and p.returncode != 0:
       increase RuntimeError(f"command failed ({p.returncode}): {cmd}")
def get_json(path, payload=None, timeout=30):
   url = f"{API}{path}"
   information = json.dumps(payload).encode() if payload just isn't None else None
   req = urllib.request.Request(url, information=information,
                                headers={"Content-Type": "software/json"})
   with urllib.request.urlopen(req, timeout=timeout) as r:
       physique = r.learn()
   return json.masses(physique) if physique else {}
def align_frames(seconds, fps=24):
   """H3 consumes body counts on the 17k+5 grid. Snap upward."""
   n = max(5, int(spherical(seconds * fps)))
   whereas n % 17 != 5:
       n += 1
   return n
def h3_canvas(side=(16, 9), megapixels=0.98, a number of=32):
   """Mirror of ComfyUI's ResolutionSelector + H3's 768*1344 space cap."""
   ar = side[0] / side[1]
   complete = megapixels * 1e6
   h = (complete / ar) ** 0.5
   w = ar * h
   cap = 768 * 1344
   if w * h > cap:
       s = (cap / (w * h)) ** 0.5
       w, h = w * s, h * s
   r = lambda v: max(a number of, int(spherical(v / a number of)) * a number of)
   return r(w), r(h)
def preflight():
   attempt:
       import torch
   besides ImportError:
       increase SystemExit("PyTorch lacking — run this in a Colab GPU runtime.")
   if not torch.cuda.is_available():
       increase SystemExit("No CUDA system. Runtime > Change runtime sort > GPU (A100).")
   title = torch.cuda.get_device_name(0)
   vram = torch.cuda.get_device_properties(0).total_memory / 1e9
   free_disk = shutil.disk_usage("/content material").free / 1e9
   bf16 = torch.cuda.is_bf16_supported()
   print(f"GPU        : {title}  ({vram:.1f} GB VRAM, bf16={bf16})")
   print(f"Free disk  : {free_disk:.1f} GB")
   if not bf16:
       increase SystemExit(
           "This GPU has no bf16 help (T4/K80). MiniMax-H3 is not going to run right here.n"
           "Switch to an A100/L4/H100 runtime."
       )
   profile = subsequent((p for p in PROFILES if vram >= p["min_vram"]), None)
   if profile is None:
       increase SystemExit(
           f"{vram:.0f} GB VRAM is beneath the ~20 GB ground for the smallest H3 construct."
       )
   if free_disk < 45:
       print("WARNING: <45 GB free. Point MODELS_ROOT at Drive or anticipate a disk-full error.")
   print(f"Profile    : {profile['name']}  (unet={profile['unet_fl']}, te={profile['te']})")
   return profile

We outline the core MiniMax-H3 configuration, mannequin profiles, technology parameters, and shared utility capabilities used all through the workflow. We calculate legitimate body counts and canvas dimensions whereas checking GPU functionality, accessible VRAM, BF16 help, and disk house earlier than inference begins. We additionally mechanically choose probably the most applicable mannequin profile so the pipeline matches the {hardware} accessible in our Colab runtime.

def install_comfy():
   cozy = Path(CFG["COMFY_DIR"])
   if CFG["SKIP_INSTALL"] and cozy.exists():
       print("Skipping set up (SKIP_INSTALL=True).")
       return
   sh("pip set up -q -U 'huggingface_hub[hf_xet]' hf_transfer websocket-client")
   if not cozy.exists():
       sh(f"git clone --depth 1 https://github.com/comfyanonymous/ComfyUI {cozy}")
  
   sh(f"pip set up -q -r {cozy}/necessities.txt")
   ver = (cozy / "comfyui_version.py")
   if ver.exists():
       print("ComfyUI:", ver.read_text().strip())
   if not (cozy / "comfy_extras" / "nodes_minimax_h3.py").exists():
       increase SystemExit("This ComfyUI checkout lacks native MiniMax-H3 nodes — replace it.")
  
   root = Path(CFG["MODELS_ROOT"])
   for sub in ("diffusion_models", "text_encoders", "vae", "loras"):
       (root / sub).mkdir(dad and mom=True, exist_ok=True)
   (cozy / "extra_model_paths.yaml").write_text(
       "minimax_h3:n"
       f"    base_path: {root}n"
       "    diffusion_models: diffusion_modelsn"
       "    text_encoders: text_encodersn"
       "    vae: vaen"
       "    loras: lorasn"
   )
   Path(CFG["OUT_DIR"]).mkdir(dad and mom=True, exist_ok=True)
def fetch(repo_id, filename, subdir):
   from huggingface_hub import hf_hub_download
   os.environ["HF_HUB_ENABLE_HF_TRANSFER"] = "1"
   dest = Path(CFG["MODELS_ROOT"]) / subdir
   goal = dest / Path(filename).title
   if goal.exists() and goal.stat().st_size > 1_000_000:
       print(f"cached  {goal.title} ({goal.stat().st_size/1e9:.1f} GB)")
       return goal
   print(f"pulling {filename} -> {dest}")
   attempt:
       p = hf_hub_download(repo_id=repo_id, filename=filename, local_dir=str(dest),
                           token=CFG["HF_TOKEN"] or None)
   besides Exception as e:
       if "401" in str(e) or "403" in str(e) or "gated" in str(e).decrease():
           increase SystemExit(
               f"Access denied for {repo_id}. Accept the MiniMax-H3 group license on the "
               "mannequin web page, create a learn token, then set CFG['HF_TOKEN']."
           ) from e
       increase
  
   p = Path(p)
   if p != goal:
       goal.mother or father.mkdir(dad and mom=True, exist_ok=True)
       shutil.transfer(str(p), str(goal))
   return goal
def download_weights(profile, mode):
   unet = profile["unet_ref"] if mode == "r2v" else profile["unet_fl"]
   fetch(REPO, f"diffusion_models/{unet}", "diffusion_models")
   fetch(REPO, f"text_encoders/{profile['te']}", "text_encoders")
   fetch(REPO, f"vae/{VAE_VIDEO}", "vae")
   fetch(REPO, f"vae/{VAE_AUDIO}", "vae")
   lora = None
   if CFG["TURBO_LORA"]:
      
       from huggingface_hub import HfApi
       lora_repo = "drbaph/MiniMax-H3-Turbo-Lora-ComfyUI"
       recordsdata = [f for f in HfApi().list_repo_files(lora_repo)
                if f.endswith(".safetensors") and "pruned" in f]
       if not recordsdata:
           recordsdata = [f for f in HfApi().list_repo_files(lora_repo) if f.endswith(".safetensors")]
       if recordsdata:
           lora = fetch(lora_repo, sorted(recordsdata)[-1], "loras").title
           print(f"turbo LoRA: {lora}")
   return unet, profile["te"], lora

We set up and configure ComfyUI, put together the exterior mannequin listing construction, and allow MiniMax-H3 help contained in the Colab atmosphere. We obtain the required diffusion mannequin, textual content encoder, video VAE, and audio VAE weights from Hugging Face whereas reusing cached recordsdata at any time when potential. We additionally optionally retrieve the Turbo LoRA configuration, permitting us to commerce some technology high quality for sooner inference when required.

class ComfyServer:
   def __init__(self, flags):
       self.flags, self.proc, self.log = flags, None, Path("/content material/comfyui.log")
   def begin(self):
       cmd = [sys.executable, "main.py",
              "--listen", "127.0.0.1", "--port", str(CFG["PORT"]),
              "--disable-auto-launch", "--preview-method", "none",
              "--output-directory", CFG["OUT_DIR"]] + self.flags
       print("$", " ".be part of(cmd))
       f = open(self.log, "wb")
       self.proc = subprocess.Popen(cmd, cwd=CFG["COMFY_DIR"], stdout=f, stderr=subprocess.STDOUT)
       deadline = time.time() + 300
       whereas time.time() < deadline:
           if self.proc.ballot() just isn't None:
               print(self.log.read_text()[-4000:])
               increase SystemExit("ComfyUI died throughout startup (log above).")
           attempt:
               stats = get_json("/system_stats", timeout=3)
               dev = stats.get("units", [{}])[0]
               print(f"server up — {dev.get('title','?')} "
                     f"{dev.get('vram_total',0)/1e9:.1f} GB complete, "
                     f"{dev.get('vram_free',0)/1e9:.1f} GB free")
               return
           besides Exception:
               time.sleep(2)
       increase SystemExit("Server didn't change into prepared in 300s. Check /content material/comfyui.log")
   def tail(self, n=3000):
       return self.log.read_text()[-n:] if self.log.exists() else ""
   def free_vram(self):
       attempt:
           get_json("/free", {"unload_models": True, "free_memory": True})
       besides Exception:
           go
   def cease(self):
       if self.proc and self.proc.ballot() is None:
           self.proc.terminate()
           attempt:
               self.proc.wait(30)
           besides subprocess.TimeoutExpired:
               self.proc.kill()
class Schema:
   """Reads /object_info so the graph is validated towards the *working* node set
   as a substitute of regardless of the docs mentioned final week."""
   def __init__(self):
       self.data = get_json("/object_info", timeout=120)
   def require(self, *courses):
       lacking = [c for c in classes if c not in self.info]
       if lacking:
           increase SystemExit(f"Missing node courses: {lacking}. Update ComfyUI to >= 0.30.0.")
   def inputs_of(self, cls):
       spec = self.data[cls]["input"]
       return listing(spec.get("required", {})) + listing(spec.get("non-compulsory", {}))
   def examine(self, cls, payload):
       recognized = set(self.inputs_of(cls))
       unknown = [k for k in payload if k not in known]
       if unknown:
           print(f"  word: {cls} doesn't declare {unknown} — declared: {sorted(recognized)}")
   def autogrow(self, cls, prefix, n):
       """Autogrow slots (ref_image_1, ref_video_1, ...) are dynamic; uncover the
       actual names if the server exposes them, in any other case fall again to 1-based."""
       discovered = sorted([k for k in self.inputs_of(cls) if k.startswith(prefix)])
       if len(discovered) >= n:
           return discovered[:n]
       return [f"{prefix}{i+1}" for i in range(n)]

We create a server-management layer that launches ComfyUI as a background subprocess and verifies that it turns into accessible by way of its API. We monitor server startup, examine GPU reminiscence statistics, free VRAM when vital, and safely terminate the server after execution. We additionally construct a schema-inspection utility that reads stay ComfyUI node definitions so we are able to validate graph inputs and dynamically uncover supported node slots.

class H3Graph:
   def __init__(self, schema, unet, te, lora=None):
       self.s, self.g, self._id = schema, {}, 0
       self.unet, self.te, self.lora = unet, te, lora
   def node(self, cls, **inputs):
       self.s.examine(cls, inputs)
       self._id += 1
       nid = str(self._id)
       self.g[nid] = {"class_type": cls, "inputs": inputs}
       return nid
  
   def _backbone(self):
       mannequin = self.node("UNETLoader", unet_name=self.unet, weight_dtype="default")
       if self.lora:
           mannequin = self.node("LoraLoaderModelOnly", mannequin=[model, 0],
                             lora_name=self.lora, strength_model=1.0)
       if CFG["SIGMA_SHIFT"]:
           sv, sa = CFG["SIGMA_SHIFT"]
           mannequin = self.node("MiniMaxH3SigmaShift", mannequin=[model, 0],
                             shift_video=float(sv), shift_audio=float(sa))
       clip = self.node("CLIPLoader", clip_name=self.te, sort="minimax", system="default")
       vvae = self.node("VAELoader", vae_name=VAE_VIDEO)
       avae = self.node("VAELoader", vae_name=VAE_AUDIO)
       return mannequin, clip, vvae, avae
   def _tail(self, mannequin, cond, latent, vvae, avae):
       turbo = bool(self.lora)
       steps = CFG["TURBO_STEPS"] if turbo else CFG["STEPS"]
       sampler_name = CFG["TURBO_SAMPLER"] if turbo else CFG["SAMPLER"]
       sched = CFG["TURBO_SCHEDULER"] if turbo else CFG["SCHEDULER"]
       noise = self.node("RandomNoise", noise_seed=int(CFG["SEED"]))
       samp = self.node("KSamplerSelect", sampler_name=sampler_name)
       sig = self.node("BasicScheduler", mannequin=[model, 0], scheduler=sched,
                       steps=steps, denoise=1.0)
       guider = self.node("BasicGuider", mannequin=[model, 0], conditioning=[cond[0], cond[1]])
       out = self.node("SamplerCustomAdvanced", noise=[noise, 0], guider=[guider, 0],
                       sampler=[samp, 0], sigmas=[sig, 0], latent_image=[latent[0], latent[1]])
      
       frames = self.node("VAEDecode", samples=[out, 0], vae=[vvae, 0])
       audio = self.node("VAEDecodeAudio", samples=[out, 0], vae=[avae, 0])
       vid = self.node("CreateVideo", pictures=[frames, 0], audio=[audio, 0], fps=24)
       self.node("SaveVideo", video=[vid, 0], filename_prefix="MiniMaxH3/h3",
                 format="auto", codec="auto")
       print(f"  sampling: {steps} steps, {sampler_name}/{sched}")
       return self.g
   def _load_image(self, uploaded_name):
       return self.node("LoadImage", picture=uploaded_name, add="picture")
  
   def t2v_or_flf2v(self, w, h, size, first=None, final=None):
       self.s.require("MiniMaxH3ImageToVideo", "SamplerCustomAdvanced", "SaveVideo")
       mannequin, clip, vvae, avae = self._backbone()
       kw = {}
       if first:
           kw["first_frame"] = [self._load_image(first), 0]  
       if final:
           kw["last_frame"] = [self._load_image(last), 0]    
       n = self.node("MiniMaxH3ImageToVideo", clip=[clip, 0], vae=[vvae, 0],
                     immediate=CFG["PROMPT"], width=w, peak=h, size=size, **kw)
       return self._tail(mannequin, (n, 0), (n, 1), vvae, avae)
   def r2v(self, w, h, size, ref_names):
       self.s.require("MiniMaxH3ReferenceToVideo")
       mannequin, clip, vvae, avae = self._backbone()
       slots = self.s.autogrow("MiniMaxH3ReferenceToVideo", "ref_image_", len(ref_names))
       refs = {slot: [self._load_image(nm), 0] for slot, nm in zip(slots, ref_names)}
       print(f"  reference slots: {listing(refs)}")
       n = self.node("MiniMaxH3ReferenceToVideo", clip=[clip, 0], vae=[vvae, 0],
                     audio_vae=[avae, 0], immediate=CFG["PROMPT"], width=w, peak=h,
                     size=size, ref_image_size=CFG["REF_IMAGE_SIZE"], **refs)
       return self._tail(mannequin, (n, 0), (n, 1), vvae, avae)

We assemble the MiniMax-H3 ComfyUI workflow graph completely in Python utilizing reusable node-building strategies. We assemble the mannequin spine, conditioning pipeline, sampler, schedulers, joint latent decoding, video creation, and output-saving levels for each normal and Turbo configurations. We additionally help text-to-video, first- and last-frame-conditioned video, and reference-image-conditioned video technology by way of the identical programmable graph structure.

def upload_image(path):
   """Multipart POST to /add/picture; returns the title LoadImage expects."""
   path = Path(path)
   if not path.exists():
       increase FileNotFoundError(path)
   boundary = uuid.uuid4().hex
   physique = (
       f"--{boundary}rnContent-Disposition: form-data; title="picture"; "
       f"filename="{path.title}"rnContent-Type: software/octet-streamrnrn"
   ).encode() + path.read_bytes() + (
       f"rn--{boundary}rnContent-Disposition: form-data; title="overwrite"rnrntrue"
       f"rn--{boundary}--rn"
   ).encode()
   req = urllib.request.Request(f"{API}/add/picture", information=physique,
                                headers={"Content-Type": f"multipart/form-data; boundary={boundary}"})
   with urllib.request.urlopen(req, timeout=120) as r:
       data = json.masses(r.learn())
   sub = data.get("subfolder") or ""
   print(f"  uploaded {path.title}")
   return f"{sub}/{data['name']}" if sub else data["name"]
def run_graph(graph, server, timeout=7200):
   """Submit, then observe the WebSocket for per-step progress."""
   import websocket
   cid = uuid.uuid4().hex
   Path("/content material/last_workflow_api.json").write_text(json.dumps(graph, indent=2))
   attempt:
       res = get_json("/immediate", {"immediate": graph, "client_id": cid})
   besides urllib.error.HTTPError as e:
       element = e.learn().decode()[:3000]
       increase SystemExit(f"Graph rejected by ComfyUI:n{element}")
   pid = res["prompt_id"]
   print(f"queued {pid} — first run masses ~37 GB of weights, be affected person")
   ws = websocket.WebSocket()
   ws.join(f"ws://127.0.0.1:{CFG['PORT']}/ws?clientId={cid}", timeout=60)
   t0, final = time.time(), ""
   attempt:
       whereas time.time() - t0 < timeout:
           attempt:
               msg = ws.recv()
           besides Exception:
               time.sleep(1)
               proceed
           if isinstance(msg, bytes):
               proceed
           d = json.masses(msg)
           t, information = d.get("sort"), d.get("information", {})
           if t == "executing" and information.get("prompt_id") == pid:
               if information.get("node") is None:
                   print(f"ndone in {time.time()-t0:.0f}s")
                   break
               cls = graph.get(information["node"], {}).get("class_type", information["node"])
               if cls != final:
                   print(f"n  -> {cls}", finish="", flush=True)
                   final = cls
           elif t == "progress":
               v, m = information.get("worth", 0), information.get("max", 1)
               print(f"r  -> {final}  {v}/{m}   ", finish="", flush=True)
           elif t == "execution_error":
               print("n--- execution error ---")
               print(json.dumps(information, indent=2)[:4000])
               print(server.tail())
               increase SystemExit("Generation failed.")
   lastly:
       ws.shut()
  
   recordsdata = []
   attempt:
       hist = get_json(f"/historical past/{pid}")
       for out in hist.get(pid, {}).get("outputs", {}).values():
           for gadgets in out.values():
               if isinstance(gadgets, listing):
                   for it in gadgets:
                       if isinstance(it, dict) and "filename" in it:
                           p = Path(CFG["OUT_DIR"]) / (it.get("subfolder") or "") / it["filename"]
                           if p.exists():
                               recordsdata.append(p)
   besides Exception:
       go
   if not recordsdata:
       cands = [p for p in Path(CFG["OUT_DIR"]).rglob("*")
                if p.suffix.decrease() in (".mp4", ".webm", ".mkv") and p.stat().st_mtime > t0]
       recordsdata = sorted(cands, key=lambda p: p.stat().st_mtime)
   return recordsdata
def fundamental():
   profile = preflight()
   install_comfy()
   mode = CFG["MODE"]
   unet, te, lora = download_weights(profile, mode)
   w, h = h3_canvas(CFG["ASPECT"], CFG["MEGAPIXELS"])
   size = align_frames(CFG["SECONDS"])
   print(f"ncanvas {w}x{h}, {size} frames "
         f"({size/24:.2f}s @24fps, grid examine {size % 17 == 5})")
   server = ComfyServer(profile["flags"])
   server.begin()
   attempt:
       schema = Schema()
       builder = H3Graph(schema, unet, te, lora)
       if mode == "r2v":
           if not CFG["REF_IMAGES"]:
               increase SystemExit("MODE='r2v' wants CFG['REF_IMAGES'] and <Picture N> tags "
                                "within the immediate.")
           names = [upload_image(p) for p in CFG["REF_IMAGES"][:9]]
           graph = builder.r2v(w, h, size, names)
       else:
           first = upload_image(CFG["FIRST_FRAME"]) if CFG["FIRST_FRAME"] else None
           final = upload_image(CFG["LAST_FRAME"]) if CFG["LAST_FRAME"] else None
           if mode == "flf2v" and not (first or final):
               increase SystemExit("MODE='flf2v' wants FIRST_FRAME and/or LAST_FRAME.")
           graph = builder.t2v_or_flf2v(w, h, size, first, final)
       print(f"graph: {len(graph)} nodes "
             f"({', '.be part of(sorted({n['class_type'] for n in graph.values()}))})")
       recordsdata = run_graph(graph, server)
       server.free_vram()
   lastly:
       server.cease()
   if not recordsdata:
       print("No output file discovered. Log tail:n", server.tail())
       return
   for f in recordsdata:
       print(f"noutput: {f}  ({f.stat().st_size/1e6:.1f} MB)")
   attempt:
       from IPython.show import Video, show
       vid = recordsdata[-1]
       if vid.stat().st_size < 60e6:
           show(Video(str(vid), embed=True, width=720))
       else:
           print("Too massive to embed — use recordsdata.obtain() or examine /content material/outputs")
   besides Exception:
       go
fundamental()

We deal with picture uploads, graph submission, WebSocket progress monitoring, output discovery, and the tutorial’s full execution circulation. We submit the generated graph to ComfyUI, monitor particular person node execution and sampling progress, gather the ensuing video recordsdata, and show manageable outputs immediately inside Colab. We lastly coordinate all earlier parts by way of the primary perform, taking the workflow from {hardware} preflight and mannequin loading to synchronized MiniMax-H3 video and audio technology.

In conclusion, we carried out a full programmable MiniMax-H3 inference pipeline that takes us from {hardware} validation and mannequin acquisition to graph execution and ultimate synchronized video-audio technology. We used ComfyUI as a headless server whereas controlling your complete workflow from Python, which provides us direct entry to configuration, mannequin loading, conditioning, sampling, decoding, server lifecycle administration, and generated outputs. We additionally made the pipeline extra sturdy by dynamically inspecting ComfyUI node schemas, adapting mannequin profiles to accessible VRAM, aligning body counts with MiniMax-H3 necessities, and supporting a number of conditioning modes by way of the identical reusable structure. By the top of the workflow, we’ve got a versatile basis that we are able to lengthen with totally different prompts, seeds, reference pictures, body constraints, LoRA acceleration, resolutions, and sampling methods whereas preserving a constant and automated MiniMax-H3 technology course of.


Check out the FULL CODES here. Also, be happy to observe us on Twitter and don’t overlook to hitch our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.

Need to accomplice with us for selling your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar and so on.? Connect with us

The submit Implementing a MiniMax-H3 Multimodal Video and Audio Generation Pipeline with ComfyUI APIs appeared first on MarkTechPost.

Similar Posts