[diffusion] chore: validate sampling params (#16677)

Co-authored-by: root <root@huchong2-0.huchong2.podvm.svc.cluster.local>

python/sglang/multimodal_gen/configs/sample/sampling_params.py

@@ -213,11 +213,85 @@ class SamplingParams:
         check if the sampling params is correct by itself
         """
         if self.prompt_path and not self.prompt_path.endswith(".txt"):
-            raise ValueError("prompt_path must be a txt file")
+            raise ValueError(
+                f"prompt_path must be a txt file, got {self.prompt_path!r}"
+            )
+
+        # These are always required to be sane regardless of pipeline.
+        if (
+            not isinstance(self.num_outputs_per_prompt, int)
+            or self.num_outputs_per_prompt <= 0
+        ):
+            raise ValueError(
+                f"num_outputs_per_prompt must be a positive int, got {self.num_outputs_per_prompt!r}"
+            )
+
+        # Used by seconds() and video writer; fps <= 0 is always invalid.
+        if not isinstance(self.fps, int) or self.fps <= 0:
+            raise ValueError(f"fps must be a positive int, got {self.fps!r}")
+
+        # num_frames is already asserted in __post_init__, but keep a friendly error here too
+        # (e.g., when validation is triggered from other code paths).
+        if not isinstance(self.num_frames, int) or self.num_frames <= 0:
+            raise ValueError(
+                f"num_frames must be a positive int, got {self.num_frames!r}"
+            )
+
+        if self.num_inference_steps is not None:
+            if (
+                not isinstance(self.num_inference_steps, int)
+                or self.num_inference_steps <= 0
+            ):
+                raise ValueError(
+                    f"num_inference_steps must be a positive int, got {self.num_inference_steps!r}"
+                )
+
+        # Numeric hyperparams should not be NaN/Inf and should be within basic ranges.
+        # Note: bool is a subclass of int; reject it explicitly to avoid silent surprises.
+        def _finite_non_negative_float(
+            name: str, value: Any, allow_none: bool = True
+        ) -> None:
+            if value is None and allow_none:
+                return
+            if isinstance(value, bool) or not isinstance(value, (int, float)):
+                raise ValueError(f"{name} must be a number, got {value!r}")
+            if not math.isfinite(float(value)):
+                raise ValueError(f"{name} must be finite, got {value!r}")
+            if float(value) < 0.0:
+                raise ValueError(f"{name} must be non-negative, got {value!r}")
+
+        _finite_non_negative_float(
+            "guidance_scale", self.guidance_scale, allow_none=True
+        )
+        _finite_non_negative_float(
+            "guidance_scale_2", self.guidance_scale_2, allow_none=True
+        )
+        _finite_non_negative_float(
+            "true_cfg_scale", self.true_cfg_scale, allow_none=True
+        )
+        _finite_non_negative_float(
+            "guidance_rescale", self.guidance_rescale, allow_none=False
+        )
+
+        if self.boundary_ratio is not None:
+            if isinstance(self.boundary_ratio, bool) or not isinstance(
+                self.boundary_ratio, (int, float)
+            ):
+                raise ValueError(
+                    f"boundary_ratio must be a number, got {self.boundary_ratio!r}"
+                )
+            if not math.isfinite(float(self.boundary_ratio)):
+                raise ValueError(
+                    f"boundary_ratio must be finite, got {self.boundary_ratio!r}"
+                )
+            if not (0.0 <= float(self.boundary_ratio) <= 1.0):
+                raise ValueError(
+                    f"boundary_ratio must be within [0, 1], got {self.boundary_ratio!r}"
+                )
 
     def check_sampling_param(self):
-        if self.prompt_path and not self.prompt_path.endswith(".txt"):
+        # Keep backward-compatibility for old call sites.
-            raise ValueError("prompt_path must be a txt file")
+        self._validate()
 
     def _validate_with_pipeline_config(self, pipeline_config):
         """

python/sglang/multimodal_gen/test/run_suite.py

@@ -27,6 +27,8 @@ SUITES = {
         "test_lora_format_adapter.py",
         # cli test
         "../cli/test_generate_t2i_perf.py",
+        # unit tests (no server needed)
+        "../test_sampling_params_validate.py",
         # add new 1-gpu test files here
     ],
     "2-gpu": [

python/sglang/multimodal_gen/test/test_sampling_params_validate.py

@@ -0,0 +1,49 @@
+import math
+import unittest
+
+from sglang.multimodal_gen.configs.sample.sampling_params import SamplingParams
+
+
+class TestSamplingParamsValidate(unittest.TestCase):
+    def test_prompt_path_suffix(self):
+        with self.assertRaisesRegex(ValueError, r"prompt_path"):
+            SamplingParams(prompt_path="bad.png")
+
+    def test_num_outputs_per_prompt_must_be_positive(self):
+        with self.assertRaisesRegex(ValueError, r"num_outputs_per_prompt"):
+            SamplingParams(num_outputs_per_prompt=0)
+
+    def test_fps_must_be_positive_int(self):
+        with self.assertRaisesRegex(ValueError, r"\bfps\b"):
+            SamplingParams(fps=0)
+        with self.assertRaisesRegex(ValueError, r"\bfps\b"):
+            SamplingParams(fps=None)  # type: ignore[arg-type]
+
+    def test_num_inference_steps_optional_but_if_set_must_be_positive(self):
+        SamplingParams(num_inference_steps=None)
+        with self.assertRaisesRegex(ValueError, r"num_inference_steps"):
+            SamplingParams(num_inference_steps=-1)
+
+    def test_guidance_scale_must_be_finite_non_negative_if_set(self):
+        SamplingParams(guidance_scale=None)
+        with self.assertRaisesRegex(ValueError, r"guidance_scale"):
+            SamplingParams(guidance_scale=math.nan)
+        with self.assertRaisesRegex(ValueError, r"guidance_scale"):
+            SamplingParams(guidance_scale=-0.1)
+
+    def test_guidance_rescale_must_be_finite_non_negative(self):
+        with self.assertRaisesRegex(ValueError, r"guidance_rescale"):
+            SamplingParams(guidance_rescale=-1.0)
+        with self.assertRaisesRegex(ValueError, r"guidance_rescale"):
+            SamplingParams(guidance_rescale=math.inf)
+
+    def test_boundary_ratio_range(self):
+        SamplingParams(boundary_ratio=None)
+        with self.assertRaisesRegex(ValueError, r"boundary_ratio"):
+            SamplingParams(boundary_ratio=1.5)
+        with self.assertRaisesRegex(ValueError, r"boundary_ratio"):
+            SamplingParams(boundary_ratio=math.nan)
+
+
+if __name__ == "__main__":
+    unittest.main()