[diffusion] feat: support unverified short edge instead of rejecting it for minimax-h3 (#35664)

This commit is contained in:
Mick
2026-08-20 16:52:44 +08:00
committed by GitHub
parent 06ad7b2b0d
commit f1b9a1f42a
4 changed files with 111 additions and 8 deletions
@@ -1,6 +1,12 @@
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
from functools import lru_cache
from sglang.multimodal_gen.runtime.utils.logging_utils import init_logger
logger = init_logger(__name__)
# Direct-encode text embeddings: {"positive":
# {"hidden_states": Tensor[text_len, 5120] bf16 cpu, "text_len": int}}
MINIMAX_H3_TEXT_EMBEDDINGS_EXTRA_KEY = "minimax_h3_text_embeddings"
@@ -24,6 +30,28 @@ MINIMAX_H3_SUPPORTED_FPS = 24
MINIMAX_H3_MIN_DURATION_SECONDS = 4.0
MINIMAX_H3_MAX_DURATION_SECONDS = 15.0
# The short edge every published MiniMax recipe and every reference output uses.
# The shape math itself is generic -- it scales the requested short edge by the
# aspect ratio, clamps to the pixel budget and rounds to the canvas multiple --
# so other values resolve to a valid canvas and generate. They are just not what
# the checkpoint was tuned and measured on.
MINIMAX_H3_RECOMMENDED_SHORT_EDGE = 768
@lru_cache(maxsize=None)
def warn_unverified_short_edge(short_edge: int) -> None:
"""Warn once per distinct value; requests repeat, the caveat does not."""
logger.warning(
"MiniMax H3 target.short_edge=%d is outside the verified configuration: "
"%d is the only short edge MiniMax publishes recipes and reference "
"outputs for. Smaller edges cost proportionally less memory and time; "
"quality and prompt adherence at this size are not covered by any "
"MiniMax or SGLang measurement.",
short_edge,
MINIMAX_H3_RECOMMENDED_SHORT_EDGE,
)
# The distilled checkpoint has exactly one positive denoise branch.
MINIMAX_H3_DEFAULT_BRANCHES: tuple = ({"name": "cond_1"},)
@@ -17,6 +17,7 @@ from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.m
MINIMAX_H3_MAX_DURATION_SECONDS,
MINIMAX_H3_MIN_DURATION_SECONDS,
MINIMAX_H3_SUPPORTED_FPS,
warn_unverified_short_edge,
)
from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3.task_profiles import (
MINIMAX_H3_CONDITION_ROLE_KEYFRAME,
@@ -83,10 +84,9 @@ def _validate_target(target: Any, *, profile: MiniMaxH3TaskProfile) -> dict[str,
# compatibility keys are ignored; only these three declared values are
# validated and emitted below.
short_edge = _require_int(target.get("short_edge"), f"{path}.short_edge")
if short_edge != 768:
raise ValueError(
f"{path}.short_edge must be 768 for minimax_h3, got {short_edge}"
)
if short_edge <= 0:
raise ValueError(f"{path}.short_edge must be positive, got {short_edge}")
warn_unverified_short_edge(short_edge)
aspect_ratio = _require_str(target.get("aspect_ratio"), f"{path}.aspect_ratio")
if profile.aspect_ratio_forced_auto and aspect_ratio != "auto":
raise ValueError(
@@ -27,6 +27,7 @@ import msgspec
from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3.constants import (
MINIMAX_H3_SUPPORTED_FPS,
warn_unverified_short_edge,
)
from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3.task_profiles import (
MINIMAX_H3_FL2VA_KEYFRAME_SIGNATURES,
@@ -100,11 +101,13 @@ def _validate_base_short_edge(value: Any) -> int:
try:
short_edge = int(value)
except (TypeError, ValueError) as exc:
raise ValueError("target.short_edge must be 768") from exc
if short_edge != MINIMAX_H3_BASE_SHORT_EDGE or value != short_edge:
raise ValueError(
f"target.short_edge must be 768 for MiniMax H3 shape policy v2, got {value!r}"
)
f"target.short_edge must be an integer, got {value!r}"
) from exc
if short_edge != value or short_edge <= 0:
raise ValueError(f"target.short_edge must be a positive integer, got {value!r}")
if short_edge != MINIMAX_H3_BASE_SHORT_EDGE:
warn_unverified_short_edge(short_edge)
return short_edge
@@ -0,0 +1,72 @@
"""A short edge other than 768 resolves and generates; it only warns."""
import pytest
from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3 import (
constants,
)
from sglang.multimodal_gen.runtime.pipelines_core.stages.model_specific_stages.minimax_h3.resolved_plan import (
MINIMAX_H3_BASE_SHORT_EDGE,
MINIMAX_H3_CANVAS_MULTIPLE,
MINIMAX_H3_MAX_PIXELS,
minimax_h3_resolve_spatial_shape,
)
@pytest.fixture(autouse=True)
def _reset_warn_cache():
constants.warn_unverified_short_edge.cache_clear()
yield
constants.warn_unverified_short_edge.cache_clear()
class TestResolveSpatialShape:
@pytest.mark.parametrize("short_edge", [352, 384, 416, 512, 768])
def test_a_smaller_short_edge_resolves_to_an_aligned_canvas(self, short_edge):
shape = minimax_h3_resolve_spatial_shape(
width=16, height=9, base_short_edge=short_edge
)
assert shape["width"] % MINIMAX_H3_CANVAS_MULTIPLE == 0
assert shape["height"] % MINIMAX_H3_CANVAS_MULTIPLE == 0
assert shape["width"] * shape["height"] <= MINIMAX_H3_MAX_PIXELS
# the short edge lands on the requested one, up to the 32px grid
assert (
abs(shape["effective_short_edge"] - short_edge) < MINIMAX_H3_CANVAS_MULTIPLE
)
def test_halving_the_short_edge_quarters_the_canvas(self):
full = minimax_h3_resolve_spatial_shape(width=16, height=9)
half = minimax_h3_resolve_spatial_shape(
width=16, height=9, base_short_edge=MINIMAX_H3_BASE_SHORT_EDGE // 2
)
full_pixels = full["width"] * full["height"]
half_pixels = half["width"] * half["height"]
assert 3.8 < full_pixels / half_pixels < 4.2
def test_a_larger_short_edge_is_capped_by_the_pixel_budget(self):
shape = minimax_h3_resolve_spatial_shape(
width=16, height=9, base_short_edge=1536
)
assert shape["size_mode"] == "area"
assert shape["width"] * shape["height"] <= MINIMAX_H3_MAX_PIXELS
@pytest.mark.parametrize("bad", [0, -768, 1.5])
def test_a_non_positive_or_fractional_short_edge_is_rejected(self, bad):
with pytest.raises(ValueError, match="short_edge"):
minimax_h3_resolve_spatial_shape(width=16, height=9, base_short_edge=bad)
class TestWarning:
def test_the_recommended_short_edge_does_not_warn(self, caplog):
with caplog.at_level("WARNING"):
minimax_h3_resolve_spatial_shape(width=16, height=9)
assert "short_edge" not in caplog.text
def test_an_unverified_short_edge_warns_once_per_value(self, caplog):
with caplog.at_level("WARNING"):
for _ in range(3):
minimax_h3_resolve_spatial_shape(
width=16, height=9, base_short_edge=384
)
assert caplog.text.count("outside the verified configuration") == 1
assert "768" in caplog.text