chenchenghao/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Add NPU basic function testcases (#19382)

Browse Source 
Co-authored-by: cy <chenyang08056032@163.com>
Co-authored-by: Cherry_ming <136634645@qq.com>
This commit is contained in:
Sugar920
2026-03-16 15:09:56 +08:00
committed by GitHub
co-authored by cy Cherry_ming
parent e96a3752a0
commit 895e56097c
87 changed files with 4587 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/registered/ascend/interface/test_npu_api.py
+732
View File
@@ -0,0 +1,732 @@
import json
import logging
import os
import shutil
import unittest
import requests
from transformers import AutoTokenizer
from sglang.srt.environ import envs
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import LLAMA_3_2_1B_INSTRUCT_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,
)
# Global variables: Manage server process and initialization status
GLOBAL_SERVER_PROCESS = None
GLOBAL_SERVER_INITIALIZED = False
OUTPUT_DIR = "./profiler_dir"
register_npu_ci(est_time=1600, suite="nightly-npu-a3-merged", nightly=True)
class TestNpuApi(CustomTestCase):
"""Testcase: Verify that the basic functions of the API interfaces work properly and the returned parameters are consistent with the configurations.
[Test Category] Interface
[Test Target] /health; /health_generate; /ping; /model_info; /server_info; /get_load; /v1/models; /v1/models/{model:path}; /generate
"""
@classmethod
def setUpClass(cls):
global GLOBAL_SERVER_PROCESS, GLOBAL_SERVER_INITIALIZED
# Start server only if not initialized
if not GLOBAL_SERVER_INITIALIZED:
cls.model = LLAMA_3_2_1B_INSTRUCT_WEIGHTS_PATH
other_args = [
"--attention-backend",
"ascend",
"--enable-return-hidden-states",
]
# Start server and save to global variable
GLOBAL_SERVER_PROCESS = popen_launch_server(
cls.model,
DEFAULT_URL_FOR_TEST,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
GLOBAL_SERVER_INITIALIZED = True
cls.base_url = DEFAULT_URL_FOR_TEST
@classmethod
def tearDownClass(cls):
# First class does not terminate server
pass
def test_api_health(self):
response = requests.get(f"{self.base_url}/health")
self.assertEqual(response.status_code, 200)
def test_api_health_generate(self):
response = requests.get(f"{self.base_url}/health_generate")
self.assertEqual(response.status_code, 200)
def test_api_ping(self):
response = requests.get(f"{self.base_url}/ping")
self.assertEqual(response.status_code, 200)
def test_api_model_info(self):
response = requests.get(f"{self.base_url}/model_info")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["model_path"], self.model)
self.assertEqual(response.json()["tokenizer_path"], self.model)
self.assertTrue(response.json()["is_generation"])
self.assertIsNone(response.json()["preferred_sampling_params"])
self.assertEqual(response.json()["weight_version"], "default")
self.assertFalse(response.json()["has_image_understanding"])
self.assertFalse(response.json()["has_audio_understanding"])
self.assertEqual(response.json()["model_type"], "llama")
self.assertEqual(response.json()["architectures"][0], "LlamaForCausalLM")
def test_api_server_info(self):
response = requests.get(f"{self.base_url}/server_info")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["model_path"], self.model)
self.assertEqual(response.json()["tokenizer_path"], self.model)
def test_api_get_load(self):
response = requests.get(f"{self.base_url}/get_load")
self.assertEqual(response.status_code, 200)
self.assertIsNone(response.json()[0]["rid"])
self.assertIsNone(response.json()[0]["http_worker_ipc"])
self.assertIsNone(response.json()[0]["dp_rank"])
self.assertGreaterEqual(response.json()[0]["num_reqs"], 0)
self.assertGreaterEqual(response.json()[0]["num_waiting_reqs"], 0)
self.assertGreaterEqual(response.json()[0]["num_tokens"], 0)
def test_api_v1_models(self):
response = requests.get(f"{self.base_url}/v1/models")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["data"][0]["id"], self.model)
self.assertEqual(response.json()["data"][0]["object"], "model")
self.assertEqual(response.json()["data"][0]["owned_by"], "sglang")
self.assertEqual(response.json()["data"][0]["root"], self.model)
self.assertEqual(response.json()["data"][0]["max_model_len"], 131072)
def test_api_v1_models_path(self):
response = requests.get(f"{self.base_url}/v1/models/{self.model}")
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["id"], self.model)
self.assertEqual(response.json()["object"], "model")
self.assertEqual(response.json()["owned_by"], "sglang")
self.assertEqual(response.json()["root"], self.model)
self.assertEqual(response.json()["max_model_len"], 131072)
def test_api_generate_single_text(self):
response = requests.post(
f"{self.base_url}/generate",
json={
"rid": "req_001",
"text": "The capital of France is",
"sampling_params": {
"temperature": 0,
"max_new_tokens": 20,
},
"return_logprob": True,
"stream": False,
"return_hidden_states": True,
},
)
self.assertEqual(response.status_code, 200)
meta_info_keys = response.json()["meta_info"].keys()
self.assertEqual("req_001", response.json()["meta_info"]["id"])
self.assertIn("Paris", response.json()["text"])
self.assertEqual(20, response.json()["meta_info"]["completion_tokens"])
self.assertIn("input_token_logprobs", meta_info_keys)
self.assertIn("output_token_logprobs", meta_info_keys)
self.assertIn("hidden_states", meta_info_keys)
def test_api_generate_batch_texts(self):
rids = ["req_1", "req_2"]
texts = [
"The capital of France is",
"What is the best time of year to visit Japan for cherry blossoms?",
]
response = requests.post(
f"{self.base_url}/generate",
json={
"rid": rids,
"text": texts,
"sampling_params": {
"temperature": 0,
"max_new_tokens": 20,
},
"return_logprob": False,
"stream": False,
"return_hidden_states": False,
},
)
self.assertEqual(response.status_code, 200)
self.assertEqual("req_1", response.json()[0]["meta_info"]["id"])
self.assertIn("Paris", response.json()[0]["text"])
self.assertEqual("req_2", response.json()[1]["meta_info"]["id"])
self.assertIn("Japan", response.json()[1]["text"])
def test_api_generate_temperature(self):
response = requests.post(
f"{self.base_url}/generate",
json={
"text": "The capital of France is",
"sampling_params": {
"temperature": 5,
"max_new_tokens": 20,
},
},
)
self.assertEqual(response.status_code, 200)
text1 = response.json()["text"]
response = requests.post(
f"{self.base_url}/generate",
json={
"text": "The capital of France is",
"sampling_params": {
"temperature": 5,
"max_new_tokens": 20,
},
},
)
self.assertEqual(response.status_code, 200)
text2 = response.json()["text"]
self.assertNotEqual(text2, text1)
def test_api_generate_input_ids(self):
text = "The capital of France is"
tokenizer = AutoTokenizer.from_pretrained(self.model)
input_ids = tokenizer(text, return_tensors="pt")["input_ids"][0].tolist()
response = requests.post(
f"{self.base_url}/generate",
json={
"rid": "req_002",
"input_ids": input_ids,
"sampling_params": {
"temperature": 0,
"max_new_tokens": 10,
},
"return_logprob": False,
"stream": True,
"return_hidden_states": False,
},
)
self.assertEqual(response.status_code, 200)
lines = response.text.strip().split("\n")
self.assertGreaterEqual(len(lines), 10)
json_data = lines[-3][6:]
data = json.loads(json_data)
meta_info_keys = data["meta_info"].keys()
self.assertEqual("req_002", data["meta_info"]["id"])
self.assertIn("Paris", data["text"])
self.assertEqual(10, data["meta_info"]["completion_tokens"])
self.assertNotIn("input_token_logprobs", meta_info_keys)
self.assertNotIn("output_token_logprobs", meta_info_keys)
self.assertNotIn("hidden_states", meta_info_keys)
class TestChatCompletionsInterface(CustomTestCase):
"""Testcase: The test is to verify whether the functions of each parameter of the v1/chat/completions interface are normal.
[Test Category] Interface
[Test Target] v1/chat/completions
"""
@classmethod
def setUpClass(cls):
# Skip initialization, directly reuse global server
cls.model = LLAMA_3_2_1B_INSTRUCT_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.additional_chat_kwargs = {}
@classmethod
def tearDownClass(cls):
# Do not terminate server
pass
def test_model_and_messages(self):
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
},
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
data = response.json()
self.assertEqual(data["model"], self.model)
self.assertIsNotNone(data["choices"][0]["message"]["reasoning_content"])
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"messages": [{"role": "user", "content": "Hello"}],
},
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
data = response.json()
self.assertEqual(data["model"], "default")
self.assertIsNotNone(data["choices"][0]["message"]["reasoning_content"])
def test_max_completion_tokens(self):
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"messages": [{"role": "user", "content": "Hello"}],
"max_completion_tokens": 1,
},
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
self.assertEqual(response.json()["choices"][0]["finish_reason"], "length")
def test_stream(self):
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
"stream": True,
},
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
has_reasoning = False
has_content = False
for line in response.iter_lines():
if line:
line = line.decode("utf-8")
if line.startswith("data:") and not line.startswith("data: [DONE]"):
data = json.loads(line[6:])
if "choices" in data and len(data["choices"]) > 0:
delta = data["choices"][0].get("delta", {})
if "reasoning_content" in delta and delta["reasoning_content"]:
has_reasoning = True
if "content" in delta and delta["content"]:
has_content = True
self.assertTrue(
has_reasoning, "Reasoning content not included in stream response"
)
self.assertTrue(has_content, "Normal content not included in stream response")
def test_temperature(self):
response1 = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [
{
"role": "user",
"content": "Please write a five-character quatrain for me.",
}
],
"temperature": 0,
},
)
self.assertEqual(response1.status_code, 200, f"Failed with: {response1.text}")
content1 = response1.json()["choices"][0]["message"]["content"]
response2 = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [
{
"role": "user",
"content": "Please write a five-character quatrain for me.",
}
],
"temperature": 0,
},
)
self.assertEqual(response2.status_code, 200, f"Failed with: {response2.text}")
content2 = response2.json()["choices"][0]["message"]["content"]
self.assertEqual(content1, content2)
response3 = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [
{
"role": "user",
"content": "Please write a five-character quatrain for me.",
}
],
"temperature": 2,
},
)
self.assertEqual(response3.status_code, 200, f"Failed with: {response3.text}")
content3 = response3.json()["choices"][0]["message"]["content"]
response4 = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [
{
"role": "user",
"content": "Please write a five-character quatrain for me.",
}
],
"temperature": 2,
},
)
self.assertEqual(response4.status_code, 200, f"Failed with: {response4.text}")
content4 = response4.json()["choices"][0]["message"]["content"]
self.assertNotEqual(content3, content4)
def test_return_hidden_states(self):
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
"return_hidden_states": True,
},
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
self.assertIn("hidden_states", response.json()["choices"][0])
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
},
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
self.assertNotIn("hidden_states", response.json()["choices"][0])
def test_top_k(self):
response1 = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [
{
"role": "user",
"content": "Please write a five-character quatrain for me.",
}
],
"top_k": 20,
},
)
self.assertEqual(response1.status_code, 200, f"Failed with: {response1.text}")
content1 = response1.json()["choices"][0]["message"]["content"]
response2 = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [
{
"role": "user",
"content": "Please write a five-character quatrain for me.",
}
],
"top_k": 20,
},
)
self.assertEqual(response2.status_code, 200, f"Failed with: {response2.text}")
content2 = response2.json()["choices"][0]["message"]["content"]
self.assertNotEqual(content1, content2)
def test_stop_token_ids(self):
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
"stop_token_ids": [1, 13],
},
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
self.assertEqual(response.json()["choices"][0]["matched_stop"], 13)
def test_rid(self):
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
"rid": "sssss",
},
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
self.assertEqual(response.json()["id"], "sssss")
class TestEnableThinking(CustomTestCase):
"""Testcase: The test is to verify whether the functions of each parameter of the v1/completions interface are normal.
[Test Category] Interface
[Test Target] v1/completions
"""
@classmethod
def setUpClass(cls):
# Skip initialization, directly reuse global server
cls.model = LLAMA_3_2_1B_INSTRUCT_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.additional_chat_kwargs = {}
logging.basicConfig(level=logging.INFO) # Initialize logging
@classmethod
def tearDownClass(cls):
# Do not terminate server
pass
def test_model_parameters_model(self):
response = requests.post(
f"{self.base_url}/v1/completions",
json={"model": self.model, "prompt": "who are you?"},
)
logging.info(f"response.json:{response.json()}")
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
data = response.json()
self.assertEqual(data["model"], self.model)
def test_model_parameters_prompt(self):
# str format
response = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?"},
)
logging.info(f"response.json:{response.json()}")
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
# list[int] format
list_int = [1, 2, 3, 4]
response1 = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": list_int},
)
logging.info(f"response1.json:{response1.json()}")
self.assertEqual(response1.status_code, 200, f"Failed with: {response1.text}")
# list[str] format
list_str = ["who is you", "hello world", "ABChello"]
response2 = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": list_str},
)
logging.info(f"response2.json:{response2.json()}")
self.assertEqual(response2.status_code, 200, f"Failed with: {response2.text}")
# list[list[int]] format
list_list_int = [[14990], [1350, 445, 14990, 1879, 899], [14623, 525, 498, 30]]
response3 = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": list_list_int},
)
logging.info(f"response3.json:{response3.json()}")
self.assertEqual(response3.status_code, 200, f"Failed with: {response3.text}")
def test_model_parameters_max_tokens(self):
response = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "max_tokens": 1},
)
logging.info(f"response.json:{response.json()}")
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
logging.info(f"finish_reason:{response.json()['choices'][0]['finish_reason']}")
self.assertEqual(response.json()["choices"][0]["finish_reason"], "length")
def test_model_parameters_stream(self):
response = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "stream": True},
)
logging.info(f"response.text:{response.text}")
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
has_text = False
logging.info("\n=== Stream With Reasoning ===")
for line in response.iter_lines():
if line:
line = line.decode("utf-8")
if line.startswith("data:") and not line.startswith("data: [DONE]"):
data = json.loads(line[6:])
if "choices" in data and len(data["choices"]) > 0:
if "text" in data["choices"][0]:
has_text = True
self.assertTrue(has_text, "Text content not included in stream response")
def test_model_parameters_temperature(self):
response = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "temperature": 0},
)
logging.info(f"response.json:{response.json()}")
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
response1 = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "temperature": 0},
)
logging.info(f"response1.json:{response1.json()}")
self.assertEqual(response1.status_code, 200, f"Failed with: {response1.text}")
self.assertEqual(
response.json()["choices"][0]["text"],
response1.json()["choices"][0]["text"],
)
response2 = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "temperature": 2},
)
logging.info(f"response2.json:{response2.json()}")
self.assertEqual(response2.status_code, 200, f"Failed with: {response2.text}")
response3 = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "temperature": 2},
)
logging.info(f"response3.json:{response3.json()}")
self.assertEqual(response3.status_code, 200, f"Failed with: {response3.text}")
self.assertNotEqual(
response2.json()["choices"][0]["text"],
response3.json()["choices"][0]["text"],
)
def test_model_parameters_hidden_states(self):
response = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "return_hidden_states": True},
)
logging.info(f"response.json:{response.json()}")
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
self.assertIn("hidden_states", response.json()["choices"][0])
def test_model_parameters_top_k(self):
response = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "top_k": 20},
)
logging.info(f"response.json:{response.json()}")
logging.info(f"response.text:{response.text}")
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
response1 = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "top_k": 20},
)
logging.info(f"response1.json:{response1.json()}")
logging.info(f"response1.text:{response1.text}")
self.assertEqual(response1.status_code, 200, f"Failed with: {response1.text}")
self.assertNotEqual(
response.json()["choices"][0]["text"],
response1.json()["choices"][0]["text"],
)
def test_model_parameters_stop_token_ids(self):
list_ids = [13]
response = requests.post(
f"{self.base_url}/v1/completions",
json={
"prompt": "who are you?",
"stop_token_ids": list_ids,
"max_tokens": 1024,
},
)
logging.info(f"response.json:{response.json()}")
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
self.assertEqual(response.json()["choices"][0]["matched_stop"], 13)
def test_model_parameters_rid(self):
response = requests.post(
f"{self.base_url}/v1/completions",
json={"prompt": "who are you?", "rid": "10086"},
)
logging.info(f"response.json:{response.json()}")
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
self.assertEqual(response.json()["id"], "10086")
class TestStartProfile(CustomTestCase):
"""Testcase: Verify the correctness of /start_profile API with different parameter combinations (start_step/num_steps) on Ascend NPU backend.
[Test Category] Interface
[Test Target] /start_profile
"""
@classmethod
def setUpClass(cls):
# Skip initialization, reuse global server + configure profiler directory
envs.SGLANG_TORCH_PROFILER_DIR.set(OUTPUT_DIR)
cls.model = LLAMA_3_2_1B_INSTRUCT_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.additional_chat_kwargs = {}
@classmethod
def tearDownClass(cls):
# Terminate server in last class
global GLOBAL_SERVER_PROCESS
if GLOBAL_SERVER_PROCESS:
kill_process_tree(GLOBAL_SERVER_PROCESS.pid)
GLOBAL_SERVER_PROCESS = None
def setUp(self):
self._clear_profile_dir()
def test_start_profile_1(self):
self._start_profile(start_step="15", num_steps=5)
self._post_request()
self._check_non_empty_profile_dir()
def test_start_profile_2(self):
self._clear_profile_dir()
self._check_empty_profile_dir()
self._start_profile()
self._post_request()
requests.post(f"{self.base_url}/stop_profile")
self._check_non_empty_profile_dir()
def test_start_profile_3(self):
self._start_profile(num_steps=5)
self._post_request()
self._check_non_empty_profile_dir()
def _start_profile(self, **kwargs):
response = requests.post(
f"{self.base_url}/start_profile",
json=kwargs if kwargs else None,
)
self.assertEqual(response.status_code, 200)
return response
def _post_request(self):
response = requests.post(
f"{self.base_url}/generate",
json={
"text": "The capital of France is",
"sampling_params": {
"temperature": 0,
"max_new_tokens": 32,
},
},
)
self.assertEqual(response.status_code, 200)
def _clear_profile_dir(self):
if os.path.isdir(OUTPUT_DIR):
shutil.rmtree(OUTPUT_DIR)
def _check_non_empty_profile_dir(self):
self.assertTrue(os.path.isdir(OUTPUT_DIR), "Profiler directory does not exist")
self.assertNotEqual(
len(os.listdir(OUTPUT_DIR)), 0, "Profiler directory is empty"
)
def _check_empty_profile_dir(self):
if os.path.isdir(OUTPUT_DIR):
self.assertEqual(
len(os.listdir(OUTPUT_DIR)), 0, "Profiler directory is not empty"
)
if __name__ == "__main__":
unittest.main()
test/registered/ascend/interface/test_npu_api_abort_request.py
+78
View File
@@ -0,0 +1,78 @@
import threading
import time
import unittest
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import LLAMA_3_2_1B_INSTRUCT_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,
)
responses = []
def send_requests(url, **kwargs):
response = requests.post(DEFAULT_URL_FOR_TEST + url, json=kwargs)
responses.append(response)
register_npu_ci(est_time=400, suite="nightly-1-npu-a3", nightly=True)
class TestNpuApi(CustomTestCase):
"""Testcase: Verify the functionality of /abort_request API to terminate a running /generate request on Ascend backend.
[Test Category] Interface
[Test Target] /abort_request
"""
@classmethod
def setUpClass(cls):
cls.model = LLAMA_3_2_1B_INSTRUCT_WEIGHTS_PATH
other_args = [
"--attention-backend",
"ascend",
]
cls.process = popen_launch_server(
cls.model,
DEFAULT_URL_FOR_TEST,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_api_abort_request(self):
# Create thread 1: Send a long-running /generate request with rid=10086
thread1 = threading.Thread(
target=send_requests,
args=("/generate",),
kwargs={
"rid": "10086",
"text": "who are you?",
"sampling_params": {"temperature": 0.0, "max_new_tokens": 1024},
},
)
# Create thread 2: Send an /abort_request to terminate the request with rid=10086
thread2 = threading.Thread(
target=send_requests, args=("/abort_request",), kwargs={"rid": "10086"}
)
thread1.start()
time.sleep(0.5)
thread2.start()
thread1.join()
thread2.join()
print(responses[1].text)
if __name__ == "__main__":
unittest.main()
test/registered/ascend/interface/test_npu_api_encode.py
+134
View File
@@ -0,0 +1,134 @@
import logging
import unittest
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import QWEN3_VL_4B_INSTRUCT_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,
)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[logging.StreamHandler()],
)
logger = logging.getLogger(__name__)
register_npu_ci(est_time=400, suite="nightly-1-npu-a3", nightly=True)
class TestNpuApi(CustomTestCase):
"""Testcase: Verify the availability and correctness of the /encode API on Ascend backend with GME_QWEN2_VL_2B_INSTRUCT model.
[Test Category] Interface
[Test Target] /encode
"""
@classmethod
def setUpClass(cls):
cls.model = QWEN3_VL_4B_INSTRUCT_WEIGHTS_PATH
other_args = [
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--tp-size",
2,
"--is-embedding",
]
cls.process = popen_launch_server(
cls.model,
DEFAULT_URL_FOR_TEST,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_api_encode_01(self):
# Test Scenario 1: Call /encode API with plain text parameter
response = requests.post(
f"{DEFAULT_URL_FOR_TEST}/encode",
json={
"rid": "2",
"text": "what is the capital of France",
"sampling_params": {
"temperature": 0,
"max_new_tokens": 200,
"top_p": 1,
},
},
)
logger.info("Test 01 response keys: %s", response.json().keys())
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["meta_info"]["id"], "2")
def test_api_encode_02(self):
# Test Scenario 2: Call /encode API with input_ids parameter
response = requests.post(
f"{DEFAULT_URL_FOR_TEST}/encode",
json={
"rid": "3",
"input_ids": [101, 7592, 2088, 102],
"sampling_params": {"temperature": 0, "max_new_tokens": 200},
},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["meta_info"]["id"], "3")
def test_api_encode_03(self):
# Test Scenario 3: Call /encode API with text and image parameters (multimodal capability verification)
response = requests.post(
f"{DEFAULT_URL_FOR_TEST}/encode",
json={
"rid": "4",
"text": "show me the words",
"image_data": "https://miaobi-lite.bj.bcebos.com/miaobi/5mao/b%27b2Ny6K%2BG5Yir5Luj56CBXzE3MzQ2MzcyNjAuMzgxNDk5NQ%3D%3D%27/0.png",
"sampling_params": {"temperature": 0, "max_new_tokens": 200},
},
)
logger.info("Test 03 response keys: %s", response.json().keys())
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()["meta_info"]["id"], "4")
def test_api_encode_04(self):
# Test Scenario 4: Call /encode API with list of rids (multiple requests) - text input
request_rids = ["5", "6", "7"]
response = requests.post(
f"{DEFAULT_URL_FOR_TEST}/encode",
json={
"rid": request_rids,
"text": [
"what is the capital of UK",
"what is the capital of Germany",
"what is the capital of Japan",
],
"sampling_params": {
"temperature": 0,
"max_new_tokens": 200,
"top_p": 1,
},
},
)
response_json = response.json()
logger.info(
"Test 04 response type: %s, first item meta_info: %s",
type(response_json),
response_json[0].get("meta_info", {}),
)
self.assertEqual(response.status_code, 200)
self.assertEqual(len(response_json), len(request_rids))
for idx, result in enumerate(response_json):
self.assertEqual(result["meta_info"]["id"], request_rids[idx])
if __name__ == "__main__":
unittest.main()
test/registered/ascend/interface/test_npu_enable_thinking.py
+194
View File
@@ -0,0 +1,194 @@
import json
import unittest
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import QWEN3_30B_A3B_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="nightly-2-npu-a3",
nightly=True,
disabled="https://github.com/Ascend/sglang/issues/32",
)
class TestEnableThinking(CustomTestCase):
"""Testcase: Testing with the 'enable_thinking' feature enabled/disabled,
both streaming and non-streaming input requests successful
[Test Category] Interface
[Test Target] /v1/chat/completions
"""
@classmethod
def setUpClass(cls):
cls.model = QWEN3_30B_A3B_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.other_args = [
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--mem-fraction-static",
0.95,
"--tp",
16,
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=cls.other_args,
)
cls.additional_chat_kwargs = {}
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_chat_completion_with_reasoning(self):
# Test non-streaming with "enable_thinking": True, reasoning_content should not be empty
client = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
"temperature": 0,
"separate_reasoning": True,
"chat_template_kwargs": {"enable_thinking": True},
**self.additional_chat_kwargs,
},
)
self.assertEqual(client.status_code, 200, f"Failed with: {client.text}")
data = client.json()
self.assertIn("choices", data)
self.assertTrue(len(data["choices"]) > 0)
self.assertIn("message", data["choices"][0])
self.assertIn("reasoning_content", data["choices"][0]["message"])
self.assertIsNotNone(data["choices"][0]["message"]["reasoning_content"])
def test_chat_completion_without_reasoning(self):
# Test non-streaming with "enable_thinking": False, reasoning_content should be empty
client = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
"temperature": 0,
"separate_reasoning": True,
"chat_template_kwargs": {"enable_thinking": False},
**self.additional_chat_kwargs,
},
)
self.assertEqual(client.status_code, 200, f"Failed with: {client.text}")
data = client.json()
self.assertIn("choices", data)
self.assertTrue(len(data["choices"]) > 0)
self.assertIn("message", data["choices"][0])
if "reasoning_content" in data["choices"][0]["message"]:
self.assertIsNone(data["choices"][0]["message"]["reasoning_content"])
def test_stream_chat_completion_with_reasoning(self):
# Test streaming with "enable_thinking": True, reasoning_content should not be empty
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
"temperature": 0,
"separate_reasoning": True,
"stream": True,
"chat_template_kwargs": {"enable_thinking": True},
**self.additional_chat_kwargs,
},
stream=True,
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
has_reasoning = False
has_content = False
print("\n=== Stream With Reasoning ===")
for line in response.iter_lines():
if line:
line = line.decode("utf-8")
if line.startswith("data:") and not line.startswith("data: [DONE]"):
data = json.loads(line[6:])
if "choices" in data and len(data["choices"]) > 0:
delta = data["choices"][0].get("delta", {})
if "reasoning_content" in delta and delta["reasoning_content"]:
has_reasoning = True
if "content" in delta and delta["content"]:
has_content = True
self.assertTrue(
has_reasoning,
"The reasoning content is not included in the stream response",
)
self.assertTrue(
has_content, "The stream response does not contain normal content"
)
def test_stream_chat_completion_without_reasoning(self):
# Test streaming with "enable_thinking": False, reasoning_content should be empty
response = requests.post(
f"{self.base_url}/v1/chat/completions",
json={
"model": self.model,
"messages": [{"role": "user", "content": "Hello"}],
"temperature": 0,
"separate_reasoning": True,
"stream": True,
"chat_template_kwargs": {"enable_thinking": False},
**self.additional_chat_kwargs,
},
stream=True,
)
self.assertEqual(response.status_code, 200, f"Failed with: {response.text}")
has_reasoning = False
has_content = False
print("\n=== Stream Without Reasoning ===")
for line in response.iter_lines():
if line:
line = line.decode("utf-8")
if line.startswith("data:") and not line.startswith("data: [DONE]"):
data = json.loads(line[6:])
if "choices" in data and len(data["choices"]) > 0:
delta = data["choices"][0].get("delta", {})
if "reasoning_content" in delta and delta["reasoning_content"]:
has_reasoning = True
if "content" in delta and delta["content"]:
has_content = True
self.assertFalse(
has_reasoning,
"The reasoning content should not be included in the stream response",
)
self.assertTrue(
has_content, "The stream response does not contain normal content"
)
if __name__ == "__main__":
unittest.main()
test/registered/ascend/interface/test_npu_matched_stop.py
+163
View File
@@ -0,0 +1,163 @@
import json
import unittest
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import LLAMA_3_1_8B_INSTRUCT_WEIGHTS_PATH
from sglang.test.ci.ci_register import register_npu_ci
from sglang.test.test_utils import (
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
MANY_NEW_TOKENS_PROMPT = """
Please write an extremely detailed and vivid fantasy story, set in a world full of intricate magic systems, political intrigue, and complex characters.
Ensure that you thoroughly describe every scene, character's motivations, and the environment. Include long, engaging dialogues and elaborate on the inner thoughts of the characters.
Each section should be as comprehensive as possible to create a rich and immersive experience for the reader.
The story should span multiple events, challenges, and character developments over time. Aim to make the story at least 3,000 words long.
"""
register_npu_ci(est_time=400, suite="nightly-1-npu-a3", nightly=True)
class TestMatchedStop(CustomTestCase):
"""Testcase: Test configuring 'matched_stop' to different values(string, EOS token, length) correctly identifies
it as a stop signal.
[Test Category] Interface
[Test Target] /v1/chat/completions; /v1/completions
"""
@classmethod
def setUpClass(cls):
cls.model = LLAMA_3_1_8B_INSTRUCT_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.other_args = [
"--max-running-requests",
10,
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--mem-fraction-static",
0.8,
]
cls.process = popen_launch_server(
cls.model, cls.base_url, timeout=300, other_args=cls.other_args
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def run_completions_generation(
self,
prompt=MANY_NEW_TOKENS_PROMPT,
max_tokens=1,
stop=None,
finish_reason=None,
matched_stop=None,
):
# Configure matched_stop to None, and use the '/v1/completions' interface
# verify that the actual termination reason matches the configured value.
payload = {
"prompt": prompt,
"model": self.model,
"temperature": 0,
"top_p": 1,
"max_tokens": max_tokens,
}
if stop is not None:
payload["stop"] = stop
response_completions = requests.post(
self.base_url + "/v1/completions",
json=payload,
)
print(json.dumps(response_completions.json()))
print("=" * 100)
assert (
response_completions.json()["choices"][0]["finish_reason"] == finish_reason
)
assert response_completions.json()["choices"][0]["matched_stop"] == matched_stop
def run_chat_completions_generation(
self,
prompt=MANY_NEW_TOKENS_PROMPT,
max_tokens=1,
stop=None,
finish_reason=None,
matched_stop=None,
):
# Configure matched_stop to None, and use the '/v1/chat/completions' interface
# verify that the actual termination reason matches the configured value.
chat_payload = {
"model": self.model,
"messages": [
{"role": "system", "content": "You are a helpful AI assistant"},
{"role": "user", "content": prompt},
],
"temperature": 0,
"top_p": 1,
"max_tokens": max_tokens,
}
if stop is not None:
chat_payload["stop"] = stop
response_chat = requests.post(
self.base_url + "/v1/chat/completions",
json=chat_payload,
)
print(json.dumps(response_chat.json()))
print("=" * 100)
assert response_chat.json()["choices"][0]["finish_reason"] == finish_reason
assert response_chat.json()["choices"][0]["matched_stop"] == matched_stop
def test_finish_stop_str(self):
# Setting finish_reason="stop",'matched_stop="\n"' allows for correct termination
self.run_completions_generation(
max_tokens=1000, stop="\n", finish_reason="stop", matched_stop="\n"
)
self.run_chat_completions_generation(
max_tokens=1000, stop="\n", finish_reason="stop", matched_stop="\n"
)
def test_finish_stop_eos(self):
# Setting matched_stop is a specific EOS end flagallows for correct identification and termination of signal
llama_format_prompt = """
<|begin_of_text|><|start_header_id|>system<|end_header_id|>
You are a helpful assistant.<|eot_id|><|start_header_id|>user<|end_header_id|>
What is 2 + 2?<|eot_id|><|start_header_id|>assistant<|end_header_id|>
"""
eos_token_id = 128009
self.run_completions_generation(
prompt=llama_format_prompt,
max_tokens=1000,
finish_reason="stop",
matched_stop=eos_token_id,
)
self.run_chat_completions_generation(
prompt="What is 2 + 2?",
max_tokens=1000,
finish_reason="stop",
matched_stop=eos_token_id,
)
def test_finish_length(self):
# Setting finish_reason="length",'matched_stop="\n"' allows for correct termination
self.run_completions_generation(
max_tokens=5, finish_reason="length", matched_stop=None
)
self.run_chat_completions_generation(
max_tokens=5, finish_reason="length", matched_stop=None
)
if __name__ == "__main__":
unittest.main()
test/registered/ascend/interface/test_npu_openai_function_calling.py
+943
View File
@@ -0,0 +1,943 @@
import json
import unittest
import openai
from sglang.srt.utils import kill_process_tree
from sglang.srt.utils.hf_transformers_utils import get_tokenizer
from sglang.test.ascend.test_ascend_utils import LLAMA_3_2_1B_INSTRUCT_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="nightly-1-npu-a3",
nightly=True,
disabled="https://github.com/Ascend/sglang/issues/39",
)
class TestOpenAIServerFunctionCalling(CustomTestCase):
"""Testcase：Verify the correctness of full-scenario OpenAI-style function calling with llama3 parser for Llama-3.2-1B-Instruct model.
Cover: Single/multi-turn calls, streaming/non-streaming returns, multi-parameter verification of tool_choice, and JSON parsing validity of function parameters.
[Test Category] Interface
[Test Target] /v1/chat/completions
"""
# NOTE: this system_message is for Llama3.2 system prompt. Without this,
# sometimes Llama3.2 gives a different tool call format such as:
# '<|python_tag|>{"type": "function", "function": "add", "parameters": {"a": "3", "b": "5"}}'
SYSTEM_MESSAGE = (
"You are a helpful assistant with tool calling capabilities. "
"Only reply with a tool call if the function exists in the library provided by the user. "
"If it doesn't exist, just reply directly in natural language. "
"When you receive a tool call response, use the output to format an answer to the original user question. "
"You have access to the following functions. "
"To call a function, please respond with JSON for a function call. "
'Respond in the format {"name": function name, "parameters": dictionary of argument name and its value}. '
"Do not use variables.\n\n"
)
@classmethod
def setUpClass(cls):
# Replace with the model name needed for testing
cls.model = LLAMA_3_2_1B_INSTRUCT_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.api_key = "sk-123456"
# Start the local OpenAI Server. If necessary, you can add other parameters such as --enable-tools.
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
api_key=cls.api_key,
other_args=[
# If your server needs extra parameters to test function calling, please add them here.
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--tool-call-parser",
"llama3",
],
)
cls.base_url += "/v1"
cls.tokenizer = get_tokenizer(cls.model)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_function_calling_format(self):
"""
Test: Whether the function call format returned by the AI is correct.
When returning a tool call, message.content should be None, and tool_calls should be a list.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
tools = [
{
"type": "function",
"function": {
"name": "add",
"description": "Compute the sum of two numbers",
"parameters": {
"type": "object",
"properties": {
"a": {
"type": "integer",
"description": "A number",
},
"b": {
"type": "integer",
"description": "A number",
},
},
"required": ["a", "b"],
},
},
}
]
messages = [
{"role": "system", "content": self.SYSTEM_MESSAGE},
{"role": "user", "content": "Compute (3+5)"},
]
response = client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=messages,
temperature=0.8,
top_p=0.8,
stream=False,
tools=tools,
)
tool_calls = response.choices[0].message.tool_calls
assert (
isinstance(tool_calls, list) and len(tool_calls) > 0
), "tool_calls should be a non-empty list"
function_name = tool_calls[0].function.name
assert function_name == "add", "Function name should be 'add'"
# This unit test is too difficult for default model. Mark it as optional unit tests so it won't trigger unless specified.
def _test_function_calling_multiturn(self):
"""
Test: Whether the function call format returned by the AI is correct.
When returning a tool call, message.content should be None, and tool_calls should be a list.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
tools = [
{
"type": "function",
"function": {
"name": "add",
"description": "Compute the sum of two numbers",
"parameters": {
"type": "object",
"properties": {
"a": {
"type": "integer",
"description": "A number",
},
"b": {
"type": "integer",
"description": "A number",
},
},
"required": ["a", "b"],
},
},
}
]
messages = [{"role": "user", "content": "Compute (3+5)"}]
response = client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=messages,
temperature=0.8,
top_p=0.8,
stream=False,
tools=tools,
)
tool_call = response.choices[0].message.tool_calls[0]
function_name = tool_call.function.name
assert function_name == "add", "Function name should be 'add'"
function_arguments = json.loads(tool_call.function.arguments)
assert function_arguments in [
{"a": 3, "b": 5},
{"a": "3", "b": "5"},
], f"Unexpected function arguments: {function_arguments}"
messages.append(response.choices[0].message)
messages.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": "8",
"name": function_name,
}
)
final_response = client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=messages,
temperature=0.8,
top_p=0.8,
stream=False,
tools=tools,
)
assert (
"8" in final_response.choices[0].message.content
), "tool_call response should have the sum 8 in the content"
def test_function_calling_streaming_simple(self):
"""
Test: Whether the function name can be correctly recognized in streaming mode.
- Expect a function call to be found, and the function name to be correct.
- Verify that streaming mode returns at least multiple chunks.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The city to find the weather for",
},
"unit": {
"type": "string",
"description": "Weather unit (celsius or fahrenheit)",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["city", "unit"],
},
},
}
]
messages = [
{"role": "system", "content": self.SYSTEM_MESSAGE},
{
"role": "user",
"content": "What is the temperature in Paris in celsius??",
},
]
response_stream = client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=messages,
temperature=0.8,
top_p=0.8,
stream=True,
tools=tools,
)
chunks = list(response_stream)
self.assertTrue(len(chunks) > 0, "Streaming should return at least one chunk")
found_function_name = False
for chunk in chunks:
choice = chunk.choices[0]
# Check whether the current chunk contains tool_calls
if choice.delta.tool_calls:
tool_call = choice.delta.tool_calls[0]
if tool_call.function.name:
self.assertEqual(
tool_call.function.name,
"get_current_weather",
"Function name should be 'get_current_weather'",
)
found_function_name = True
break
self.assertTrue(
found_function_name,
"Target function name 'get_current_weather' was not found in the streaming chunks",
)
finish_reason = chunks[-1].choices[0].finish_reason
self.assertEqual(
finish_reason,
"tool_calls",
"Final response of function calling should have finish_reason 'tool_calls'",
)
def test_function_calling_streaming_args_parsing(self):
"""
Test: Whether the function call arguments returned in streaming mode can be correctly concatenated into valid JSON.
- The user request requires multiple parameters.
- AI may return the arguments in chunks that need to be concatenated.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
tools = [
{
"type": "function",
"function": {
"name": "add",
"description": "Compute the sum of two integers",
"parameters": {
"type": "object",
"properties": {
"a": {
"type": "integer",
"description": "First integer",
},
"b": {
"type": "integer",
"description": "Second integer",
},
},
"required": ["a", "b"],
},
"strict": True, # Llama-3.2-1B is flaky in tool call. It won't always respond with parameters unless we set strict.
},
}
]
messages = [
{"role": "system", "content": self.SYSTEM_MESSAGE},
{"role": "user", "content": "Please sum 5 and 7, just call the function."},
]
response_stream = client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=messages,
temperature=0.9,
top_p=0.9,
stream=True,
tools=tools,
)
argument_fragments = []
chunks = list(response_stream)
function_name = None
for chunk in chunks:
choice = chunk.choices[0]
if choice.delta.tool_calls:
tool_call = choice.delta.tool_calls[0]
# Record the function name on first occurrence
function_name = tool_call.function.name or function_name
# In case of multiple chunks, JSON fragments may need to be concatenated
if tool_call.function.arguments is not None:
argument_fragments.append(tool_call.function.arguments)
self.assertEqual(function_name, "add", "Function name should be 'add'")
joined_args = "".join(argument_fragments)
self.assertTrue(
len(joined_args) > 0,
"No parameter fragments were returned in the function call",
)
finish_reason = chunks[-1].choices[0].finish_reason
self.assertEqual(
finish_reason,
"tool_calls",
"Final response of function calling should have finish_reason 'tool_calls'",
)
# Check whether the concatenated JSON is valid
try:
args_obj = json.loads(joined_args)
except json.JSONDecodeError:
self.fail(
"The concatenated tool call arguments are not valid JSON, parsing failed"
)
self.assertIn("a", args_obj, "Missing parameter 'a'")
self.assertIn("b", args_obj, "Missing parameter 'b'")
self.assertEqual(str(args_obj["a"]), "5", "Parameter a should be 5")
self.assertEqual(str(args_obj["b"]), "7", "Parameter b should be 7")
def test_function_call_strict(self):
"""
Test: Whether the strict mode of function calling works as expected.
- When strict mode is enabled, the AI should not return a function call if the function name is not recognized.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
tools = [
{
"type": "function",
"function": {
"name": "sub",
"description": "Compute the difference of two integers",
"parameters": {
"type": "object",
"properties": {
"int_a": {
"type": "integer",
"description": "First integer",
},
"int_b": {
"type": "integer",
"description": "Second integer",
},
},
"required": ["int_a", "int_b"],
},
"strict": True,
},
}
]
messages = [
{"role": "user", "content": "Please compute 5 - 7, using your tool."}
]
response = client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=messages,
temperature=0.8,
top_p=0.8,
stream=False,
tools=tools,
)
tool_calls = response.choices[0].message.tool_calls
function_name = tool_calls[0].function.name
arguments = tool_calls[0].function.arguments
args_obj = json.loads(arguments)
self.assertEqual(function_name, "sub", "Function name should be 'sub'")
self.assertEqual(str(args_obj["int_a"]), "5", "Parameter int_a should be 5")
self.assertEqual(str(args_obj["int_b"]), "7", "Parameter int_b should be 7")
def test_function_call_required(self):
"""
Test: Whether tool_choice: "required" works as expected
- When tool_choice == "required", the model should return one or more tool_calls.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
tools = [
{
"type": "function",
"function": {
"name": "sub",
"description": "Compute the difference of two integers",
"parameters": {
"type": "object",
"properties": {
"int_a": {
"type": "integer",
"description": "First integer",
},
"int_b": {
"type": "integer",
"description": "Second integer",
},
},
"required": ["int_a", "int_b"],
},
"strict": True,
},
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "use this to get latest weather information for a city given its name",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "name of the city to get weather for",
}
},
"required": ["city"],
},
},
},
]
messages = [{"role": "user", "content": "What is the capital of France?"}]
response = client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=messages,
temperature=0.8,
top_p=0.8,
stream=False,
tools=tools,
tool_choice="required",
)
tool_calls = response.choices[0].message.tool_calls
self.assertIsNotNone(tool_calls, "No tool_calls in the response")
function_name = tool_calls[0].function.name
arguments = tool_calls[0].function.arguments
args_obj = json.loads(arguments)
self.assertEqual(
function_name,
"get_weather",
f"Function name should be 'get_weather', got: {function_name}",
)
self.assertIn(
"city", args_obj, f"Function arguments should have 'city', got: {args_obj}"
)
# Make the test more robust by checking type and accepting valid responses
city_value = args_obj["city"]
self.assertIsInstance(
city_value,
str,
f"Parameter city should be a string, got: {type(city_value)}",
)
self.assertTrue(
"Paris" in city_value or "France" in city_value,
f"Parameter city should contain either 'Paris' or 'France', got: {city_value}",
)
def test_function_call_specific(self):
"""
Test: Whether tool_choice: ToolChoice works as expected
- When tool_choice is a specific ToolChoice, the model should return one or more tool_calls.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
tools = [
{
"type": "function",
"function": {
"name": "sub",
"description": "Compute the difference of two integers",
"parameters": {
"type": "object",
"properties": {
"int_a": {
"type": "integer",
"description": "First integer",
},
"int_b": {
"type": "integer",
"description": "Second integer",
},
},
"required": ["int_a", "int_b"],
},
"strict": True,
},
},
{
"type": "function",
"function": {
"name": "get_weather",
"description": "use this to get latest weather information for a city given its name",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "name of the city to get weather for",
}
},
"required": ["city"],
},
},
},
]
messages = [{"role": "user", "content": "What is the capital of France?"}]
response = client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=messages,
temperature=0.8,
top_p=0.8,
stream=False,
tools=tools,
tool_choice={"type": "function", "function": {"name": "get_weather"}},
)
tool_calls = response.choices[0].message.tool_calls
self.assertIsNotNone(tool_calls, "No tool_calls in the response")
function_name = tool_calls[0].function.name
arguments = tool_calls[0].function.arguments
args_obj = json.loads(arguments)
self.assertEqual(
function_name, "get_weather", "Function name should be 'get_weather'"
)
self.assertIn("city", args_obj, "Function arguments should have 'city'")
def test_streaming_multiple_choices_finish_reason(self):
"""
Test: Verify that each choice gets its own finish_reason chunk in streaming mode with n > 1.
This tests the fix for the bug where only the last index got a finish_reason chunk.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city and state, e.g. San Francisco, CA",
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["location"],
},
},
}
]
messages = [
{"role": "user", "content": "What is the weather like in Los Angeles?"}
]
# Request with n=2 to get multiple choices
response_stream = client.chat.completions.create(
model=self.model,
messages=messages,
max_tokens=2048,
temperature=0.8,
stream=True,
tools=tools,
tool_choice="required", # Force tool calls
n=2, # Multiple choices
)
chunks = list(response_stream)
# Track finish_reason chunks for each index
finish_reason_chunks = {}
for chunk in chunks:
if chunk.choices:
for choice in chunk.choices:
if choice.finish_reason is not None:
index = choice.index
if index not in finish_reason_chunks:
finish_reason_chunks[index] = []
finish_reason_chunks[index].append(choice.finish_reason)
# Verify we got finish_reason chunks for both indices
self.assertEqual(
len(finish_reason_chunks),
2,
f"Expected finish_reason chunks for 2 indices, got {len(finish_reason_chunks)}",
)
# Verify both index 0 and 1 have finish_reason
self.assertIn(
0, finish_reason_chunks, "Missing finish_reason chunk for index 0"
)
self.assertIn(
1, finish_reason_chunks, "Missing finish_reason chunk for index 1"
)
# Verify the finish_reason is "tool_calls" since we forced tool calls
for index, reasons in finish_reason_chunks.items():
self.assertEqual(
reasons[-1], # Last finish_reason for this index
"tool_calls",
f"Expected finish_reason 'tool_calls' for index {index}, got {reasons[-1]}",
)
def test_function_calling_streaming_no_tool_call(self):
"""
Test: Whether the finish_reason is stop in streaming mode when no tool call is given.
- Expect no function call to be found.
- Verify that finish_reason is stop
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
tools = [
{
"type": "function",
"function": {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The city to find the weather for",
},
"unit": {
"type": "string",
"description": "Weather unit (celsius or fahrenheit)",
"enum": ["celsius", "fahrenheit"],
},
},
"required": ["city", "unit"],
},
},
}
]
messages = [{"role": "user", "content": "Who are you?"}]
response_stream = client.chat.completions.create(
model=self.model,
max_tokens=2048,
messages=messages,
temperature=0.8,
top_p=0.8,
stream=True,
tools=tools,
tool_choice="none",
)
chunks = list(response_stream)
self.assertTrue(len(chunks) > 0, "Streaming should return at least one chunk")
found_tool_call = False
for chunk in chunks:
choice = chunk.choices[0]
# Check whether the current chunk contains tool_calls
found_tool_call = choice.delta.tool_calls is not None
self.assertFalse(
found_tool_call,
"Shouldn't have any tool_call in the streaming chunks",
)
finish_reason = chunks[-1].choices[0].finish_reason
self.assertEqual(
finish_reason,
"stop",
"Final response of no function calling should have finish_reason 'stop'",
)
def test_streaming_multiple_choices_without_tools(self):
"""
Test: Verify that each choice gets its own finish_reason chunk without tool calls.
This tests the fix for regular content streaming with multiple choices.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
messages = [{"role": "user", "content": "Say hello in one word."}]
# Request with n=2 to get multiple choices, no tools
response_stream = client.chat.completions.create(
model=self.model,
messages=messages,
temperature=0.8,
stream=True,
max_tokens=10, # Keep it short
n=2, # Multiple choices
)
chunks = list(response_stream)
# Track finish_reason chunks for each index
finish_reason_chunks = {}
for chunk in chunks:
if chunk.choices:
for choice in chunk.choices:
if choice.finish_reason is not None:
index = choice.index
if index not in finish_reason_chunks:
finish_reason_chunks[index] = []
finish_reason_chunks[index].append(choice.finish_reason)
# Verify we got finish_reason chunks for both indices
self.assertEqual(
len(finish_reason_chunks),
2,
f"Expected finish_reason chunks for 2 indices, got {len(finish_reason_chunks)}",
)
# Verify both index 0 and 1 have finish_reason
self.assertIn(
0, finish_reason_chunks, "Missing finish_reason chunk for index 0"
)
self.assertIn(
1, finish_reason_chunks, "Missing finish_reason chunk for index 1"
)
# Verify the finish_reason is "stop" (regular completion)
for index, reasons in finish_reason_chunks.items():
self.assertIn(
reasons[-1],
["stop", "length"], # Could be either depending on how model responds
f"Expected finish_reason 'stop' or 'length' for index {index}, got {reasons[-1]}",
)
class TestOpenAIPythonicFunctionCalling(CustomTestCase):
"""Testcase：Verify the functionality of Python-style list-format function calling with pythonic parser for Llama-3.2-1B-Instruct model on Ascend NPU backend.
Cover: Explicit format prompt verification, streaming call index integrity, and return validity of parallel tool calls.
[Test Category] Interface
[Test Target] /v1/chat/completions
"""
PYTHONIC_TOOLS = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather for a given location.",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The name of the city or location.",
}
},
"required": ["location"],
},
},
},
{
"type": "function",
"function": {
"name": "get_tourist_attractions",
"description": "Get a list of top tourist attractions for a given city.",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "The name of the city to find attractions for.",
}
},
"required": ["city"],
},
},
},
]
PYTHONIC_MESSAGES = [
{
"role": "system",
"content": (
"You are a travel assistant. "
"When asked to call functions, ALWAYS respond ONLY with a python list of function calls, "
"using this format: [func_name1(param1=value1, param2=value2), func_name2(param=value)]. "
"Do NOT use JSON, do NOT use variables, do NOT use any other format. "
"Here is an example:\n"
'[get_weather(location="Paris"), get_tourist_attractions(city="Paris")]'
),
},
{
"role": "user",
"content": (
"I'm planning a trip to Tokyo next week. What's the weather like and what are some top tourist attractions? "
"Propose parallel tool calls at once, using the python list of function calls format as shown above."
),
},
]
@classmethod
def setUpClass(cls):
cls.model = LLAMA_3_2_1B_INSTRUCT_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.api_key = "sk-123456"
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
api_key=cls.api_key,
other_args=[
"--attention-backend",
"ascend",
"--disable-cuda-graph",
"--tool-call-parser",
"pythonic",
],
)
cls.base_url += "/v1"
cls.tokenizer = get_tokenizer(cls.model)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_pythonic_tool_call_prompt(self):
"""
Test: Explicit prompt for pythonic tool call format without chat template.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
response = client.chat.completions.create(
model=self.model,
messages=self.PYTHONIC_MESSAGES,
tools=self.PYTHONIC_TOOLS,
temperature=0.1,
stream=False,
)
tool_calls = response.choices[0].message.tool_calls
self.assertIsInstance(tool_calls, list, "No tool_calls found")
self.assertGreaterEqual(len(tool_calls), 1)
names = [tc.function.name for tc in tool_calls]
self.assertTrue(
"get_weather" in names or "get_tourist_attractions" in names,
f"Function name '{names}' should container either 'get_weather' or 'get_tourist_attractions'",
)
def test_pythonic_tool_call_streaming(self):
"""
Test: Streaming pythonic tool call format; assert tool_call index is present.
"""
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
response_stream = client.chat.completions.create(
model=self.model,
messages=self.PYTHONIC_MESSAGES,
tools=self.PYTHONIC_TOOLS,
temperature=0.1,
stream=True,
)
found_tool_calls = False
found_index = False
found_names = set()
for chunk in response_stream:
choice = chunk.choices[0]
if getattr(choice.delta, "tool_calls", None):
found_tool_calls = True
tool_call = choice.delta.tool_calls[0]
if hasattr(tool_call, "index") or (
isinstance(tool_call, dict) and "index" in tool_call
):
found_index = True
found_names.add(str(tool_call.function.name))
self.assertTrue(found_tool_calls, "No tool_calls found in streaming response")
self.assertTrue(found_index, "No index field found in any streamed tool_call")
self.assertTrue(
"get_weather" in found_names or "get_tourist_attractions" in found_names,
f"Function name '{found_names}' should container either 'get_weather' or 'get_tourist_attractions'",
)
if __name__ == "__main__":
unittest.main()
test/registered/ascend/interface/test_npu_openai_server_ignore_eos.py
+101
View File
@@ -0,0 +1,101 @@
import unittest
import openai
from sglang.srt.utils import kill_process_tree
from sglang.srt.utils.hf_transformers_utils import get_tokenizer
from sglang.test.ascend.test_ascend_utils import LLAMA_3_2_1B_INSTRUCT_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="nightly-2-npu-a3", nightly=True)
class TestOpenAIServerIgnoreEOS(CustomTestCase):
"""Testcase: Test 'ignore_eos' is True, the EOS is ignored and continue reasoning
[Test Category] Interface
[Test Target] ignore_eos
"""
@classmethod
def setUpClass(cls):
cls.model = LLAMA_3_2_1B_INSTRUCT_WEIGHTS_PATH
cls.base_url = DEFAULT_URL_FOR_TEST
cls.api_key = "sk-123456"
cls.other_args = [
"--attention-backend",
"ascend",
"--disable-cuda-graph",
]
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
api_key=cls.api_key,
other_args=cls.other_args,
)
cls.base_url += "/v1"
cls.tokenizer = get_tokenizer(cls.model)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_ignore_eos(self):
client = openai.Client(api_key=self.api_key, base_url=self.base_url)
max_tokens = 200
response_default = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Count from 1 to 20."},
],
temperature=0,
max_tokens=max_tokens,
extra_body={"ignore_eos": False},
)
response_ignore_eos = client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Count from 1 to 20."},
],
temperature=0,
max_tokens=max_tokens,
extra_body={"ignore_eos": True},
)
default_tokens = len(
self.tokenizer.encode(response_default.choices[0].message.content)
)
ignore_eos_tokens = len(
self.tokenizer.encode(response_ignore_eos.choices[0].message.content)
)
# Check if ignore_eos resulted in more tokens or exactly max_tokens
# The ignore_eos response should either:
# 1. Have more tokens than the default response (if default stopped at EOS before max_tokens)
# 2. Have exactly max_tokens (if it reached the max_tokens limit)
self.assertTrue(
ignore_eos_tokens > default_tokens or ignore_eos_tokens >= max_tokens,
f"ignore_eos did not generate more tokens: {ignore_eos_tokens} vs {default_tokens}",
)
self.assertEqual(
response_ignore_eos.choices[0].finish_reason,
"length",
f"Expected finish_reason='length' for ignore_eos=True, got {response_ignore_eos.choices[0].finish_reason}",
)
if __name__ == "__main__":
unittest.main()
test/registered/ascend/interface/test_npu_penalty.py
+111
View File
@@ -0,0 +1,111 @@
import json
import random
import unittest
from concurrent.futures import ThreadPoolExecutor
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ascend.test_ascend_utils import LLAMA_3_2_1B_INSTRUCT_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="nightly-1-npu-a3", nightly=True)
class TestPenalty(CustomTestCase):
"""Testcase：Verify successful processing of inference requests with three specific mechanisms(frequency_penalty, presence_penalty, min_new_tokens).
[Test Category] Interface
[Test Target] /generate
"""
@classmethod
def setUpClass(cls):
cls.model = LLAMA_3_2_1B_INSTRUCT_WEIGHTS_PATH
other_args = [
"--attention-backend",
"ascend",
"--disable-cuda-graph",
]
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=other_args,
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def run_decode(self, sampling_params):
# Send inference request with specified sampling/penalty parameters.
return_logprob = True
top_logprobs_num = 5
return_text = True
n = 1
response = requests.post(
self.base_url + "/generate",
json={
# prompt that is supposed to generate < 32 tokens
"text": "<|start_header_id|>user<|end_header_id|>\n\nWhat is the answer for 1 + 1 = ?<|eot_id|><|start_header_id|>assistant<|end_header_id|>\n\n",
"sampling_params": {
"max_new_tokens": 48,
"n": n,
**sampling_params,
},
"return_logprob": return_logprob,
"top_logprobs_num": top_logprobs_num,
"return_text_in_logprobs": return_text,
"logprob_start_len": 0,
},
)
self.assertEqual(response.status_code, 200)
print(json.dumps(response.json()))
print("=" * 100)
def test_default_values(self):
self.run_decode({})
def test_frequency_penalty(self):
self.run_decode({"frequency_penalty": 2})
def test_min_new_tokens(self):
self.run_decode({"min_new_tokens": 16})
def test_presence_penalty(self):
self.run_decode({"presence_penalty": 2})
def test_penalty_mixed(self):
args = [
{},
{},
{},
{"frequency_penalty": 2},
{"presence_penalty": 1},
{"min_new_tokens": 16},
{"frequency_penalty": 0.2},
{"presence_penalty": 0.4},
{"min_new_tokens": 8},
{"frequency_penalty": 0.4, "presence_penalty": 0.8},
{"frequency_penalty": 0.4, "min_new_tokens": 12},
{"presence_penalty": 0.8, "min_new_tokens": 12},
{"presence_penalty": -0.3, "frequency_penalty": 1.3, "min_new_tokens": 32},
{"presence_penalty": 0.3, "frequency_penalty": -1.3, "min_new_tokens": 32},
]
random.shuffle(args * 5)
with ThreadPoolExecutor(8) as executor:
list(executor.map(self.run_decode, args))
if __name__ == "__main__":
unittest.main(verbosity=3)