diff --git a/python/sglang/test/kits/mmmu_vlm_kit.py b/python/sglang/test/kits/mmmu_vlm_kit.py index 61ea70804..a2dd6a537 100644 --- a/python/sglang/test/kits/mmmu_vlm_kit.py +++ b/python/sglang/test/kits/mmmu_vlm_kit.py @@ -198,11 +198,7 @@ class MMMUMixin: # Run evaluation self.run_mmmu_eval(self.model, output_path) - # Get the result file - # Search recursively for JSON result files (lmms-eval v0.4.1+ creates subdirectories) result_files = glob.glob(f"{output_path}/**/*.json", recursive=True) - if not result_files: - result_files = glob.glob(f"{output_path}/*.json") if not result_files: raise FileNotFoundError(f"No JSON result files found in {output_path}") @@ -389,11 +385,7 @@ class MMMUMultiModelTestBase(CustomTestCase): # Run evaluation self.run_mmmu_eval(model.model, output_path) - # Get the result file - # Search recursively for JSON result files (lmms-eval v0.4.1+ creates subdirectories) result_files = glob.glob(f"{output_path}/**/*.json", recursive=True) - if not result_files: - result_files = glob.glob(f"{output_path}/*.json") if not result_files: raise FileNotFoundError(f"No JSON result files found in {output_path}") diff --git a/test/registered/distributed/test_epd_disaggregation.py b/test/registered/distributed/test_epd_disaggregation.py index a0bb64e74..a115db259 100644 --- a/test/registered/distributed/test_epd_disaggregation.py +++ b/test/registered/distributed/test_epd_disaggregation.py @@ -14,7 +14,7 @@ from grpc_health.v1 import health_pb2, health_pb2_grpc from sglang.srt.utils import kill_process_tree from sglang.srt.utils.network import get_zmq_socket_on_host from sglang.test.ci.ci_register import register_cuda_ci -from sglang.test.kits.mmmu_vlm_kit import _run_lmms_eval_with_retry +from sglang.test.kits.mmmu_vlm_kit import MMMUMixin from sglang.test.server_fixtures.disaggregation_fixture import ( PDDisaggregationServerBase, ) @@ -619,13 +619,18 @@ class TestEPDDisaggregationOmni(PDDisaggregationServerBase): @unittest.skipIf(is_in_ci(), "Skipping in CI to reduce multi-GPU runtime") -class TestEPDDisaggregationOneEncoder(PDDisaggregationServerBase): +class TestEPDDisaggregationOneEncoder(MMMUMixin, PDDisaggregationServerBase): """Test EPD disaggregation with single encode server""" + # Qwen2.5-VL-3B-Instruct scores ~0.40 on the 50-sample MMMU subset. + accuracy = 0.40 + mmmu_args = ["--limit", "50"] + @classmethod def setUpClass(cls): super().setUpClass() cls.model = DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST + cls.base_url = cls.lb_url # MMMUMixin reads this for OPENAI_API_BASE cls.encode_port = f"{int(cls.lb_port) + 300}" cls.encode_url = f"http://{cls.base_host}:{cls.encode_port}" @@ -744,75 +749,6 @@ class TestEPDDisaggregationOneEncoder(PDDisaggregationServerBase): except Exception as e: print(f"Error killing process: {e}") - def run_mmmu_eval(self, model_version: str, output_path: str, limit: str = "50"): - """ - Evaluate a VLM on the MMMU validation set with lmms-eval. - Reference: test_vlm_models.py - - Args: - model_version: Model version/checkpoint to evaluate - output_path: Path to save evaluation results - limit: Number of samples to evaluate (default: "50" for CI time constraints) - """ - model = "openai_compatible" - tp = 1 - tasks = "mmmu_val" - batch_size = 32 - log_suffix = "openai_compatible" - os.makedirs(output_path, exist_ok=True) - - model_args = f'model_version="{model_version}",tp={tp}' - - cmd = [ - "python3", - "-m", - "lmms_eval", - "--model", - model, - "--model_args", - model_args, - "--tasks", - tasks, - "--batch_size", - str(batch_size), - "--log_samples", - "--log_samples_suffix", - log_suffix, - "--output_path", - str(output_path), - "--limit", - limit, - ] - - _run_lmms_eval_with_retry(cmd, timeout=3600) - - def test_mmmu(self): - """Test MMMU evaluation with EPD disaggregation""" - import glob - import json - - output_path = "./logs/epd_one_encoder_mmmu" - self.run_mmmu_eval(self.model, output_path) - - # Get the result file - result_files = glob.glob(f"{output_path}/**/*.json", recursive=True) - if not result_files: - result_files = glob.glob(f"{output_path}/*.json") - - if not result_files: - self.fail(f"No JSON result files found in {output_path}") - - result_file_path = result_files[0] - with open(result_file_path, "r") as f: - result = json.load(f) - print(f"MMMU result: {result}") - - mmmu_accuracy = result["results"]["mmmu_val"]["mmmu_acc,none"] - print(f"MMMU accuracy: {mmmu_accuracy:.4f}") - - # for qwen2.5-vl-3b-instruct, the accuracy is 0.40 - self.assertGreater(mmmu_accuracy, 0.40) - @unittest.skipIf( is_in_ci(), @@ -997,16 +933,21 @@ class TestEPDDisaggregationQwen35(PDDisaggregationServerBase): ) -class TestEPDDisaggregationMultiEncoders(PDDisaggregationServerBase): +class TestEPDDisaggregationMultiEncoders(MMMUMixin, PDDisaggregationServerBase): """ Test EPD disaggregation with multiple encode servers for load balancing. Both encode servers run on GPU 0 (different ports) for testing load distribution. """ + # Qwen2.5-VL-3B-Instruct scores ~0.40 on the 50-sample MMMU subset. + accuracy = 0.40 + mmmu_args = ["--limit", "50"] + @classmethod def setUpClass(cls): super().setUpClass() cls.model = DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST + cls.base_url = cls.lb_url # MMMUMixin reads this for OPENAI_API_BASE cls.encode_port1 = f"{int(cls.lb_port) + 300}" cls.encode_port2 = f"{int(cls.lb_port) + 301}" cls.encode_url1 = f"http://{cls.base_host}:{cls.encode_port1}" @@ -1147,83 +1088,20 @@ class TestEPDDisaggregationMultiEncoders(PDDisaggregationServerBase): except Exception as e: print(f"Error killing process: {e}") - def run_mmmu_eval(self, model_version: str, output_path: str, limit: str = "50"): - """ - Evaluate a VLM on the MMMU validation set with lmms-eval. - Reference: test_vlm_models.py - - Args: - model_version: Model version/checkpoint to evaluate - output_path: Path to save evaluation results - limit: Number of samples to evaluate (default: "50" for CI time constraints) - """ - model = "openai_compatible" - tp = 1 - tasks = "mmmu_val" - batch_size = 32 - log_suffix = "openai_compatible" - os.makedirs(output_path, exist_ok=True) - - model_args = f'model_version="{model_version}",tp={tp}' - - cmd = [ - "python3", - "-m", - "lmms_eval", - "--model", - model, - "--model_args", - model_args, - "--tasks", - tasks, - "--batch_size", - str(batch_size), - "--log_samples", - "--log_samples_suffix", - log_suffix, - "--output_path", - str(output_path), - "--limit", - limit, - ] - - _run_lmms_eval_with_retry(cmd, timeout=3600) - - def test_mmmu(self): - """Test MMMU evaluation with EPD disaggregation (multiple encoders)""" - import glob - import json - - output_path = "./logs/epd_multi_encoder_mmmu" - self.run_mmmu_eval(self.model, output_path) - - # Get the result file - result_files = glob.glob(f"{output_path}/**/*.json", recursive=True) - if not result_files: - result_files = glob.glob(f"{output_path}/*.json") - - if not result_files: - self.fail(f"No JSON result files found in {output_path}") - - result_file_path = result_files[0] - with open(result_file_path, "r") as f: - result = json.load(f) - print(f"MMMU result (multi encoder): {result}") - - mmmu_accuracy = result["results"]["mmmu_val"]["mmmu_acc,none"] - print(f"MMMU accuracy (multi encoder): {mmmu_accuracy:.4f}") - # for qwen2.5-vl-3b-instruct, the accuracy is 0.40 - self.assertGreater(mmmu_accuracy, 0.40) - @unittest.skipIf(is_in_ci(), "Skipping in CI to reduce multi-GPU runtime") -class TestEPDDisaggregationGrpcEncoderMMMU(PDDisaggregationServerBase): +class TestEPDDisaggregationGrpcEncoderMMMU(MMMUMixin, PDDisaggregationServerBase): """Test MMMU evaluation with gRPC encoder in EPD mode.""" + # Qwen2.5-VL-3B-Instruct scores ~0.40 on the 50-sample MMMU subset. + accuracy = 0.40 + mmmu_args = ["--limit", "50"] + @classmethod def setUpClass(cls): super().setUpClass() cls.model = DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST + cls.base_url = cls.lb_url # MMMUMixin reads this for OPENAI_API_BASE cls.encode_port = f"{int(cls.lb_port) + 304}" cls.encode_url = f"grpc://{cls.base_host}:{cls.encode_port}" @@ -1373,63 +1251,6 @@ class TestEPDDisaggregationGrpcEncoderMMMU(PDDisaggregationServerBase): except Exception as e: print(f"Error killing process: {e}") - def run_mmmu_eval(self, model_version: str, output_path: str, limit: str = "50"): - model = "openai_compatible" - tp = 1 - tasks = "mmmu_val" - batch_size = 32 - log_suffix = "openai_compatible" - os.makedirs(output_path, exist_ok=True) - - model_args = f'model_version="{model_version}",tp={tp}' - - cmd = [ - "python3", - "-m", - "lmms_eval", - "--model", - model, - "--model_args", - model_args, - "--tasks", - tasks, - "--batch_size", - str(batch_size), - "--log_samples", - "--log_samples_suffix", - log_suffix, - "--output_path", - str(output_path), - "--limit", - limit, - ] - - _run_lmms_eval_with_retry(cmd, timeout=3600) - - def test_mmmu(self): - import glob - import json - - output_path = "./logs/epd_grpc_encoder_mmmu" - self.run_mmmu_eval(self.model, output_path) - - result_files = glob.glob(f"{output_path}/**/*.json", recursive=True) - if not result_files: - result_files = glob.glob(f"{output_path}/*.json") - - if not result_files: - self.fail(f"No JSON result files found in {output_path}") - - result_file_path = result_files[0] - with open(result_file_path, "r") as f: - result = json.load(f) - print(f"MMMU result (grpc encoder): {result}") - - mmmu_accuracy = result["results"]["mmmu_val"]["mmmu_acc,none"] - print(f"MMMU accuracy (grpc encoder): {mmmu_accuracy:.4f}") - # for qwen2.5-vl-3b-instruct, the accuracy is 0.40 - self.assertGreater(mmmu_accuracy, 0.40) - @unittest.skipIf(is_in_ci(), "Skipping in CI to reduce multi-GPU runtime") class TestEPDDisaggregationGrpcEncoderOnly(PDDisaggregationServerBase): diff --git a/test/registered/vlm/test_encoder_dp.py b/test/registered/vlm/test_encoder_dp.py index dd0f2669c..de96ac24b 100644 --- a/test/registered/vlm/test_encoder_dp.py +++ b/test/registered/vlm/test_encoder_dp.py @@ -1,21 +1,11 @@ -import glob -import json -import os import random +import tempfile import unittest from types import SimpleNamespace -from sglang.srt.utils import kill_process_tree from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci -from sglang.test.kits.mmmu_vlm_kit import _run_lmms_eval_with_retry -from sglang.test.test_utils import ( - DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, - DEFAULT_URL_FOR_TEST, - CustomTestCase, - is_in_amd_ci, - is_in_ci, - popen_launch_server, -) +from sglang.test.kits.mmmu_vlm_kit import MMMUMultiModelTestBase +from sglang.test.test_utils import is_in_ci register_cuda_ci(est_time=500, suite="nightly-4-gpu", nightly=True) register_amd_ci(est_time=500, suite="nightly-amd-4-gpu", nightly=True) @@ -28,228 +18,27 @@ MODELS = [ ] -# Set default mem_fraction_static to 0.8 -DEFAULT_MEM_FRACTION_STATIC = 0.8 - - -class TestVLMEncoderDP(CustomTestCase): - parsed_args = None # Class variable to store args - - @classmethod - def setUpClass(cls): - # Removed argument parsing from here - cls.base_url = DEFAULT_URL_FOR_TEST - cls.api_key = "sk-123456" - cls.time_out = DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH - - if cls.parsed_args is None: - cls.parsed_args = SimpleNamespace( - mem_fraction_static=DEFAULT_MEM_FRACTION_STATIC - ) - - # Set OpenAI API key and base URL environment variables. Needed for lmm-evals to work. - os.environ["OPENAI_API_KEY"] = cls.api_key - os.environ["OPENAI_API_BASE"] = f"{cls.base_url}/v1" - - def run_mmmu_eval( - self, - model_version: str, - output_path: str, - *, - env: dict | None = None, - ): - """ - Evaluate a VLM on the MMMU validation set with lmms‑eval. - Only `model_version` (checkpoint) and `chat_template` vary; - We are focusing only on the validation set due to resource constraints. - """ - # -------- fixed settings -------- - model = "openai_compatible" - tp = 1 - tasks = "mmmu_val" - batch_size = 32 - log_suffix = "openai_compatible" - os.makedirs(output_path, exist_ok=True) - - # -------- compose --model_args -------- - model_args = f'model_version="{model_version}",' f"tp={tp}" - - # -------- build command list -------- - cmd = [ - "python3", - "-m", - "lmms_eval", - "--model", - model, - "--model_args", - model_args, - "--tasks", - tasks, - "--batch_size", - str(batch_size), - "--log_samples", - "--log_samples_suffix", - log_suffix, - "--output_path", - str(output_path), - ] - - _run_lmms_eval_with_retry(cmd, timeout=3600) - - def _run_vlm_mmmu_test( - self, - model, - output_path, - test_name="", - custom_env=None, - log_level="info", - capture_output=False, - ): - """ - Common method to run VLM MMMU benchmark test. - - Args: - model: Model to test - output_path: Path for output logs - test_name: Optional test name for logging - custom_env: Optional custom environment variables - log_level: Log level for server (default: "info") - capture_output: Whether to capture server stdout/stderr - """ - print(f"\nTesting model: {model.model}{test_name}") - - process = None - mmmu_accuracy = 0 # Initialize to handle potential exceptions - server_output = "" - - try: - # Prepare environment variables - process_env = os.environ.copy() - if custom_env: - process_env.update(custom_env) - if not is_in_amd_ci(): - process_env["SGLANG_USE_CUDA_IPC_TRANSPORT"] = "1" - - # Prepare stdout/stderr redirection if needed - stdout_file = None - stderr_file = None - if capture_output: - stdout_file = open("/tmp/server_stdout.log", "w") - stderr_file = open("/tmp/server_stderr.log", "w") - - # Launch server for testing - process = popen_launch_server( - model.model, - base_url=self.base_url, - timeout=self.time_out, - api_key=self.api_key, - other_args=[ - "--trust-remote-code", - "--cuda-graph-max-bs", - "32", - "--mm-enable-dp-encoder", - "--tp=4", - "--mem-fraction-static", - str(self.parsed_args.mem_fraction_static), # Use class variable - "--log-level", - log_level, - ], - env=process_env, - return_stdout_stderr=( - (stdout_file, stderr_file) if capture_output else None - ), - ) - - # Run evaluation - self.run_mmmu_eval(model.model, output_path) - - # Get the result file - # Search recursively for JSON result files (lmms-eval v0.4.1+ creates subdirectories) - result_files = glob.glob(f"{output_path}/**/*.json", recursive=True) - if not result_files: - result_files = glob.glob(f"{output_path}/*.json") - - if not result_files: - raise FileNotFoundError(f"No JSON result files found in {output_path}") - - result_file_path = result_files[0] - - with open(result_file_path, "r") as f: - result = json.load(f) - print(f"Result{test_name}\n: {result}") - - # Process the result - mmmu_accuracy = result["results"]["mmmu_val"]["mmmu_acc,none"] - print( - f"Model {model.model} achieved accuracy{test_name}: {mmmu_accuracy:.4f}" - ) - - # Capture server output if requested - if capture_output and process: - server_output = self._read_output_from_files() - - # Assert performance meets expected threshold - self.assertGreaterEqual( - mmmu_accuracy, - model.mmmu_accuracy, - f"Model {model.model} accuracy ({mmmu_accuracy:.4f}) below expected threshold ({model.mmmu_accuracy:.4f}){test_name}", - ) - - return server_output - - except Exception as e: - print(f"Error testing {model.model}{test_name}: {e}") - self.fail(f"Test failed for {model.model}{test_name}: {e}") - - finally: - # Ensure process cleanup happens regardless of success/failure - if process is not None and process.poll() is None: - print(f"Cleaning up process {process.pid}") - try: - kill_process_tree(process.pid) - except Exception as e: - print(f"Error killing process: {e}") - - # clean up temporary files - if capture_output: - if stdout_file: - stdout_file.close() - if stderr_file: - stderr_file.close() - for filename in ["/tmp/server_stdout.log", "/tmp/server_stderr.log"]: - try: - if os.path.exists(filename): - os.remove(filename) - except Exception as e: - print(f"Error removing {filename}: {e}") - - def _read_output_from_files(self): - output_lines = [] - - log_files = [ - ("/tmp/server_stdout.log", "[STDOUT]"), - ("/tmp/server_stderr.log", "[STDERR]"), - ] - for filename, tag in log_files: - try: - if os.path.exists(filename): - with open(filename, "r") as f: - for line in f: - output_lines.append(f"{tag} {line.rstrip()}") - except Exception as e: - print(f"Error reading {tag.lower()} file: {e}") - - return "\n".join(output_lines) +class TestVLMEncoderDP(MMMUMultiModelTestBase): + # --cuda-graph-max-bs 32 last-wins over the kit's default 64. + other_args = [ + "--mm-enable-dp-encoder", + "--tp=4", + "--cuda-graph-max-bs", + "32", + ] def test_vlm_mmmu_benchmark(self): - """Test VLM models against MMMU benchmark.""" models_to_test = MODELS if is_in_ci(): models_to_test = [random.choice(MODELS)] for model in models_to_test: - self._run_vlm_mmmu_test(model, "./logs") + # Per-model temp dir avoids cross-test cached results. + with tempfile.TemporaryDirectory( + prefix=f"encoder_dp_{model.model.replace('/', '_')}_" + ) as output_path: + self._run_vlm_mmmu_test(model, output_path) if __name__ == "__main__":