Implementing a MiniMax-H3 Multimodal Video and Audio Generation Pipeline with ComfyUI APIs
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.check(cls, inputs)
self._id += 1
nid = str(self._id)
self.g[nid] = {“class_type”: cls, “inputs”: inputs}
return nid
def _backbone(self):
model = self.node(“UNETLoader”, unet_name=self.unet, weight_dtype=”default”)
if self.lora:
model = self.node(“LoraLoaderModelOnly”, model=[model, 0],
lora_name=self.lora, strength_model=1.0)
if CFG[“SIGMA_SHIFT”]:
sv, sa = CFG[“SIGMA_SHIFT”]
model = self.node(“MiniMaxH3SigmaShift”, model=[model, 0],
shift_video=float(sv), shift_audio=float(sa))
clip = self.node(“CLIPLoader”, clip_name=self.te, type=”minimax”, device=”default”)
vvae = self.node(“VAELoader”, vae_name=VAE_VIDEO)
avae = self.node(“VAELoader”, vae_name=VAE_AUDIO)
return model, clip, vvae, avae
def _tail(self, model, 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”, model=[model, 0], scheduler=sched,
steps=steps, denoise=1.0)
guider = self.node(“BasicGuider”, model=[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”, images=[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”, image=uploaded_name, upload=”image”)
def t2v_or_flf2v(self, w, h, length, first=None, last=None):
self.s.require(“MiniMaxH3ImageToVideo”, “SamplerCustomAdvanced”, “SaveVideo”)
model, clip, vvae, avae = self._backbone()
kw = {}
if first:
kw[“first_frame”] = [self._load_image(first), 0]
if last:
kw[“last_frame”] = [self._load_image(last), 0]
n = self.node(“MiniMaxH3ImageToVideo”, clip=[clip, 0], vae=[vvae, 0],
prompt=CFG[“PROMPT”], width=w, height=h, length=length, **kw)
return self._tail(model, (n, 0), (n, 1), vvae, avae)
def r2v(self, w, h, length, ref_names):
self.s.require(“MiniMaxH3ReferenceToVideo”)
model, 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: {list(refs)}”)
n = self.node(“MiniMaxH3ReferenceToVideo”, clip=[clip, 0], vae=[vvae, 0],
audio_vae=[avae, 0], prompt=CFG[“PROMPT”], width=w, height=h,
length=length, ref_image_size=CFG[“REF_IMAGE_SIZE”], **refs)
return self._tail(model, (n, 0), (n, 1), vvae, avae)
