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

[XPU] Add registry mechanism for XPU CI tests (#25405)

Browse Source 
Co-authored-by: Ma Mingfei <mingfei.ma@intel.com>
This commit is contained in:
vikram singh shekhawat
2026-05-27 08:56:59 +08:00
committed by GitHub
co-authored by Ma Mingfei
parent 87c3171aaa
commit 737c6cd6d1
9 changed files with 156 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/srt/run_suite.py
+4 -10
View File
@@ -87,16 +87,10 @@ suite_xeon = {
],
}
# Add Intel XPU tests
# NOTE: please sort the test cases alphabetically by the test file name
suite_xpu = {
"per-commit-xpu": [
TestFile("xpu/test_deepseek_ocr.py", 360),
TestFile("xpu/test_deepseek_ocr_triton.py", 360),
# TestFile("xpu/test_internvl.py"),
TestFile("xpu/test_intel_xpu_backend.py"),
],
}
# XPU tests migrated to test/registered/xpu/ using register_xpu_ci().
# The legacy per-commit-xpu suite is replaced by stage-a-test-1-gpu-xpu
# and stage-b-test-1-gpu-xpu in test/run_suite.py (registry-based).
suite_xpu = {}
suites.update(suite_amd)
suites.update(suite_arm64)
test/srt/xpu/test_deepseek_ocr.py
-110
View File
@@ -1,110 +0,0 @@
"""
python3 -m unittest test_deepseek_ocr.py
"""
import json
import os
import unittest
from pathlib import Path
import requests
from sglang.srt.utils import kill_process_tree
from sglang.srt.utils.hf_transformers import get_tokenizer
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
class TestDeepSeekOCR(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = "deepseek-ai/DeepSeek-OCR"
cls.tokenizer = get_tokenizer(cls.model)
cls.base_url = DEFAULT_URL_FOR_TEST
cls.image_path = str(
(Path(__file__).resolve().parents[3] / "examples/assets/example_image.png")
)
if not os.path.exists(cls.image_path):
raise FileNotFoundError(f"Image not found: {cls.image_path}")
cls.common_args = [
"--device",
"xpu",
"--attention-backend",
"intel_xpu",
]
os.environ["SGLANG_USE_SGL_XPU"] = "1"
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
*cls.common_args,
],
)
@classmethod
def tearDownClass(cls):
"""Fixture that is run once after all tests in the class."""
if hasattr(cls, "process") and cls.process:
cls.process.terminate()
try:
cls.process.wait(timeout=30)
except Exception:
# Force kill if it didn't exit cleanly in time
kill_process_tree(cls.process.pid)
def get_request_json(self, max_new_tokens=32, n=1):
response = requests.post(
self.base_url + "/generate",
json={
"text": "<image>\n<|grounding|>Convert the document to pure text.",
"image_data": self.image_path,
"sampling_params": {
"temperature": 0 if n == 1 else 0.5,
"max_new_tokens": max_new_tokens,
},
},
)
return response.json()
def run_decode(
self,
max_new_tokens=128,
n=1,
):
ret = self.get_request_json(max_new_tokens=max_new_tokens, n=n)
print(json.dumps(ret, indent=2))
def assert_one_item(item):
if item["meta_info"]["finish_reason"]["type"] == "stop":
self.assertEqual(
item["meta_info"]["finish_reason"]["matched"],
self.tokenizer.eos_token_id,
)
elif item["meta_info"]["finish_reason"]["type"] == "length":
self.assertEqual(
len(item["output_ids"]), item["meta_info"]["completion_tokens"]
)
self.assertEqual(len(item["output_ids"]), max_new_tokens)
# Determine whether to assert a single item or multiple items based on n
if n == 1:
assert_one_item(ret)
else:
self.assertEqual(len(ret), n)
for i in range(n):
assert_one_item(ret[i])
print("=" * 100)
def test_moe(self):
self.run_decode()
if __name__ == "__main__":
unittest.main()
test/srt/xpu/test_deepseek_ocr_triton.py
-53
View File
@@ -1,53 +0,0 @@
"""
python3 -m unittest test_deepseek_ocr_triton.py
"""
import os
import unittest
from pathlib import Path
from test_deepseek_ocr import TestDeepSeekOCR
from sglang.srt.utils.hf_transformers import get_tokenizer
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
popen_launch_server,
)
# TODO: Temporarily disable this test and re-enable it after Triton-XPU is upgraded.
@unittest.skip("Temporarily disabled until Triton-XPU upgrade")
class TestDeepSeekOCRTriton(TestDeepSeekOCR):
@classmethod
def setUpClass(cls):
cls.model = "deepseek-ai/DeepSeek-OCR"
cls.tokenizer = get_tokenizer(cls.model)
cls.base_url = DEFAULT_URL_FOR_TEST
cls.image_path = str(
(Path(__file__).resolve().parents[3] / "examples/assets/example_image.png")
)
if not os.path.exists(cls.image_path):
raise FileNotFoundError(f"Image not found: {cls.image_path}")
cls.common_args = [
"--device",
"xpu",
"--attention-backend",
"intel_xpu",
]
os.environ["SGLANG_USE_SGL_XPU"] = "0"
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
*cls.common_args,
],
)
# Prevent pytest from collecting the imported base test class here.
del TestDeepSeekOCR
if __name__ == "__main__":
unittest.main()
test/srt/xpu/test_intel_xpu_backend.py
-83
View File
@@ -1,83 +0,0 @@
"""
Usage:
python3 -m unittest test_intel_xpu_backend.TestIntelXPUBackend.test_latency_qwen_model
"""
import unittest
from functools import wraps
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_TEST_FP8_WITH_MOE,
DEFAULT_SMALL_MODEL_NAME_FOR_TEST_BASE,
DEFAULT_SMALL_MODEL_NAME_FOR_TEST_QWEN,
CustomTestCase,
is_in_ci,
run_bench_one_batch,
)
def intel_xpu_benchmark(
extra_args=None, min_throughput=None, mem_fraction_static="0.4"
):
def decorator(test_func):
@wraps(test_func)
def wrapper(self):
common_args = [
"--disable-radix-cache",
"--trust-remote-code",
"--mem-fraction-static",
str(mem_fraction_static),
"--batch-size",
"1",
"--device",
"xpu",
]
ci_args = ["--input", "64", "--output", "4"] if is_in_ci() else []
full_args = common_args + ci_args + (extra_args or [])
model = test_func(self)
prefill_latency, decode_throughput, decode_latency = run_bench_one_batch(
model, full_args
)
print(f"{model=}")
print(f"{prefill_latency=}")
print(f"{decode_throughput=}")
print(f"{decode_latency=}")
if is_in_ci() and min_throughput is not None:
self.assertGreater(decode_throughput, min_throughput)
return wrapper
return decorator
class TestIntelXPUBackend(CustomTestCase):
@intel_xpu_benchmark(min_throughput=10, mem_fraction_static="0.3")
def test_latency_qwen_model(self):
return DEFAULT_SMALL_MODEL_NAME_FOR_TEST_QWEN
@intel_xpu_benchmark(
["--attention-backend", "intel_xpu", "--page-size", "128"],
mem_fraction_static="0.5",
)
def test_attention_backend(self):
return DEFAULT_SMALL_MODEL_NAME_FOR_TEST_BASE
@intel_xpu_benchmark(
[
"--json-model-override-args",
'{"num_hidden_layers": 4}',
"--decode-attention-backend",
"intel_xpu",
],
min_throughput=32,
)
def test_mla_decode_attention_backend(self):
return DEFAULT_MODEL_NAME_FOR_TEST_FP8_WITH_MOE
if __name__ == "__main__":
unittest.main()