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

[NPU] Add PR test cases (#32392)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: sglang-npu-bot <sglangnpu@163.com>
This commit is contained in:
Sugar920
2026-08-02 23:00:02 +08:00
committed by GitHub
co-authored by Claude sglang-npu-bot
parent 27b15349e5
commit f8e62a9224
24 changed files with 2267 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/pr-test-npu.yml
+169 -11
View File
@@ -236,10 +236,10 @@ jobs:
cd test
python3 run_suite.py --hw npu --suite stage-b-test-2-npu-a2 --auto-partition-id ${{ matrix.part }} --auto-partition-size 2
stage-b-test-4-npu-a3:
stage-b-test-1-npu-a3:
needs: [check-changes, pr-gate, set-image-config]
if: needs.check-changes.outputs.main_package == 'true'
runs-on: linux-aarch64-a3-4
runs-on: linux-aarch64-a3-2
container:
image: ${{ needs.set-image-config.outputs.CANN_image_a3 }}
steps:
@@ -283,13 +283,13 @@ jobs:
STREAMS_PER_DEVICE: 32
run: |
cd test
python3 run_suite.py --hw npu --suite stage-b-test-4-npu-a3 --timeout-per-file 3600
python3 run_suite.py --hw npu --suite stage-b-test-1-npu-a3 --timeout-per-file 3600
stage-b-test-16-npu-a3:
stage-b-test-2-npu-a3:
needs: [check-changes, pr-gate, set-image-config]
if: needs.check-changes.outputs.main_package == 'true'
runs-on: linux-aarch64-a3-16
runs-on: linux-aarch64-a3-2
container:
image: ${{ needs.set-image-config.outputs.CANN_image_a3 }}
steps:
@@ -332,6 +332,161 @@ jobs:
PYTORCH_NPU_ALLOC_CONF: "expandable_segments:True"
STREAMS_PER_DEVICE: 32
run: |
cd test
python3 run_suite.py --hw npu --suite stage-b-test-2-npu-a3 --timeout-per-file 3600
stage-b-test-4-npu-a3:
needs: [check-changes, pr-gate, set-image-config]
if: needs.check-changes.outputs.main_package == 'true'
runs-on: linux-aarch64-a3-4
strategy:
fail-fast: false
matrix:
part: [ 0, 1 ]
container:
image: ${{ needs.set-image-config.outputs.CANN_image_a3 }}
steps:
- name: Checkout code
uses: actions/checkout@v4
with:
ref: ${{ inputs.ref || github.ref }}
- name: Mark repository safe
run: |
git config --system --add safe.directory ${GITHUB_WORKSPACE}
- name: Install dependencies
env:
TORCH_CACHE_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local/whl/cpu"
PYPI_CACHE_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple"
UV_INDEX_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple"
GITHUB_PROXY_URL: "https://gh-proxy.test.osinfra.cn/"
RUSTUP_CACHE_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local:8082"
run: |
# speed up by using infra cache services
CACHING_URL="cache-service.nginx-pypi-cache.svc.cluster.local"
sed -Ei "s@(ports|archive).ubuntu.com@${CACHING_URL}:8081@g" /etc/apt/sources.list
pip config set global.index-url http://${CACHING_URL}/pypi/simple
pip config set global.trusted-host "${CACHING_URL}"
bash scripts/ci/npu/npu_ci_install_dependency.sh a3
# copy required file from our daily cache
cp ~/.cache/modelscope/hub/datasets/otavia/ShareGPT_Vicuna_unfiltered/ShareGPT_V3_unfiltered_cleaned_split.json /tmp
# copy gsm8k dataset
cp ~/.cache/modelscope/hub/datasets/tmp/test.jsonl /tmp
- name: Run test
timeout-minutes: 120
env:
SGLANG_USE_MODELSCOPE: true
SGLANG_IS_IN_CI: true
HF_ENDPOINT: https://hf-mirror.com
TORCH_EXTENSIONS_DIR: /tmp/torch_extensions
PYTORCH_NPU_ALLOC_CONF: "expandable_segments:True"
STREAMS_PER_DEVICE: 32
run: |
pip install sglang_router
cd test
python3 run_suite.py --hw npu --suite stage-b-test-4-npu-a3 --auto-partition-id ${{ matrix.part }} --auto-partition-size 2 --timeout-per-file 3600
stage-b-test-8-npu-a3:
needs: [check-changes, pr-gate, set-image-config]
if: needs.check-changes.outputs.main_package == 'true'
runs-on: linux-aarch64-a3-8
container:
image: ${{ needs.set-image-config.outputs.CANN_image_a3 }}
steps:
- name: Checkout code
uses: actions/checkout@v4
with:
ref: ${{ inputs.ref || github.ref }}
- name: Mark repository safe
run: |
git config --system --add safe.directory ${GITHUB_WORKSPACE}
- name: Install dependencies
env:
TORCH_CACHE_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local/whl/cpu"
PYPI_CACHE_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple"
UV_INDEX_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple"
GITHUB_PROXY_URL: "https://gh-proxy.test.osinfra.cn/"
RUSTUP_CACHE_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local:8082"
run: |
# speed up by using infra cache services
CACHING_URL="cache-service.nginx-pypi-cache.svc.cluster.local"
sed -Ei "s@(ports|archive).ubuntu.com@${CACHING_URL}:8081@g" /etc/apt/sources.list
pip config set global.index-url http://${CACHING_URL}/pypi/simple
pip config set global.trusted-host "${CACHING_URL}"
bash scripts/ci/npu/npu_ci_install_dependency.sh a3
# copy required file from our daily cache
cp ~/.cache/modelscope/hub/datasets/otavia/ShareGPT_Vicuna_unfiltered/ShareGPT_V3_unfiltered_cleaned_split.json /tmp
# copy gsm8k dataset
cp ~/.cache/modelscope/hub/datasets/tmp/test.jsonl /tmp
- name: Run test
timeout-minutes: 60
env:
SGLANG_USE_MODELSCOPE: true
SGLANG_IS_IN_CI: true
HF_ENDPOINT: https://hf-mirror.com
TORCH_EXTENSIONS_DIR: /tmp/torch_extensions
PYTORCH_NPU_ALLOC_CONF: "expandable_segments:True"
STREAMS_PER_DEVICE: 32
run: |
cd test
python3 run_suite.py --hw npu --suite stage-b-test-8-npu-a3 --timeout-per-file 3600
stage-b-test-16-npu-a3:
needs: [check-changes, pr-gate, set-image-config]
if: needs.check-changes.outputs.main_package == 'true'
runs-on: linux-aarch64-a3-16
container:
image: ${{ needs.set-image-config.outputs.CANN_image_a3 }}
steps:
- name: Checkout code
uses: actions/checkout@v4
with:
ref: ${{ inputs.ref || github.ref }}
- name: Mark repository safe
run: |
git config --system --add safe.directory ${GITHUB_WORKSPACE}
- name: Install dependencies
env:
TORCH_CACHE_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local/whl/cpu"
PYPI_CACHE_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple"
UV_INDEX_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local/pypi/simple"
GITHUB_PROXY_URL: "https://gh-proxy.test.osinfra.cn/"
RUSTUP_CACHE_URL: "http://cache-service.nginx-pypi-cache.svc.cluster.local:8082"
run: |
# speed up by using infra cache services
CACHING_URL="cache-service.nginx-pypi-cache.svc.cluster.local"
sed -Ei "s@(ports|archive).ubuntu.com@${CACHING_URL}:8081@g" /etc/apt/sources.list
pip config set global.index-url http://${CACHING_URL}/pypi/simple
pip config set global.trusted-host "${CACHING_URL}"
bash scripts/ci/npu/npu_ci_install_dependency.sh a3
# copy required file from our daily cache
cp ~/.cache/modelscope/hub/datasets/otavia/ShareGPT_Vicuna_unfiltered/ShareGPT_V3_unfiltered_cleaned_split.json /tmp
# copy gsm8k dataset
cp ~/.cache/modelscope/hub/datasets/tmp/test.jsonl /tmp
- name: Run test
timeout-minutes: 120
env:
SGLANG_USE_MODELSCOPE: true
SGLANG_IS_IN_CI: true
HF_ENDPOINT: https://hf-mirror.com
TORCH_EXTENSIONS_DIR: /tmp/torch_extensions
PYTORCH_NPU_ALLOC_CONF: "expandable_segments:True"
STREAMS_PER_DEVICE: 32
run: |
apt-get install -y libssl-dev
pip install sglang_router
cd test
python3 run_suite.py --hw npu --suite stage-b-test-16-npu-a3 --timeout-per-file 3600
@@ -537,27 +692,27 @@ jobs:
# NPU accuracy tests
- name: glm4_7_flash_1p_gsm8k
runner: linux-aarch64-a3-2
test_case: test/registered/ascend/accuracy/glm4_7_flash/test_npu_glm4_7_flash_1p_gsm8k.py
test_case: test/registered/npu/accuracy/glm4_7_flash/test_npu_glm4_7_flash_1p_gsm8k.py
test_type: 'accuracy'
- name: qwen3_vl_30b_a3b_bf16_2p_gsm8k
runner: linux-aarch64-a3-4
test_case: test/registered/ascend/accuracy/qwen3_vl_30b_a3b/test_npu_qwen3_vl_30b_a3b_bf16_2p_gsm8k.py
test_case: test/registered/npu/accuracy/qwen3_vl_30b_a3b/test_npu_qwen3_vl_30b_a3b_bf16_2p_gsm8k.py
test_type: 'accuracy'
- name: glm5_top64_pruned_bf16_8p_gsm8k
runner: linux-aarch64-a3-16
test_case: test/registered/ascend/accuracy/glm5_top64_pruned/test_npu_glm5_top64_pruned_bf16_8p_gsm8k.py
test_case: test/registered/npu/accuracy/glm5_top64_pruned/test_npu_glm5_top64_pruned_bf16_8p_gsm8k.py
test_type: 'accuracy'
- name: moonshotai_moonlight_16b_a3b_bf16_1p_gsm8k
runner: linux-aarch64-a3-2
test_case: test/registered/ascend/accuracy/moonshotai_moonlight_16b_a3b/test_npu_moonlight_16b_a3b_bf16_1p_gsm8k.py
test_case: test/registered/npu/accuracy/moonshotai_moonlight_16b_a3b/test_npu_moonlight_16b_a3b_bf16_1p_gsm8k.py
test_type: 'accuracy'
- name: qwen3_5_9b_bf16_1p_gsm8k
runner: linux-aarch64-a3-2
test_case: test/registered/ascend/accuracy/qwen3_5_9b/test_npu_qwen3_5_9b_bf16_1p_gsm8k.py
test_case: test/registered/npu/accuracy/qwen3_5_9b/test_npu_qwen3_5_9b_bf16_1p_gsm8k.py
test_type: 'accuracy'
- name: qwen3_vl_8b_bf16_2p_gsm8k
runner: linux-aarch64-a3-4
test_case: test/registered/ascend/accuracy/qwen3_vl_8b/test_npu_qwen3_vl_8b_bf16_2p_gsm8k.py
test_case: test/registered/npu/accuracy/qwen3_vl_8b/test_npu_qwen3_vl_8b_bf16_2p_gsm8k.py
test_type: 'accuracy'
uses: ./.github/workflows/nightly-test-npu-e2e-single-node.yml
with:
@@ -578,8 +733,11 @@ jobs:
stage-a-unit-test-npu,
stage-b-test-1-npu-a2,
stage-b-test-1-npu-a3,
stage-b-test-2-npu-a2,
stage-b-test-2-npu-a3,
stage-b-test-4-npu-a3,
stage-b-test-8-npu-a3,
stage-b-test-16-npu-a3,
multimodal-gen-test-1-npu-a3,
python/sglang/test/ascend/npu_eval_accuracy_kit.py
+49
View File
@@ -0,0 +1,49 @@
"""NPU accuracy test mixin.
Wraps :class:`~sglang.test.kits.eval_accuracy_kit.GSM8KMixin` with:
* 1% accuracy tolerance (threshold * 0.99)
* Automatic retry (up to 3 attempts)
Delegates to ``super().test_gsm8k()`` so community-side changes to eval
setup, backend selection, or parameters are inherited automatically.
Usage::
from sglang.test.ascend.npu_eval_accuracy_kit import NPUGSM8KMixin
class TestFoo(CustomTestCase, NPUGSM8KMixin):
gsm8k_accuracy_thres = 0.6
"""
from sglang.test.kits.eval_accuracy_kit import GSM8KMixin
_NPU_ACCURACY_TOLERANCE = 0.99
_NPU_MAX_ACCURACY_ATTEMPTS = 3
class NPUGSM8KMixin(GSM8KMixin):
"""NPU GSM8K accuracy mixin — 1% tolerance + up to 3 retries."""
def test_gsm8k(self):
threshold = self.gsm8k_score_threshold
if threshold != threshold: # NaN → legacy alias
threshold = self.gsm8k_accuracy_thres
relaxed = threshold * _NPU_ACCURACY_TOLERANCE
# Patch both so the parent picks up the relaxed threshold.
self.gsm8k_score_threshold = relaxed
self.gsm8k_accuracy_thres = relaxed
for attempt in range(_NPU_MAX_ACCURACY_ATTEMPTS):
try:
super().test_gsm8k()
return
except AssertionError:
if attempt == _NPU_MAX_ACCURACY_ATTEMPTS - 1:
raise
print(
f"[{type(self).__name__}] GSM8K attempt "
f"{attempt + 1}/{_NPU_MAX_ACCURACY_ATTEMPTS} failed, "
f"retrying..."
)
python/sglang/test/ascend/test_ascend_utils.py
+1 -1
View File
@@ -212,7 +212,7 @@ QWEN3_NEXT_80B_A3B_INSTRUCT_WEIGHTS_PATH = os.path.join(
MODEL_WEIGHTS_DIR, "Qwen/Qwen3-Next-80B-A3B-Instruct"
)
QWEN3_32B_EAGLE3_WEIGHTS_PATH = os.path.join(
MODEL_WEIGHTS_DIR, "Zjcxy-SmartAI/Qwen3-32B-Eagle3"
MODEL_WEIGHTS_DIR, "Zjcxy-SmartAI/Eagle3-Qwen3-32B-zh"
)
QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH = os.path.join(
MODEL_WEIGHTS_DIR, "aleoyang/Qwen3-32B-w8a8-MindIE"
test/registered/ascend/accuracy/glm4_7_flash/test_npu_glm4_7_flash_1p_gsm8k.py → test/registered/npu/accuracy/glm4_7_flash/test_npu_glm4_7_flash_1p_gsm8k.py
View File
test/registered/ascend/accuracy/glm5_top64_pruned/test_npu_glm5_top64_pruned_bf16_8p_gsm8k.py → test/registered/npu/accuracy/glm5_top64_pruned/test_npu_glm5_top64_pruned_bf16_8p_gsm8k.py
View File
test/registered/ascend/accuracy/moonshotai_moonlight_16b_a3b/test_npu_moonlight_16b_a3b_bf16_1p_gsm8k.py → test/registered/npu/accuracy/moonshotai_moonlight_16b_a3b/test_npu_moonlight_16b_a3b_bf16_1p_gsm8k.py
View File
test/registered/ascend/accuracy/qwen3_5_9b/test_npu_qwen3_5_9b_bf16_1p_gsm8k.py → test/registered/npu/accuracy/qwen3_5_9b/test_npu_qwen3_5_9b_bf16_1p_gsm8k.py
View File
test/registered/ascend/accuracy/qwen3_vl_30b_a3b/test_npu_qwen3_vl_30b_a3b_bf16_2p_gsm8k.py → test/registered/npu/accuracy/qwen3_vl_30b_a3b/test_npu_qwen3_vl_30b_a3b_bf16_2p_gsm8k.py
View File
test/registered/ascend/accuracy/qwen3_vl_8b/test_npu_qwen3_vl_8b_bf16_2p_gsm8k.py → test/registered/npu/accuracy/qwen3_vl_8b/test_npu_qwen3_vl_8b_bf16_2p_gsm8k.py
View File
test/registered/npu/basic_function/dp_attn/test_npu_dp_attention.py
+199
View File
@@ -0,0 +1,199 @@
import unittest
import requests
from sglang.lang.chat_template import get_chat_template_by_model_path
from sglang.srt.environ import envs
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.npu_eval_accuracy_kit import NPUGSM8KMixin
from sglang.test.ascend.test_ascend_utils import (
DEEPSEEK_V2_LITE_W8A8_WEIGHTS_PATH,
KIMI_VL_A3B_INSTRUCT_WEIGHTS_PATH,
QWEN3_32B_WEIGHTS_PATH,
)
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.kits.ebnf_constrained_kit import EBNFConstrainedMixin
from sglang.test.kits.json_constrained_kit import JSONConstrainedMixin
from sglang.test.kits.radix_cache_server_kit import run_radix_attention_test
from sglang.test.kits.regex_constrained_kit import RegexConstrainedMixin
from sglang.test.test_utils import (
DEFAULT_IMAGE_URL,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_npu_ci(est_time=400, suite="stage-b-test-4-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-4-npu-a3", nightly=True)
class TestDPAttentionDP2TP2(
CustomTestCase,
NPUGSM8KMixin,
JSONConstrainedMixin,
EBNFConstrainedMixin,
RegexConstrainedMixin,
):
gsm8k_accuracy_thres = 0.6
@classmethod
def setUpClass(cls):
cls.model = QWEN3_32B_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls._env_override = envs.SGLANG_DISABLE_CONSECUTIVE_PREFILL_OVERLAP.override(
True
)
cls._env_override.__enter__()
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",
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--enable-torch-compile",
"--torch-compile-max-bs",
"2",
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
cls._env_override.__exit__(None, None, None)
class TestDPAttentionMixedChunk(
CustomTestCase,
NPUGSM8KMixin,
):
gsm8k_accuracy_thres = 0.35
@classmethod
def setUpClass(cls):
cls.model = DEEPSEEK_V2_LITE_W8A8_WEIGHTS_PATH
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",
"2",
"--enable-dp-attention",
"--dp",
"2",
"--enable-mixed-chunk",
"--chunked-prefill-size",
"256",
"--attention-backend",
"ascend",
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
class TestDPRetract(
CustomTestCase,
JSONConstrainedMixin,
EBNFConstrainedMixin,
RegexConstrainedMixin,
):
@classmethod
def setUpClass(cls):
cls.model = DEEPSEEK_V2_LITE_W8A8_WEIGHTS_PATH
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",
"2",
"--enable-dp-attention",
"--dp",
"2",
"--max-total-tokens",
"4500",
"--max-running-requests",
"128",
"--chunked-prefill-size",
"256",
"--attention-backend",
"ascend",
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_radix_attention(self):
with envs.SGLANG_TEST_RETRACT.override(True):
run_radix_attention_test(self.base_url)
self.assertIsNone(self.process.poll())
class TestDPAttentionDP2TP2VLM(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = KIMI_VL_A3B_INSTRUCT_WEIGHTS_PATH
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",
"2",
"--enable-dp-attention",
"--dp",
"2",
"--attention-backend",
"ascend",
"--disable-cuda-graph",
],
)
@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/registered/npu/basic_function/parallel_strategy/data_parallelism/test_npu_load_balance_method.py
+124
View File
@@ -0,0 +1,124 @@
import random
import unittest
from types import SimpleNamespace
from urllib.parse import urlparse
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import DEEPSEEK_R1_0528_W8A8_WEIGHTS_PATH
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.few_shot_gsm8k import run_eval as run_eval_few_shot_gsm8k
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
is_in_ci,
popen_launch_server,
)
register_npu_ci(est_time=700, suite="stage-b-test-16-npu-a3", nightly=False)
register_npu_ci(est_time=700, suite="nightly-16-npu-a3", nightly=True)
class TestDPAttentionRoundBinLoadBalance(CustomTestCase):
"""Testcase：Verify that the model accuracy did not decrease when --load-balance-method is set to round_robin, auto,
total_requests or total_tokens in PD mixed scenario
[Test Category] Parameter
[Test Target] --load-balance-method
"""
mode = "round_robin"
@classmethod
def setUpClass(cls):
cls.model_path = DEEPSEEK_R1_0528_W8A8_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.url = urlparse(cls.base_url)
other_args = [
"--trust-remote-code",
"--tp",
"16",
"--enable-dp-attention",
"--dp",
"2",
"--enable-torch-compile",
"--torch-compile-max-bs",
"2",
"--load-balance-method",
cls.mode,
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--quantization",
"modelslim",
"--mem-fraction-static",
"0.75",
]
cls.process = popen_launch_server(
cls.model_path,
cls.base_url,
timeout=3 * DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
env={
"TRANSFORMERS_VERBOSITY": "error",
},
)
@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=f"http://{self.url.hostname}",
port=int(self.url.port),
)
metrics = run_eval_few_shot_gsm8k(args)
self.assertGreaterEqual(
metrics["accuracy"],
0.95,
)
def test_server_info(self):
response = requests.get(f"{self.base_url}/get_server_info")
self.assertEqual(response.status_code, 200)
self.assertIn(self.mode, response.text)
class _TestDPAttentionAutoLoadBalance(TestDPAttentionRoundBinLoadBalance):
mode = "auto"
class _TestDPAttentionTotalRequestsLoadBalance(TestDPAttentionRoundBinLoadBalance):
mode = "total_requests"
class _TestDPAttentionTotalTokensLoadBalance(TestDPAttentionRoundBinLoadBalance):
mode = "total_tokens"
if __name__ == "__main__":
# To reduce the CI execution time.
if is_in_ci():
loader = unittest.TestLoader()
suite = unittest.TestSuite()
RUN_FLAG = [
TestDPAttentionRoundBinLoadBalance,
_TestDPAttentionAutoLoadBalance,
_TestDPAttentionTotalRequestsLoadBalance,
_TestDPAttentionTotalTokensLoadBalance,
]
suite.addTests(loader.loadTestsFromTestCase(random.choice(RUN_FLAG)))
runner = unittest.TextTestRunner()
runner.run(suite)
else:
unittest.main()
test/registered/npu/basic_function/parallel_strategy/data_parallelism/test_npu_load_balance_method_pd_disaggregation.py
+215
View File
@@ -0,0 +1,215 @@
import itertools
import json
import os
import random
import unittest
from time import sleep
from types import SimpleNamespace
from typing import List, Type
from urllib.parse import urlparse
import requests
from sglang.test.ascend.disaggregation_utils import TestDisaggregationBase
from sglang.test.ascend.test_ascend_utils import (
QWEN3_30B_A3B_INSTRUCT_2507_WEIGHTS_PATH,
)
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.few_shot_gsm8k import run_eval as run_eval_few_shot_gsm8k
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
is_in_ci,
popen_launch_pd_server,
)
register_npu_ci(est_time=3600, suite="stage-b-test-16-npu-a3", nightly=False)
register_npu_ci(est_time=3600, suite="nightly-16-npu-a3", nightly=True)
load_balance_method_options = [
"auto",
"round_robin",
"total_requests",
"total_tokens",
"follow_bootstrap_room",
]
all_params = list(itertools.product(load_balance_method_options, repeat=2))
class BaseTestNPULoadBalanceMethodDPDisaggregation(TestDisaggregationBase):
"""Testcase：Verify that the model accuracy did not decrease when --load-balance-method is set to round_robin, auto,
total_requests, total_tokens or follow_bootstrap_room in PD disaggregation scenario
[Test Category] Parameter
[Test Target] --load-balance-method
"""
params = ("auto", "auto")
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.prefill_load_balance_method, cls.decode_load_balance_method = cls.params
cls.model = QWEN3_30B_A3B_INSTRUCT_2507_WEIGHTS_PATH
os.environ["ASCEND_MF_STORE_URL"] = "tcp://127.0.0.1:24666"
# Non blocking start servers
cls.start_prefill()
cls.start_decode()
# Block until both
cls.wait_server_ready(cls.prefill_url + "/health")
cls.wait_server_ready(cls.decode_url + "/health")
cls.launch_lb()
cls.url = urlparse(cls.lb_url)
@classmethod
def start_prefill(cls):
prefill_args = [
"--disaggregation-mode",
"prefill",
"--tp-size",
"2",
"--enable-dp-attention",
"--dp",
"2",
"--load-balance-method",
cls.prefill_load_balance_method,
"--disaggregation-transfer-backend",
"ascend",
"--disable-cuda-graph",
"--attention-backend",
"ascend",
"--mem-fraction-static",
0.8,
"--dist-init-addr",
"127.0.0.1:10100",
"--base-gpu-id",
4,
]
cls.process_prefill = popen_launch_pd_server(
cls.model,
cls.prefill_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=prefill_args,
env={
"TRANSFORMERS_VERBOSITY": "error",
},
)
@classmethod
def start_decode(cls):
decode_args = [
"--disaggregation-mode",
"decode",
"--base-gpu-id",
2,
"--tp-size",
"2",
"--enable-dp-attention",
"--dp",
"2",
"--load-balance-method",
cls.decode_load_balance_method,
"--disaggregation-transfer-backend",
"ascend",
"--disable-cuda-graph",
"--attention-backend",
"ascend",
"--mem-fraction-static",
0.8,
"--dist-init-addr",
"127.0.0.1:10000",
]
cls.process_decode = popen_launch_pd_server(
cls.model,
cls.decode_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=decode_args,
env={
"TRANSFORMERS_VERBOSITY": "error",
},
)
def test_gsm8k(self):
args = SimpleNamespace(
num_shots=5,
data_path=None,
num_questions=200,
max_new_tokens=512,
parallel=128,
host=f"http://{self.url.hostname}",
port=int(self.url.port),
)
metrics = run_eval_few_shot_gsm8k(args)
self.assertGreaterEqual(
metrics["accuracy"],
# 0.95 with 0.02 tolerable fluctuation
0.93,
)
def test_server_info(self):
response = requests.get(f"{self.lb_url}/get_server_info")
self.assertEqual(response.status_code, 200)
data = json.loads(response.text)
self.assertEqual(
(
"follow_bootstrap_room"
if self.prefill_load_balance_method == "auto"
else self.prefill_load_balance_method
),
data.get("prefill")[0].get("load_balance_method"),
)
self.assertEqual(
(
"round_robin"
if self.decode_load_balance_method == "auto"
else self.decode_load_balance_method
),
data.get("decode")[0].get("load_balance_method"),
)
@classmethod
def tearDownClass(cls):
os.environ.pop("ASCEND_MF_STORE_URL")
super().tearDownClass()
# wait for server release source
sleep(10)
TestClassType = Type[BaseTestNPULoadBalanceMethodDPDisaggregation]
all_test_classes: List[TestClassType] = [BaseTestNPULoadBalanceMethodDPDisaggregation]
for index, param_tuple in enumerate(all_params):
if param_tuple == BaseTestNPULoadBalanceMethodDPDisaggregation.params:
continue
prefill_load_balance_method, decode_load_balance_method = param_tuple
class_name = f"Test_{index:02d}_prefill_{prefill_load_balance_method}_decode_{decode_load_balance_method}"
new_class = type(
class_name,
(BaseTestNPULoadBalanceMethodDPDisaggregation,),
{"params": param_tuple},
)
all_test_classes.append(new_class)
if __name__ == "__main__":
if is_in_ci():
RUN_COUNT = 3
loader = unittest.TestLoader()
suite = unittest.TestSuite()
selected_classes = random.sample(
all_test_classes, min(RUN_COUNT, len(all_test_classes))
)
for cls in selected_classes:
suite.addTests(loader.loadTestsFromTestCase(cls))
runner = unittest.TextTestRunner(verbosity=2)
runner.run(suite)
else:
unittest.main()
test/registered/npu/basic_function/parallel_strategy/expert_parallelism/test_npu_eplb_min_rebalancing_utilization_threshold.py
+162
View File
@@ -0,0 +1,162 @@
import os
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import QWEN3_30B_A3B_W8A8_WEIGHTS_PATH
from sglang.test.ci.ci_register import register_npu_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_npu_ci(est_time=400, suite="stage-b-test-8-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-8-npu-a3", nightly=True)
class TestEplbMinRebalancingUtilizationThresholdBase(CustomTestCase):
"""
Testcase：Validates that rebalancing operations are triggered or skipped based on the configured
--eplb-min-rebalancing-utilization-threshold value and current load balance.
[Test Category] Parameter
[Test Target] --eplb-min-rebalancing-utilization-threshold, --eplb-rebalance-layers-per-chunk
"""
model = QWEN3_30B_A3B_W8A8_WEIGHTS_PATH
accuracy = 0.86
common_args = [
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--trust-remote-code",
"--chunked-prefill-size",
"1024",
"--tp-size",
"8",
"--quantization",
"modelslim",
"--moe-a2a-backend",
"deepep",
"--deepep-mode",
"normal",
"--enable-eplb",
"--ep-num-redundant-experts",
16,
"--eplb-rebalance-num-iterations",
50,
"--expert-distribution-recorder-buffer-size",
50,
"--enable-expert-distribution-metrics",
"--eplb-rebalance-layers-per-chunk",
"1",
]
out_file_path = "./rebalance_out_log.txt"
err_file_path = "./rebalance_err_log.txt"
log_info = "Skipped ep rebalancing: current GPU utilization"
test_args = ["--eplb-min-rebalancing-utilization-threshold", 0.05]
@classmethod
def setUpClass(cls):
if hasattr(cls, "out_file_path"):
cls.out_file = open(cls.out_file_path, "w+", encoding="utf-8")
if hasattr(cls, "err_file_path"):
cls.err_file = open(cls.err_file_path, "w+", encoding="utf-8")
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=cls.common_args + cls.test_args,
env={
"SGLANG_ENABLE_JIT_DEEPGEMM": "0",
"SGLANG_EXPERT_LOCATION_UPDATER_CANARY": "1",
"HCCL_BUFFSIZE": "1024",
"SGLANG_DEEPEP_BF16_DISPATCH": "1",
"TRANSFORMERS_VERBOSITY": "error",
"DEEP_NORMAL_MODE_USE_INT8_QUANT": "1",
**os.environ,
},
return_stdout_stderr=(cls.out_file, cls.err_file),
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
cls.out_file.close()
cls.err_file.close()
os.remove("./rebalance_out_log.txt")
os.remove("./rebalance_err_log.txt")
def test_gsm8k(self):
args = SimpleNamespace(
max_tokens=512,
base_url=DEFAULT_URL_FOR_TEST,
model=self.model,
eval_name="gsm8k",
api="completion",
num_examples=200,
num_threads=128,
num_shots=5,
)
metrics = run_eval(args)
self.assertGreaterEqual(
metrics["score"],
self.accuracy,
f'Accuracy of {self.model} is {str(metrics["score"])}, is lower than {self.accuracy}',
)
"""
Testcase：When the configuration --eplb-min-rebalancing-utilization-threshold is set to 0.05, if the load balance
exceeds this threshold, rebalancing operations are skipped.
"""
self.err_file.seek(0)
content = self.err_file.read()
self.assertIn(self.log_info, content)
self.assertIn("[EPLBManager] rebalance start", content)
class TestEplbMinRebalancingUtilizationThreshold095(
TestEplbMinRebalancingUtilizationThresholdBase
):
"""
Testcase：When the configuration --eplb-min-rebalancing-utilization-threshold is set to 0.95, if load balancing
is less than or equal to this threshold, rebalancing operations are triggered.
"""
log_info = "rebalance end"
test_args = ["--eplb-min-rebalancing-utilization-threshold", 0.95]
@classmethod
def setUpClass(cls):
if hasattr(cls, "out_file_path"):
cls.out_file = open(cls.out_file_path, "w+", encoding="utf-8")
if hasattr(cls, "err_file_path"):
cls.err_file = open(cls.err_file_path, "w+", encoding="utf-8")
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=cls.common_args + cls.test_args,
env={
"SGLANG_ENABLE_JIT_DEEPGEMM": "0",
"SGLANG_EXPERT_LOCATION_UPDATER_CANARY": "1",
"HCCL_BUFFSIZE": "1024",
"SGLANG_NPU_DISABLE_ACL_FORMAT_WEIGHT": "1",
"TRANSFORMERS_VERBOSITY": "error",
"DEEP_NORMAL_MODE_USE_INT8_QUANT": "1",
**os.environ,
},
return_stdout_stderr=(cls.out_file, cls.err_file),
)
if __name__ == "__main__":
unittest.main(verbosity=2)
test/registered/npu/basic_function/parallel_strategy/expert_parallelism/test_npu_expert_distribution_recorder_mode.py
+130
View File
@@ -0,0 +1,130 @@
import glob
import os
import unittest
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import (
QWEN3_30B_A3B_INSTRUCT_2507_WEIGHTS_PATH,
run_command,
)
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_npu_ci(est_time=400, suite="stage-b-test-2-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-2-npu-a3", nightly=True)
class TestExpertDistributionRecorderModeStatic(CustomTestCase):
"""Testcase: Verify set the parameter --expert-distribution-recorder-mode，
will generate .pt file and the inference request successfully.
[Test Category] Parameter
[Test Target] --expert-distribution-recorder-mode
"""
expert_distribution_recorder_mode = "stat"
path = "/tmp/pt"
@classmethod
def setUpClass(cls):
cls.process = popen_launch_server(
QWEN3_30B_A3B_INSTRUCT_2507_WEIGHTS_PATH,
DEFAULT_URL_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--trust-remote-code",
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--mem-fraction-static",
"0.8",
"--tp-size",
"2",
"--expert-parallel-size",
"2",
"--enable-eplb",
"--moe-a2a-backend",
"deepep",
"--deepep-mode",
"normal",
"--expert-distribution-recorder-mode",
cls.expert_distribution_recorder_mode,
],
env={
"SGLANG_NPU_DISABLE_ACL_FORMAT_WEIGHT": "1",
"HCCL_BUFFSIZE": "1024",
"SGLANG_EXPERT_DISTRIBUTION_RECORDER_DIR": f"{cls.path}",
"TRANSFORMERS_VERBOSITY": "error",
},
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
run_command(f"rm -rf {cls.path}")
def test_recorder_mode(self):
# Start recording
requests.post(f"{DEFAULT_URL_FOR_TEST}/start_expert_distribution_record")
response = requests.post(
f"{DEFAULT_URL_FOR_TEST}/generate",
json={
"text": "The capital of France is",
"sampling_params": {
"temperature": 0,
"max_new_tokens": 32,
},
},
)
self.assertEqual(
response.status_code, 200, "The request status code is not 200."
)
self.assertIn(
"Paris", response.text, "The inference result does not include Paris."
)
# Stop recording
requests.post(f"{DEFAULT_URL_FOR_TEST}/stop_expert_distribution_record")
# Export the .pt file
requests.post(f"{DEFAULT_URL_FOR_TEST}/dump_expert_distribution_record")
# Check distribution_recorder_files
distribution_recorder_suffixes = ["*.pt"]
distribution_recorder_files = []
for suffix in distribution_recorder_suffixes:
distribution_recorder_files.extend(
glob.glob(os.path.join(self.path, "**", suffix), recursive=True)
)
self.assertGreater(
len(distribution_recorder_files),
0,
msg=f"No distribution recorder",
)
class TestExpertDistributionRecorderModeStatApprox(
TestExpertDistributionRecorderModeStatic
):
expert_distribution_recorder_mode = "stat_approx"
class TestExpertDistributionRecorderPerPass(TestExpertDistributionRecorderModeStatic):
expert_distribution_recorder_mode = "per_pass"
class TestExpertDistributionRecorderPerToken(TestExpertDistributionRecorderModeStatic):
expert_distribution_recorder_mode = "per_token"
if __name__ == "__main__":
unittest.main()
test/registered/npu/basic_function/parallel_strategy/expert_parallelism/test_npu_moe_dense_tp_size.py
+79
View File
@@ -0,0 +1,79 @@
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import DEEPSEEK_CODER_V2_LITE_WEIGHTS_PATH
from sglang.test.ci.ci_register import register_npu_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_npu_ci(est_time=400, suite="stage-b-test-2-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-2-npu-a3", nightly=True)
class TestAscendMoeDenseTPSize(CustomTestCase):
"""Testcase: Verify that the model accuracy remains uncompromised when the parameter --moe-dense-tp-size is configured to 1.
[Test Category] Parameter
[Test Target] --moe-dense-tp-size, --eplb-algorithm
"""
@classmethod
def setUpClass(cls):
cls.model = DEEPSEEK_CODER_V2_LITE_WEIGHTS_PATH
cls.process = popen_launch_server(
cls.model,
DEFAULT_URL_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--trust-remote-code",
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--mem-fraction-static",
"0.5",
"--tp-size",
"2",
"--expert-parallel-size",
"2",
"--enable-eplb",
"--moe-a2a-backend",
"deepep",
"--deepep-mode",
"normal",
"--eplb-algorithm",
"deepseek",
"--moe-dense-tp-size",
"1",
],
env={
"HCCL_BUFFSIZE": "1024",
},
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_gsm8k(self):
args = SimpleNamespace(
max_tokens=512,
base_url=DEFAULT_URL_FOR_TEST,
model=self.model,
eval_name="gsm8k",
api="completion",
num_examples=200,
num_threads=128,
num_shots=5,
)
metrics = run_eval(args)
self.assertGreater(metrics["score"], 0.79)
if __name__ == "__main__":
unittest.main()
test/registered/npu/basic_function/parallel_strategy/expert_parallelism/test_npu_moe_runner_backend.py
+92
View File
@@ -0,0 +1,92 @@
import unittest
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import DEEPSEEK_CODER_V2_LITE_WEIGHTS_PATH
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_npu_ci(est_time=400, suite="stage-b-test-2-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-2-npu-a3", nightly=True)
class TestMoreRunnerBackendTriton(CustomTestCase):
"""Testcase：Verify set --moe-runner-backend, the inference request is successfully processed.
[Test Category] Parameter
[Test Target] --ep, --enable-eplb, --moe-runner-backend, --deepep-mode, --ep-num-redundant-experts, --moe-a2a-backend,
--expert-distribution-recorder-buffer-size
"""
model = DEEPSEEK_CODER_V2_LITE_WEIGHTS_PATH
moe_runner_backend = "triton"
@classmethod
def setUpClass(cls):
cls.process = popen_launch_server(
cls.model,
DEFAULT_URL_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--trust-remote-code",
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--mem-fraction-static",
"0.5",
"--tp-size",
"2",
"--ep",
"2",
"--enable-eplb",
"--moe-a2a-backend",
"deepep",
"--deepep-mode",
"normal",
"--ep-num-redundant-experts",
"4",
"--expert-distribution-recorder-buffer-size",
"50",
"--moe-runner-backend",
cls.moe_runner_backend,
],
env={
"HCCL_BUFFSIZE": "1024",
},
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_moe_runner_backend(self):
response = requests.post(
f"{DEFAULT_URL_FOR_TEST}/generate",
json={
"text": "The capital of France is",
"sampling_params": {
"temperature": 0,
"max_new_tokens": 32,
},
},
)
self.assertEqual(
response.status_code, 200, "The request status code is not 200."
)
self.assertIn(
"Paris", response.text, "The inference result does not include Paris."
)
class TestMoreRunnerBackendTritonDefault(TestMoreRunnerBackendTriton):
moe_runner_backend = "auto"
if __name__ == "__main__":
unittest.main()
test/registered/npu/basic_function/pd_disaggregation/test_npu_pd_disaggregation.py
+270
View File
@@ -0,0 +1,270 @@
import logging
import os
import random
import shutil
import tempfile
import time
import unittest
from typing import Dict
import requests
from sglang.bench_serving import get_tokenizer
from sglang.test.ascend.test_ascend_utils import QWEN3_32B_WEIGHTS_PATH
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.server_fixtures.disaggregation_fixture import (
PDDisaggregationServerBase,
)
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
popen_launch_pd_server,
popen_with_error_check,
)
register_npu_ci(est_time=400, suite="stage-b-test-16-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-16-npu-a3", nightly=True)
class DisaggregationHiCacheBase(PDDisaggregationServerBase):
"""Testcase: All parameters for testing the PD_disaggregation feature were configured, inference was successful, and cache_tokens continued to grow..
[Test Category] Functional
[Test Target] PD disaggregatio on NPU
--disaggregation-mode; --disaggregation-transfer-backend; --disaggregation-decode-polling-interval;
--disaggregation-decode-enable-offload-kvcache; --num-reserved-decode-tokens; --disaggregation-bootstrap-port;
"""
@classmethod
def setUpClass(cls):
super(DisaggregationHiCacheBase, cls).setUpClass()
cls.model = QWEN3_32B_WEIGHTS_PATH
cls.tokenizer = get_tokenizer(cls.model)
cls.temp_dir = tempfile.mkdtemp()
cls.bootstrap_port = "8996"
cls.start_prefill()
cls.start_decode()
# Block until both
cls.wait_server_ready(cls.prefill_url + "/health")
cls.wait_server_ready(cls.decode_url + "/health")
cls.launch_router()
cls.wait_server_ready(cls.lb_url + "/health")
time.sleep(10)
@classmethod
def start_prefill(cls):
# Prefill with HiCache enabled
prefill_args = [
"--trust-remote-code",
"--attention-backend",
"ascend",
"--disaggregation-mode",
"prefill",
"--disaggregation-transfer-backend",
"ascend",
"--disaggregation-bootstrap-port",
cls.bootstrap_port,
"--tp-size",
"2",
"--enable-hierarchical-cache",
"--hicache-io-backend",
"kernel_ascend",
"--hicache-mem-layout",
"page_first_direct",
"--hicache-storage-backend",
"file",
"--hicache-storage-prefetch-policy",
"wait_complete",
"--mem-fraction-static",
"0.9",
"--disable-cuda-graph",
]
env = {
**os.environ,
"SGLANG_HICACHE_FILE_BACKEND_STORAGE_DIR": cls.temp_dir,
"ASCEND_MF_STORE_URL": "tcp://127.0.0.1:24667",
}
cls.process_prefill = popen_launch_pd_server(
cls.model,
cls.prefill_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=prefill_args,
env=env,
)
@classmethod
def start_decode(cls):
pass
@classmethod
def launch_router(cls):
lb_command = [
"python3",
"-m",
"sglang_router.launch_router",
"--pd-disaggregation",
"--prefill",
cls.prefill_url,
cls.bootstrap_port,
"--decode",
cls.decode_url,
"--host",
cls.base_host,
"--port",
cls.lb_port,
]
cls.process_lb = popen_with_error_check(lb_command)
cls.wait_server_ready(cls.lb_url + "/health")
def gen_prompt(self, token_num: int) -> str:
# Generate a string consisting of random tokens.
all_available_tokens = list(self.tokenizer.get_vocab().values())
selected_tokens = random.choices(all_available_tokens, k=token_num)
return self.tokenizer.decode(selected_tokens)
def send_request(
self, prompt: str, max_tokens: int = 100, temperature: float = 0.0
) -> Dict:
# Send a generate request and return response
response = requests.post(
f"{self.lb_url}/generate",
json={
"text": prompt,
"sampling_params": {
"temperature": temperature,
"max_new_tokens": max_tokens,
"ignore_eos": True,
},
},
timeout=60,
)
self.assertEqual(
response.status_code,
200,
f"Request failed: {response.status_code} - {response.text}",
)
return response.json()
def trigger_offloading_and_flush(self):
# Helper method to trigger offloading and flush cache
# Trigger offloading
self.send_request(self.gen_prompt(1), max_tokens=150)
# Flush device cache to force remote storage access
time.sleep(2)
requests.post(self.prefill_url + "/flush_cache")
class TestDisaggregationDecodeWithHiCache(DisaggregationHiCacheBase):
"""Decode startup parameters"""
ascend_devices = os.environ.get("ASCEND_RT_VISIBLE_DEVICES", "0,1,2,3")
base_gpu_id = (
ascend_devices.split(",")[2] if len(ascend_devices.split(",")) >= 3 else "2"
)
@classmethod
def start_decode(cls):
decode_args = [
"--trust-remote-code",
"--attention-backend",
"ascend",
"--disaggregation-mode",
"decode",
"--disaggregation-transfer-backend",
"ascend",
"--tp-size",
2,
"--mem-fraction-static",
"0.9",
"--base-gpu-id",
cls.base_gpu_id,
"--disaggregation-decode-enable-offload-kvcache",
"--hicache-io-backend",
"kernel_ascend",
"--hicache-mem-layout",
"page_first_direct",
"--hicache-storage-backend",
"file",
"--hicache-storage-prefetch-policy",
"wait_complete",
"--num-reserved-decode-tokens",
128,
"--disaggregation-decode-polling-interval",
2,
]
env = {
**os.environ,
"SGLANG_HICACHE_FILE_BACKEND_STORAGE_DIR": cls.temp_dir,
"ASCEND_MF_STORE_URL": "tcp://127.0.0.1:24667",
}
cls.process_decode = popen_launch_pd_server(
cls.model,
cls.decode_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=decode_args,
env=env,
)
def test_prefill_cache_hit(self):
"""Test that prefill cache works with repeated queries"""
repeated_prompt = self.gen_prompt(800)
self.send_request(repeated_prompt, max_tokens=100)
# Flush cache
self.trigger_offloading_and_flush()
# Second request - should hit cache (faster)
response2 = self.send_request(repeated_prompt, max_tokens=100)
# Assert cached tokens cnt
self.assertGreater(response2["meta_info"]["cached_tokens"], 700)
def test_multi_turn_conversation_cache(self):
# Test multi-turn conversation scenario with cache hit improvement
logging.warning("====================Testing request=======================")
initial_prompt = self.gen_prompt(300)
response1 = self.send_request(initial_prompt, max_tokens=200, temperature=0.1)
current_context = initial_prompt + response1["text"]
previous_cached_tokens = 0
for turn in range(2, 5):
logging.warning(f"\nTurn {turn}: Continuing from previous context")
response = self.send_request(
current_context, max_tokens=200, temperature=0.1
)
cached_tokens = response["meta_info"]["cached_tokens"]
logging.warning(f"Turn {turn} cached tokens: {cached_tokens}")
logging.warning(
f"Improvement: {cached_tokens - previous_cached_tokens} tokens"
)
# Assert cache improvement
self.assertGreater(
cached_tokens,
previous_cached_tokens,
f"Turn {turn} should have more cached tokens than turn {turn - 1}",
)
# Update context and cached tokens for next iteration
current_context += response["text"]
previous_cached_tokens = cached_tokens
# Flush prefill cache
self.trigger_offloading_and_flush()
@classmethod
def tearDownClass(cls):
super().tearDownClass()
if os.path.exists(cls.temp_dir):
shutil.rmtree(cls.temp_dir)
if __name__ == "__main__":
unittest.main()
test/registered/npu/basic_function/speculative_inference/test_npu_basic_sanity_eagle3.py
+73
View File
@@ -0,0 +1,73 @@
import unittest
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import (
QWEN3_8B_EAGLE3_WEIGHTS_PATH,
QWEN3_8B_WEIGHTS_PATH,
)
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.kits.basic_api_contract_kit import BasicAPIContractMixin
from sglang.test.kits.basic_decode_correctness_kit import BasicDecodeCorrectnessMixin
from sglang.test.kits.basic_scheduler_stress_kit import BasicSchedulerStressMixin
from sglang.test.kits.eval_accuracy_kit import GSM8KMixin
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_npu_ci(est_time=200, suite="stage-b-test-1-npu-a3", nightly=False)
register_npu_ci(est_time=200, suite="nightly-1-npu-a3", nightly=True)
class TestBasicSanityEagle3(
BasicAPIContractMixin,
BasicDecodeCorrectnessMixin,
BasicSchedulerStressMixin,
GSM8KMixin,
CustomTestCase,
):
served_model_name = QWEN3_8B_WEIGHTS_PATH
model = QWEN3_8B_WEIGHTS_PATH
gsm8k_num_questions = 1400
gsm8k_accuracy_thres = 0.74
@classmethod
def setUpClass(cls):
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
QWEN3_8B_WEIGHTS_PATH,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--dtype",
"float16",
"--attention-backend",
"ascend",
"--speculative-algorithm",
"EAGLE3",
"--speculative-draft-model-path",
QWEN3_8B_EAGLE3_WEIGHTS_PATH,
"--speculative-num-steps",
"1",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"2",
"--cuda-graph-max-bs",
"4",
"--mem-fraction-static",
"0.7",
"--disable-piecewise-cuda-graph",
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
if __name__ == "__main__":
unittest.main()
test/registered/npu/basic_function/speculative_inference/test_npu_eagle3.py
+1
View File
@@ -9,6 +9,7 @@ from sglang.test.ascend.test_ascend_utils import (
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.test_utils import CustomTestCase
register_npu_ci(est_time=400, suite="stage-b-test-1-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-1-npu-a3", nightly=True)
test/registered/npu/basic_function/speculative_inference/test_npu_speculative_attention_mode.py
+178
View File
@@ -0,0 +1,178 @@
import os
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import (
QWEN3_32B_EAGLE3_WEIGHTS_PATH,
QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH,
)
from sglang.test.ci.ci_register import register_npu_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_npu_ci(
est_time=400,
suite="stage-b-test-4-npu-a3",
nightly=False,
)
register_npu_ci(
est_time=400,
suite="nightly-4-npu-a3",
nightly=True,
)
class TestNpuSpeculativeAttentionMode(CustomTestCase):
"""Testcase: Verify that model inference accuracy remains uncompromised when launching the server
with --speculative-attention-mode set to 'decode' and 'prefill' respectively.
[Test Category] Parameter
[Test Target] --speculative-attention-mode
"""
def _run_gsm8k_eval(self):
"""Helper method to run GSM8K evaluation and return metrics."""
eval_args = SimpleNamespace(
base_url=DEFAULT_URL_FOR_TEST,
eval_name="gsm8k",
api="completion",
model=QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH,
num_examples=1319,
num_threads=128,
max_tokens=512,
num_shots=5,
temperature=0.0,
)
return run_eval(eval_args)
def test_speculative_attention_mode_decode(self):
"""Test --speculative-attention-mode decode without PD disaggregation."""
args = [
"--trust-remote-code",
"--attention-backend",
"ascend",
"--device",
"npu",
"--quantization",
"modelslim",
"--disable-radix-cache",
"--speculative-draft-model-quantization",
"unquant",
"--speculative-algorithm",
"EAGLE3",
"--speculative-draft-model-path",
QWEN3_32B_EAGLE3_WEIGHTS_PATH,
"--speculative-num-steps",
"4",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"5",
"--speculative-attention-mode",
"decode",
"--tp-size",
"4",
"--mem-fraction-static",
"0.7",
"--disable-cuda-graph",
"--dtype",
"bfloat16",
]
env = os.environ.copy()
env.update(
{
"SGLANG_ENABLE_OVERLAP_PLAN_STREAM": "1",
"SGLANG_ENABLE_SPEC_V2": "1",
"TRANSFORMERS_VERBOSITY": "error",
}
)
process = popen_launch_server(
QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH,
DEFAULT_URL_FOR_TEST,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH * 3,
other_args=args,
env=env,
)
try:
metrics = self._run_gsm8k_eval()
self.assertGreaterEqual(
metrics["score"],
0.83,
f"GSM8K score {metrics['score']} below threshold 0.83",
)
finally:
kill_process_tree(process.pid)
def test_speculative_attention_mode_prefill(self):
"""Test --speculative-attention-mode prefill without PD disaggregation."""
args = [
"--trust-remote-code",
"--attention-backend",
"ascend",
"--device",
"npu",
"--quantization",
"modelslim",
"--disable-radix-cache",
"--speculative-draft-model-quantization",
"unquant",
"--speculative-algorithm",
"EAGLE3",
"--speculative-draft-model-path",
QWEN3_32B_EAGLE3_WEIGHTS_PATH,
"--speculative-num-steps",
"4",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"5",
"--speculative-attention-mode",
"prefill",
"--tp-size",
"4",
"--mem-fraction-static",
"0.7",
"--disable-cuda-graph",
"--dtype",
"bfloat16",
]
env = os.environ.copy()
env.update(
{
"SGLANG_ENABLE_OVERLAP_PLAN_STREAM": "1",
"SGLANG_ENABLE_SPEC_V2": "1",
"TRANSFORMERS_VERBOSITY": "error",
}
)
process = popen_launch_server(
QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH,
DEFAULT_URL_FOR_TEST,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH * 3,
other_args=args,
env=env,
)
try:
metrics = self._run_gsm8k_eval()
self.assertGreaterEqual(
metrics["score"],
0.83,
f"GSM8K score {metrics['score']} below threshold 0.83",
)
finally:
kill_process_tree(process.pid)
if __name__ == "__main__":
unittest.main()
test/registered/npu/basic_function/speculative_inference/test_npu_speculative_draft_attention_backend.py
+109
View File
@@ -0,0 +1,109 @@
import os
import unittest
from types import SimpleNamespace
from urllib.parse import urlparse
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import (
DEEPSEEK_R1_0528_W4A8_PER_CHANNEL_WEIGHTS_PATH,
)
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_npu_ci(est_time=400, suite="stage-b-test-16-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-16-npu-a3", nightly=True)
MODEL_PATH = DEEPSEEK_R1_0528_W4A8_PER_CHANNEL_WEIGHTS_PATH
class TestAscendSpeculativeDraftAttentionAndMoeRunner(CustomTestCase):
"""Testcase: Test configuration '--speculative-draft-attention-backend' and '--speculative-moe-runner-backend' on the GSM8K dataset is no less than 0.9.
[Test Category] Parameter
[Test Target] --speculative-draft-attention-backend; --speculative-moe-runner-backend
"""
os.environ["DEEP_NORMAL_MODE_USE_INT8_QUANT"] = "1"
os.environ["HCCL_BUFFSIZE"] = "2048"
os.environ["SGLANG_ENABLE_OVERLAP_PLAN_SITEAM"] = "1"
os.environ["SGLANG_ENABLE_SPEC_V2"] = "1"
env = os.environ.copy()
@classmethod
def setUpClass(cls):
cls.models = MODEL_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.url = urlparse(DEFAULT_URL_FOR_TEST)
cls.base_url = DEFAULT_URL_FOR_TEST
cls.common_args = [
"--trust-remote-code",
"--attention-backend",
"ascend",
"--quantization",
"modelslim",
"--mem-fraction-static",
0.7,
"--disable-radix-cache",
"--chunked-prefill-size",
32768,
"--tp-size",
16,
"--speculative-algorithm",
"NEXTN",
"--speculative-num-steps",
1,
"--speculative-eagle-topk",
1,
"--speculative-num-draft-tokens",
2,
"--moe-a2a-backend",
"deepep",
"--deepep-mode",
"auto",
"--max-running-requests",
64,
"--speculative-draft-attention-backend",
"ascend",
"--speculative-moe-runner-backend",
"auto",
]
cls.process = popen_launch_server(
MODEL_PATH,
cls.base_url,
timeout=1500,
other_args=cls.common_args,
env=cls.env,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_a_gsm8k(self):
args = SimpleNamespace(
base_url=self.base_url,
eval_name="gsm8k",
api="completion",
num_examples=1319,
num_threads=128,
max_tokens=512,
num_shots=5,
)
metrics = run_eval(args)
score = metrics["score"]
print(f"GSM8K score for {MODEL_PATH}: {score:.4f}")
self.assertIsNotNone(score, "GSM8K evaluation returned no score")
self.assertIsInstance(score, float, "Score should be a float")
self.assertGreaterEqual(score, 0.9, f"GSM8K score {score} below threshold 0.9")
if __name__ == "__main__":
unittest.main()
test/registered/npu/basic_function/speculative_inference/test_npu_speculative_moe_a2a_backend.py
+99
View File
@@ -0,0 +1,99 @@
import os
import unittest
from types import SimpleNamespace
from urllib.parse import urlparse
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import (
DEEPSEEK_R1_0528_W8A8_WEIGHTS_PATH,
)
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_npu_ci(est_time=400, suite="stage-b-test-16-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-16-npu-a3", nightly=True)
class TestAscendDistTimeout(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = DEEPSEEK_R1_0528_W8A8_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.accuracy = 0.95
cls.url = urlparse(DEFAULT_URL_FOR_TEST)
os.environ["HCCL_BUFFSIZE"] = "2048"
os.environ["SGLANG_ENABLE_OVERLAP_PLAN_STREAM"] = "1"
os.environ["SGLANG_ENABLE_SPEC_V2"] = "1"
os.environ["SGLANG_NPU_FUSED_MOE_MODE"] = "1"
os.environ["TRANSFORMERS_VERBOSITY"] = "error"
cls.env = os.environ.copy()
cls.common_args = [
"--trust-remote-code",
"--attention-backend",
"ascend",
"--quantization",
"modelslim",
"--mem-fraction-static",
0.8,
"--disable-radix-cache",
"--chunked-prefill-size",
2048,
"--tp-size",
16,
"--disable-cuda-graph",
"--speculative-moe-a2a-backend",
"ascend_fuseep",
"--speculative-num-steps",
1,
"--speculative-eagle-topk",
1,
"--speculative-num-draft-tokens",
2,
"--moe-a2a-backend",
"ascend_fuseep",
"--deepep-mode",
"auto",
"--speculative-draft-model-quantization",
"unquant",
]
def test_a_gsm8k(self):
print(f"##=== Testing accuracy: {self.model} ===##")
process = popen_launch_server(
self.model,
self.base_url,
timeout=1500,
other_args=self.common_args,
env=self.env,
)
try:
args = SimpleNamespace(
base_url=self.base_url,
eval_name="gsm8k",
api="completion",
num_examples=1319,
num_threads=128,
max_tokens=512,
num_shots=5,
temperature=0.0,
)
metrics = run_eval(args)
self.assertGreaterEqual(
metrics["score"],
self.accuracy,
f"GSM8K score {metrics['score']} below threshold {self.accuracy}",
)
finally:
kill_process_tree(process.pid)
if __name__ == "__main__":
unittest.main()
test/registered/npu/basic_function/speculative_inference/test_npu_speculative_multi_npu.py
+160
View File
@@ -0,0 +1,160 @@
import os
import unittest
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import (
QWEN3_32B_EAGLE3_WEIGHTS_PATH,
QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH,
)
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.test_utils import (
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_npu_ci(est_time=400, suite="stage-b-test-4-npu-a3", nightly=False)
register_npu_ci(est_time=400, suite="nightly-4-npu-a3", nightly=True)
_ASCEND_BACKEND = "ascend"
_COMMON_ARGS = [
"--trust-remote-code",
"--attention-backend",
_ASCEND_BACKEND,
"--quantization",
"modelslim",
"--disable-radix-cache",
"--speculative-draft-model-quantization",
"unquant",
"--speculative-algorithm",
"EAGLE3",
"--speculative-draft-model-path",
QWEN3_32B_EAGLE3_WEIGHTS_PATH,
"--speculative-num-steps",
"4",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"5",
"--speculative-attention-mode",
"decode",
"--mem-fraction-static",
"0.7",
"--disable-cuda-graph",
"--dtype",
"bfloat16",
]
# --speculative-draft-load-format dummy: initializes draft weights with random values.
# Valid options: auto (default), dummy.
# dummy is used for profiling; output quality is not guaranteed.
# --speculative-draft-model-revision main: specifies a branch/tag/commit for the
# draft model. Valid range: any valid git ref string; None means default branch.
_SERVER_ARGS = _COMMON_ARGS + [
"--tp-size",
"4",
"--speculative-draft-load-format",
"dummy",
"--speculative-draft-model-revision",
"main",
]
class TestNpuSpeculativeDraftParams(CustomTestCase):
"""Test --speculative-draft-load-format and --speculative-draft-model-revision
with 4-card TP.
[Test Category] Parameter & Multi-NPU
[Test Target]
--tp-size 4;
--speculative-draft-load-format dummy;
--speculative-draft-model-revision main
[Model]
Target: aleoyang/Qwen3-32B-w8a8-MindIE
Draft: Qwen/Qwen3-32B-Eagle3 (dummy weights, revision=main)
"""
@classmethod
def setUpClass(cls) -> None:
cls.base_url = DEFAULT_URL_FOR_TEST
env = os.environ.copy()
env.update(
{
"SGLANG_ENABLE_OVERLAP_PLAN_STREAM": "1",
"SGLANG_ENABLE_SPEC_V2": "1",
}
)
cls.process = popen_launch_server(
QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH * 5,
other_args=_SERVER_ARGS,
env=env,
)
@classmethod
def tearDownClass(cls) -> None:
kill_process_tree(cls.process.pid)
def test_draft_params_via_server_info(self):
"""Verify draft load format and revision are set correctly via /server_info."""
# 1. Health check
health_resp = requests.get(f"{self.base_url}/health", timeout=10)
self.assertEqual(health_resp.status_code, 200)
# 2. Get server info and assert parameter values
# Note: /get_server_info is deprecated, use /server_info instead.
info_resp = requests.get(f"{self.base_url}/server_info", timeout=10)
self.assertEqual(info_resp.status_code, 200)
info = info_resp.json()
load_format = info.get("speculative_draft_load_format")
revision = info.get("speculative_draft_model_revision")
self.assertEqual(
load_format,
"dummy",
"speculative_draft_load_format should be 'dummy'",
)
self.assertEqual(
revision,
"main",
"speculative_draft_model_revision should be 'main'",
)
# 3. Simple generation test to ensure service is functional
prompt = "What is the capital of France?"
resp = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 64,
"temperature": 0,
},
timeout=60,
)
self.assertEqual(resp.status_code, 200)
data = resp.json()
self.assertIn("choices", data)
content = data["choices"][0]["message"]["content"]
self.assertGreater(len(content.strip()), 0)
# With dummy draft weights, the main model still produces correct answer.
self.assertIn(
"paris",
content.lower(),
f"Expected 'Paris' in response, but got: {content[:200]}",
)
print(f"Q: {prompt}")
print(f"A: {content}")
if __name__ == "__main__":
unittest.main()
test/registered/npu/basic_function/speculative_inference/test_npu_speculative_token_map.py
+157
View File
@@ -0,0 +1,157 @@
import os
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import (
FR_SPEC_TOKEN_MAP_PATH,
LLAMA_3_8B_EAGLE_WEIGHTS_PATH,
LLAMA_3_8B_INSTRUCT_WEIGHTS_PATH,
QWEN3_32B_EAGLE3_WEIGHTS_PATH,
QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH,
)
from sglang.test.ci.ci_register import register_npu_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_npu_ci(est_time=300, suite="stage-b-test-4-npu-a3", nightly=False)
register_npu_ci(est_time=300, suite="nightly-4-npu-a3", nightly=True)
class TestNpuSpeculativeTokenMap(CustomTestCase):
"""Test --speculative-token-map with EAGLE3 (ignored) and EAGLE (enabled).
Both cases run GSM8K evaluation to ensure accuracy does not degrade.
"""
def test_eagle3_ignores_token_map_gsm8k(self):
"""EAGLE3 ignores token map; GSM8K accuracy should meet threshold."""
args = [
"--trust-remote-code",
"--attention-backend",
"ascend",
"--quantization",
"modelslim",
"--disable-radix-cache",
"--speculative-algorithm",
"EAGLE3",
"--speculative-draft-model-path",
QWEN3_32B_EAGLE3_WEIGHTS_PATH,
"--speculative-draft-model-quantization",
"unquant",
"--speculative-num-steps",
"4",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"5",
"--speculative-attention-mode",
"decode",
"--speculative-token-map",
"/nonexistent/token_map.pt", # ignored
"--tp-size",
"4",
"--mem-fraction-static",
"0.7",
"--disable-cuda-graph",
"--dtype",
"bfloat16",
]
env = os.environ.copy()
env.update(
{
"SGLANG_ENABLE_OVERLAP_PLAN_STREAM": "1",
"SGLANG_ENABLE_SPEC_V2": "1",
}
)
process = popen_launch_server(
QWEN3_32B_W8A8_MINDIE_WEIGHTS_PATH,
DEFAULT_URL_FOR_TEST,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH * 3,
other_args=args,
env=env,
)
try:
eval_args = SimpleNamespace(
base_url=DEFAULT_URL_FOR_TEST,
eval_name="gsm8k",
api="completion",
num_examples=1319,
num_threads=128,
max_tokens=512,
num_shots=5,
temperature=0.0,
)
metrics = run_eval(eval_args)
self.assertGreaterEqual(metrics["score"], 0.83)
finally:
kill_process_tree(process.pid)
def test_eagle_with_valid_token_map_gsm8k(self):
"""EAGLE (EAGLE-2) with valid token map; GSM8K accuracy should meet threshold."""
args = [
"--trust-remote-code",
"--attention-backend",
"ascend",
"--disable-radix-cache",
"--speculative-algorithm",
"EAGLE",
"--speculative-draft-model-path",
LLAMA_3_8B_EAGLE_WEIGHTS_PATH,
"--speculative-num-steps",
"5",
"--speculative-eagle-topk",
"1",
"--speculative-num-draft-tokens",
"8",
"--speculative-token-map",
FR_SPEC_TOKEN_MAP_PATH,
"--tp-size",
"4",
"--mem-fraction-static",
"0.7",
"--disable-cuda-graph",
"--dtype",
"float16",
]
env = os.environ.copy()
env.update(
{
"SGLANG_ENABLE_OVERLAP_PLAN_STREAM": "1",
"SGLANG_ENABLE_SPEC_V2": "1",
}
)
process = popen_launch_server(
LLAMA_3_8B_INSTRUCT_WEIGHTS_PATH,
DEFAULT_URL_FOR_TEST,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH * 3,
other_args=args,
env=env,
)
try:
eval_args = SimpleNamespace(
base_url=DEFAULT_URL_FOR_TEST,
eval_name="gsm8k",
api="completion",
num_examples=1319,
num_threads=128,
max_tokens=512,
num_shots=5,
temperature=0.0,
)
metrics = run_eval(eval_args)
self.assertGreaterEqual(
metrics["score"], 0.75
) # adjust threshold as needed
finally:
kill_process_tree(process.pid)
if __name__ == "__main__":
unittest.main()