video: serve MiniMax-H3 through an isolated worker, and clone voices in the studio

MiniMax-H3 is a 33B flow-matching DiT that denoises video and audio in ONE packed
sequence. diffusers ships it as Modular blocks only (MiniMaxH3ModularPipeline,
0.40+) — there is no DiffusionPipeline half, so it fits neither
_detect_pipeline_class() nor this venv's diffusers 0.38. It runs in an isolated
venv, the same way pyannote, NeMo-Canary and vLLM do:

  tools/h3_service.py   generation inside the isolated venv (HTTP)
  codai/api/h3_worker.py venv bootstrap, process lifecycle, eviction handle
  requirements-h3.txt   the venv's pins (diffusers>=0.40, transformers 5.x)

_generate_h3() adapts the request to the checkpoint's contracts instead of
forwarding it, and reports every adaptation in the response warnings: guidance is
baked in (negative_prompt and guidance_scale dropped), 24 fps is fixed and
num_frames snaps up to the next 17n+5 within 5-15s, canvas axes round to 32, and
v2v/interp are rejected. character_profiles become native ref2va references (max
9) so H3 skips the identity-keyframe bridge; an explicit keyframe wins, since
fl2va and ref2va are separate transformer partitions. The worker picks a workflow
per request and never loads without one — that pulls both 61.7GB partitions.

The worker registers as an evictable, VRAM-tracked model; eviction stops the
subprocess, the only way the transformer's pages actually come back.

