minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

ci: prune per-commit CUDA tests — move 25 files + 13 testcases to test/manual/ (#24721)

Browse Source
This commit is contained in:
Alison Shao
2026-05-08 15:53:23 -07:00
committed by GitHub
parent aefd8e257f
commit 5fbec0e445
45 changed files with 818 additions and 525 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/manual/4-gpu-models/test_qwen35_fp4_triton.py
+77
View File
@@ -0,0 +1,77 @@
import unittest
from sglang.test.accuracy_test_runner import AccuracyTestParams
from sglang.test.ci.ci_register import register_cuda_ci
# This eval harness applies the chat_template, which is critical for qwen3.5
# to get good accuracy on gsm8k
from sglang.test.run_combined_tests import run_combined_tests
from sglang.test.test_utils import (
CustomTestCase,
ModelLaunchSettings,
)
register_cuda_ci(est_time=720, suite="stage-c-test-4-gpu-b200")
QWEN35_FP4_MODEL = "nvidia/Qwen3.5-397B-A17B-NVFP4"
ACC_THRESHOLDS = {QWEN35_FP4_MODEL: {"gsm8k": 0.95}}
class TestQwen35FP4(CustomTestCase):
def test_gsm8k(self):
base_args = [
"--tp-size",
"4",
"--chunked-prefill-size",
"2048",
"--mamba-scheduler-strategy",
"extra_buffer",
"--mamba-track-interval",
"128",
"--mamba-ssm-dtype",
"bfloat16",
"--max-running-requests",
"128",
"--reasoning-parser",
"qwen3",
"--attention-backend",
"trtllm_mha",
"--quantization",
"modelopt_fp4",
"--model-loader-extra-config",
'{"enable_multithread_load": true,"num_threads": 64}',
]
variants = [
ModelLaunchSettings(
QWEN35_FP4_MODEL,
extra_args=base_args,
variant="Triton",
),
# TODO: Fix this and re-enable it
# ModelLaunchSettings(
# QWEN35_FP4_MODEL,
# extra_args=base_args + ["--linear-attn-decode-backend", "flashinfer"],
# variant="FlashInfer",
# ),
]
run_combined_tests(
models=variants,
test_name="Qwen3.5-397B-A17B-NVFP4",
accuracy_params=AccuracyTestParams(
dataset="gsm8k",
baseline_accuracy=ACC_THRESHOLDS[QWEN35_FP4_MODEL]["gsm8k"],
num_examples=200,
num_threads=128,
max_tokens=16000,
thinking_mode="qwen3",
temperature=0.6,
top_p=0.95,
top_k=20,
),
)
if __name__ == "__main__":
unittest.main()
test/manual/4-gpu-models/test_qwen35_models_archived.py
+168
View File
@@ -0,0 +1,168 @@
"""Archived test classes split out of test/registered/4-gpu-models/test_qwen35_models.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/4-gpu-models/test_qwen35_models_archived.py`.
"""
import unittest
from types import SimpleNamespace
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.accuracy_test_runner import AccuracyTestParams
from sglang.test.kits.reasoning_kit import ReasoningTokenUsageMixin
# This eval harness applies the chat_template, which is critical for qwen3.5
# to get good accuracy on gsm8k
from sglang.test.run_combined_tests import run_combined_tests
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
ModelLaunchSettings,
popen_launch_server,
)
QWEN35_FP4_MODEL = "nvidia/Qwen3.5-397B-A17B-NVFP4"
ACC_THRESHOLDS = {QWEN35_FP4_MODEL: {"gsm8k": 0.95}}
class TestQwen35FP4(CustomTestCase):
def test_gsm8k(self):
base_args = [
"--tp-size",
"4",
"--chunked-prefill-size",
"2048",
"--mamba-scheduler-strategy",
"extra_buffer",
"--mamba-track-interval",
"128",
"--mamba-ssm-dtype",
"bfloat16",
"--max-running-requests",
"128",
"--reasoning-parser",
"qwen3",
"--attention-backend",
"trtllm_mha",
"--quantization",
"modelopt_fp4",
"--model-loader-extra-config",
'{"enable_multithread_load": true,"num_threads": 64}',
]
variants = [
ModelLaunchSettings(
QWEN35_FP4_MODEL,
extra_args=base_args,
variant="Triton",
),
# TODO: Fix this and re-enable it
# ModelLaunchSettings(
# QWEN35_FP4_MODEL,
# extra_args=base_args + ["--linear-attn-decode-backend", "flashinfer"],
# variant="FlashInfer",
# ),
]
run_combined_tests(
models=variants,
test_name="Qwen3.5-397B-A17B-NVFP4",
accuracy_params=AccuracyTestParams(
dataset="gsm8k",
baseline_accuracy=ACC_THRESHOLDS[QWEN35_FP4_MODEL]["gsm8k"],
num_examples=200,
num_threads=128,
max_tokens=16000,
thinking_mode="qwen3",
temperature=0.6,
top_p=0.95,
top_k=20,
),
)
class TestQwen35FP4MTP(ReasoningTokenUsageMixin, CustomTestCase):
reasoning_parser_name = "qwen3"
@classmethod
def setUpClass(cls):
cls.model = QWEN35_FP4_MODEL
cls.base_url = DEFAULT_URL_FOR_TEST
cls.init_reasoning_token_verifier()
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--tp-size",
"4",
"--chunked-prefill-size",
"2048",
"--mamba-scheduler-strategy",
"extra_buffer",
"--mamba-track-interval",
"128",
"--mamba-ssm-dtype",
"bfloat16",
"--max-running-requests",
"128",
"--reasoning-parser",
"qwen3",
"--attention-backend",
"trtllm_mha",
"--quantization",
"modelopt_fp4",
"--speculative-algorithm",
"NEXTN",
"--speculative-num-steps",
"3",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"4",
"--mem-fraction-static",
"0.8",
"--model-loader-extra-config",
'{"enable_multithread_load": true,"num_threads": 64}',
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
model=self.model,
eval_name="gsm8k",
num_shots=5,
num_examples=200,
max_tokens=16000,
num_threads=128,
repeat=1,
temperature=0.6,
top_p=0.95,
top_k=20,
base_url=self.base_url,
host="http://127.0.0.1",
port=int(self.base_url.split(":")[-1]),
)
metrics = run_eval(args)
print(f"{metrics=}")
self.assertGreaterEqual(metrics["score"], ACC_THRESHOLDS[self.model]["gsm8k"])
server_info = requests.get(self.base_url + "/server_info")
avg_spec_accept_length = server_info.json()["internal_states"][0][
"avg_spec_accept_length"
]
print(f"{avg_spec_accept_length=}")
self.assertGreater(avg_spec_accept_length, 3.3)
if __name__ == "__main__":
unittest.main()
test/manual/4-gpu-models/test_qwen3_next_models.py
+34
View File
@@ -0,0 +1,34 @@
import unittest
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.kits.eval_accuracy_kit import GSM8KMixin
from sglang.test.kits.kl_divergence_kit import KLDivergenceMixin
from sglang.test.kits.prefix_cache_branching_kit import PrefixCacheBranchingMixin
from sglang.test.server_fixtures.default_fixture import DefaultServerBase
register_cuda_ci(est_time=142, suite="stage-c-test-4-gpu-h100")
QWEN3_NEXT_MODEL = "Qwen/Qwen3-Next-80B-A3B-Instruct"
class TestQwen3Next(
GSM8KMixin, KLDivergenceMixin, PrefixCacheBranchingMixin, DefaultServerBase
):
model = QWEN3_NEXT_MODEL
cache_chunk_size = 64
gsm8k_accuracy_thres = 0.93
kl_div_thres = 0.0025
other_args = [
"--tp-size",
"4",
"--chunked-prefill-size",
"2048",
"--mamba-scheduler-strategy",
"extra_buffer",
"--mamba-track-interval",
"128",
]
if __name__ == "__main__":
unittest.main()
test/manual/4-gpu-models/test_qwen3_next_models_mtp_archived.py
+44
View File
@@ -0,0 +1,44 @@
"""Archived test classes split out of test/registered/4-gpu-models/test_qwen3_next_models_mtp.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/4-gpu-models/test_qwen3_next_models_mtp_archived.py`.
"""
import unittest
from sglang.test.kits.eval_accuracy_kit import GSM8KMixin
from sglang.test.kits.kl_divergence_kit import KLDivergenceMixin
from sglang.test.server_fixtures.default_fixture import DefaultServerBase
QWEN3_NEXT_MODEL = "Qwen/Qwen3-Next-80B-A3B-Instruct"
class TestQwen3NextMTP(GSM8KMixin, KLDivergenceMixin, DefaultServerBase):
model = QWEN3_NEXT_MODEL
gsm8k_accuracy_thres = 0.93
kl_div_thres = 0.0025
other_args = [
"--trust-remote-code",
"--speculative-algorithm",
"NEXTN",
"--speculative-num-steps",
"3",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"4",
"--mem-fraction-static",
"0.8",
"--tp",
"4",
"--chunked-prefill-size",
"2048",
"--mamba-scheduler-strategy",
"no_buffer",
"--disable-radix-cache",
]
if __name__ == "__main__":
unittest.main()
test/manual/8-gpu-models/test_deepseek_v3_basic.py
+85
View File
@@ -0,0 +1,85 @@
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.send_one import BenchArgs, send_one_prompt
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,
write_github_step_summary,
)
register_cuda_ci(est_time=301, suite="stage-c-test-8-gpu-h200")
FULL_DEEPSEEK_V3_MODEL_PATH = "deepseek-ai/DeepSeek-V3-0324"
class TestDeepseekV3Basic(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = FULL_DEEPSEEK_V3_MODEL_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--tp",
"8",
"--model-loader-extra-config",
'{"enable_multithread_load": true, "num_threads": 64}',
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH * 5,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_a_gsm8k(
self,
): # Append an "a" to make this test run first (alphabetically) to warm up the server
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=1400,
num_threads=1400,
num_shots=8,
)
metrics = run_eval(args)
print(f"{metrics=}")
if is_in_ci():
write_github_step_summary(
f"### test_gsm8k (deepseek-v3)\n" f'{metrics["score"]=:.3f}\n'
)
self.assertGreater(metrics["score"], 0.935)
def test_bs_1_speed(self):
args = BenchArgs(port=int(self.base_url.split(":")[-1]), max_new_tokens=2048)
acc_length, speed = send_one_prompt(args)
print(f"{speed=:.2f}")
if is_in_ci():
write_github_step_summary(
f"### test_bs_1_speed (deepseek-v3)\n" f"{speed=:.2f} token/s\n"
)
if is_in_amd_ci():
self.assertGreater(speed, 12)
else:
self.assertGreater(speed, 75)
if __name__ == "__main__":
unittest.main()
test/manual/8-gpu-models/test_dsa_models_basic.py
+262
View File
@@ -0,0 +1,262 @@
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.send_one import BenchArgs, send_one_prompt
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
register_cuda_ci(est_time=1047, suite="stage-c-test-8-gpu-h200")
DEEPSEEK_V32_MODEL_PATH = "deepseek-ai/DeepSeek-V3.2"
GLM5_MODEL_PATH = "zai-org/GLM-5-FP8"
class TestDeepseekV32DP(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = DEEPSEEK_V32_MODEL_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--tp",
"8",
"--dp",
"8",
"--enable-dp-attention",
"--model-loader-extra-config",
'{"enable_multithread_load": true, "num_threads": 64}',
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_a_gsm8k(
self,
): # Append an "a" to make this test run first (alphabetically) to warm up the server
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=1400,
num_threads=1400,
num_shots=20,
)
metrics = run_eval(args)
print(f"{metrics=}")
if is_in_ci():
write_github_step_summary(
f"### test_gsm8k (deepseek-v32)\n" f'{metrics["score"]=:.3f}\n'
)
self.assertGreater(metrics["score"], 0.935)
def test_bs_1_speed(self):
args = BenchArgs(port=int(self.base_url.split(":")[-1]), max_new_tokens=2048)
acc_length, speed = send_one_prompt(args)
print(f"{speed=:.2f}")
if is_in_ci():
write_github_step_summary(
f"### test_bs_1_speed (deepseek-v32)\n" f"{speed=:.2f} token/s\n"
)
self.assertGreater(speed, 50)
class TestDeepseekV32TP(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = DEEPSEEK_V32_MODEL_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--tp",
"8",
"--model-loader-extra-config",
'{"enable_multithread_load": true, "num_threads": 64}',
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_a_gsm8k(
self,
): # Append an "a" to make this test run first (alphabetically) to warm up the server
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=1400,
num_threads=1400,
num_shots=20,
)
metrics = run_eval(args)
print(f"{metrics=}")
if is_in_ci():
write_github_step_summary(
f"### test_gsm8k (deepseek-v32)\n" f'{metrics["score"]=:.3f}\n'
)
self.assertGreater(metrics["score"], 0.935)
def test_bs_1_speed(self):
args = BenchArgs(port=int(self.base_url.split(":")[-1]), max_new_tokens=2048)
acc_length, speed = send_one_prompt(args)
print(f"{speed=:.2f}")
if is_in_ci():
write_github_step_summary(
f"### test_bs_1_speed (deepseek-v32)\n" f"{speed=:.2f} token/s\n"
)
self.assertGreater(speed, 80)
class TestGLM5DP(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = GLM5_MODEL_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--tp",
"8",
"--dp",
"8",
"--enable-dp-attention",
"--model-loader-extra-config",
'{"enable_multithread_load": true, "num_threads": 64}',
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_a_gsm8k(
self,
): # Append an "a" to make this test run first (alphabetically) to warm up the server
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=1400,
num_threads=1400,
num_shots=20,
)
metrics = run_eval(args)
print(f"{metrics=}")
if is_in_ci():
write_github_step_summary(
f"### test_gsm8k (glm-5)\n" f'{metrics["score"]=:.3f}\n'
)
self.assertGreater(metrics["score"], 0.935)
def test_bs_1_speed(self):
args = BenchArgs(port=int(self.base_url.split(":")[-1]), max_new_tokens=2048)
acc_length, speed = send_one_prompt(args)
print(f"{speed=:.2f}")
if is_in_ci():
write_github_step_summary(
f"### test_bs_1_speed (glm-5)\n" f"{speed=:.2f} token/s\n"
)
self.assertGreater(speed, 40)
class TestGLM5TP(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = GLM5_MODEL_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--tp",
"8",
"--model-loader-extra-config",
'{"enable_multithread_load": true, "num_threads": 64}',
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_a_gsm8k(
self,
): # Append an "a" to make this test run first (alphabetically) to warm up the server
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=1400,
num_threads=1400,
num_shots=20,
)
metrics = run_eval(args)
print(f"{metrics=}")
if is_in_ci():
write_github_step_summary(
f"### test_gsm8k (glm-5)\n" f'{metrics["score"]=:.3f}\n'
)
self.assertGreater(metrics["score"], 0.935)
def test_bs_1_speed(self):
args = BenchArgs(port=int(self.base_url.split(":")[-1]), max_new_tokens=2048)
acc_length, speed = send_one_prompt(args)
print(f"{speed=:.2f}")
if is_in_ci():
write_github_step_summary(
f"### test_bs_1_speed (glm-5)\n" f"{speed=:.2f} token/s\n"
)
self.assertGreater(speed, 60)
if __name__ == "__main__":
unittest.main()
test/manual/attention/test_fa3.py
+266
View File
@@ -0,0 +1,266 @@
import unittest
from types import SimpleNamespace
import requests
from sglang.srt.environ import envs
from sglang.srt.utils import get_device_sm, kill_process_tree
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_DRAFT_MODEL_EAGLE3,
DEFAULT_MODEL_NAME_FOR_TEST,
DEFAULT_MODEL_NAME_FOR_TEST_MLA,
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
# FlashAttention3 integration tests (requires SM 90+ / H100)
# Multiple test classes: FA3, FA3+MLA, FA3+SpecDecode variants
register_cuda_ci(est_time=551, suite="stage-b-test-1-gpu-large")
GSM_DATASET_PATH = None
# In case of some machine lack internet connection, we can set OFFLINE_MODE to True.
OFFLINE_MODE = False
# Change the path below when OFFLINE_MODE is True.
OFFLINE_PATH_DICT = {
DEFAULT_MODEL_NAME_FOR_TEST: "/shared/public/elr-models/meta-llama/Meta-Llama-3.1-8B-Instruct",
DEFAULT_DRAFT_MODEL_EAGLE3: "/shared/public/elr-models/jamesliu1/sglang-EAGLE3-Llama-3.1-Instruct-8B",
DEFAULT_MODEL_NAME_FOR_TEST_MLA: "/shared/public/sharing/deepseek/dsv3-test/snapshots/",
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN: "/shared/public/sharing/deepseek/dsv3-test-NextN/snapshots/",
GSM_DATASET_PATH: "/shared/public/data/gsm8k/test.jsonl",
}
if OFFLINE_MODE:
DEFAULT_MODEL_NAME_FOR_TEST = OFFLINE_PATH_DICT[DEFAULT_MODEL_NAME_FOR_TEST]
DEFAULT_DRAFT_MODEL_EAGLE3 = OFFLINE_PATH_DICT[DEFAULT_DRAFT_MODEL_EAGLE3]
DEFAULT_MODEL_NAME_FOR_TEST_MLA = OFFLINE_PATH_DICT[DEFAULT_MODEL_NAME_FOR_TEST_MLA]
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN = OFFLINE_PATH_DICT[
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN
]
GSM_DATASET_PATH = OFFLINE_PATH_DICT[GSM_DATASET_PATH]
# Default server arguments shared across all tests
DEFAULT_SERVER_ARGS = [
"--trust-remote-code",
"--cuda-graph-max-bs",
"8",
"--attention-backend",
"fa3",
]
"""
Integration test for python/sglang/srt/layers/attention/flashattention_backend.py
"""
@unittest.skipIf(get_device_sm() < 90, "Test requires CUDA SM 90 or higher")
class BaseFlashAttentionTest(CustomTestCase):
"""Base class for testing FlashAttention3."""
model = DEFAULT_MODEL_NAME_FOR_TEST
base_url = DEFAULT_URL_FOR_TEST
accuracy_threshold = 0.65 # derived tests need to override this
speculative_decode = False
spec_decode_threshold = 1.0 # derived spec decoding tests need to override this
@classmethod
def get_server_args(cls):
"""Return the arguments for the server launch. Override in subclasses."""
return DEFAULT_SERVER_ARGS
@classmethod
def setUpClass(cls):
# disable deep gemm precompile to make launch server faster
# please don't do this if you want to make your inference workload faster
with (
envs.SGLANG_JIT_DEEPGEMM_PRECOMPILE.override(False),
envs.SGLANG_ENABLE_JIT_DEEPGEMM.override(False),
):
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=cls.get_server_args(),
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
requests.get(self.base_url + "/flush_cache")
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=100,
num_threads=128,
num_shots=4,
gsm8k_data_path=GSM_DATASET_PATH,
)
metrics = run_eval(args)
print(f"{metrics=}")
# Use the appropriate metric key based on the test class
metric_key = "score"
self.assertGreater(metrics[metric_key], self.accuracy_threshold)
if self.speculative_decode:
server_info = requests.get(self.base_url + "/server_info").json()
avg_spec_accept_length = server_info["internal_states"][0][
"avg_spec_accept_length"
]
print(f"{avg_spec_accept_length=}")
self.assertGreater(avg_spec_accept_length, self.spec_decode_threshold)
class TestFlashAttention3MLA(BaseFlashAttentionTest):
"""Test FlashAttention3 with MLA, e.g. deepseek v3 test model"""
accuracy_threshold = 0.60
model = DEFAULT_MODEL_NAME_FOR_TEST_MLA
@classmethod
def get_server_args(cls):
return DEFAULT_SERVER_ARGS
class TestFlashAttention3SpeculativeDecode(BaseFlashAttentionTest):
"""Test FlashAttention3 with speculative decode enabled with Llama 3.1 8B and its eagle3 model"""
model = DEFAULT_MODEL_NAME_FOR_TEST
accuracy_threshold = 0.65
speculative_decode = True
spec_decode_threshold = 1.5
@classmethod
def get_server_args(cls):
args = DEFAULT_SERVER_ARGS
args.extend(
[
"--cuda-graph-max-bs",
"4",
"--speculative-algorithm",
"EAGLE3",
"--speculative-draft-model-path",
DEFAULT_DRAFT_MODEL_EAGLE3,
"--speculative-num-steps",
"3",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"4",
"--dtype",
"float16",
]
)
return args
class TestFlashAttention3SpeculativeDecodeTopk(BaseFlashAttentionTest):
"""Tests FlashAttention3 with enhanced speculative decoding using Llama 3.1 8B and EAGLE3.
This test will be using top-k value > 1 which would verify the other branches of the FA3 code
"""
model = DEFAULT_MODEL_NAME_FOR_TEST
accuracy_threshold = 0.65
speculative_decode = True
spec_decode_threshold = 1.6
@classmethod
def get_server_args(cls):
args = DEFAULT_SERVER_ARGS
args.extend(
[
"--cuda-graph-max-bs",
"4",
"--speculative-algorithm",
"EAGLE3",
"--speculative-draft-model-path",
DEFAULT_DRAFT_MODEL_EAGLE3,
"--speculative-num-steps",
"5",
"--speculative-eagle-topk",
"4",
"--speculative-num-draft-tokens",
"8",
"--dtype",
"float16",
]
)
return args
class TestFlashAttention3MLASpeculativeDecode(BaseFlashAttentionTest):
"""Test FlashAttention3 with speculative decode enabled with deepseek v3 test model and its nextN model"""
model = DEFAULT_MODEL_NAME_FOR_TEST_MLA
accuracy_threshold = 0.60
speculative_decode = True
spec_decode_threshold = 2.5
@classmethod
def get_server_args(cls):
args = DEFAULT_SERVER_ARGS
args.extend(
[
"--cuda-graph-max-bs",
"4",
"--speculative-algorithm",
"EAGLE",
"--speculative-draft-model-path",
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN,
"--speculative-num-steps",
"3",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"4",
]
)
return args
class TestFlashAttention3MLASpeculativeDecodeTopk(BaseFlashAttentionTest):
"""Test FlashAttention3 with speculative decode enabled with deepseek v3 test model and its nextN model
This test will be using top-k value > 1 which would verify the other branches of the FA3 code
"""
model = DEFAULT_MODEL_NAME_FOR_TEST_MLA
accuracy_threshold = 0.60
speculative_decode = True
spec_decode_threshold = 2.95
@classmethod
def get_server_args(cls):
args = DEFAULT_SERVER_ARGS
args.extend(
[
"--cuda-graph-max-bs",
"4",
"--speculative-algorithm",
"EAGLE",
"--speculative-draft-model-path",
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN,
"--speculative-num-steps",
"5",
"--speculative-eagle-topk",
"4",
"--speculative-num-draft-tokens",
"8",
]
)
return args
if __name__ == "__main__":
unittest.main()
test/manual/attention/test_local_attn.py
+77
View File
@@ -0,0 +1,77 @@
import os
import unittest
from types import SimpleNamespace
import requests
from sglang.srt.utils import get_device_sm, kill_process_tree
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_TEST_LOCAL_ATTENTION,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
# Local attention with FA3 (requires SM 90+ / H100, tp=4)
register_cuda_ci(est_time=217, suite="stage-c-test-4-gpu-h100")
@unittest.skipIf(get_device_sm() < 90, "Test requires CUDA SM 90 or higher")
class TestFlashAttention3LocalAttn(CustomTestCase):
model = DEFAULT_MODEL_NAME_FOR_TEST_LOCAL_ATTENTION
base_url = DEFAULT_URL_FOR_TEST
accuracy_threshold = 0.90
@classmethod
def get_server_args(cls):
return [
"--cuda-graph-max-bs",
"2",
"--attention-backend",
"fa3",
"--tp",
"4",
"--context-length",
"1000000",
]
@classmethod
def setUpClass(cls):
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=cls.get_server_args(),
env=os.environ,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
requests.get(self.base_url + "/flush_cache")
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=100,
num_threads=128,
num_shots=4,
)
metrics = run_eval(args)
print(f"{metrics=}")
# Use the appropriate metric key based on the test class
metric_key = "score"
self.assertGreater(metrics[metric_key], self.accuracy_threshold)
if __name__ == "__main__":
unittest.main()
test/manual/core/test_gpt_oss_1gpu.py
+35
View File
@@ -0,0 +1,35 @@
import unittest
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.gpt_oss_common import BaseTestGptOss
register_cuda_ci(est_time=408, suite="stage-b-test-1-gpu-large")
register_amd_ci(est_time=750, suite="stage-b-test-1-gpu-small-amd-mi35x")
class TestGptOss1Gpu(BaseTestGptOss):
def test_mxfp4_20b(self):
self.run_test(
model_variant="20b",
quantization="mxfp4",
expected_score_of_reasoning_effort={
"low": 0.34,
"medium": 0.34,
"high": 0.27, # TODO investigate
},
)
def test_bf16_20b(self):
self.run_test(
model_variant="20b",
quantization="bf16",
expected_score_of_reasoning_effort={
"low": 0.34,
"medium": 0.34,
"high": 0.27, # TODO investigate
},
)
if __name__ == "__main__":
unittest.main()
test/manual/distributed/test_dp_attention_archived.py
+101
View File
@@ -0,0 +1,101 @@
"""Archived test classes split out of test/registered/distributed/test_dp_attention.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/distributed/test_dp_attention_archived.py`.
"""
import unittest
from types import SimpleNamespace
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.kits.ebnf_constrained_kit import EBNFConstrainedMixin
from sglang.test.kits.json_constrained_kit import JSONConstrainedMixin
from sglang.test.kits.regex_constrained_kit import RegexConstrainedMixin
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_TEST_MLA,
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
is_in_amd_ci,
popen_launch_server,
)
class TestDPAttentionDP2TP2DeepseekV3MTP(
CustomTestCase,
JSONConstrainedMixin,
EBNFConstrainedMixin,
RegexConstrainedMixin,
):
@classmethod
def setUpClass(cls):
cls.model = DEFAULT_MODEL_NAME_FOR_TEST_MLA
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--disable-radix",
"--speculative-algorithm",
"EAGLE",
"--speculative-num-steps",
"2",
"--speculative-eagle-topk",
"4",
"--speculative-num-draft-tokens",
"4",
"--speculative-draft-model-path",
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN,
"--tp-size",
"2",
"--enable-dp-attention",
"--dp-size",
"2",
]
if not is_in_amd_ci():
other_args += ["--mem-frac", "0.7"]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
requests.get(self.base_url + "/flush_cache")
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.60)
server_info = requests.get(self.base_url + "/server_info")
avg_spec_accept_length = server_info.json()["internal_states"][0][
"avg_spec_accept_length"
]
print(
f"###test_gsm8k (deepseek-v3 mtp + dp):\n"
f"accuracy={metrics['score']=:.3f}\n"
f"{avg_spec_accept_length=:.3f}\n"
)
self.assertGreater(avg_spec_accept_length, 2.5)
if __name__ == "__main__":
unittest.main()
test/manual/distributed/test_dp_attention_large.py
+192
View File
@@ -0,0 +1,192 @@
import unittest
from types import SimpleNamespace
import requests
from sglang.lang.chat_template import get_chat_template_by_model_path
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.ebnf_constrained_kit import EBNFConstrainedMixin
from sglang.test.kits.json_constrained_kit import JSONConstrainedMixin
from sglang.test.kits.regex_constrained_kit import RegexConstrainedMixin
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_IMAGE_URL,
DEFAULT_MLA_MODEL_NAME_FOR_TEST,
DEFAULT_MODEL_NAME_FOR_TEST_MLA,
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
is_in_amd_ci,
popen_launch_server,
)
register_cuda_ci(est_time=245, suite="stage-c-test-4-gpu-h100")
register_amd_ci(est_time=350, suite="stage-c-test-4-gpu-amd")
@unittest.skipIf(
is_in_amd_ci(),
"DeepSeek MLA forward_mla NameError on AMD (batched_gemm not defined)",
)
class TestDPAttentionDP2TP4(
CustomTestCase,
JSONConstrainedMixin,
EBNFConstrainedMixin,
RegexConstrainedMixin,
):
@classmethod
def setUpClass(cls):
cls.model = DEFAULT_MLA_MODEL_NAME_FOR_TEST
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--trust-remote-code",
"--tp=4",
"--enable-dp-attention",
"--dp=2",
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
num_examples=None,
num_threads=1024,
)
metrics = run_eval(args)
print(f"{metrics=}")
self.assertGreater(metrics["score"], 0.8)
@unittest.skipIf(
is_in_amd_ci(),
"DeepSeek MTP forward_mla NameError on AMD + needs 8 GPUs",
)
class TestDPAttentionDP2TP2DeepseekV3MTP(
CustomTestCase,
JSONConstrainedMixin,
EBNFConstrainedMixin,
RegexConstrainedMixin,
):
@classmethod
def setUpClass(cls):
cls.model = DEFAULT_MODEL_NAME_FOR_TEST_MLA
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--disable-radix",
"--speculative-algorithm=EAGLE",
"--speculative-num-steps=2",
"--speculative-eagle-topk=4",
"--speculative-num-draft-tokens=4",
"--speculative-draft-model-path",
DEFAULT_MODEL_NAME_FOR_TEST_MLA_NEXTN,
"--tp-size=4",
"--enable-dp-attention",
"--dp-size=2",
]
if not is_in_amd_ci():
other_args += ["--mem-frac", "0.7"]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
requests.get(self.base_url + "/flush_cache")
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.60)
server_info = requests.get(self.base_url + "/server_info")
avg_spec_accept_length = server_info.json()["internal_states"][0][
"avg_spec_accept_length"
]
print(
f"###test_gsm8k (deepseek-v3 mtp + dp):\n"
f"accuracy={metrics['score']=:.3f}\n"
f"{avg_spec_accept_length=:.3f}\n"
)
self.assertGreater(avg_spec_accept_length, 2.5)
@unittest.skipIf(
is_in_amd_ci(),
"Qwen3-VL-30B-A3B-Instruct OOMs at TP=4 DP=2 on MI325 4-GPU runners",
)
class TestDPAttentionDP2TP4VLM(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "Qwen/Qwen3-VL-30B-A3B-Instruct"
cls.base_url = DEFAULT_URL_FOR_TEST
cls.image_url = DEFAULT_IMAGE_URL
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--trust-remote-code",
"--tp",
"4",
"--enable-dp-attention",
"--dp",
"2",
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_vlm_generate(self):
chat_template = get_chat_template_by_model_path(self.model)
prompt = f"{chat_template.image_token}What is in this image?"
response = requests.post(
self.base_url + "/generate",
json={
"text": prompt,
"image_data": [self.image_url],
"sampling_params": {
"temperature": 0,
"max_new_tokens": 16,
},
},
)
response.raise_for_status()
response_json = response.json()
print(response_json)
self.assertIn("output_ids", response_json)
self.assertGreater(len(response_json["output_ids"]), 0)
if __name__ == "__main__":
unittest.main()
test/manual/eval/test_eval_accuracy_large.py
+46
View File
@@ -0,0 +1,46 @@
"""
Usage:
python -m unittest test_eval_accuracy_large.TestEvalAccuracyLarge.test_mmlu
"""
import unittest
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.eval_accuracy_kit import HumanEvalMixin, MGSMEnMixin, MMLUMixin
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_cuda_ci(est_time=496, suite="stage-b-test-1-gpu-small")
register_amd_ci(est_time=420, suite="stage-b-test-1-gpu-small-amd")
class TestEvalAccuracyLarge(CustomTestCase, MMLUMixin, HumanEvalMixin, MGSMEnMixin):
mmlu_score_threshold = 0.70
humaneval_score_threshold = 0.64
humaneval_score_threshold_amd = 0.60
mgsm_en_score_threshold = 0.835
@classmethod
def setUpClass(cls):
cls.model = DEFAULT_MODEL_NAME_FOR_TEST
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=["--log-level-http", "warning"],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
if __name__ == "__main__":
unittest.main()
test/manual/lora/test_lora_backend.py
+83
View File
@@ -0,0 +1,83 @@
# Copyright 2023-2024 SGLang Team
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
import multiprocessing as mp
import os
import unittest
from typing import List
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.lora_utils import (
ALL_OTHER_LORA_MODELS,
BACKENDS,
CI_LORA_MODELS,
DEFAULT_PROMPTS,
TORCH_DTYPES,
LoRAModelCase,
run_lora_test_one_by_one,
)
from sglang.test.test_utils import CustomTestCase, is_in_ci
register_cuda_ci(est_time=224, suite="stage-b-test-1-gpu-small")
register_amd_ci(
est_time=200,
suite="stage-b-test-1-gpu-small-amd",
disabled="see https://github.com/sgl-project/sglang/issues/13107",
)
class TestLoRABackend(CustomTestCase):
def _run_backend_on_model_cases(self, model_cases: List[LoRAModelCase]):
for model_case in model_cases:
# If skip_long_prompt is True, filter out prompts longer than 1000 characters
prompts = (
DEFAULT_PROMPTS
if not model_case.skip_long_prompt
else [p for p in DEFAULT_PROMPTS if len(p) < 1000]
)
for torch_dtype in TORCH_DTYPES:
for backend in BACKENDS:
run_lora_test_one_by_one(
prompts,
model_case,
torch_dtype,
max_new_tokens=32,
backend=backend,
)
def test_ci_lora_models(self):
self._run_backend_on_model_cases(CI_LORA_MODELS)
def test_all_lora_models(self):
if is_in_ci():
return
# Retain ONLY_RUN check here
filtered_models = []
for model_case in ALL_OTHER_LORA_MODELS:
if "ONLY_RUN" in os.environ and os.environ["ONLY_RUN"] != model_case.base:
continue
filtered_models.append(model_case)
self._run_backend_on_model_cases(filtered_models)
if __name__ == "__main__":
try:
mp.set_start_method("spawn")
except RuntimeError:
pass
unittest.main(warnings="ignore")
test/manual/mla/test_flashmla_archived.py
+73
View File
@@ -0,0 +1,73 @@
"""Archived test classes split out of test/registered/mla/test_flashmla.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/mla/test_flashmla_archived.py`.
"""
"""
Usage:
python3 test/registered/mla/test_flashmla.py
"""
import unittest
from types import SimpleNamespace
import torch
from sglang.srt.utils import kill_process_tree
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_TEST_MLA,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
popen_launch_server,
)
# FlashMLA attention backend tests with MTP speculative decoding
class TestFlashMLAAttnBackend(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.model = DEFAULT_MODEL_NAME_FOR_TEST_MLA
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = ["--trust-remote-code"]
if torch.cuda.is_available() and torch.version.cuda:
other_args.extend(
[
"--cuda-graph-max-bs",
"2",
"--attention-backend",
"flashmla",
]
)
# Use longer timeout for DeepGEMM JIT compilation which can take 10-20 minutes
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH * 2,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.60)
if __name__ == "__main__":
unittest.main()
test/manual/mla/test_mla.py
+46
View File
@@ -0,0 +1,46 @@
import unittest
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.eval_accuracy_kit import MGSMEnMixin
from sglang.test.test_utils import (
DEFAULT_MLA_MODEL_NAME_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
# MLA attention test with MGSM evaluation
register_cuda_ci(est_time=181, suite="stage-b-test-1-gpu-large")
register_amd_ci(est_time=1100, suite="stage-b-test-1-gpu-small-amd")
class TestMLA(CustomTestCase, MGSMEnMixin):
mgsm_en_score_threshold = 0.8
@classmethod
def setUpClass(cls):
cls.model = DEFAULT_MLA_MODEL_NAME_FOR_TEST
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--trust-remote-code",
"--enable-torch-compile",
"--torch-compile-max-bs",
"4",
"--chunked-prefill-size",
"256",
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
if __name__ == "__main__":
unittest.main()
test/manual/mla/test_mla_deepseek_v3.py
+211
View File
@@ -0,0 +1,211 @@
import os
import unittest
from types import SimpleNamespace
import requests
from sglang.srt.utils import is_cuda, is_hip, kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
is_in_ci,
popen_launch_server,
)
# DeepSeek-V3 MLA tests with torch compile, FA3, and MTP speculative decoding
register_cuda_ci(est_time=543, suite="stage-b-test-1-gpu-large")
register_amd_ci(
est_time=221,
suite="stage-b-test-1-gpu-small-amd",
disabled="see https://github.com/sgl-project/sglang/issues/12574",
)
class TestMLADeepseekV3(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "lmsys/sglang-ci-dsv3-test"
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = ["--trust-remote-code", "--chunked-prefill-size", "256"]
if is_cuda():
other_args.extend(["--enable-torch-compile", "--cuda-graph-max-bs", "2"])
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.60)
@unittest.skipIf(is_in_ci(), "To reduce the CI execution time.")
class TestMLADeepseekV3DisableFusedFunc(CustomTestCase):
@classmethod
def setUpClass(cls):
os.environ["SGLANG_CI_DISABLE_MOE_FUSED_FUNC"] = "1"
cls.model = "lmsys/sglang-ci-dsv3-test"
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = ["--trust-remote-code", "--chunked-prefill-size", "256"]
if is_cuda():
other_args.extend(["--cuda-graph-max-bs", "2"])
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.62)
@unittest.skipIf(is_hip(), "FA is not available.")
class TestMLADeepseekV3Fa3Fp8Kvcache(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "lmsys/sglang-ci-dsv3-test"
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--chunked-prefill-size",
"256",
"--kv-cache-dtype",
"fp8_e4m3",
]
if is_cuda():
other_args.extend(
[
"--attention-backend",
"fa3",
"--mem-fraction-static",
"0.8",
"--cuda-graph-max-bs",
"2",
]
)
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.60)
class TestDeepseekV3MTP(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "lmsys/sglang-ci-dsv3-test"
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--cuda-graph-max-bs",
"2",
"--disable-radix",
"--enable-torch-compile",
"--torch-compile-max-bs",
"1",
"--speculative-algorithm",
"EAGLE",
"--speculative-num-steps",
"2",
"--speculative-eagle-topk",
"4",
"--speculative-num-draft-tokens",
"4",
]
# This test runs first (alphabetically) and needs longer timeout for
# DeepGEMM JIT compilation which is required for DeepSeek-V3's FP8 MoE layers
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH * 2,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
requests.get(self.base_url + "/flush_cache")
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.60)
server_info = requests.get(self.base_url + "/server_info")
avg_spec_accept_length = server_info.json()["internal_states"][0][
"avg_spec_accept_length"
]
print(f"{avg_spec_accept_length=}")
self.assertGreater(avg_spec_accept_length, 2.5)
if __name__ == "__main__":
unittest.main()
test/manual/mla/test_mla_flashinfer_archived.py
+68
View File
@@ -0,0 +1,68 @@
"""Archived test classes split out of test/registered/mla/test_mla_flashinfer.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/mla/test_mla_flashinfer_archived.py`.
"""
import unittest
from types import SimpleNamespace
import torch
from sglang.srt.utils import kill_process_tree
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
# FlashInfer MLA backend tests with MTP speculative decoding
class TestFlashinferMLA(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "lmsys/sglang-ci-dsv3-test"
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = ["--trust-remote-code"]
if torch.cuda.is_available() and torch.version.cuda:
other_args.extend(
[
"--enable-torch-compile",
"--cuda-graph-max-bs",
"4",
"--attention-backend",
"flashinfer",
]
)
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.615)
if __name__ == "__main__":
unittest.main()
test/manual/mla/test_mla_int8_deepseek_v3_archived.py
+113
View File
@@ -0,0 +1,113 @@
"""Archived test classes split out of test/registered/mla/test_mla_int8_deepseek_v3.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/mla/test_mla_int8_deepseek_v3_archived.py`.
"""
import unittest
from types import SimpleNamespace
import torch
from sglang.srt.utils import kill_process_tree
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
is_in_ci,
popen_launch_server,
)
# DeepSeek-V3 INT8 quantization tests (channel and block INT8)
class TestMLADeepseekV3ChannelInt8(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "lmsys/sglang-ci-dsv3-channel-int8-test"
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = ["--trust-remote-code"]
if torch.cuda.is_available() and torch.version.cuda:
other_args.extend(
[
"--cuda-graph-max-bs",
"16",
"--enable-torch-compile",
"--torch-compile-max-bs",
"2",
]
)
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreaterEqual(metrics["score"], 0.61)
@unittest.skipIf(is_in_ci(), "To reduce the CI execution time.")
class TestMLADeepseekV3BlockInt8(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "lmsys/sglang-ci-dsv3-block-int8-test"
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = ["--trust-remote-code"]
if torch.cuda.is_available() and torch.version.cuda:
other_args.extend(
[
"--cuda-graph-max-bs",
"16",
"--enable-torch-compile",
"--torch-compile-max-bs",
"2",
]
)
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.62)
if __name__ == "__main__":
unittest.main()
test/manual/models/test_nvidia_nemotron_3_nano_archived.py
+47
View File
@@ -0,0 +1,47 @@
"""Archived test classes split out of test/registered/models/test_nvidia_nemotron_3_nano.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/models/test_nvidia_nemotron_3_nano_archived.py`.
"""
import unittest
from sglang.test.kits.lm_eval_kit import LMEvalMixin
from sglang.test.server_fixtures.default_fixture import DefaultServerBase
NEMOTRON_3_NANO_THINKING_ARGS = [
"--trust-remote-code",
"--tool-call-parser",
"qwen3_coder",
"--reasoning-parser",
"deepseek-r1",
]
class TestNvidiaNemotron3Nano30BBF16(LMEvalMixin, DefaultServerBase):
"""Test Nemotron-3-Nano-30B BF16 model with lm-eval GSM8K evaluation."""
model = "nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16"
model_config_name = "lm_eval_configs/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16.yaml"
other_args = [
"--tp-size",
"2",
] + NEMOTRON_3_NANO_THINKING_ARGS
class TestNvidiaNemotron3Nano30BBF16FlashInfer(LMEvalMixin, DefaultServerBase):
"""Test Nemotron-3-Nano-30B BF16 model with lm-eval GSM8K evaluation using flashinfer mamba backend."""
model = "nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16"
model_config_name = "lm_eval_configs/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16.yaml"
other_args = [
"--tp-size",
"2",
"--mamba-backend",
"flashinfer",
] + NEMOTRON_3_NANO_THINKING_ARGS
if __name__ == "__main__":
unittest.main()
test/manual/models/test_nvidia_nemotron_nano_v2.py
+99
View File
@@ -0,0 +1,99 @@
import unittest
from sglang.srt.utils import is_blackwell
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.kits.eval_accuracy_kit import GSM8KMixin
from sglang.test.server_fixtures.default_fixture import DefaultServerBase
register_cuda_ci(est_time=249, suite="stage-b-test-2-gpu-large")
class TestNvidiaNemotronNanoV2BF16(GSM8KMixin, DefaultServerBase):
model = "nvidia/NVIDIA-Nemotron-Nano-9B-v2"
gsm8k_accuracy_thres = 0.87
other_args = ["--max-mamba-cache-size", "256"]
class TestNvidiaNemotronNanoV2BF16PP(GSM8KMixin, DefaultServerBase):
model = "nvidia/NVIDIA-Nemotron-Nano-9B-v2"
gsm8k_accuracy_thres = 0.87
other_args = ["--max-mamba-cache-size", "256", "--pp-size", "2"]
class TestNvidiaNemotronNanoV2FP8(GSM8KMixin, DefaultServerBase):
gsm8k_accuracy_thres = 0.87
model = "nvidia/NVIDIA-Nemotron-Nano-9B-v2-FP8"
other_args = ["--max-mamba-cache-size", "256"]
@unittest.skipIf(not is_blackwell(), "NVFP4 only supported on blackwell")
class TestNvidiaNemotronNanoV2NVFP4(GSM8KMixin, DefaultServerBase):
gsm8k_accuracy_thres = 0.855
model = "nvidia/NVIDIA-Nemotron-Nano-9B-v2-NVFP4"
other_args = ["--max-mamba-cache-size", "256"]
@unittest.skip(
"STANDALONE speculative decoding does not yet support target and draft models "
"with different hidden sizes (Nemotron-9B: 4480, Llama-3.2-1B: 2048)"
)
class TestNvidiaNemotronNanoV2SpeculativeDecoding(GSM8KMixin, DefaultServerBase):
gsm8k_accuracy_thres = 0.87
model = "nvidia/NVIDIA-Nemotron-Nano-9B-v2"
other_args = [
"--speculative-algorithm",
"STANDALONE",
"--speculative-num-steps",
"2",
"--speculative-eagle-topk",
"3",
"--speculative-num-draft-tokens",
"5",
"--speculative-draft-model-path",
"meta-llama/Llama-3.2-1B",
"--speculative-draft-load-format",
"dummy",
"--max-running-requests",
"8",
"--max-total-tokens",
"2048",
"--json-model-override-args",
'{"vocab_size": 131072}',
]
@unittest.skip(
"STANDALONE speculative decoding does not yet support target and draft models "
"with different hidden sizes (Nemotron-9B: 4480, Llama-3.2-1B: 2048)"
)
class TestNvidiaNemotronNanoV2SpeculativeDecodingBF16Cache(
GSM8KMixin, DefaultServerBase
):
gsm8k_accuracy_thres = 0.87
model = "nvidia/NVIDIA-Nemotron-Nano-9B-v2"
other_args = [
"--speculative-algorithm",
"STANDALONE",
"--speculative-num-steps",
"2",
"--speculative-eagle-topk",
"3",
"--speculative-num-draft-tokens",
"5",
"--speculative-draft-model-path",
"meta-llama/Llama-3.2-1B",
"--speculative-draft-load-format",
"dummy",
"--max-running-requests",
"8",
"--max-total-tokens",
"2048",
"--json-model-override-args",
'{"vocab_size": 131072}',
"--mamba-ssm-dtype",
"bfloat16",
]
if __name__ == "__main__":
unittest.main()
test/manual/models/test_nvidia_nemotron_nano_v2_vl.py
+33
View File
@@ -0,0 +1,33 @@
import unittest
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.kits.eval_accuracy_kit import GSM8KMixin
from sglang.test.kits.mmmu_vlm_kit import MMMUMixin
from sglang.test.server_fixtures.default_fixture import DefaultServerBase
from sglang.test.server_fixtures.mmmu_fixture import MMMUServerBase
# NVIDIA Nemotron Nano V2 VL model tests (CUDA only)
# GSM8k + MMMU evaluation
register_cuda_ci(est_time=256, suite="stage-b-test-1-gpu-large")
MODEL = "nvidia/NVIDIA-Nemotron-Nano-12B-v2-VL-BF16"
class TestNvidiaNemotronNanoV2VLTextOnly(GSM8KMixin, DefaultServerBase):
gsm8k_accuracy_thres = 0.85
model = MODEL
other_args = ["--max-mamba-cache-size", "256", "--trust-remote-code"]
class TestNvidiaNemotronNanoV2VLMMMU(MMMUMixin, MMMUServerBase):
accuracy = 0.444
model = MODEL
other_args = ["--max-mamba-cache-size", "128", "--trust-remote-code"]
mmmu_args = ["--limit=0.1"]
"""`--limit=0.1`: 10 percent of each task - this is fine for testing since the nominal result isn't interesting - this run is just to prevent relative regressions."""
if __name__ == "__main__":
unittest.main()
test/manual/models/test_qwen_models.py
+83
View File
@@ -0,0 +1,83 @@
# Qwen model tests
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.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_cuda_ci(est_time=108, suite="stage-b-test-1-gpu-small")
register_amd_ci(est_time=130, suite="stage-b-test-1-gpu-small-amd")
class TestQwen2(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "Qwen/Qwen2-7B-Instruct"
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(f"{metrics=}")
self.assertGreater(metrics["score"], 0.78)
class TestQwen2FP8(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "neuralmagic/Qwen2-7B-Instruct-FP8"
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=200,
num_threads=128,
)
metrics = run_eval(args)
print(f"{metrics=}")
self.assertGreater(metrics["score"], 0.78)
if __name__ == "__main__":
unittest.main()
test/manual/perf/test_bench_one_batch_1gpu.py
+98
View File
@@ -0,0 +1,98 @@
import os
import re
import subprocess
import unittest
import numpy as np
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_TEST,
DEFAULT_SMALL_MODEL_NAME_FOR_TEST,
CustomTestCase,
is_in_ci,
kill_process_tree,
run_bench_one_batch,
write_github_step_summary,
)
register_cuda_ci(est_time=95, suite="stage-b-test-1-gpu-large")
register_amd_ci(est_time=120, suite="stage-b-test-1-gpu-large-amd")
class TestBenchOneBatch1GPU(CustomTestCase):
def test_bs1_small(self):
_, output_throughput, _ = run_bench_one_batch(
DEFAULT_SMALL_MODEL_NAME_FOR_TEST, ["--cuda-graph-max-bs", "2"]
)
self.assertGreater(output_throughput, 50)
def test_bs1_default(self):
env = os.environ.copy()
env["SGLANG_ENABLE_METRICS_DEVICE_TIMER"] = "1"
command = [
"python3",
"-m",
"sglang.bench_offline_throughput",
"--num-prompts",
"1",
"--dataset-name",
"random",
"--random-input-len",
"256",
"--random-output-len",
"1024",
"--model-path",
DEFAULT_MODEL_NAME_FOR_TEST,
"--cuda-graph-max-bs",
"2",
]
print(f"command={' '.join(command)}")
process = subprocess.Popen(
command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env
)
try:
stdout, stderr = process.communicate()
output = stdout.decode(errors="backslashreplace")
error = stderr.decode(errors="backslashreplace")
print(f"Output: {output}", flush=True)
print(f"Error: {error}", flush=True)
output_throughput = -1
for line in output.split("\n"):
if "Last generation throughput (tok/s):" in line:
output_throughput = float(line.split(":")[-1])
finally:
kill_process_tree(process.pid)
if is_in_ci():
write_github_step_summary(
f"### test_bs1_default (llama-3.1-8b)\n"
f"output_throughput: {output_throughput:.2f} token/s\n"
)
self.assertGreater(output_throughput, 135)
fwd_occupancy_values = []
for line in error.split("\n"):
match = re.search(r"fwd occupancy:\s*([\d.]+|nan)%", line)
if match:
val = match.group(1)
if val != "nan":
fwd_occupancy_values.append(float(val))
print(f"{fwd_occupancy_values=}", flush=True)
self.assertGreater(
len(fwd_occupancy_values), 0, "No fwd occupancy values found in logs"
)
fwd_occupancy_p90 = float(np.percentile(fwd_occupancy_values, 90))
print(f"{fwd_occupancy_p90=}", flush=True)
self.assertGreater(fwd_occupancy_p90, 97.5)
if __name__ == "__main__":
unittest.main()
test/manual/piecewise_cuda_graph/test_piecewise_cuda_graph_support_1_gpu_archived.py
+63
View File
@@ -0,0 +1,63 @@
"""Archived test classes split out of test/registered/piecewise_cuda_graph/test_piecewise_cuda_graph_support_1_gpu.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/piecewise_cuda_graph/test_piecewise_cuda_graph_support_1_gpu_archived.py`.
"""
import unittest
from sglang.srt.utils import kill_process_tree
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
SimpleNamespace,
popen_launch_server,
)
# CI Registration
class TestPiecewiseCudaGraphInternVL25(CustomTestCase):
"""Test piecewise CUDA graph with InternVL2.5-8B model"""
@classmethod
def setUpClass(cls):
cls.model = "OpenGVLab/InternVL2_5-8B"
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--enforce-piecewise-cuda-graph",
"--disable-radix-cache",
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k_accuracy(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
num_examples=None,
num_threads=1024,
)
metrics = run_eval(args)
print(f"GSM8K Accuracy: {metrics['score']:.3f}")
# Baseline (no piecewise CUDA graph): 0.571 — this eval uses 5-shot
# concatenated text via chat API, which scores lower than reported
# benchmarks (~77.8%) that use proper CoT chat format. The threshold
# is set 5% below observed to catch catastrophic regressions.
self.assertGreaterEqual(metrics["score"], 0.54)
if __name__ == "__main__":
unittest.main()
test/manual/prefill_only/test_cross_encoder_models.py
+97
View File
@@ -0,0 +1,97 @@
import multiprocessing as mp
import random
import unittest
import torch
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.runners import TEST_RERANK_QUERY_DOCS, HFRunner, SRTRunner
from sglang.test.test_utils import CustomTestCase, is_in_ci
# Cross encoder model tests
register_cuda_ci(est_time=125, suite="stage-b-test-1-gpu-small")
register_amd_ci(est_time=150, suite="stage-b-test-1-gpu-small-amd")
MODELS = [
("cross-encoder/ms-marco-MiniLM-L6-v2", 1, 1e-2),
("BAAI/bge-reranker-v2-m3", 1, 1e-2),
]
ATTENTION_BACKEND = ["torch_native", "triton"]
TORCH_DTYPES = [torch.float32]
class TestCrossEncoderModels(CustomTestCase):
@classmethod
def setUpClass(cls):
mp.set_start_method("spawn", force=True)
def assert_close_prefill_logits(
self,
prompts,
model_path,
tp_size,
torch_dtype,
score_tolerance,
attention_backend,
) -> None:
with HFRunner(
model_path,
torch_dtype=torch_dtype,
model_type="cross_encoder",
) as hf_runner:
hf_scores = hf_runner.forward(prompts).scores
with SRTRunner(
model_path,
tp_size=tp_size,
torch_dtype=torch_dtype,
model_type="cross_encoder",
attention_backend=attention_backend,
chunked_prefill_size=-1,
disable_radix_cache=True,
) as srt_runner:
srt_scores = srt_runner.forward(prompts).scores
for i in range(len(srt_scores)):
score_difference = abs(hf_scores[i] - srt_scores[i])
assert (
score_difference < score_tolerance
), "cross encoder scores are not all close"
def preprocess_prompts(self, prompt):
processed_prompts = []
query = prompt["query"]
documents = prompt["documents"]
for document in documents:
processed_prompts.append([query, document])
return processed_prompts
def test_prefill_logits(self):
models_to_test = MODELS
if is_in_ci():
models_to_test = [random.choice(MODELS)]
for model, tp_size, prefill_tolerance in models_to_test:
for attention_backend in ATTENTION_BACKEND:
for queryDocs in TEST_RERANK_QUERY_DOCS:
prompts = self.preprocess_prompts(queryDocs)
for torch_dtype in TORCH_DTYPES:
self.assert_close_prefill_logits(
prompts,
model,
tp_size,
torch_dtype,
prefill_tolerance,
attention_backend,
)
if __name__ == "__main__":
unittest.main()
test/manual/prefill_only/test_encoder_embedding_models.py
+168
View File
@@ -0,0 +1,168 @@
import multiprocessing as mp
import random
import time
import unittest
import torch
from transformers import AutoConfig, AutoTokenizer
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.runners import DEFAULT_PROMPTS, HFRunner, SRTRunner
from sglang.test.test_utils import CustomTestCase, get_similarities, is_in_ci
# Encoder embedding model tests (CUDA only)
# Copyright 2023-2024 SGLang Team
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
# python -m unittest test_encoder_embedding_models.TestEncoderEmbeddingModels.test_prefill_logits
register_cuda_ci(est_time=444, suite="stage-b-test-1-gpu-small")
MODELS = [("BAAI/bge-small-en", 1, 1e-5), ("BAAI/bge-m3", 1, 1e-5)]
ATTENTION_BACKEND = ["torch_native", "triton", "flashinfer"]
BATCH_SIZE = [1, 2]
TORCH_DTYPES = [torch.float32, torch.float16]
sgl_to_st_ratio = []
class TestEncoderEmbeddingModels(CustomTestCase):
@classmethod
def setUpClass(cls):
mp.set_start_method("spawn", force=True)
def _truncate_prompts(self, prompts, model_path):
config = AutoConfig.from_pretrained(model_path)
max_length = getattr(config, "max_position_embeddings", 512) - 20
tokenizer = AutoTokenizer.from_pretrained(model_path)
truncated_prompts = []
for prompt in prompts:
tokens = tokenizer(prompt, return_tensors="pt", truncation=False)
if len(tokens.input_ids[0]) > max_length:
truncated_text = tokenizer.decode(
tokens.input_ids[0][: max_length - 1], skip_special_tokens=True
)
truncated_prompts.append(truncated_text)
else:
truncated_prompts.append(prompt)
return truncated_prompts
def assert_close_prefill_logits(
self,
prompts,
model_path,
tp_size,
torch_dtype,
prefill_tolerance,
attention_backend,
batch_size,
) -> None:
truncated_prompts = self._truncate_prompts(prompts, model_path)
truncated_prompts = truncated_prompts * batch_size
with HFRunner(
model_path,
torch_dtype=torch_dtype,
model_type="embedding",
) as hf_runner:
# warm up
hf_outputs = hf_runner.forward(truncated_prompts)
st_start_time = time.perf_counter()
hf_outputs = hf_runner.forward(truncated_prompts)
st_end_time = time.perf_counter()
with SRTRunner(
model_path,
tp_size=tp_size,
torch_dtype=torch_dtype,
model_type="embedding",
attention_backend=attention_backend,
chunked_prefill_size=-1,
disable_radix_cache=True,
) as srt_runner:
# warm up
srt_outputs = srt_runner.forward(truncated_prompts)
sgl_start_time = time.perf_counter()
srt_outputs = srt_runner.forward(truncated_prompts)
sgl_end_time = time.perf_counter()
transformer_time = st_end_time - st_start_time
sgl_time = sgl_end_time - sgl_start_time
sgl_to_st_ratio.append(sgl_time / transformer_time)
for i in range(len(truncated_prompts)):
hf_logits = torch.Tensor(hf_outputs.embed_logits[i])
srt_logits = torch.Tensor(srt_outputs.embed_logits[i])
similarity = torch.tensor(get_similarities(hf_logits, srt_logits))
# If something is wrong, uncomment this to observe similarity.
# print("similarity diff", abs(similarity - 1))
if len(truncated_prompts[i]) <= 1000:
assert torch.all(
abs(similarity - 1) < prefill_tolerance
), "embeddings are not all close"
def test_prefill_logits(self):
models_to_test = MODELS
if is_in_ci():
models_to_test = [random.choice(MODELS)]
for model, tp_size, prefill_tolerance in models_to_test:
for attention_backend in ATTENTION_BACKEND:
for batch_size in BATCH_SIZE:
for torch_dtype in TORCH_DTYPES:
# NOTE: FlashInfer currently has limitations with head_dim = 32 or
# other dimensions.
# The FlashInfer head_dim limitation itself is tracked here:
# https://github.com/flashinfer-ai/flashinfer/issues/1048
#
# Flashinfer does not support torch.float32 for dtype_q, so skip it
if attention_backend == "flashinfer":
if (
model == "BAAI/bge-small-en"
or torch_dtype == torch.float32
):
continue
self.assert_close_prefill_logits(
DEFAULT_PROMPTS,
model,
tp_size,
torch_dtype,
prefill_tolerance,
attention_backend,
batch_size,
)
for i in range(len(BATCH_SIZE)):
print(
"bacth size: ",
BATCH_SIZE[i] * 5,
"sgl_time/st_time",
round(sgl_to_st_ratio[i], 3),
)
if __name__ == "__main__":
unittest.main()
test/manual/quant/test_autoround.py
+65
View File
@@ -0,0 +1,65 @@
"""
Usage:
python3 -m unittest test_autoround.TestAutoRound.test_mmlu
"""
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_AUTOROUND_MODEL_NAME_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_cuda_ci(est_time=99, suite="stage-b-test-1-gpu-large")
class TestAutoRound(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.base_url = DEFAULT_URL_FOR_TEST
@classmethod
def tearDownClass(cls):
pass
def test_mmlu(self):
device = "auto"
for model in DEFAULT_AUTOROUND_MODEL_NAME_FOR_TEST:
with self.subTest(model=model):
print(f"\n[INFO] Launching server for model: {model}")
process = popen_launch_server(
model,
self.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=["--trust-remote-code", "--quantization", "auto-round"],
device=device,
)
try:
args = SimpleNamespace(
base_url=self.base_url,
model=model,
eval_name="mmlu",
num_examples=32,
num_threads=32,
device=device,
)
metrics = run_eval(args)
if "Llama" in model:
self.assertGreaterEqual(metrics["score"], 0.6)
else:
self.assertGreaterEqual(metrics["score"], 0.25)
finally:
kill_process_tree(process.pid)
print(f"[INFO] Server for {model} stopped.")
if __name__ == "__main__":
unittest.main()
test/manual/quant/test_awq_archived.py
+57
View File
@@ -0,0 +1,57 @@
"""Archived test classes split out of test/registered/quant/test_awq.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/quant/test_awq_archived.py`.
"""
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
is_in_amd_ci,
popen_launch_server,
)
@unittest.skipIf(is_in_amd_ci(), "AWQ Marlin is not supported on AMD GPUs")
class TestAWQMarlinFloat16(CustomTestCase):
"""
Verify that the model can be loaded with float16 dtype and awq_marlin quantization
"""
@classmethod
def setUpClass(cls):
cls.model = "QuantTrio/Qwen3-VL-30B-A3B-Instruct-AWQ"
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=["--dtype", "float16", "--quantization", "awq_marlin"],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_mmlu(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="mmlu",
num_examples=64,
num_threads=32,
)
metrics = run_eval(args)
self.assertGreater(metrics["score"], 0.85)
if __name__ == "__main__":
unittest.main()
test/manual/quant/test_deepseek_v32_fp4_4gpu.py
+162
View File
@@ -0,0 +1,162 @@
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.send_one import BenchArgs, send_one_prompt
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
CustomTestCase,
is_in_ci,
popen_launch_server,
write_github_step_summary,
)
register_cuda_ci(est_time=874, suite="stage-c-test-4-gpu-b200")
FULL_DEEPSEEK_V3_FP4_MODEL_PATH = "nvidia/DeepSeek-V3.2-NVFP4"
SERVER_LAUNCH_TIMEOUT = 1200
class TestDeepseekV32FP4DP(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = FULL_DEEPSEEK_V3_FP4_MODEL_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--tp",
"4",
"--dp",
"4",
"--enable-dp-attention",
"--moe-runner-backend",
"flashinfer_trtllm",
"--quantization",
"modelopt_fp4",
"--tool-call-parser",
"deepseekv32",
"--reasoning-parser",
"deepseek-v3",
"--model-loader-extra-config",
'{"enable_multithread_load": true,"num_threads": 64}',
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=SERVER_LAUNCH_TIMEOUT,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_a_gsm8k(
self,
): # Append an "a" to make this test run first (alphabetically) to warm up the server
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=500,
num_threads=500,
num_shots=20,
)
metrics = run_eval(args)
print(f"{metrics=}")
if is_in_ci():
write_github_step_summary(
f"### test_gsm8k (deepseek-v3-fp4)\n" f'{metrics["score"]=:.3f}\n'
)
self.assertGreater(metrics["score"], 0.93)
def test_bs_1_speed(self):
args = BenchArgs(port=int(self.base_url.split(":")[-1]), max_new_tokens=2048)
acc_length, speed = send_one_prompt(args)
print(f"{acc_length=:.2f} {speed=:.2f}")
if is_in_ci():
write_github_step_summary(
f"### test_bs_1_speed (deepseek-v32 mtp)\n"
f"{acc_length=:.2f}\n"
f"{speed=:.2f} token/s\n"
)
self.assertGreater(speed, 60)
class TestDeepseekV32FP4TP(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = FULL_DEEPSEEK_V3_FP4_MODEL_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--tp",
"4",
"--moe-runner-backend",
"flashinfer_trtllm",
"--quantization",
"modelopt_fp4",
"--tool-call-parser",
"deepseekv32",
"--reasoning-parser",
"deepseek-v3",
"--model-loader-extra-config",
'{"enable_multithread_load": true,"num_threads": 64}',
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=SERVER_LAUNCH_TIMEOUT,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_a_gsm8k(
self,
): # Append an "a" to make this test run first (alphabetically) to warm up the server
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=500,
num_threads=500,
num_shots=20,
)
metrics = run_eval(args)
print(f"{metrics=}")
if is_in_ci():
write_github_step_summary(
f"### test_gsm8k (deepseek-v3-fp4)\n" f'{metrics["score"]=:.3f}\n'
)
self.assertGreater(metrics["score"], 0.93)
def test_bs_1_speed(self):
args = BenchArgs(port=int(self.base_url.split(":")[-1]), max_new_tokens=2048)
acc_length, speed = send_one_prompt(args)
print(f"{acc_length=:.2f} {speed=:.2f}")
if is_in_ci():
write_github_step_summary(
f"### test_bs_1_speed (deepseek-v32 mtp)\n"
f"{acc_length=:.2f}\n"
f"{speed=:.2f} token/s\n"
)
self.assertGreater(speed, 90)
if __name__ == "__main__":
unittest.main()
test/manual/quant/test_eval_fp8_accuracy.py
+118
View File
@@ -0,0 +1,118 @@
import unittest
from types import SimpleNamespace
from sglang.srt.utils import is_hip, kill_process_tree
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_ACCURACY_TEST_FP8,
DEFAULT_MODEL_NAME_FOR_DYNAMIC_QUANT_ACCURACY_TEST_FP8,
DEFAULT_MODEL_NAME_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_cuda_ci(est_time=351, suite="stage-b-test-1-gpu-large")
register_amd_ci(est_time=600, suite="stage-b-test-1-gpu-small-amd")
class TestEvalFP8Accuracy(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = DEFAULT_MODEL_NAME_FOR_ACCURACY_TEST_FP8
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model, cls.base_url, timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_mmlu(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="mmlu",
num_examples=64,
num_threads=32,
temperature=0.1,
)
metrics = run_eval(args)
if is_hip():
# Another threshold for AMD because fp8 dtype is difference
self.assertGreaterEqual(metrics["score"], 0.60)
else:
self.assertGreaterEqual(metrics["score"], 0.60)
class TestEvalFP8DynamicQuantAccuracy(CustomTestCase):
def _run_test(self, model, other_args, expected_score):
base_url = DEFAULT_URL_FOR_TEST
other_args = other_args or []
process = popen_launch_server(
model,
base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
try:
args = SimpleNamespace(
base_url=base_url,
model=model,
eval_name="mmlu",
num_examples=64,
num_threads=32,
temperature=0.1,
)
metrics = run_eval(args)
self.assertGreaterEqual(metrics["score"], expected_score)
finally:
kill_process_tree(process.pid)
def test_mmlu_offline_only(self):
"""Test with offline quantization only."""
self._run_test(
model=DEFAULT_MODEL_NAME_FOR_DYNAMIC_QUANT_ACCURACY_TEST_FP8,
other_args=[],
expected_score=0.64,
)
def test_mmlu_offline_and_online_override(self):
"""Test with both offline and online quantization."""
self._run_test(
model=DEFAULT_MODEL_NAME_FOR_DYNAMIC_QUANT_ACCURACY_TEST_FP8,
other_args=["--quantization", "w8a8_fp8"],
# inference will use sgl kernel w/ online quant override
# we observed that the accuracy is higher then offline only
expected_score=0.64,
)
def test_mmlu_online_only(self):
"""Test with online quantization only."""
self._run_test(
model=DEFAULT_MODEL_NAME_FOR_TEST,
# inference will use sgl kernel w/ online quantization only
# we observed that the accuracy is higher then offline only
other_args=["--quantization", "w8a8_fp8"],
expected_score=0.64,
)
def test_mmlu_fp16_baseline(self):
"""Test with unquantized fp16 baseline."""
self._run_test(
model=DEFAULT_MODEL_NAME_FOR_TEST,
other_args=[],
expected_score=0.64,
)
if __name__ == "__main__":
unittest.main()
test/manual/quant/test_nvfp4_gemm_archived.py
+74
View File
@@ -0,0 +1,74 @@
"""Archived test classes split out of test/registered/quant/test_nvfp4_gemm.py.
Originally registered with `register_cuda_ci(...)`. Moved here as part of
the per-commit pruning effort to keep the code reachable manually.
Run with `python3 test/manual/quant/test_nvfp4_gemm_archived.py`.
"""
import unittest
from types import SimpleNamespace
from urllib.parse import urlparse
from sglang.srt.utils import get_device_sm, kill_process_tree
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
popen_launch_server,
try_cached_model,
)
MODEL_PATH = "nvidia/Llama-3.1-8B-Instruct-NVFP4"
class FP4GemmBase:
backend = None
@classmethod
def setUpClass(cls):
if cls.backend is None:
raise NotImplementedError("Subclass must set 'backend' attribute")
cls.model = try_cached_model(MODEL_PATH)
cls.base_url = DEFAULT_URL_FOR_TEST
other_args = [
"--trust-remote-code",
"--quantization",
"modelopt_fp4",
"--fp4-gemm-backend",
cls.backend,
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
parsed_url = urlparse(self.base_url)
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="gsm8k",
api="completion",
max_tokens=512,
num_examples=1319,
num_threads=200,
)
metrics = run_eval(args)
print(metrics)
self.assertGreater(metrics["score"], 0.64)
@unittest.skipIf(get_device_sm() < 100, "Test requires CUDA SM 100 or higher")
class TestFP4GemmAuto(FP4GemmBase, unittest.TestCase):
backend = "auto"
if __name__ == "__main__":
unittest.main()
test/manual/quant/test_quantization.py
+144
View File
@@ -0,0 +1,144 @@
import json
import unittest
import warnings
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_NIGHTLY_EVAL_QUANT_TP1,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
is_in_ci,
popen_launch_server,
write_github_step_summary,
write_results_to_json,
)
register_cuda_ci(est_time=460, suite="stage-b-test-1-gpu-large")
MODEL_SCORE_THRESHOLDS = {
# Baselines observed with gsm8k 5-shot concatenated format via chat API,
# which scores lower than reported benchmarks using proper CoT format.
# Thresholds set 5% below observed to catch catastrophic regressions.
"hugging-quants/Meta-Llama-3.1-8B-Instruct-AWQ-INT4": 0.74, # observed: 0.781
"hugging-quants/Meta-Llama-3.1-8B-Instruct-GPTQ-INT4": 0.74, # observed: 0.785
"hugging-quants/Mixtral-8x7B-Instruct-v0.1-AWQ-INT4": 0.36, # observed: 0.380
}
def parse_models(model_string):
return [model.strip() for model in model_string.split(",") if model.strip()]
def popen_launch_server_wrapper(base_url, model, is_fp8, is_tp2):
other_args = ["--log-level-http", "warning", "--trust-remote-code"]
if is_fp8:
if "Llama-3" in model or "gemma-2" in model:
other_args.extend(["--kv-cache-dtype", "fp8_e5m2"])
elif "Qwen2-72B-Instruct-FP8" in model:
other_args.extend(["--quantization", "fp8"])
elif "neuralmagic/Mixtral-8x7B-Instruct-v0.1-FP8" in model:
other_args.extend([])
else:
other_args.extend(["--quantization", "fp8", "--kv-cache-dtype", "fp8_e5m2"])
if is_tp2:
other_args.extend(["--tp", "2"])
if "DeepSeek" in model:
other_args.extend(["--mem-frac", "0.85"])
process = popen_launch_server(
model,
base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
return process
def check_model_scores(results):
failed_models = []
summary = " | model | score | threshold |\n"
summary += "| ----- | ----- | --------- |\n"
for model, score in results:
threshold = MODEL_SCORE_THRESHOLDS.get(model)
if threshold is None:
print(f"Warning: No threshold defined for model {model}")
continue
if score < threshold:
failed_models.append(
f"\nScore Check Failed: {model}\n"
f"Model {model} score ({score:.4f}) is below threshold ({threshold:.4f})"
)
line = f"| {model} | {score} | {threshold} |\n"
summary += line
print(summary)
if is_in_ci():
write_github_step_summary(
f"### TestNightlyGsm8KEval for awq, gptq, gguf\n{summary}"
)
if failed_models:
raise AssertionError("\n".join(failed_models))
class TestNightlyGsm8KEval(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.model_groups = [
(parse_models(DEFAULT_MODEL_NAME_FOR_NIGHTLY_EVAL_QUANT_TP1), False, False),
]
cls.base_url = DEFAULT_URL_FOR_TEST
def test_gsm8k_all_models(self):
warnings.filterwarnings(
"ignore", category=ResourceWarning, message="unclosed.*socket"
)
is_first = True
all_results = []
for model_group, is_fp8, is_tp2 in self.model_groups:
for model in model_group:
with self.subTest(model=model):
process = popen_launch_server_wrapper(
self.base_url, model, is_fp8, is_tp2
)
args = SimpleNamespace(
base_url=self.base_url,
model=model,
eval_name="gsm8k",
num_examples=None,
num_threads=1024,
)
metrics = run_eval(args)
print(
f"{'=' * 42}\n{model} - metrics={metrics} score={metrics['score']}\n{'=' * 42}\n"
)
write_results_to_json(model, metrics, "w" if is_first else "a")
is_first = False
all_results.append((model, metrics["score"]))
kill_process_tree(process.pid)
try:
with open("results.json", "r") as f:
print("\nFinal Results from results.json:")
print(json.dumps(json.load(f), indent=2))
except Exception as e:
print(f"Error reading results.json: {e}")
# Check all scores after collecting all results
check_model_scores(all_results)
if __name__ == "__main__":
unittest.main()
test/manual/scheduler/test_no_chunked_prefill.py
+34
View File
@@ -0,0 +1,34 @@
import unittest
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_TEST,
CustomTestCase,
run_bench_serving,
run_mmlu_test,
)
register_cuda_ci(est_time=131, suite="stage-b-test-1-gpu-large")
register_amd_ci(est_time=108, suite="stage-b-test-1-gpu-small-amd")
class TestNoChunkedPrefill(CustomTestCase):
def test_no_chunked_prefill(self):
run_mmlu_test(
disable_radix_cache=False, enable_mixed_chunk=False, chunked_prefill_size=-1
)
def test_no_chunked_prefill_without_radix_cache(self):
res = run_bench_serving(
model=DEFAULT_MODEL_NAME_FOR_TEST,
num_prompts=10,
request_rate=float("inf"),
other_server_args=["--disable-radix-cache", "--chunked-prefill-size", "-1"],
)
assert res["completed"] == 10
if __name__ == "__main__":
unittest.main()
test/manual/scheduler/test_no_overlap_scheduler.py
+39
View File
@@ -0,0 +1,39 @@
"""
Usage:
python3 -m unittest test_overlap_schedule.TestOverlapSchedule.test_radix_attention_chunked_prefill
python3 test_overlap_schedule.py
"""
import unittest
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.test_utils import CustomTestCase, run_mmlu_test
register_cuda_ci(est_time=267, suite="stage-b-test-1-gpu-large")
register_amd_ci(est_time=275, suite="stage-b-test-1-gpu-small-amd")
class TestOverlapSchedule(CustomTestCase):
def test_no_radix_attention_chunked_prefill(self):
run_mmlu_test(
disable_radix_cache=True, chunked_prefill_size=32, disable_overlap=True
)
def test_no_radix_attention_no_chunked_prefill(self):
run_mmlu_test(
disable_radix_cache=True, chunked_prefill_size=-1, disable_overlap=True
)
def test_radix_attention_chunked_prefill(self):
run_mmlu_test(
disable_radix_cache=False, chunked_prefill_size=32, disable_overlap=True
)
def test_radix_attention_no_chunked_prefill(self):
run_mmlu_test(
disable_radix_cache=False, chunked_prefill_size=-1, disable_overlap=True
)
if __name__ == "__main__":
unittest.main()
test/manual/spec/eagle/test_eagle3_basic.py
+67
View File
@@ -0,0 +1,67 @@
import unittest
from types import SimpleNamespace
import requests
from sglang.srt.utils import is_hip
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
from sglang.test.run_eval import run_eval
from sglang.test.server_fixtures.eagle_fixture import EagleServerBase
from sglang.test.test_utils import (
DEFAULT_DRAFT_MODEL_EAGLE3,
DEFAULT_TARGET_MODEL_EAGLE3,
)
register_cuda_ci(est_time=88, suite="stage-b-test-1-gpu-small")
register_amd_ci(est_time=50, suite="stage-b-test-1-gpu-small")
_is_hip = is_hip()
class TestEagle3Basic(EagleServerBase):
target_model = DEFAULT_TARGET_MODEL_EAGLE3
draft_model = DEFAULT_DRAFT_MODEL_EAGLE3
spec_algo = "EAGLE3"
spec_steps = 2
spec_topk = 1
spec_tokens = 3
extra_args = (
[
"--dtype=float16",
"--chunked-prefill-size",
1024,
"--attention-backend",
"aiter",
]
if _is_hip
else ["--dtype=float16", "--chunked-prefill-size", 1024]
)
def test_mmlu(self):
"""Override to add EAGLE-specific assertions"""
args = SimpleNamespace(
base_url=self.base_url,
model=self.target_model,
eval_name="mmlu",
num_examples=64,
num_threads=32,
)
metrics = run_eval(args)
self.assertGreaterEqual(metrics["score"], 0.72)
server_info = requests.get(self.base_url + "/server_info").json()
avg_spec_accept_length = server_info["internal_states"][0][
"avg_spec_accept_length"
]
print(f"{avg_spec_accept_length=}")
if _is_hip:
self.assertGreater(avg_spec_accept_length, 2.24)
else:
self.assertGreater(avg_spec_accept_length, 2.26)
if __name__ == "__main__":
unittest.main()