Files
sglang/test/srt/test_pp_single_node.py
2025-05-11 10:55:06 -07:00

96 lines
2.5 KiB
Python

"""
Usage:
python3 -m unittest test_pp_single_node.TestPPAccuracy.test_gsm8k
python3 -m unittest test_pp_single_node.TestFixedBugs.test_chunked_prefill_with_small_bs
"""
import os
import time
import unittest
from types import SimpleNamespace
from sglang.bench_one_batch_server import BenchArgs as OneBatchBenchArgs
from sglang.srt.server_args import ServerArgs
from sglang.srt.utils import kill_process_tree
from sglang.test.few_shot_gsm8k import run_eval
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
popen_launch_server,
run_bench_one_batch_server,
)
class TestPPAccuracy(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.base_url = "http://127.0.0.1:23333"
cls.process = popen_launch_server(
DEFAULT_MODEL_NAME_FOR_TEST,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--tp-size",
2,
"--pp-size",
4,
"--chunked-prefill-size",
256,
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
num_shots=5,
data_path=None,
num_questions=200,
max_new_tokens=512,
parallel=128,
host="http://127.0.0.1",
port=int(self.base_url.split(":")[-1]),
)
metrics = run_eval(args)
print(f"{metrics=}")
self.assertGreater(metrics["accuracy"], 0.74)
# Wait a little bit so that the memory check happens.
time.sleep(5)
class TestFixedBugs(unittest.TestCase):
def test_chunked_prefill_with_small_bs(self):
model = DEFAULT_MODEL_NAME_FOR_TEST
server_args = ServerArgs(model_path=model)
bench_args = OneBatchBenchArgs(
batch_size=(1,),
input_len=(1,),
output_len=(1,),
base_url=DEFAULT_URL_FOR_TEST,
)
other_server_args = [
"--tp-size",
2,
"--pp-size",
2,
"--chunked-prefill",
256,
"--max-running-requests",
2,
]
run_bench_one_batch_server(
model,
DEFAULT_URL_FOR_TEST,
server_args,
bench_args,
other_server_args,
)
if __name__ == "__main__":
unittest.main()