Character studio: clone the voice off the same footage the face came from
(/v1/audio/voices/extract) with `extract --voice`, the new `voice` subcommand or
the UI checkbox, then `--say` a line that is lip-synced in that cloned voice.
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY
parent fa462b62
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Fully-managed MiniMax-H3 worker (video + native audio generation).
H3 is a 33B flow-matching DiT that denoises video and audio in ONE packed sequence.
diffusers ships it as Modular blocks only (``MiniMaxH3ModularPipeline``, 0.40+), so
it fits neither ``_detect_pipeline_class()`` nor the main venv's diffusers 0.38 —
it runs in an isolated venv, exactly like pyannote / NeMo-Canary / vLLM.
This module owns the venv + process lifecycle and the HTTP call; the generation
itself lives in ``tools/h3_service.py``. The service is registered with the model
manager as an evictable, VRAM-tracked model, so a big LLM can push H3 off the card
(eviction calls the handle's ``cleanup()``, which stops the subprocess).
Config keys (models.json entry for the H3 model, all optional)::
"backend": "h3", # or an id/path containing "minimax-h3"
"h3_venv": "/opt/coderai/h3_venv",
"dtype": "bfloat16",
"offload_strategy": "group", # group | leaf | '' (group is the default: 33B)
"device_map": "", # per-component device_map instead of offload
"used_vram_gb": 22.0, # what to reserve/evict for before starting
"gpu_device": 0
"""
import collections
import os
import socket
import subprocess
import sys
import threading
import time
from pathlib import Path
from typing import Optional
_REPO_ROOT = Path(__file__).resolve().parents[2]
_SERVICE_SCRIPT = _REPO_ROOT / "tools" / "h3_service.py"
_REQUIREMENTS = _REPO_ROOT / "requirements-h3.txt"
_BAKED_VENV = Path("/opt/coderai/h3_venv")
# Default reservation before starting the service. H3 is 33B; with group offload
# only the executing group is resident, but the VAEs, the text encoder and the
# activations still want room. Override per model with `used_vram_gb`.
DEFAULT_VRAM_GB = 22.0
_lock = threading.RLock()
_services: dict = {} # model -> {"proc","port","url"}
_bootstrapped = False
# ── detection ─────────────────────────────────────────────────────────────────
def is_h3_model(model_name: str = "", config: dict = None) -> bool:
"""True when this model must go through the H3 worker.
Either an explicit ``"backend": "h3"`` in models.json, or a name/path that
carries the checkpoint's identity (MiniMax-H3, minimax_h3, minimax-h3-…).
"""
cfg = config or {}
backend = str(cfg.get("backend") or "").strip().lower()
if backend in ("h3", "minimax-h3", "minimax_h3"):
return True
if backend and backend not in ("diffusers", "video", ""):
return False
haystack = " ".join(str(x or "") for x in (
model_name, cfg.get("model_path"), cfg.get("model_id"), cfg.get("path")))
h = haystack.lower().replace("_", "-")
return "minimax-h3" in h
def resolve_model_path(model_name: str, config: dict = None) -> str:
cfg = config or {}
for key in ("model_path", "path", "model_id", "repo_id"):
value = (cfg.get(key) or "").strip() if isinstance(cfg.get(key), str) else ""
if value:
return os.path.expanduser(value)
return model_name or "MiniMaxAI/MiniMax-H3"
# ── venv ──────────────────────────────────────────────────────────────────────
def resolve_venv_dir(config: dict = None) -> Path:
configured = ((config or {}).get("h3_venv") or "").strip()
if configured:
return Path(os.path.expanduser(configured))
explicit = os.environ.get("CODERAI_H3_VENV")
if explicit:
return Path(os.path.expanduser(explicit))
if (_BAKED_VENV / ("Scripts" if os.name == "nt" else "bin")).exists():
return _BAKED_VENV
cache = os.environ.get("CODERAI_CACHE_DIR") or ("/cache" if os.path.isdir("/cache") else "")
if cache and os.path.isdir(cache):
return Path(cache) / "h3_venv"
return Path(os.path.expanduser("~/.coderai/h3_venv"))
def _venv_python(venv: Path) -> Path:
return venv / ("Scripts" if os.name == "nt" else "bin") / (
"python.exe" if os.name == "nt" else "python")
def _diffusers_ok(py: Path) -> bool:
"""The venv is usable only if its diffusers actually carries the H3 blocks."""
try:
return subprocess.run(
[str(py), "-c", "from diffusers import MiniMaxH3ModularPipeline"],
capture_output=True).returncode == 0
except Exception:
return False
def _bootstrap_venv(config: dict = None) -> Path:
global _bootstrapped
venv = resolve_venv_dir(config)
py = _venv_python(venv)
if _bootstrapped and py.exists():
return py
if py.exists() and _diffusers_ok(py):
_bootstrapped = True
return py
if not py.exists():
print(f"[h3] creating isolated venv at {venv} …", flush=True)
venv.parent.mkdir(parents=True, exist_ok=True)
subprocess.run([sys.executable, "-m", "venv", str(venv)], check=True)
py = _venv_python(venv)
if not _diffusers_ok(py):
if not _REQUIREMENTS.exists():
raise RuntimeError(f"H3 requirements file missing: {_REQUIREMENTS}")
print("[h3] installing diffusers>=0.40 + torch into the isolated venv "
"(first run downloads several GB, this takes a while) …", flush=True)
subprocess.run([str(py), "-m", "pip", "install", "-U", "pip"], check=True)
subprocess.run([str(py), "-m", "pip", "install", "-r", str(_REQUIREMENTS)],
check=True)
if not _diffusers_ok(py):
raise RuntimeError(
"H3 venv built but 'from diffusers import MiniMaxH3ModularPipeline' "
"still fails — MiniMax-H3 needs diffusers >= 0.40.0")
_bootstrapped = True
return py
# ── service lifecycle ─────────────────────────────────────────────────────────
def _free_port() -> int:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("127.0.0.1", 0))
port = s.getsockname()[1]
s.close()
return port
def _pump_logs(proc, tail):
for line in proc.stdout:
line = line.rstrip()
if line:
tail.append(line)
print(f"[h3] {line}", flush=True)
def _health_ok(url: str) -> bool:
import requests
try:
r = requests.get(url + "/health", timeout=5)
return r.ok and bool(r.json().get("ok"))
except Exception:
return False
def ensure_service(model_path: str, config: dict = None,
ready_timeout: float = 1800.0) -> str:
"""Start (or reuse) the H3 service for one checkpoint; return its base URL."""
config = config or {}
with _lock:
svc = _services.get(model_path)
if svc and svc["proc"].poll() is None and _health_ok(svc["url"]):
return svc["url"]
if svc and svc["proc"].poll() is not None:
_services.pop(model_path, None)
py = _bootstrap_venv(config)
port = _free_port()
url = f"http://127.0.0.1:{port}"
env = dict(os.environ)
try:
from codai.models.cache import get_hf_hub_cache_dir
hub = get_hf_hub_cache_dir()
env["HF_HUB_CACHE"] = hub
env["HUGGINGFACE_HUB_CACHE"] = hub
except Exception:
pass
gpu = config.get("gpu_device")
if gpu is not None:
env["CUDA_VISIBLE_DEVICES"] = str(gpu)
offload = str(config.get("offload_strategy") or "group").lower()
if offload not in ("group", "leaf", ""):
offload = "group" # 'model'/'sequential'/'auto' have no modular equivalent
cmd = [str(py), str(_SERVICE_SCRIPT),
"--model", model_path,
"--host", "127.0.0.1", "--port", str(port),
"--dtype", str(config.get("dtype") or "bfloat16"),
"--offload", offload]
if config.get("device_map"):
cmd += ["--device-map", str(config["device_map"])]
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, bufsize=1, env=env, cwd=str(_REPO_ROOT))
tail = collections.deque(maxlen=15)
threading.Thread(target=_pump_logs, args=(proc, tail), daemon=True).start()
_services[model_path] = {"proc": proc, "port": port, "url": url}
def _tail_msg():
joined = " | ".join(list(tail)[-5:]).strip()
return f". Last output: {joined}" if joined else ""
deadline = time.time() + ready_timeout
while time.time() < deadline:
if proc.poll() is not None:
_services.pop(model_path, None)
raise RuntimeError(
f"H3 worker exited (code {proc.returncode}) before becoming ready"
+ _tail_msg())
if _health_ok(url):
print(f"[h3] service ready for {model_path} at {url}", flush=True)
return url
time.sleep(2)
stop_service(model_path)
raise RuntimeError(f"H3 worker for {model_path} did not become ready in time"
+ _tail_msg())
def is_running(model_path: str) -> bool:
with _lock:
svc = _services.get(model_path)
return bool(svc and svc["proc"].poll() is None)
def stop_service(model_path: str) -> None:
with _lock:
svc = _services.pop(model_path, None)
if not svc:
return
proc = svc["proc"]
if proc.poll() is None:
try:
proc.terminate()
proc.wait(timeout=15)
except Exception:
pass
if proc.poll() is None:
try:
proc.kill()
except Exception:
pass
print(f"[h3] service for {model_path} stopped", flush=True)
def stop_all() -> None:
for name in list(_services.keys()):
stop_service(name)
# ── model-manager integration ─────────────────────────────────────────────────
class H3Handle:
"""Evictable handle registered with the model manager.
``cleanup()`` is what the manager calls to reclaim VRAM: it stops the whole
subprocess, which is the only way to be sure the 33B transformer's pages are
gone (dropping the pipeline inside the worker would leave the process's CUDA
context and allocator arenas behind).
"""
def __init__(self, model_path: str, url: str):
self.model_path = model_path
self.url = url
def cleanup(self):
try:
stop_service(self.model_path)
except Exception:
pass
# The manager's RAM/VRAM accounting probes these on loaded models.
def to(self, *args, **kwargs):
return self
def acquire(model_name: str, config: dict = None) -> H3Handle:
"""Ensure the worker runs and is registered as an evictable, VRAM-tracked model."""
from codai.models.manager import multi_model_manager
cfg = config or {}
model_path = resolve_model_path(model_name, cfg)
key = f"h3:{model_path}"
# Idempotent: reuses a healthy worker, re-spawns a dead one (new port).
url = ensure_service(model_path, cfg)
def _loader():
return H3Handle(model_path, url)
needed = float(cfg.get("used_vram_gb") or DEFAULT_VRAM_GB)
# acquire_stt_backend is the manager's generic "register an out-of-band backend
# as an evictable, VRAM-tracked model" entry point (STT is just where it was
# first needed) — same call the diarization worker uses.
handle = multi_model_manager.acquire_stt_backend(key, needed, True, _loader,
keep_resident=False)
handle.url = url # a re-spawned worker may sit on a new port
return handle
def generate(model_name: str, payload: dict, config: dict = None,
timeout: float = 7200.0) -> dict:
"""Run one generation on the worker. Returns the service's JSON response."""
import requests
handle = acquire(model_name, config)
resp = requests.post(handle.url + "/generate", json=payload, timeout=timeout)
if not resp.ok:
detail = resp.text[:800]
raise RuntimeError(f"H3 generation failed ({resp.status_code}): {detail}")
data = resp.json()
if data.get("error"):
raise RuntimeError(f"H3 generation failed: {data['error']}")
return data
import atexit as _atexit
_atexit.register(stop_all)
......@@ -3576,9 +3576,157 @@ async def get_video_progress():
# Main generation endpoint
# =============================================================================
#: Video pipeline families that can do IP-Adapter themselves, so they don't need the
#: keyframe bridge. Everything else (Wan, CogVideoX, LTX, SVD, …) does.
_NATIVE_IP_ADAPTER_VIDEO = ('animatediff', 'animate-diff')
#: Video pipeline families that condition on character references themselves, so they
#: don't need the keyframe bridge: AnimateDiff has real IP-Adapter, and MiniMax-H3's
#: ref2va workflow takes the reference images directly (up to 9). Everything else
#: (Wan, CogVideoX, LTX, SVD, …) needs the bridge.
_NATIVE_IP_ADAPTER_VIDEO = ('animatediff', 'animate-diff', 'minimax-h3', 'minimax_h3')
async def _generate_h3(request: VideoGenerationRequest, model_name: str,
model_cfg: dict, http_request,
identity_note: Optional[str] = None):
"""Generate on MiniMax-H3 through the isolated worker.
H3 differs from every other video model here in ways the request has to be
translated for, not just forwarded:
* It is guidance-distilled — no negative prompt, no guidance scale. Both are
dropped (with a warning, so a caller that set them isn't left guessing).
* Video and audio are denoised jointly, so the returned mp4 already HAS a
soundtrack. The post-processing chain still runs on top for lip sync,
dialogs, subtitles and upscaling.
* Character identity is native: the profile's reference images go in as
ref2va references (up to 9) instead of through the keyframe bridge.
* Frame count and canvas are checkpoint contracts (24 fps, 17n+5 frames,
5-15 s, axes multiple of 32); the worker snaps them and reports what it
actually generated.
"""
from codai.api import h3_worker
warnings: list = []
if identity_note:
warnings.append(identity_note)
mode = (request.mode or 't2v').lower()
if mode in ('v2v', 'interp'):
raise HTTPException(status_code=400, detail=(
f"MiniMax-H3 has no '{mode}' workflow — it supports t2va (prompt), "
"fl2va (first and/or last keyframe) and ref2va (reference media)."))
if request.negative_prompt:
warnings.append("MiniMax-H3 is guidance-distilled: negative_prompt ignored")
if request.guidance_scale is not None:
warnings.append("MiniMax-H3 is guidance-distilled: guidance_scale ignored")
def _as_b64(ref: str) -> str:
"""Normalise a base64 / data-URI / URL reference to plain base64."""
return base64.b64encode(_decode_b64_or_url(ref)).decode()
payload = {
"prompt": request.prompt or "",
"num_inference_steps": request.num_inference_steps or 50,
}
if request.num_frames:
payload["num_frames"] = int(request.num_frames)
if request.fps and int(request.fps) != 24:
warnings.append("MiniMax-H3 generates at a fixed 24 fps; fps ignored")
if request.width and request.height:
payload["width"], payload["height"] = int(request.width), int(request.height)
if request.seed is not None:
payload["seed"] = int(request.seed)
first = request.init_image or request.image
if first:
payload["image"] = _as_b64(first)
if request.end_image:
payload["last_image"] = _as_b64(request.end_image)
# ── native identity conditioning (ref2va) ───────────────────────────────
refs: list = []
try:
if request.character_profiles:
from codai.api.characters import resolve_character_profiles
for img in resolve_character_profiles(list(request.character_profiles)):
refs.append({"type": "image", "data": _as_b64(img)})
except Exception as e:
print(f" [h3] character profile resolution failed: {e}")
for img in (request.character_references or []):
refs.append({"type": "image", "data": _as_b64(img)})
for entry in (request.characters or []):
for img in (entry.get("images") or []):
refs.append({"type": "image", "data": _as_b64(img)})
if refs:
# The checkpoint takes at most 9 image references; keep the first ones,
# which is the order the caller considered most representative.
if len(refs) > 9:
warnings.append(f"MiniMax-H3 takes 9 image references; used the first 9 of {len(refs)}")
refs = refs[:9]
if payload.get("image") or payload.get("last_image"):
# fl2va and ref2va are different transformer partitions — a request
# can't be both. The explicit keyframe wins; say so.
warnings.append("MiniMax-H3: keyframe supplied, so character references were dropped "
"(fl2va and ref2va are separate checkpoints)")
else:
payload["references"] = refs
if getattr(request, 'loras', None):
adapters = []
for lora in request.loras:
path = getattr(lora, 'model', None) or getattr(lora, 'path', None)
if path:
adapters.append({"path": path, "weight": float(getattr(lora, 'weight', 1.0) or 1.0)})
if adapters:
payload["loras"] = adapters
_vid_progress_reset(int(payload["num_inference_steps"]))
try:
data = await asyncio.to_thread(h3_worker.generate, model_name, payload, model_cfg)
finally:
_vid_progress_done()
mp4_bytes = base64.b64decode(data.get("mp4_b64") or "")
if not mp4_bytes:
raise HTTPException(status_code=500, detail="MiniMax-H3 returned no video")
print(f" [h3] {data.get('workflow')}: {data.get('num_frames')} frames "
f"({data.get('seconds')}s @ {data.get('fps')}fps), {len(mp4_bytes)/1e6:.1f} MB")
temps: list = []
try:
needs_post = any([
request.upscale_output, request.interpolate_output, request.add_audio,
request.tts_text, request.dialogs, request.lip_sync,
request.generate_subtitles, request.burn_subtitles,
])
if needs_post:
mp4_bytes = await asyncio.get_event_loop().run_in_executor(
None, _postprocess_video, mp4_bytes, request, http_request, temps, warnings)
finally:
for path in temps:
try:
if os.path.isfile(path):
os.unlink(path)
except Exception:
pass
result = _save_file(mp4_bytes, 'mp4', http_request)
try:
from codai.api.archive import archive_manager
asyncio.get_event_loop().create_task(asyncio.to_thread(
archive_manager.save_generation,
"video", "/v1/video/generations", request.model, request.prompt or "",
{"mode": request.mode, "workflow": data.get("workflow"),
"num_frames": data.get("num_frames"), "fps": data.get("fps"),
"width": data.get("width"), "height": data.get("height"),
"num_inference_steps": payload["num_inference_steps"],
"seed": request.seed},
[(mp4_bytes, "mp4")]))
except Exception:
pass
resp = VideoGenerationResponse(created=int(time.time()), data=[result])
if warnings:
resp.warnings = warnings
return resp
async def _maybe_build_identity_keyframe(request: VideoGenerationRequest,
......@@ -3733,6 +3881,20 @@ async def video_generations(request: VideoGenerationRequest,
# Always define _model_cfg — it's used later regardless of whether we load now.
_model_cfg = model_info.get('config') or {}
# ── MiniMax-H3 ───────────────────────────────────────────────────────────
# H3 is Modular-Diffusers-only and needs diffusers >= 0.40, so it runs in an
# isolated venv behind codai/api/h3_worker.py rather than through the loader
# below. Branch before any of it: the worker owns the weights, the VRAM
# reservation and the eviction hook.
try:
from codai.api import h3_worker
_is_h3 = h3_worker.is_h3_model(model_name, _model_cfg)
except Exception:
_is_h3 = False
if _is_h3:
return await _generate_h3(request, model_name, _model_cfg, http_request,
_identity_note)
# Refuse to load onto a poisoned CUDA context — it would just re-assert.
if getattr(multi_model_manager, 'cuda_context_poisoned', False):
raise HTTPException(status_code=503, detail=(
......
......@@ -116,6 +116,15 @@ def detect_model_capabilities(model_name: str) -> ModelCapabilities:
return caps
# ── Video generation ─────────────────────────────────────────────────────
# MiniMax-H3 generates video with a natively-synchronised soundtrack (one DiT
# denoises both), so it carries the audio capability too. Served through the
# isolated worker in codai/api/h3_worker.py, not the diffusers loader.
if 'minimax-h3' in n or 'minimax_h3' in n:
caps.video_generation = True
caps.image_to_video = True
caps.audio_generation = True
return caps
if any(x in n for x in ['cogvideox', 'cogvideo', 'ltx-video', 'ltxvideo',
'hunyuan-video', 'mochi-1', 'dynamicrafter',
'animatediff', 'text2video', 'modelscope-t2v',
......
# MiniMax-H3 (video + native audio)
H3 is a 33B flow-matching DiT that denoises **video and audio in one packed
sequence** — every generation comes back as an mp4 that already has a synchronised
soundtrack, not a dub. It is served through an isolated worker rather than the
normal diffusers loader, for one blunt reason: diffusers ships H3 as Modular
Diffusers blocks only (`MiniMaxH3Blocks` / `MiniMaxH3ModularPipeline`, 0.40+),
there is no `DiffusionPipeline` half, and the main venv is on diffusers 0.38.
```
/v1/video/generations ──► codai/api/video.py::_generate_h3
└─► codai/api/h3_worker.py (venv + process + eviction)
└─► tools/h3_service.py (isolated venv, HTTP)
└─► ModularPipeline.from_pretrained(..., workflow=…)
```
Same shape as the pyannote, NeMo-Canary and vLLM workers.
## Configure a model
Add an entry to `models.json`. Detection is by name (anything containing
`minimax-h3` / `minimax_h3`) or by an explicit backend:
```json
{
"minimax-h3": {
"type": "video",
"backend": "h3",
"model_path": "MiniMaxAI/MiniMax-H3",
"offload_strategy": "group",
"dtype": "bfloat16",
"used_vram_gb": 22.0
}
}
```
| key | meaning |
|---|---|
| `offload_strategy` | `group` (default), `leaf` (smaller groups, slower), or `""` for straight-to-GPU. The diffusers CPU-offload strategies have no modular equivalent and are normalised to `group`. |
| `device_map` | per-component device map (`auto`, `balanced`) instead of group offload |
| `used_vram_gb` | reserved before start, and what eviction frees. Default 22. |
| `h3_venv` | override the isolated venv location |
| `gpu_device` | pins `CUDA_VISIBLE_DEVICES` for the worker |
The venv builds itself on first use from `requirements-h3.txt` (several GB) under
`/opt/coderai/h3_venv`, the `/cache` mount, or `~/.coderai/h3_venv`.
## What the request translation does
H3 has hard checkpoint contracts, so `_generate_h3` adapts the request instead of
forwarding it, and reports every adaptation in the response `warnings`:
- **Guidance-distilled** — no negative prompt, no guidance scale. Both are dropped.
- **24 fps, fixed.** `num_frames` is snapped **up** to the next `17n + 5` the video
VAE can decode, clamped to 5–15 s. `fps` on the request is ignored.
- **Canvas** axes are rounded down to a multiple of 32 (checkpoint short edge 768,
area cap 768×1344).
- **Three workflows**, each loading its own ~61.7 GB transformer partition:
`t2va` (prompt), `fl2va` (first and/or last keyframe, `transformer/`), `ref2va`
(up to 12 references, `transformer_ref/`). The worker picks one per request and
**reloads** when it changes — loading with no workflow pulls *both* partitions.
- **Identity is native**: `character_profiles` go in as `ref2va` image references
(max 9), so H3 skips the identity-keyframe bridge other video models need. An
explicit keyframe wins over references, because fl2va and ref2va are different
checkpoints and a request cannot be both.
- `v2v` and `interp` are rejected — H3 has no such workflow.
Post-processing (upscale, interpolation, dialogs, lip sync, subtitles) runs on top
of the returned mp4 exactly as for any other video model, so a cloned voice line
still lands over H3's own soundtrack.
## LoRAs
`loras` on the request are forwarded to the worker and loaded with
`MiniMaxH3LoraLoaderMixin`, with `set_adapters` weights. Public recipes for
training one: ~32 face crops at 512px, rank/alpha 16, LR 1e-4, ~1000 steps with
the audio loss zeroed (stills carry no audio); or 50–200 clips of 3–15 s at
*exactly* 24.000 fps with the audio kept, rank 16, 1500–5000 steps.
## Operational notes
- **Weights are ~62 GB per partition.** Budget disk and first-run download time.
- The worker is registered as an evictable, VRAM-tracked model; eviction stops the
whole subprocess, which is the only way the transformer's pages actually go back.
- Switching workflow within one model reloads the other partition — batch requests
by workflow if you care about wall-clock.
- The service logs under `[h3]`; failures surface with the worker's last output.
# Isolated venv for MiniMax-H3 video+audio generation.
#
# H3 exists ONLY as Modular Diffusers blocks (MiniMaxH3Blocks /
# MiniMaxH3ModularPipeline), which landed in diffusers 0.40.0. The main coderai
# venv is on diffusers 0.38.x, and moving the whole stack to 0.40 would shift the
# floor under every Wan / LTX / Z-Image / SDXL pipeline at once — so H3 gets its
# own venv, the same way pyannote, NeMo-Canary and vLLM do.
# codai/api/h3_worker.py builds it (on demand under ~/.coderai/h3_venv or the
# /cache mount) and runs tools/h3_service.py inside it.
# Do NOT merge these into requirements.txt.
# The transformer is 33B and the text encoder is Qwen3-VL, so this venv wants a
# recent torch. Left unpinned deliberately: unlike pyannote there is no known-bad
# version to dodge, and the H3 blocks track diffusers main.
--extra-index-url https://download.pytorch.org/whl/cu130
torch
torchvision
torchaudio
# The floor for MiniMax-H3 support.
diffusers>=0.40.0
# Qwen3VLForConditionalGeneration (H3's text encoder) needs transformers 5.x.
transformers>=5.0
accelerate>=1.0
safetensors>=0.4
huggingface_hub>=0.34
# Media: export_to_video writes the frames, soundfile writes the jointly-generated
# soundtrack, and ffmpeg (system binary, already on PATH in the image) muxes them.
imageio>=2.34
imageio-ffmpeg>=0.5
soundfile>=0.12
Pillow>=10.0
numpy
# peft carries the LoRA adapters MiniMaxH3LoraLoaderMixin loads.
peft>=0.14
......@@ -10,6 +10,8 @@ keeps local copies of the results:
- GET /v1/characters list saved profiles
- GET /v1/characters/{name} fetch a profile's reference images
- PATCH/DELETE /v1/characters/ prune bad references / remove a profile
- POST /v1/audio/voices/extract clone the voice off the same footage, so the
character can speak in their own voice
- POST /v1/loras/train optionally train an identity LoRA straight from
the profile's crops (`character: <name>`), for
models where IP-Adapter alone drifts
......@@ -65,6 +67,7 @@ DEFAULT_OUT_DIR = os.environ.get("CODERAI_CHARACTER_OUT", "character_output")
IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif"}
VIDEO_EXTS = {".mp4", ".mkv", ".mov", ".avi", ".webm", ".m4v"}
AUDIO_EXTS = {".wav", ".mp3", ".m4a", ".aac", ".flac", ".ogg", ".opus"}
DEFAULT_NEGATIVE = (
"different person, face morphing, distorted face, extra fingers, blurry, "
......@@ -264,6 +267,19 @@ class CoderAIClient:
def delete_character(self, name: str) -> dict[str, Any]:
return self._delete(f"/v1/characters/{urllib.parse.quote(name)}")
# ── voices ──────────────────────────────────────────────────────────────
def list_voices(self) -> list[dict[str, Any]]:
try:
return self._get("/v1/audio/voices").get("voices", [])
except Exception:
return []
def extract_voice(self, name: str, description: str, media: str,
is_video: bool, transcript: str = "") -> dict[str, Any]:
body = {"name": name, "description": description, "transcript": transcript}
body["video" if is_video else "audio"] = media
return self._post("/v1/audio/voices/extract", body)
# ── LoRAs ───────────────────────────────────────────────────────────────
def list_loras(self) -> list[dict[str, Any]]:
try:
......@@ -358,7 +374,8 @@ class CharacterStudio:
# ── step 1: extraction ──────────────────────────────────────────────────
def extract(self, name: str, sources: list[str], description: str = "",
max_images: int = 5, emit=log) -> dict[str, Any]:
max_images: int = 5, emit=log, with_voice: bool = False,
transcript: str = "") -> dict[str, Any]:
images, videos = split_sources(sources)
emit(f"Extracting '{name}' from {len(images)} image(s) and {len(videos)} video(s)...")
result = self.client.extract_character(name, description, images, videos, max_images)
......@@ -369,8 +386,49 @@ class CharacterStudio:
emit(f"Contact sheet: {mirrored['contact_sheet']}")
else:
emit("Contact sheet skipped (install Pillow to get one)")
if with_voice:
try:
mirrored["voice"] = self.extract_voice_profile(
name, sources, emit=emit, transcript=transcript)
except Exception as exc:
emit(f"Voice clone failed ({exc}) — the character is still usable")
return mirrored
# ── step 1a: the voice off the same footage ─────────────────────────────
def extract_voice_profile(self, name: str, sources: list[str], emit=log,
transcript: str = "") -> str | None:
"""Clone a voice profile from the first audio-bearing source.
The face and the voice come from the same clip, so the character can speak
in their own voice later (`--say`), instead of a stock TTS voice. Silently
returns None when only stills were supplied — there is nothing to clone.
"""
media, is_video = None, True
for item in sources:
if item.startswith(("http://", "https://")):
if kind_of(Path(urllib.parse.urlparse(item).path)) == "video":
media = item
break
continue
if item.startswith("data:"):
if item[5:].split(";", 1)[0].startswith(("video/", "audio/")):
media, is_video = item, item[5:].startswith("video/")
break
continue
path = Path(item)
if kind_of(path) == "video" or path.suffix.lower() in AUDIO_EXTS:
media = data_uri_for_file(path)
is_video = kind_of(path) == "video"
break
if not media:
emit("No video/audio source — skipping the voice clone (stills carry no voice)")
return None
voice = f"{safe_slug(name)}_voice"
emit(f"Cloning voice '{voice}' from the source footage...")
self.client.extract_voice(voice, f"voice of {name}", media, is_video, transcript)
emit(f"Voice profile ready: {voice}")
return voice
# ── step 1b: an identity LoRA trained from the same crops ───────────────
def _lora_jobs_path(self) -> Path:
return self.out_dir / "lora_jobs.json"
......@@ -550,6 +608,22 @@ class CharacterStudio:
body["keyframe_steps"] = int(opts["keyframe_steps"])
if opts.get("camera_motion"):
body["camera_motion"] = opts["camera_motion"]
say = (opts.get("say") or "").strip()
if say:
# A saved voice profile makes the server clone THAT voice for the line;
# a bare id (e.g. af_sarah) is a stock TTS voice. Lip sync is applied to
# the generated face either way.
voice = (opts.get("voice") or f"{safe_slug(name)}_voice").strip()
body["dialogs"] = [{"character": name, "voice": voice, "text": say,
"lip_sync": bool(opts.get("lip_sync", True)),
"speed": float(opts.get("speech_speed") or 1.0)}]
body["lip_sync"] = bool(opts.get("lip_sync", True))
if opts.get("lip_sync_method"):
body["lip_sync_method"] = opts["lip_sync_method"]
body["add_audio"] = True
body["audio_type"] = "speech"
emit(f"Dialog: '{say[:60]}' in voice '{voice}'"
+ (" with lip sync" if body["lip_sync"] else ""))
adapters = lora_specs(opts.get("loras"), float(opts.get("lora_weight") or 1.0))
if adapters:
body["loras"] = adapters
......@@ -643,6 +717,8 @@ class CharacterStudio:
description=payload.get("description") or "",
max_images=int(payload.get("max_images") or 5),
emit=emit,
with_voice=bool(payload.get("with_voice")),
transcript=payload.get("transcript") or "",
)
self._job_update(job_id, status="done", progress=100, result=profile)
except Exception as exc:
......@@ -691,6 +767,7 @@ class CharacterStudio:
return {
"characters": self.client.list_characters(),
"loras": self.client.list_loras(),
"voices": self.client.list_voices(),
"models": [
{"id": m.get("id"), "capabilities": m.get("capabilities") or []}
for m in models
......@@ -790,6 +867,7 @@ video{width:100%;border-radius:14px;margin-top:12px;background:#000}
<div class="drop" id="drop">Drop files here, or click to choose<br><span class="muted">jpg &middot; png &middot; webp &middot; mp4 &middot; mov &middot; mkv &middot; webm</span></div>
<input id="picker" type="file" multiple accept="image/*,video/*" style="display:none">
<div class="files" id="filelist"></div>
<div class="chk"><input id="with_voice" type="checkbox"><span>Also clone the voice from the same footage (needs a video/audio source)</span></div>
<button class="btn" id="btn_extract">Extract character</button>
<div class="warn">Files are base64-encoded into the request — keep source clips short (a few seconds is plenty; frames are sampled evenly).</div>
</div>
......@@ -858,6 +936,13 @@ video{width:100%;border-radius:14px;margin-top:12px;background:#000}
<div><label>Apply LoRAs (ctrl-click for several)</label><select id="video_loras" multiple size="4"></select></div>
<div><label>LoRA weight</label><input id="lora_weight" type="number" step="0.05" value="1.0"></div>
</div>
<label>Spoken line <span class="muted">(lip-synced, in the cloned voice)</span></label>
<input id="say" placeholder="I told you I could ride">
<div class="row3">
<div><label>Voice</label><select id="voice"></select></div>
<div><label>Speed</label><input id="speech_speed" type="number" step="0.05" value="1.0"></div>
<div><label>Lip sync</label><select id="lip_sync"><option value="1">on</option><option value="0">off</option></select></div>
</div>
<label>Negative prompt</label>
<input id="negative_prompt" placeholder="(server default)">
<div class="chk"><input id="portrait" type="checkbox"><span>Render a still identity portrait first (fast sanity check before the slow video)</span></div>
......@@ -901,6 +986,9 @@ async function loadState(){
const keep=$('charsel').value;
$('charsel').innerHTML=options(names, names.includes(keep)?keep:(state.characters[0]||{}).name);
$('video_model').innerHTML=options(capModels('video_generation'), state.defaults.video_model);
const voiceNames=(state.voices||[]).map(v=>v.name||v.id).filter(Boolean);
const keptVoice=$('voice').value;
$('voice').innerHTML='<option value="">(character\'s own voice)</option>'+options(voiceNames, keptVoice);
const loraNames=(state.loras||[]).map(l=>l.name||l.id).filter(Boolean);
const keptLoras=selected($('video_loras'));
$('video_loras').innerHTML=loraNames.map(n=>`<option ${keptLoras.includes(n)?'selected':''}>${esc(n)}</option>`).join('')
......@@ -948,7 +1036,8 @@ $('btn_extract').onclick=async()=>{
try{
const d=await api('/api/extract',{method:'POST',body:JSON.stringify({
name:$('name').value, description:$('description').value,
max_images:+$('max_images').value||5, sources:pending.map(f=>f.data)})});
max_images:+$('max_images').value||5, with_voice:$('with_voice').checked,
sources:pending.map(f=>f.data)})});
watch(d.job_id, async j=>{pending=[]; renderFiles(); await loadState(); if(j.result) {$('charsel').value=j.result.name; loadRefs(j.result.name)} $('btn_extract').disabled=false});
}catch(e){logln('Error: '+e.message); $('btn_extract').disabled=false}
};
......@@ -990,6 +1079,8 @@ $('btn_video').onclick=async()=>{
character_strength:+$('character_strength').value, keyframe_identity:$('keyframe_identity').value,
camera_motion:$('camera_motion').value, negative_prompt:$('negative_prompt').value,
loras:selected($('video_loras')), lora_weight:+$('lora_weight').value,
say:$('say').value, voice:$('voice').value,
lip_sync:$('lip_sync').value==='1', speech_speed:+$('speech_speed').value,
portrait:$('portrait').checked};
const d=await api('/api/video',{method:'POST',body:JSON.stringify(body)});
watch(d.job_id, j=>{
......@@ -1137,6 +1228,12 @@ def add_video_options(parser: argparse.ArgumentParser) -> None:
"tilt-up", "tilt-down", "rotate"])
parser.add_argument("--portrait", action="store_true",
help="Also render a still identity portrait (fast check before the slow video)")
parser.add_argument("--say", default="",
help="Have the character speak this line, lip-synced, in their cloned voice")
parser.add_argument("--voice", default="",
help="Voice profile or TTS voice id for --say (default: <character>_voice)")
parser.add_argument("--no-lip-sync", action="store_true", help="Speak without lip sync")
parser.add_argument("--speech-speed", type=float, default=1.0)
parser.add_argument("--lora", action="append", default=[], metavar="NAME[:WEIGHT]",
help="Apply a trained LoRA to the video (repeatable, e.g. alice_identity:0.9)")
parser.add_argument("--image-lora", action="append", default=[], metavar="NAME[:WEIGHT]",
......@@ -1161,6 +1258,10 @@ def video_opts(args: argparse.Namespace) -> dict[str, Any]:
"camera_motion": args.camera_motion,
"loras": args.lora,
"image_loras": args.image_lora,
"say": args.say,
"voice": args.voice,
"lip_sync": not args.no_lip_sync,
"speech_speed": args.speech_speed,
}
......@@ -1212,8 +1313,9 @@ def build_parser() -> argparse.ArgumentParser:
" %(prog)s show alice\n"
" %(prog)s prune alice --drop 2 --drop 4 # a bystander got picked up\n"
" %(prog)s train alice --train-steps 1200 # identity LoRA from those crops\n"
" %(prog)s voice alice --source clips/alice.mp4 # clone her voice too\n"
" %(prog)s video alice --prompt 'riding a horse along a stormy beach' \\\n"
" --lora alice_identity:0.9\n\n"
" --lora alice_identity:0.9 --say 'I told you I could ride'\n\n"
" # browser UI\n"
" %(prog)s web --browser\n"
),
......@@ -1232,6 +1334,10 @@ def build_parser() -> argparse.ArgumentParser:
help="Photo, video, directory, glob or URL (repeatable)")
p_extract.add_argument("--description", default="", help="Free-text description saved with the profile")
p_extract.add_argument("--max-images", type=int, default=5, help="Max reference crops to keep")
p_extract.add_argument("--voice", action="store_true",
help="Also clone the voice from the same footage (needs a video/audio source)")
p_extract.add_argument("--transcript", default="",
help="Transcript of the reference audio (auto-transcribed when omitted)")
sub.add_parser("list", help="List character profiles on the server")
......@@ -1251,6 +1357,12 @@ def build_parser() -> argparse.ArgumentParser:
p_train.add_argument("--seed", type=int, default=42)
add_train_options(p_train)
p_voice = sub.add_parser("voice", help="Clone a voice profile from a character's footage")
p_voice.add_argument("name")
p_voice.add_argument("--source", action="append", required=True, metavar="PATH|URL",
help="Video or audio carrying the voice (repeatable)")
p_voice.add_argument("--transcript", default="")
p_video = sub.add_parser("video", help="Generate a new video starring a saved character")
p_video.add_argument("name")
add_video_options(p_video)
......@@ -1260,6 +1372,9 @@ def build_parser() -> argparse.ArgumentParser:
p_demo.add_argument("--source", action="append", required=True, metavar="PATH|URL")
p_demo.add_argument("--description", default="")
p_demo.add_argument("--max-images", type=int, default=5)
p_demo.add_argument("--clone-voice", dest="clone_voice", action="store_true",
help="Also clone the voice from the same footage (use --say to speak with it)")
p_demo.add_argument("--transcript", default="")
p_demo.add_argument("--train", action="store_true",
help="Also train an identity LoRA from the crops and apply it to the video")
add_video_options(p_demo)
......@@ -1320,9 +1435,18 @@ def main(argv: list[str] | None = None) -> int:
log(f"Deleted character '{args.name}'")
return 0
if cmd == "voice":
voice = studio.extract_voice_profile(args.name, expand_sources(args.source),
transcript=args.transcript)
if not voice:
return 1
log(f"Use it with: video {args.name} --prompt '...' --say 'hello' --voice {voice}")
return 0
if cmd == "extract":
studio.extract(args.name, expand_sources(args.source),
description=args.description, max_images=args.max_images)
description=args.description, max_images=args.max_images,
with_voice=args.voice, transcript=args.transcript)
log("Review the crops, drop any stray faces with `prune`, then run `video`.")
return 0
......@@ -1341,7 +1465,8 @@ def main(argv: list[str] | None = None) -> int:
if cmd == "demo":
log("[1/3] extracting the character from the supplied footage")
profile = studio.extract(args.name, expand_sources(args.source),
description=args.description, max_images=args.max_images)
description=args.description, max_images=args.max_images,
with_voice=args.clone_voice, transcript=args.transcript)
if args.portrait:
log("[2/3] identity portrait")
studio.make_portrait(args.name, args.prompt, video_opts(args))
......
#!/usr/bin/env python3
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""MiniMax-H3 video+audio generation service (runs in the isolated H3 venv).
H3 needs diffusers >= 0.40, where it exists ONLY as Modular Diffusers blocks —
`MiniMaxH3Blocks` / `MiniMaxH3ModularPipeline`, reached through
`ModularPipeline.from_pretrained(...)`. There is no `DiffusionPipeline` half, so it
cannot go through coderai's `_detect_pipeline_class()` loader; and diffusers 0.40
can't be dropped into the main venv without moving every other pipeline's floor.
Hence the same shape as the pyannote/NeMo workers: an isolated venv, this script as
a local HTTP service, and codai/api/h3_worker.py owning its lifecycle.
Model facts this service encodes (they are checkpoint contracts, not preferences):
* Fixed 24 fps. `num_frames` is snapped UP to the next `17n + 5` the video VAE can
decode, and the resulting duration must land between 5 and 15 seconds.
* Canvas axes are multiples of 32; the released checkpoint's short edge is 768
with an area cap of 768*1344.
* The checkpoint is guidance-distilled: there is NO negative prompt and NO
guidance scale, and every step is a single forward pass.
* Video and audio are denoised jointly in one packed sequence, so a generation
always returns a soundtrack — muxed into the returned mp4.
* Three workflows, each loading its own transformer partition (~61.7 GB each):
t2va – prompt only
fl2va – prompt + first and/or last keyframe (`transformer/`)
ref2va– prompt + up to 12 omni references (`transformer_ref/`)
Loading WITHOUT a workflow pulls both partitions, so this service always picks
one from the request and reloads when the workflow changes.
Endpoints:
GET /health → {"ok":true,"loaded":bool,"workflow":str|null,"model":str}
POST /generate → {"mp4_b64":..., "fps":24, "num_frames":N, "width":W, "height":H,
"sampling_rate":SR, "workflow":...}
POST /unload → drop the pipeline and free VRAM (the process stays up)
"""
import argparse
import base64
import json
import os
import subprocess
import sys
import tempfile
import threading
import time
import traceback
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
H3_FPS = 24
H3_FRAMES_CHUNK = 17 # video VAE clip length
H3_FRAMES_REMAINDER = 5 # decodable counts are 17n + 5
H3_MIN_SECONDS = 5
H3_MAX_SECONDS = 15
H3_CANVAS_MULTIPLE = 32
_state = {
"pipe": None,
"workflow": None,
"loras": [], # [(path, weight)] currently fused
"model": None,
"dtype": "bfloat16",
"device_map": "",
"offload": "",
"lock": threading.RLock(),
}
def log(msg: str) -> None:
print(f"[h3] {msg}", flush=True)
# ── geometry ──────────────────────────────────────────────────────────────────
def snap_frames(num_frames: int) -> int:
"""Snap up to the next 17n+5 the video VAE can decode, clamped to 5-15 s."""
n = max(1, int(num_frames or 0))
lo = H3_MIN_SECONDS * H3_FPS # 120
hi = H3_MAX_SECONDS * H3_FPS # 360
n = max(lo, min(hi, n))
if n % H3_FRAMES_CHUNK != H3_FRAMES_REMAINDER:
n += (H3_FRAMES_REMAINDER - n % H3_FRAMES_CHUNK) % H3_FRAMES_CHUNK
while n > hi: # snapping may overshoot the 15 s ceiling
n -= H3_FRAMES_CHUNK
return n
def snap_axis(value: int, default: int) -> int:
v = int(value or default)
v = max(H3_CANVAS_MULTIPLE, v)
return v - (v % H3_CANVAS_MULTIPLE)
# ── model lifecycle ───────────────────────────────────────────────────────────
def pick_workflow(body: dict) -> str:
if body.get("references"):
return "ref2va"
if body.get("image") or body.get("last_image"):
return "fl2va"
return "t2va"
def _torch_dtype(name: str):
import torch
return {"bfloat16": torch.bfloat16, "float16": torch.float16,
"float32": torch.float32}.get((name or "").lower(), torch.bfloat16)
def unload(reason: str = "") -> None:
with _state["lock"]:
if _state["pipe"] is None:
return
log(f"unloading pipeline{(' (' + reason + ')') if reason else ''}")
_state["pipe"] = None
_state["workflow"] = None
_state["loras"] = []
try:
import gc
import torch
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
torch.cuda.ipc_collect()
except Exception:
pass
def load_pipeline(workflow: str):
"""Load (or reuse) the modular pipeline for one workflow."""
import torch
from diffusers import ModularPipeline
with _state["lock"]:
if _state["pipe"] is not None and _state["workflow"] == workflow:
return _state["pipe"]
if _state["pipe"] is not None:
# A different workflow means the OTHER transformer partition — there is
# no way to keep both on a consumer card.
unload(f"switching workflow {_state['workflow']} -> {workflow}")
model = _state["model"]
dtype = _torch_dtype(_state["dtype"])
log(f"loading {model} (workflow={workflow}, dtype={_state['dtype']}) …")
t0 = time.time()
pipe = ModularPipeline.from_pretrained(model, workflow=workflow)
load_kw = {"dtype": dtype}
if _state["device_map"]:
# Per-component device_map, forwarded to each from_pretrained.
load_kw["device_map"] = _state["device_map"]
pipe.load_components(workflow=workflow, **load_kw)
offload = (_state["offload"] or "").lower()
if offload in ("group", "leaf"):
_apply_group_offload(pipe, dtype, leaf=(offload == "leaf"))
elif not _state["device_map"]:
device = "cuda" if torch.cuda.is_available() else "cpu"
log(f"moving components to {device}")
pipe.to(device)
_state["pipe"] = pipe
_state["workflow"] = workflow
log(f"ready in {time.time() - t0:.0f}s")
return pipe
def _apply_group_offload(pipe, dtype, leaf: bool = False) -> None:
"""Stream the big modules between CPU and GPU a group at a time.
33B of weights don't fit on a consumer card; group offload keeps only the
executing group resident. Applied per component because a ModularPipeline has
no pipeline-level offload helper.
"""
import torch
if not torch.cuda.is_available():
log("group offload requested but no CUDA device — staying on CPU")
return
kind = "leaf_level" if leaf else "block_level"
for name in ("transformer", "transformer_ref", "text_encoder", "vae", "audio_vae"):
comp = getattr(pipe, name, None)
if comp is None or not hasattr(comp, "enable_group_offload"):
continue
try:
kw = dict(onload_device=torch.device("cuda"),
offload_device=torch.device("cpu"),
offload_type=kind, use_stream=True)
if not leaf:
kw["num_blocks_per_group"] = 1
comp.enable_group_offload(**kw)
log(f"group offload ({kind}) enabled on {name}")
except Exception as exc:
log(f"group offload failed on {name} ({exc}); moving it to cuda instead")
try:
comp.to("cuda")
except Exception:
pass
def apply_loras(pipe, loras: list) -> None:
"""Fuse the requested LoRAs (MiniMaxH3LoraLoaderMixin), replacing any others."""
want = [(str(l.get("path") or ""), float(l.get("weight") or 1.0))
for l in (loras or []) if l.get("path")]
if want == _state["loras"]:
return
try:
pipe.unload_lora_weights()
except Exception:
pass
_state["loras"] = []
if not want:
return
names, weights = [], []
for i, (path, weight) in enumerate(want):
adapter = f"lora_{i}"
log(f"loading LoRA {path} @ {weight}")
pipe.load_lora_weights(path, adapter_name=adapter)
names.append(adapter)
weights.append(weight)
try:
pipe.set_adapters(names, weights)
except Exception as exc:
log(f"set_adapters failed ({exc}) — adapters load at weight 1.0")
_state["loras"] = want
# ── media helpers ─────────────────────────────────────────────────────────────
def _decode_to_file(data: str, suffix: str) -> str:
"""Write a base64 / data-URI payload to a temp file and return its path."""
if data.startswith("data:"):
data = data.split(",", 1)[1]
raw = base64.b64decode(data)
fd, path = tempfile.mkstemp(suffix=suffix)
os.write(fd, raw)
os.close(fd)
return path
def _load_image(data: str):
from PIL import Image
path = _decode_to_file(data, ".png")
try:
return Image.open(path).convert("RGB")
finally:
try:
os.unlink(path)
except OSError:
pass
def build_references(refs: list, temps: list) -> list:
"""Turn the request's reference list into H3 reference dataclasses.
Order matters: it labels them in the prompt presentation and lays them out on
the shared rotary clock, so the request order is preserved exactly.
"""
from diffusers.modular_pipelines.minimax_h3 import (
MiniMaxH3AudioReference, MiniMaxH3ImageReference, MiniMaxH3VideoReference)
kinds = {"image": (MiniMaxH3ImageReference, ".png"),
"video": (MiniMaxH3VideoReference, ".mp4"),
"audio": (MiniMaxH3AudioReference, ".wav")}
out = []
for ref in refs or []:
kind = (ref.get("type") or "image").lower()
if kind not in kinds:
raise ValueError(f"unknown reference type '{kind}' (image|video|audio)")
cls, suffix = kinds[kind]
path = _decode_to_file(ref.get("data") or "", suffix)
temps.append(path)
out.append(cls.from_file(path))
return out
def mux(frames, audio, sampling_rate) -> bytes:
"""Write frames at 24 fps and mux the jointly-generated soundtrack in."""
from diffusers.utils import export_to_video
tmpdir = tempfile.mkdtemp(prefix="h3_")
video_path = os.path.join(tmpdir, "video.mp4")
export_to_video(frames, video_path, fps=H3_FPS)
out_path = video_path
if audio is not None and sampling_rate:
try:
import numpy as np
import soundfile as sf
wav = audio
if hasattr(wav, "detach"):
wav = wav.detach().to("cpu").float().numpy()
wav = np.asarray(wav)
while wav.ndim > 2: # (1, 2, samples) -> (2, samples)
wav = wav[0]
if wav.ndim == 2: # channels-major -> soundfile's (n, ch)
wav = wav.T
audio_path = os.path.join(tmpdir, "audio.wav")
sf.write(audio_path, wav, int(sampling_rate))
muxed = os.path.join(tmpdir, "muxed.mp4")
proc = subprocess.run(
["ffmpeg", "-y", "-i", video_path, "-i", audio_path,
"-c:v", "copy", "-c:a", "aac", "-b:a", "192k", "-shortest", muxed],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if proc.returncode == 0 and os.path.getsize(muxed):
out_path = muxed
else:
log("audio mux failed; returning the silent video: "
+ proc.stderr.decode("utf-8", "replace")[-300:])
except Exception as exc:
log(f"audio mux skipped ({exc}); returning the silent video")
with open(out_path, "rb") as fh:
data = fh.read()
import shutil
shutil.rmtree(tmpdir, ignore_errors=True)
return data
# ── generation ────────────────────────────────────────────────────────────────
def generate(body: dict) -> dict:
import torch
prompt = (body.get("prompt") or "").strip()
if not prompt:
raise ValueError("prompt is required (every H3 workflow needs one)")
workflow = body.get("workflow") or pick_workflow(body)
pipe = load_pipeline(workflow)
apply_loras(pipe, body.get("loras"))
num_frames = snap_frames(body.get("num_frames") or H3_MIN_SECONDS * H3_FPS)
kwargs = {
"prompt": prompt,
"num_frames": num_frames,
"num_inference_steps": int(body.get("num_inference_steps") or 50),
"output_type": "pil",
}
if body.get("width") and body.get("height"):
kwargs["width"] = snap_axis(body["width"], 1344)
kwargs["height"] = snap_axis(body["height"], 768)
seed = body.get("seed")
if seed is not None:
device = "cuda" if torch.cuda.is_available() else "cpu"
kwargs["generator"] = torch.Generator(device=device).manual_seed(int(seed))
temps: list = []
try:
if body.get("image"):
kwargs["image"] = _load_image(body["image"])
if body.get("last_image"):
kwargs["last_image"] = _load_image(body["last_image"])
if body.get("references"):
kwargs["references"] = build_references(body["references"], temps)
log(f"generating: workflow={workflow} frames={num_frames} "
f"steps={kwargs['num_inference_steps']} "
f"size={kwargs.get('width', 'auto')}x{kwargs.get('height', 'auto')}")
t0 = time.time()
out = pipe(output=["videos", "audio", "sampling_rate"], **kwargs)
took = time.time() - t0
videos = out["videos"] if isinstance(out, dict) else out.videos
audio = (out.get("audio") if isinstance(out, dict) else getattr(out, "audio", None))
rate = (out.get("sampling_rate") if isinstance(out, dict)
else getattr(out, "sampling_rate", None))
frames = videos[0] if (videos and isinstance(videos, (list, tuple))) else videos
mp4 = mux(frames, audio, rate)
log(f"done in {took:.0f}s ({len(mp4) / 1e6:.1f} MB)")
return {
"mp4_b64": base64.b64encode(mp4).decode(),
"fps": H3_FPS,
"num_frames": num_frames,
"width": kwargs.get("width"),
"height": kwargs.get("height"),
"sampling_rate": rate,
"workflow": workflow,
"seconds": round(num_frames / H3_FPS, 2),
}
finally:
for path in temps:
try:
os.unlink(path)
except OSError:
pass
# ── HTTP ──────────────────────────────────────────────────────────────────────
def make_handler():
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, fmt, *args):
return
def _json(self, payload, status=200):
body = json.dumps(payload).encode()
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
try:
self.wfile.write(body)
except BrokenPipeError:
pass
def _read_json(self):
length = int(self.headers.get("Content-Length") or 0)
return json.loads(self.rfile.read(length) or b"{}") if length else {}
def do_GET(self):
if self.path.startswith("/health"):
self._json({"ok": True,
"loaded": _state["pipe"] is not None,
"workflow": _state["workflow"],
"model": _state["model"]})
else:
self._json({"error": "not found"}, 404)
def do_POST(self):
try:
if self.path.startswith("/generate"):
self._json(generate(self._read_json()))
elif self.path.startswith("/unload"):
unload("requested")
self._json({"ok": True})
else:
self._json({"error": "not found"}, 404)
except ValueError as exc:
# Bad request (missing prompt, unknown reference type) — not a fault.
self._json({"error": str(exc)}, 400)
except Exception as exc:
traceback.print_exc()
self._json({"error": f"{type(exc).__name__}: {exc}"}, 500)
return Handler
def main(argv=None) -> int:
ap = argparse.ArgumentParser(description="MiniMax-H3 generation service")
ap.add_argument("--model", default="MiniMaxAI/MiniMax-H3",
help="HF id or local path of the H3 checkpoint")
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--port", type=int, required=True)
ap.add_argument("--dtype", default="bfloat16", choices=["bfloat16", "float16", "float32"])
ap.add_argument("--device-map", default="",
help="Per-component device_map passed to from_pretrained (e.g. 'auto', 'balanced')")
ap.add_argument("--offload", default="group", choices=["", "group", "leaf"],
help="Group-offload the big modules (default: group — 33B does not fit a consumer card)")
ap.add_argument("--preload", action="store_true",
help="Load the t2va workflow at startup instead of on first request")
args = ap.parse_args(argv)
_state["model"] = args.model
_state["dtype"] = args.dtype
_state["device_map"] = args.device_map
_state["offload"] = args.offload
if args.preload:
try:
load_pipeline("t2va")
except Exception as exc:
log(f"preload failed: {exc}")
server = ThreadingHTTPServer((args.host, args.port), make_handler())
log(f"service listening on http://{args.host}:{args.port} (model {args.model})")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
return 0
if __name__ == "__main__":
sys.exit(main())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment