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

Migrate tokenizer tests to test/registered/tokenizer/ (#16457)

Browse Source
This commit is contained in:
Alison Shao
2026-01-07 13:21:16 -08:00
committed by GitHub
parent 0c474273c5
commit 0241e0460f
4 changed files with 11 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/srt/run_suite.py
-6
View File
@@ -11,10 +11,7 @@ suites = {
"per-commit-1-gpu": [
TestFile("test_evs.py", 20),
TestFile("test_external_models.py", 30),
TestFile("test_jinja_template_utils.py", 7),
TestFile("test_modelopt_loader.py", 11),
TestFile("test_multi_tokenizer.py", 230),
TestFile("test_skip_tokenizer_init.py", 77),
TestFile("test_utils_update_weights.py", 29),
TestFile("test_video_utils.py", 5),
TestFile("test_modelopt_export.py", 9),
@@ -107,10 +104,7 @@ suite_amd = {
# TestFile("lora/test_lora_qwen3.py", 97), # Disabled temporarily, see https://github.com/sgl-project/sglang/issues/13107
TestFile("test_bench_typebaseddispatcher.py", 10),
TestFile("test_external_models.py", 45),
TestFile("test_jinja_template_utils.py", 1),
TestFile("test_multi_tokenizer.py", 345),
TestFile("test_rope_rocm.py", 3),
TestFile("test_skip_tokenizer_init.py", 117),
# TestFile("test_torch_compile_moe.py", 210), # Disabled temporarily, see https://github.com/sgl-project/sglang/issues/13107
TestFile("test_type_based_dispatcher.py", 10),
TestFile("test_video_utils.py", 8),
test/srt/test_jinja_template_utils.py
-310
View File
@@ -1,310 +0,0 @@
"""
Unit tests for Jinja chat template utils.
"""
import unittest
from sglang.srt.parser.jinja_template_utils import (
detect_jinja_template_content_format,
process_content_for_template_format,
)
from sglang.test.test_utils import CustomTestCase
class TestTemplateContentFormatDetection(CustomTestCase):
"""Test template content format detection functionality."""
def test_detect_llama4_openai_format(self):
"""Test detection of llama4-style template (should be 'openai' format)."""
llama4_pattern = """
{%- for message in messages %}
{%- if message['content'] is string %}
{{- message['content'] }}
{%- else %}
{%- for content in message['content'] %}
{%- if content['type'] == 'image' %}
{{- '<|image|>' }}
{%- elif content['type'] == 'text' %}
{{- content['text'] | trim }}
{%- endif %}
{%- endfor %}
{%- endif %}
{%- endfor %}
"""
result = detect_jinja_template_content_format(llama4_pattern)
self.assertEqual(result, "openai")
def test_detect_deepseek_string_format(self):
"""Test detection of deepseek-style template (should be 'string' format)."""
deepseek_pattern = """
{%- for message in messages %}
{%- if message['role'] == 'user' %}
{{- '<|User|>' + message['content'] + '<|Assistant|>' }}
{%- endif %}
{%- endfor %}
"""
result = detect_jinja_template_content_format(deepseek_pattern)
self.assertEqual(result, "string")
def test_detect_invalid_template(self):
"""Test handling of invalid template (should default to 'string')."""
invalid_pattern = "{{{{ invalid jinja syntax }}}}"
result = detect_jinja_template_content_format(invalid_pattern)
self.assertEqual(result, "string")
def test_detect_empty_template(self):
"""Test handling of empty template (should default to 'string')."""
result = detect_jinja_template_content_format("")
self.assertEqual(result, "string")
def test_detect_msg_content_pattern(self):
"""Test detection of template with msg.content pattern (should be 'openai' format)."""
msg_content_pattern = """
[gMASK]<sop>
{%- for msg in messages %}
{%- if msg.role == 'system' %}
<|system|>
{{ msg.content }}
{%- elif msg.role == 'user' %}
<|user|>{{ '\n' }}
{%- if msg.content is string %}
{{ msg.content }}
{%- else %}
{%- for item in msg.content %}
{%- if item.type == 'video' or 'video' in item %}
<|begin_of_video|><|video|><|end_of_video|>
{%- elif item.type == 'image' or 'image' in item %}
<|begin_of_image|><|image|><|end_of_image|>
{%- elif item.type == 'text' %}
{{ item.text }}
{%- endif %}
{%- endfor %}
{%- endif %}
{%- elif msg.role == 'assistant' %}
{%- if msg.metadata %}
<|assistant|>{{ msg.metadata }}
{{ msg.content }}
{%- else %}
<|assistant|>
{{ msg.content }}
{%- endif %}
{%- endif %}
{%- endfor %}
{% if add_generation_prompt %}<|assistant|>
{% endif %}
"""
result = detect_jinja_template_content_format(msg_content_pattern)
self.assertEqual(result, "openai")
def test_detect_m_content_pattern(self):
"""Test detection of template with m.content pattern (should be 'openai' format)."""
msg_content_pattern = """
[gMASK]<sop>
{%- for m in messages %}
{%- if m.role == 'system' %}
<|system|>
{{ m.content }}
{%- elif m.role == 'user' %}
<|user|>{{ '\n' }}
{%- if m.content is string %}
{{ m.content }}
{%- else %}
{%- for item in m.content %}
{%- if item.type == 'video' or 'video' in item %}
<|begin_of_video|><|video|><|end_of_video|>
{%- elif item.type == 'image' or 'image' in item %}
<|begin_of_image|><|image|><|end_of_image|>
{%- elif item.type == 'text' %}
{{ item.text }}
{%- endif %}
{%- endfor %}
{%- endif %}
{%- elif m.role == 'assistant' %}
{%- if m.metadata %}
<|assistant|>{{ m.metadata }}
{{ m.content }}
{%- else %}
<|assistant|>
{{ m.content }}
{%- endif %}
{%- endif %}
{%- endfor %}
{% if add_generation_prompt %}<|assistant|>
{% endif %}
"""
result = detect_jinja_template_content_format(msg_content_pattern)
self.assertEqual(result, "openai")
def test_process_content_openai_format(self):
"""Test content processing for openai format."""
msg_dict = {
"role": "user",
"content": [
{"type": "text", "text": "Look at this image:"},
{
"type": "image_url",
"image_url": {"url": "http://example.com/image.jpg"},
},
{"type": "text", "text": "What do you see?"},
],
}
image_data = []
video_data = []
audio_data = []
modalities = []
result = process_content_for_template_format(
msg_dict, "openai", image_data, video_data, audio_data, modalities
)
# Check that image_data was extracted
self.assertEqual(len(image_data), 1)
self.assertEqual(image_data[0].url, "http://example.com/image.jpg")
# Check that content was normalized
expected_content = [
{"type": "text", "text": "Look at this image:"},
{"type": "image"}, # normalized from image_url
{"type": "text", "text": "What do you see?"},
]
self.assertEqual(result["content"], expected_content)
self.assertEqual(result["role"], "user")
def test_process_content_string_format(self):
"""Test content processing for string format."""
msg_dict = {
"role": "user",
"content": [
{"type": "text", "text": "Hello"},
{
"type": "image_url",
"image_url": {"url": "http://example.com/image.jpg"},
},
{"type": "text", "text": "world"},
],
}
image_data = []
video_data = []
audio_data = []
modalities = []
result = process_content_for_template_format(
msg_dict, "string", image_data, video_data, audio_data, modalities
)
# For string format, should flatten to text only
self.assertEqual(result["content"], "Hello world")
self.assertEqual(result["role"], "user")
# Image data should not be extracted for string format
self.assertEqual(len(image_data), 0)
def test_process_content_with_audio(self):
"""Test content processing with audio content."""
msg_dict = {
"role": "user",
"content": [
{"type": "text", "text": "Listen to this:"},
{
"type": "audio_url",
"audio_url": {"url": "http://example.com/audio.mp3"},
},
],
}
image_data = []
video_data = []
audio_data = []
modalities = []
result = process_content_for_template_format(
msg_dict, "openai", image_data, video_data, audio_data, modalities
)
# Check that audio_data was extracted
self.assertEqual(len(audio_data), 1)
self.assertEqual(audio_data[0], "http://example.com/audio.mp3")
# Check that content was normalized
expected_content = [
{"type": "text", "text": "Listen to this:"},
{"type": "audio"}, # normalized from audio_url
]
self.assertEqual(result["content"], expected_content)
def test_process_content_already_string(self):
"""Test processing content that's already a string."""
msg_dict = {"role": "user", "content": "Hello world"}
image_data = []
video_data = []
audio_data = []
modalities = []
result = process_content_for_template_format(
msg_dict, "openai", image_data, video_data, audio_data, modalities
)
# Should pass through unchanged
self.assertEqual(result["content"], "Hello world")
self.assertEqual(result["role"], "user")
self.assertEqual(len(image_data), 0)
def test_process_content_with_modalities(self):
"""Test content processing with modalities field."""
msg_dict = {
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": "http://example.com/image.jpg"},
"modalities": ["vision"],
}
],
}
image_data = []
video_data = []
audio_data = []
modalities = []
result = process_content_for_template_format(
msg_dict, "openai", image_data, video_data, audio_data, modalities
)
# Check that modalities was extracted
self.assertEqual(len(modalities), 1)
self.assertEqual(modalities[0], ["vision"])
def test_process_content_filter_none_values(self):
"""Test that None values are filtered out of processed messages."""
msg_dict = {
"role": "user",
"content": "Hello",
"name": None,
"tool_call_id": None,
}
image_data = []
video_data = []
audio_data = []
modalities = []
result = process_content_for_template_format(
msg_dict, "string", image_data, video_data, audio_data, modalities
)
# None values should be filtered out
expected_keys = {"role", "content"}
self.assertEqual(set(result.keys()), expected_keys)
if __name__ == "__main__":
unittest.main()
test/srt/test_multi_tokenizer.py
-83
View File
@@ -1,83 +0,0 @@
import unittest
from types import SimpleNamespace
from sglang.srt.utils import kill_process_tree
from sglang.test.run_eval import run_eval
from sglang.test.test_utils import (
DEFAULT_MODEL_NAME_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
auto_config_device,
get_benchmark_args,
is_in_ci,
popen_launch_server,
run_benchmark,
write_github_step_summary,
)
class TestMultiTokenizer(CustomTestCase):
# from test_hicache.py
@classmethod
def setUpClass(cls):
cls.model = DEFAULT_MODEL_NAME_FOR_TEST
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=[
"--tokenizer-worker-num",
8,
"--mem-fraction-static",
0.7,
],
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def test_mmlu(self):
args = SimpleNamespace(
base_url=self.base_url,
model=self.model,
eval_name="mmlu",
num_examples=64,
num_threads=32,
)
metrics = run_eval(args)
self.assertGreaterEqual(metrics["score"], 0.65)
def test_multi_tokenizer_ttft(self):
# from test_bench_serving.py run_bench_serving
args = get_benchmark_args(
base_url=self.base_url,
dataset_name="random",
dataset_path="",
tokenizer=None,
num_prompts=100,
random_input_len=4096,
random_output_len=2048,
sharegpt_context_len=None,
request_rate=1,
disable_stream=False,
disable_ignore_eos=False,
seed=0,
device=auto_config_device(),
lora_name=None,
)
res = run_benchmark(args)
if is_in_ci():
write_github_step_summary(
f"### test_multi_tokenizer_ttft\n"
f"median_e2e_latency_ms: {res['median_e2e_latency_ms']:.2f} ms\n"
)
self.assertLess(res["median_e2e_latency_ms"], 11000)
self.assertLess(res["median_ttft_ms"], 86)
self.assertLess(res["median_itl_ms"], 10)
if __name__ == "__main__":
unittest.main()
test/srt/test_skip_tokenizer_init.py
-244
View File
@@ -1,244 +0,0 @@
"""
python3 -m unittest test_skip_tokenizer_init.TestSkipTokenizerInit.test_parallel_sample
python3 -m unittest test_skip_tokenizer_init.TestSkipTokenizerInit.run_decode_stream
"""
import json
import unittest
import requests
from transformers import AutoProcessor, AutoTokenizer
from sglang.lang.chat_template import get_chat_template_by_model_path
from sglang.srt.utils import kill_process_tree
from sglang.test.test_utils import (
DEFAULT_IMAGE_URL,
DEFAULT_SMALL_MODEL_NAME_FOR_TEST,
DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
download_image_with_retry,
popen_launch_server,
)
class TestSkipTokenizerInit(CustomTestCase):
@classmethod
def setUpClass(cls):
cls.model = DEFAULT_SMALL_MODEL_NAME_FOR_TEST
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
other_args=["--skip-tokenizer-init", "--stream-output"],
)
cls.eos_token_id = [119690]
cls.tokenizer = AutoTokenizer.from_pretrained(
DEFAULT_SMALL_MODEL_NAME_FOR_TEST, use_fast=False
)
@classmethod
def tearDownClass(cls):
kill_process_tree(cls.process.pid)
def run_decode(
self,
prompt_text="The capital of France is",
max_new_tokens=32,
return_logprob=False,
top_logprobs_num=0,
n=1,
):
input_ids = self.get_input_ids(prompt_text)
request = self.get_request_json(
input_ids=input_ids,
return_logprob=return_logprob,
top_logprobs_num=top_logprobs_num,
max_new_tokens=max_new_tokens,
stream=False,
n=n,
)
response = requests.post(
self.base_url + "/generate",
json=request,
)
ret = response.json()
print(json.dumps(ret, indent=2))
def assert_one_item(item):
if item["meta_info"]["finish_reason"]["type"] == "stop":
self.assertEqual(
item["meta_info"]["finish_reason"]["matched"],
self.tokenizer.eos_token_id,
)
elif item["meta_info"]["finish_reason"]["type"] == "length":
self.assertEqual(
len(item["output_ids"]), item["meta_info"]["completion_tokens"]
)
self.assertEqual(len(item["output_ids"]), max_new_tokens)
self.assertEqual(item["meta_info"]["prompt_tokens"], len(input_ids))
if return_logprob:
num_input_logprobs = len(input_ids) - request["logprob_start_len"]
if num_input_logprobs > len(input_ids):
num_input_logprobs -= len(input_ids)
self.assertEqual(
len(item["meta_info"]["input_token_logprobs"]),
num_input_logprobs,
f'{len(item["meta_info"]["input_token_logprobs"])} mismatch with {len(input_ids)}',
)
self.assertEqual(
len(item["meta_info"]["output_token_logprobs"]),
max_new_tokens,
)
# Determine whether to assert a single item or multiple items based on n
if n == 1:
assert_one_item(ret)
else:
self.assertEqual(len(ret), n)
for i in range(n):
assert_one_item(ret[i])
print("=" * 100)
def run_decode_stream(self, return_logprob=False, top_logprobs_num=0, n=1):
max_new_tokens = 32
input_ids = self.get_input_ids("The capital of France is")
requests.post(self.base_url + "/flush_cache")
response = requests.post(
self.base_url + "/generate",
json=self.get_request_json(
input_ids=input_ids,
max_new_tokens=max_new_tokens,
return_logprob=return_logprob,
top_logprobs_num=top_logprobs_num,
stream=False,
n=n,
),
)
ret = response.json()
print(json.dumps(ret))
output_ids = ret["output_ids"]
print("output from non-streaming request:")
print(output_ids)
print(self.tokenizer.decode(output_ids, skip_special_tokens=True))
requests.post(self.base_url + "/flush_cache")
response_stream = requests.post(
self.base_url + "/generate",
json=self.get_request_json(
input_ids=input_ids,
return_logprob=return_logprob,
top_logprobs_num=top_logprobs_num,
stream=True,
n=n,
),
)
response_stream_json = []
for line in response_stream.iter_lines():
print(line)
if line.startswith(b"data: ") and line[6:] != b"[DONE]":
response_stream_json.append(json.loads(line[6:]))
out_stream_ids = []
for x in response_stream_json:
out_stream_ids += x["output_ids"]
print("output from streaming request:")
print(out_stream_ids)
print(self.tokenizer.decode(out_stream_ids, skip_special_tokens=True))
assert output_ids == out_stream_ids
def test_simple_decode(self):
self.run_decode()
def test_parallel_sample(self):
self.run_decode(n=3)
def test_logprob(self):
for top_logprobs_num in [0, 3]:
self.run_decode(return_logprob=True, top_logprobs_num=top_logprobs_num)
def test_eos_behavior(self):
self.run_decode(max_new_tokens=256)
def test_simple_decode_stream(self):
self.run_decode_stream()
def get_input_ids(self, prompt_text) -> list[int]:
input_ids = self.tokenizer(prompt_text, return_tensors="pt")["input_ids"][
0
].tolist()
return input_ids
def get_request_json(
self,
input_ids,
max_new_tokens=32,
return_logprob=False,
top_logprobs_num=0,
stream=False,
n=1,
):
return {
"input_ids": input_ids,
"sampling_params": {
"temperature": 0 if n == 1 else 0.5,
"max_new_tokens": max_new_tokens,
"n": n,
"stop_token_ids": self.eos_token_id,
},
"stream": stream,
"return_logprob": return_logprob,
"top_logprobs_num": top_logprobs_num,
"logprob_start_len": 0,
}
class TestSkipTokenizerInitVLM(TestSkipTokenizerInit):
@classmethod
def setUpClass(cls):
cls.image_url = DEFAULT_IMAGE_URL
cls.image = download_image_with_retry(cls.image_url)
cls.model = DEFAULT_SMALL_VLM_MODEL_NAME_FOR_TEST
cls.tokenizer = AutoTokenizer.from_pretrained(cls.model, use_fast=False)
cls.processor = AutoProcessor.from_pretrained(cls.model, trust_remote_code=True)
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=["--skip-tokenizer-init"],
)
cls.eos_token_id = [cls.tokenizer.eos_token_id]
def get_input_ids(self, _prompt_text) -> list[int]:
chat_template = get_chat_template_by_model_path(self.model)
text = f"{chat_template.image_token}What is in this picture?"
inputs = self.processor(
text=[text],
images=[self.image],
return_tensors="pt",
)
return inputs.input_ids[0].tolist()
def get_request_json(self, *args, **kwargs):
ret = super().get_request_json(*args, **kwargs)
ret["image_data"] = [self.image_url]
ret["logprob_start_len"] = (
-1
) # Do not try to calculate logprobs of image embeddings.
return ret
def test_simple_decode_stream(self):
# TODO mick
pass
if __name__ == "__main__":
unittest.main()