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

[CICD] [prefill-only] Consolidate prefill-only model E2E tests (#22405)

Browse Source
This commit is contained in:
Sundara Raman Ramachandran
2026-04-09 00:54:34 -07:00
committed by GitHub
parent 9709192ce9
commit a64905a7b8
16 changed files with 757 additions and 1398 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/CODEOWNERS
+1 -1
View File
@@ -68,7 +68,7 @@
/sgl-model-gateway/src/tool_parser @slin1237 @CatherineSue /sgl-model-gateway/src/tool_parser @slin1237 @CatherineSue
/sgl-model-gateway/src/wasm @slin1237 /sgl-model-gateway/src/wasm @slin1237
/sgl-model-gateway/examples/wasm @slin1237 /sgl-model-gateway/examples/wasm @slin1237
/test/registered/core/test_score_api.py @sundar24295s @chanh @fortunecookiee /test/registered/prefill_only @sundar24295s @chanh @fortunecookiee
/benchmark/prefill_only/bench_score.py @sundar24295s @chanh @fortunecookiee /benchmark/prefill_only/bench_score.py @sundar24295s @chanh @fortunecookiee
/test/srt/ascend @ping1jing2 @iforgetmyname /test/srt/ascend @ping1jing2 @iforgetmyname
/test/srt/test_modelopt* @Edwardf0t1 /test/srt/test_modelopt* @Edwardf0t1
test/registered/core/test_score_api.py
-604
View File
@@ -1,604 +0,0 @@
import unittest
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from sglang.srt.entrypoints.engine import Engine
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.test_utils import DEFAULT_SMALL_MODEL_NAME_FOR_TEST, CustomTestCase
register_cuda_ci(est_time=260, suite="stage-b-test-1-gpu-large")
TEST_MODEL_NAME = DEFAULT_SMALL_MODEL_NAME_FOR_TEST
class TestScoreAPI(CustomTestCase):
"""Test the scoring API functionality."""
def setUp(self):
"""Set up each test case."""
self.engine = Engine(model_path=TEST_MODEL_NAME)
def tearDown(self):
"""Clean up after each test case."""
if self.engine is not None:
self.engine.shutdown()
torch.cuda.empty_cache()
def compute_hf_scores(
self, query, items, label_token_ids, apply_softmax=False, item_first=False
):
"""Compute scores using direct HuggingFace model inference.
Returns probabilities for each token ID, optionally normalized with softmax.
Args:
query: The query text
items: List of item texts
label_token_ids: List of token IDs to compute probabilities for
apply_softmax: Whether to normalize probabilities using softmax
item_first: If True, prepend items to query. Otherwise append items to query.
"""
# Initialize HF model and tokenizer
tokenizer = AutoTokenizer.from_pretrained(
TEST_MODEL_NAME, trust_remote_code=True
)
model = AutoModelForCausalLM.from_pretrained(
TEST_MODEL_NAME, trust_remote_code=True
)
try:
scores = []
for item in items:
# Construct full text based on item_first parameter
full_text = f"{item}{query}" if item_first else f"{query}{item}"
inputs = tokenizer(full_text, return_tensors="pt").to(model.device)
# Get logits for the last token
with torch.no_grad():
outputs = model(**inputs)
last_token_logits = outputs.logits[0, -1]
# Get logits for just our target tokens
target_logits = last_token_logits[label_token_ids]
# Apply softmax over just the target tokens
target_probs = torch.softmax(target_logits, dim=-1)
# Convert to list of probabilities in order of label_token_ids
probs = [target_probs[i].item() for i in range(len(label_token_ids))]
scores.append(probs)
return scores
finally:
# Clean up HF resources
model.cpu()
del model
del tokenizer
torch.cuda.empty_cache()
def _get_token_ids(self, tokens):
"""Helper method to get token IDs for a list of tokens."""
tokenizer = AutoTokenizer.from_pretrained(
TEST_MODEL_NAME, trust_remote_code=True
)
try:
label_token_ids = []
for token in tokens:
encoding = tokenizer(token, add_special_tokens=False)
token_ids = encoding["input_ids"]
label_token_ids.append(token_ids[0])
return label_token_ids
finally:
del tokenizer
def _compare_scores(self, hf_scores, sglang_scores, label_token_ids, case_name=""):
"""Helper method to compare scores between HF and SGLang using relative tolerance."""
self.assertEqual(
len(hf_scores),
len(sglang_scores),
f"Score lengths don't match for {case_name}",
)
# Use a relative tolerance of 1%
TOLERANCE = 0.01
for hf_score_list, sglang_score_list in zip(hf_scores, sglang_scores):
self.assertEqual(
len(hf_score_list),
len(sglang_score_list),
f"Score list lengths don't match for {case_name}",
)
for hf_score, sglang_score in zip(hf_score_list, sglang_score_list):
diff = abs(hf_score - sglang_score)
self.assertLessEqual(
diff,
TOLERANCE,
msg=f"Scores differ by {diff:.2%} ({case_name}): "
f"HF={hf_score:.6f}, SGLang={sglang_score:.6f}",
)
self.assertGreaterEqual(
sglang_score, 0, f"SGLang score {sglang_score:.6f} not in [0,1]"
)
self.assertLessEqual(
sglang_score, 1, f"SGLang score {sglang_score:.6f} not in [0,1]"
)
self.assertAlmostEqual(
sum(sglang_score_list),
1.0,
places=6,
msg=f"SGLang scores don't sum to 1 ({case_name}): {sum(sglang_score_list):.6f}",
)
def test_score_consistency(self):
"""Test that SGLang scoring matches direct HuggingFace model scoring."""
# Define test cases
test_cases = [
{
"name": "default case",
"query": "I pledge allegiance",
"items": ["", " to"],
"item_first": False,
},
{
"name": "item_first case",
"query": " is a city",
"items": ["Tokyo", "Japan"],
"item_first": True,
},
]
# Common tokens to test for all cases
tokens = [" to", " the"]
label_token_ids = self._get_token_ids(tokens)
# Run each test case
for case in test_cases:
# Get scores from SGLang
sglang_scores = self.engine.score(
query=case["query"],
items=case["items"],
label_token_ids=label_token_ids,
apply_softmax=True,
item_first=case["item_first"],
).scores
# Get scores from HuggingFace using the same parameters
hf_scores = self.compute_hf_scores(
query=case["query"],
items=case["items"],
label_token_ids=label_token_ids,
apply_softmax=True,
item_first=case["item_first"],
)
# Compare scores
self._compare_scores(
hf_scores, sglang_scores, label_token_ids, case["name"]
)
def test_score_batch_handling(self):
"""Test that batch scoring works correctly."""
# Test with different batch sizes
batch_sizes = [1, 2, 4, 8]
label_token_ids = [1, 2, 3]
for batch_size in batch_sizes:
texts = [f"test {i}" for i in range(batch_size)]
scores = self.engine.score(
query="The test was",
items=texts,
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
self.assertEqual(
len(scores),
batch_size,
f"Expected {batch_size} scores, got {len(scores)}",
)
# Verify each score list has the correct length
for score_list in scores:
self.assertEqual(
len(score_list),
len(label_token_ids),
f"Score list length {len(score_list)} doesn't match label_token_ids length {len(label_token_ids)}",
)
self.assertTrue(
all(isinstance(v, float) for v in score_list),
"All scores should be floats",
)
self.assertAlmostEqual(
1.0, sum(score_list), 6, "Scores should sum to 1"
)
def test_score_request_construction(self):
"""Test that scoring requests are constructed to avoid decode phase."""
from unittest.mock import patch
# Capture the internal request to verify optimization
captured_requests = []
original_gen = self.engine.tokenizer_manager.generate_request
async def mock_generate_request(req, request=None):
captured_requests.append(req)
async for result in original_gen(req, request):
yield result
# Patch the generate_request method
with patch.object(
self.engine.tokenizer_manager,
"generate_request",
side_effect=mock_generate_request,
):
# Run a scoring request
query = "What is the capital of"
items = ["France", "Germany"]
label_token_ids = [1, 2, 3]
scores = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
# Verify we got results
self.assertEqual(len(scores), len(items))
# Verify the captured request has decode-avoiding properties
self.assertEqual(len(captured_requests), 1)
request = captured_requests[0]
# Key assertions for decode phase avoidance:
# 1. max_new_tokens should be 0 (prevents token generation)
# Handle both single and batch request cases
if isinstance(request.sampling_params, dict):
max_new_tokens = request.sampling_params.get("max_new_tokens", 0)
elif isinstance(request.sampling_params, list):
# For batch requests, check the first item
max_new_tokens = request.sampling_params[0].get("max_new_tokens", 0)
else:
max_new_tokens = getattr(request.sampling_params, "max_new_tokens", 0)
self.assertEqual(
max_new_tokens, 0, "max_new_tokens should be 0 to avoid decode phase"
)
# 2. Should have token_ids_logprob for scoring
# Handle both single and batch request cases
if (
isinstance(request.token_ids_logprob, list)
and len(request.token_ids_logprob) > 0
and isinstance(request.token_ids_logprob[0], list)
):
# Batch case: token_ids_logprob is a list of lists
# Each item in the batch should have the same label_token_ids
for item_token_ids in request.token_ids_logprob:
self.assertEqual(
item_token_ids,
label_token_ids,
"Each batch item should have label_token_ids for scoring",
)
else:
# Single request case
self.assertEqual(
request.token_ids_logprob,
label_token_ids,
"Should have label_token_ids for scoring",
)
# 3. Should request logprobs but not stream
self.assertTrue(
request.return_logprob, "Should request logprobs for scoring"
)
self.assertFalse(request.stream, "Scoring requests should not stream")
def test_multi_item_scoring_basic(self):
"""Test basic multi-item scoring functionality."""
# Test with a simple query and items
query = "What is the capital of California? Answer Yes or No for each of the following options:"
items = ["Sacramento", "San Jose", "San Francisco"]
label_token_ids = [9454, 2753] # "Yes" and "No" tokens
# Get scores using SGLang
result = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
)
scores = result.scores
prompt_tokens = result.prompt_tokens
# Verify we get the expected number of scores
self.assertEqual(len(scores), len(items), "Should get one score list per item")
self.assertGreater(prompt_tokens, 0, "Should have positive prompt_tokens")
# Verify each score list has the correct length
for i, score_list in enumerate(scores):
self.assertEqual(
len(score_list),
len(label_token_ids),
f"Item {i} should have {len(label_token_ids)} scores",
)
# Verify scores are probabilities (sum to 1)
self.assertAlmostEqual(
sum(score_list),
1.0,
places=6,
msg=f"Scores for item {i} should sum to 1",
)
# Verify all scores are non-negative
for j, score in enumerate(score_list):
self.assertGreaterEqual(
score, 0, f"Score {j} for item {i} should be non-negative"
)
def test_multi_item_scoring_consistency(self):
"""Test that multi-item scoring gives consistent results."""
query = "Choose the best option:"
items = ["Option A", "Option B", "Option C"]
label_token_ids = [1, 2, 3]
# Run the same test multiple times
scores1 = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
scores2 = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
# Results should be identical (deterministic)
self.assertEqual(len(scores1), len(scores2), "Should get same number of items")
for i, (s1, s2) in enumerate(zip(scores1, scores2)):
self.assertEqual(
len(s1), len(s2), f"Item {i} should have same number of scores"
)
for j, (score1, score2) in enumerate(zip(s1, s2)):
self.assertAlmostEqual(
score1,
score2,
places=6,
msg=f"Score {j} for item {i} should be identical",
)
def test_multi_item_scoring_different_sizes(self):
"""Test multi-item scoring with different numbers of items."""
query = "Rate each option:"
label_token_ids = [1, 2, 3, 4, 5]
# Test with different numbers of items
test_cases = [
["Single item"],
["Item 1", "Item 2"],
["A", "B", "C", "D"],
["X", "Y", "Z", "W", "V", "U"],
]
for items in test_cases:
with self.subTest(items=items):
scores = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
self.assertEqual(
len(scores), len(items), f"Should get {len(items)} score lists"
)
for i, score_list in enumerate(scores):
self.assertEqual(
len(score_list),
len(label_token_ids),
f"Item {i} should have {len(label_token_ids)} scores",
)
self.assertAlmostEqual(sum(score_list), 1.0, places=6)
def test_multi_item_scoring_empty_items(self):
"""Test multi-item scoring with empty items list."""
query = "Test query"
items = []
label_token_ids = [1, 2]
result = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
)
scores = result.scores
prompt_tokens = result.prompt_tokens
self.assertEqual(len(scores), 0, "Should return empty list for empty items")
self.assertEqual(
prompt_tokens, 0, "Should return 0 prompt_tokens for empty items"
)
def test_multi_item_scoring_single_item(self):
"""Test multi-item scoring with single item (should work like regular scoring)."""
query = "Complete this sentence: The capital of France is"
items = ["Paris"]
label_token_ids = [1, 2, 3]
result = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
)
scores = result.scores
prompt_tokens = result.prompt_tokens
self.assertEqual(len(scores), 1, "Should get one score list")
self.assertEqual(
len(scores[0]), len(label_token_ids), "Should have correct number of scores"
)
self.assertAlmostEqual(sum(scores[0]), 1.0, places=6)
self.assertGreater(prompt_tokens, 0, "Should have positive prompt_tokens")
def test_multi_item_scoring_different_queries(self):
"""Test multi-item scoring with different types of queries."""
items = ["Yes", "No"]
label_token_ids = [1, 2]
test_queries = [
"Is this true?",
"Choose the correct answer:",
"What is the best option?",
"Select all that apply:",
"", # Empty query
]
for query in test_queries:
with self.subTest(query=query):
scores = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
self.assertEqual(
len(scores),
len(items),
f"Should get {len(items)} score lists for query: '{query}'",
)
for i, score_list in enumerate(scores):
self.assertEqual(len(score_list), len(label_token_ids))
self.assertAlmostEqual(sum(score_list), 1.0, places=6)
def test_multi_item_scoring_different_label_tokens(self):
"""Test multi-item scoring with different label token sets."""
query = "Choose the best option:"
items = ["Option A", "Option B"]
test_label_tokens = [
[1, 2], # Two tokens
[1, 2, 3, 4], # Four tokens
[1], # Single token
[1, 2, 3, 4, 5, 6, 7, 8], # Many tokens
]
for label_token_ids in test_label_tokens:
with self.subTest(label_tokens=label_token_ids):
scores = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
self.assertEqual(len(scores), len(items))
for i, score_list in enumerate(scores):
self.assertEqual(
len(score_list),
len(label_token_ids),
f"Item {i} should have {len(label_token_ids)} scores",
)
self.assertAlmostEqual(sum(score_list), 1.0, places=6)
def test_multi_item_scoring_without_softmax(self):
"""Test multi-item scoring without softmax normalization."""
query = "Rate each option:"
items = ["Good", "Bad", "Neutral"]
label_token_ids = [1, 2, 3]
scores = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=False, # No softmax
).scores
self.assertEqual(len(scores), len(items))
for i, score_list in enumerate(scores):
self.assertEqual(len(score_list), len(label_token_ids))
# Without softmax, scores don't need to sum to 1
# But they should still be valid logits/probabilities
for j, score in enumerate(score_list):
self.assertIsInstance(
score, (int, float), f"Score {j} for item {i} should be numeric"
)
def test_multi_item_scoring_large_batch(self):
"""Test multi-item scoring with a large number of items."""
query = "Classify each item:"
items = [f"Item {i}" for i in range(20)] # 20 items
label_token_ids = [1, 2, 3]
scores = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
self.assertEqual(len(scores), len(items), "Should handle large batches")
for i, score_list in enumerate(scores):
self.assertEqual(len(score_list), len(label_token_ids))
self.assertAlmostEqual(sum(score_list), 1.0, places=6)
def test_multi_item_scoring_unicode(self):
"""Test multi-item scoring with unicode characters."""
query = "选择最佳选项："
items = ["选项A", "选项B", "选项C"]
label_token_ids = [1, 2, 3]
scores = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
self.assertEqual(len(scores), len(items))
for i, score_list in enumerate(scores):
self.assertEqual(len(score_list), len(label_token_ids))
self.assertAlmostEqual(sum(score_list), 1.0, places=6)
def test_multi_item_scoring_error_handling(self):
"""Test multi-item scoring error handling."""
query = "Test query"
items = ["Item 1", "Item 2"]
label_token_ids = [1, 2]
# Test with invalid label_token_ids
with self.assertRaises((ValueError, TypeError)):
self.engine.score(
query=query,
items=items,
label_token_ids="invalid", # Should be list of ints
apply_softmax=True,
)
# Test with None items
with self.assertRaises((ValueError, TypeError)):
self.engine.score(
query=query,
items=None,
label_token_ids=label_token_ids,
apply_softmax=True,
)
if __name__ == "__main__":
unittest.main()
test/registered/core/test_score_classification.py
-351
View File
@@ -1,351 +0,0 @@
"""Tests for Scoring API with SequenceClassification models.
Covers both single-item and multi-item scoring (MIS) for classification
models. Uses json_model_override_args to override a regular Qwen3 model's
architecture to Qwen3ForSequenceClassification so we can validate the
scoring pipeline without needing a dedicated classification checkpoint
(the score head gets randomly initialised, which is fine for shape /
pipeline correctness).
"""
import json
import unittest
import torch
from sglang.srt.entrypoints.engine import Engine
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.test_utils import CustomTestCase
register_cuda_ci(est_time=120, suite="stage-b-test-1-gpu-small")
# A lightweight Qwen3 checkpoint whose backbone weights load cleanly into
# Qwen3ForSequenceClassification (the classification head is random).
TEST_BASE_MODEL = "Qwen/Qwen3-0.6B"
# <|endoftext|> for Qwen3 tokenizer — used as MIS delimiter
QWEN3_ENDOFTEXT_TOKEN_ID = 151643
class TestScoreClassification(CustomTestCase):
"""Single-item scoring with a SequenceClassification model."""
NUM_LABELS = 2
@classmethod
def setUpClass(cls):
override_args = json.dumps(
{
"architectures": ["Qwen3ForSequenceClassification"],
"num_labels": cls.NUM_LABELS,
}
)
cls.engine = Engine(
model_path=TEST_BASE_MODEL,
disable_radix_cache=True,
json_model_override_args=override_args,
mem_fraction_static=0.15,
)
@classmethod
def tearDownClass(cls):
if hasattr(cls, "engine") and cls.engine is not None:
cls.engine.shutdown()
torch.cuda.empty_cache()
def test_basic_single_item(self):
"""Each item gets a score vector of length num_labels."""
scores = self.engine.score(
query="Rate each option:",
items=["Option A", "Option B"],
apply_softmax=True,
).scores
self.assertEqual(len(scores), 2)
for i, score_list in enumerate(scores):
self.assertEqual(len(score_list), self.NUM_LABELS)
self.assertAlmostEqual(
sum(score_list),
1.0,
places=5,
msg=f"Softmax scores for item {i} should sum to 1",
)
for val in score_list:
self.assertGreaterEqual(val, 0.0)
self.assertLessEqual(val, 1.0)
def test_single_item_edge_case(self):
"""Single item in the list."""
scores = self.engine.score(
query="Evaluate:",
items=["Only item"],
apply_softmax=True,
).scores
self.assertEqual(len(scores), 1)
self.assertEqual(len(scores[0]), self.NUM_LABELS)
self.assertAlmostEqual(sum(scores[0]), 1.0, places=5)
def test_raw_logits_without_softmax(self):
"""Without softmax, returns raw logits (no probability constraints)."""
scores = self.engine.score(
query="Evaluate:",
items=["Alpha", "Beta"],
apply_softmax=False,
).scores
self.assertEqual(len(scores), 2)
for score_list in scores:
self.assertEqual(len(score_list), self.NUM_LABELS)
for val in score_list:
self.assertTrue(
isinstance(val, (int, float)),
f"Expected numeric score, got {type(val)}",
)
def test_deterministic(self):
"""Identical inputs yield near-identical scores (fp16 non-determinism allowed)."""
kwargs = dict(query="Evaluate:", items=["alpha", "beta", "gamma"])
scores1 = self.engine.score(**kwargs).scores
scores2 = self.engine.score(**kwargs).scores
self.assertEqual(len(scores1), len(scores2))
for s1, s2 in zip(scores1, scores2):
for v1, v2 in zip(s1, s2):
self.assertAlmostEqual(v1, v2, places=1)
def test_tokenized_inputs(self):
"""Pre-tokenized query and items work the same as text inputs."""
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained(TEST_BASE_MODEL)
query_text = "Rate this:"
items_text = ["Good", "Bad"]
text_scores = self.engine.score(
query=query_text,
items=items_text,
apply_softmax=True,
).scores
query_ids = tokenizer.encode(query_text)
items_ids = [tokenizer.encode(item) for item in items_text]
token_scores = self.engine.score(
query=query_ids,
items=items_ids,
apply_softmax=True,
).scores
self.assertEqual(len(text_scores), len(token_scores))
for txt_s, tok_s in zip(text_scores, token_scores):
for t, k in zip(txt_s, tok_s):
self.assertAlmostEqual(t, k, places=4)
def test_label_token_ids_ignored(self):
"""SequenceClassification models ignore label_token_ids (no crash)."""
scores = self.engine.score(
query="Evaluate:",
items=["Test item"],
label_token_ids=[1, 2, 3],
apply_softmax=True,
).scores
self.assertEqual(len(scores), 1)
self.assertEqual(len(scores[0]), self.NUM_LABELS)
class TestScoreClassificationMIS(CustomTestCase):
"""Multi-item scoring (MIS) with a SequenceClassification model.
MIS packs all items into one sequence separated by a delimiter token.
The score_and_pool function extracts per-item scores at delimiter
positions.
"""
NUM_LABELS = 2
@classmethod
def setUpClass(cls):
override_args = json.dumps(
{
"architectures": ["Qwen3ForSequenceClassification"],
"num_labels": cls.NUM_LABELS,
}
)
cls.engine = Engine(
model_path=TEST_BASE_MODEL,
disable_radix_cache=True,
chunked_prefill_size=-1,
multi_item_scoring_delimiter=QWEN3_ENDOFTEXT_TOKEN_ID,
json_model_override_args=override_args,
mem_fraction_static=0.15,
)
@classmethod
def tearDownClass(cls):
if hasattr(cls, "engine") and cls.engine is not None:
cls.engine.shutdown()
torch.cuda.empty_cache()
def test_mis_basic(self):
"""MIS produces one score vector per item."""
items = ["Option A", "Option B", "Option C"]
scores = self.engine.score(
query="Rate each option:",
items=items,
apply_softmax=True,
).scores
self.assertEqual(len(scores), len(items))
for i, score_list in enumerate(scores):
self.assertEqual(
len(score_list),
self.NUM_LABELS,
f"Item {i} should have {self.NUM_LABELS} scores",
)
self.assertAlmostEqual(
sum(score_list),
1.0,
places=5,
msg=f"Scores for item {i} should sum to 1",
)
for val in score_list:
self.assertGreaterEqual(val, 0.0)
self.assertLessEqual(val, 1.0)
def test_mis_many_items(self):
"""Stress test: 10 items."""
items = [f"Item {i}" for i in range(10)]
scores = self.engine.score(
query="Classify each:",
items=items,
apply_softmax=True,
).scores
self.assertEqual(len(scores), len(items))
for score_list in scores:
self.assertEqual(len(score_list), self.NUM_LABELS)
self.assertAlmostEqual(sum(score_list), 1.0, places=5)
def test_mis_single_item(self):
"""Edge case: single item through MIS path."""
scores = self.engine.score(
query="Evaluate:",
items=["Single item"],
apply_softmax=True,
).scores
self.assertEqual(len(scores), 1)
self.assertEqual(len(scores[0]), self.NUM_LABELS)
self.assertAlmostEqual(sum(scores[0]), 1.0, places=5)
def test_items_produce_distinct_scores(self):
"""Different items must produce different score vectors.
Even with a randomly initialised classification head, different
item texts produce different hidden states, so scores should
differ. This catches bugs where all delimiter tokens share the
same pooled representation.
"""
items = [
"Option A is about cats",
"Option B is about dogs",
"Option C is about fish",
]
scores = self.engine.score(query="Rate each option:", items=items).scores
self.assertEqual(len(scores), len(items))
all_identical = all(scores[0] == s for s in scores[1:])
self.assertFalse(
all_identical,
f"All {len(items)} items returned identical scores — "
f"MIS delimiter indexing is likely broken. Scores: {scores[0]}",
)
def test_deterministic(self):
"""Identical MIS requests return identical scores."""
kwargs = dict(
query="Evaluate:",
items=["alpha", "beta", "gamma"],
)
scores1 = self.engine.score(**kwargs).scores
scores2 = self.engine.score(**kwargs).scores
self.assertEqual(scores1, scores2)
def test_softmax_valid(self):
"""With softmax, each item's scores form a valid probability distribution."""
items = ["Option A", "Option B", "Option C"]
scores = self.engine.score(
query="Rate each option:",
items=items,
apply_softmax=True,
).scores
for i, score_list in enumerate(scores):
self.assertEqual(len(score_list), self.NUM_LABELS)
for val in score_list:
self.assertGreaterEqual(val, 0.0)
self.assertLessEqual(val, 1.0)
self.assertAlmostEqual(
sum(score_list),
1.0,
places=6,
msg=f"Softmax scores for item {i} don't sum to 1: {sum(score_list)}",
)
class TestScoreClassificationMISAdvanced(CustomTestCase):
"""Advanced MIS tests with more labels to stress tensor shape handling."""
NUM_LABELS = 12
@classmethod
def setUpClass(cls):
override_args = json.dumps(
{
"architectures": ["Qwen3ForSequenceClassification"],
"num_labels": cls.NUM_LABELS,
}
)
cls.engine = Engine(
model_path=TEST_BASE_MODEL,
disable_radix_cache=True,
chunked_prefill_size=-1,
multi_item_scoring_delimiter=QWEN3_ENDOFTEXT_TOKEN_ID,
json_model_override_args=override_args,
mem_fraction_static=0.15,
)
@classmethod
def tearDownClass(cls):
if hasattr(cls, "engine") and cls.engine is not None:
cls.engine.shutdown()
torch.cuda.empty_cache()
def test_many_labels_shape(self):
"""Verify correct shape with many labels (catches 2D tensor bugs)."""
items = [f"Item {i}" for i in range(5)]
scores = self.engine.score(
query="Classify:",
items=items,
apply_softmax=True,
).scores
self.assertEqual(len(scores), len(items))
for score_list in scores:
self.assertEqual(len(score_list), self.NUM_LABELS)
self.assertAlmostEqual(sum(score_list), 1.0, places=5)
def test_many_items_distinct(self):
"""15 items should not all produce identical scores."""
items = [f"City {i}" for i in range(15)]
scores = self.engine.score(query="Classify each city:", items=items).scores
self.assertEqual(len(scores), len(items))
unique_count = len({tuple(s) for s in scores})
self.assertGreater(unique_count, 1, "All 15 items returned identical scores")
if __name__ == "__main__":
unittest.main()
test/registered/embedding/test_input_embeddings.py → test/registered/input_embedding/test_input_embeddings.py
View File
test/registered/embedding/test_input_embeds_chunked.py → test/registered/input_embedding/test_input_embeds_chunked.py
View File
test/registered/models/test_cross_encoder_models.py → test/registered/prefill_only/test_cross_encoder_models.py
View File
test/registered/unit/managers/test_embed_overrides.py → test/registered/prefill_only/test_embed_overrides.py
View File
test/registered/embedding/test_embedding_models.py → test/registered/prefill_only/test_embedding_models.py
View File
test/registered/embedding/test_encoder_embedding_models.py → test/registered/prefill_only/test_encoder_embedding_models.py
View File
test/registered/embedding/test_openai_embedding.py → test/registered/prefill_only/test_openai_embedding.py
View File
test/registered/models/test_reward_models.py → test/registered/prefill_only/test_reward_models.py
View File
test/registered/prefill_only/test_score_api.py
+234
View File
@@ -0,0 +1,234 @@
"""HTTP layer tests for the /v1/score endpoint.
Two test classes, each with its own server instance:
TestCausalLMScoringHTTP — basic endpoint: schema defaults, response
structure, error rejection (no MIS delimiter)
TestCausalLMMISScoringHTTP — MIS mode: validates --multi-item-scoring-delimiter
CLI flag wiring and per-item output shape
Engine-level correctness (numerical accuracy, batching, edge cases) lives in
test_score_engine.py. These tests focus on the HTTP integration seam:
Pydantic schema defaults, FastAPI routing, and server argument wiring.
"""
import unittest
import requests
from sglang.srt.utils import kill_process_tree
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.test_utils import (
DEFAULT_SMALL_MODEL_NAME_FOR_TEST,
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
DEFAULT_URL_FOR_TEST,
CustomTestCase,
popen_launch_server,
)
register_cuda_ci(est_time=160, suite="stage-b-test-1-gpu-small")
_MODEL = DEFAULT_SMALL_MODEL_NAME_FOR_TEST # Llama-3.2-1B-Instruct
# <|eot_id|> for Llama-3.x Instruct — used as MIS delimiter
_LLAMA3_EOT_TOKEN_ID = 128009
# ---------------------------------------------------------------------------
# Basic scoring (no MIS delimiter)
# ---------------------------------------------------------------------------
class TestCausalLMScoringHTTP(CustomTestCase):
"""Validates /v1/score HTTP integration — schema, defaults, and error handling.
Starts a plain CausalLM server (no --multi-item-scoring-delimiter) to test
the HTTP layer in isolation: response envelope shape, the apply_softmax
default (False), and Pydantic validation errors on malformed input.
"""
@classmethod
def setUpClass(cls):
cls.model = _MODEL
cls.base_url = DEFAULT_URL_FOR_TEST
cls.process = popen_launch_server(
cls.model,
cls.base_url,
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
)
@classmethod
def tearDownClass(cls):
if hasattr(cls, "process") and cls.process:
kill_process_tree(cls.process.pid)
def _post(self, payload):
return requests.post(self.base_url + "/v1/score", json=payload)
def test_response_envelope(self):
"""Response JSON contains scores, model, and object='scoring'."""
resp = self._post(
{
"query": "The capital of France is",
"items": ["Paris", "Berlin"],
"label_token_ids": [1, 2],
"apply_softmax": True,
"model": self.model,
}
)
self.assertEqual(resp.status_code, 200)
body = resp.json()
self.assertIn("scores", body)
self.assertIn("model", body)
self.assertEqual(body["object"], "scoring")
def test_apply_softmax_false_by_default(self):
"""Without apply_softmax=True, raw log-probs are returned (do not sum to 1)."""
resp = self._post(
{
"query": "The capital of France is",
"items": ["Paris"],
"label_token_ids": [1, 2],
# apply_softmax intentionally omitted — default is False
"model": self.model,
}
)
self.assertEqual(resp.status_code, 200)
scores = resp.json()["scores"]
self.assertEqual(len(scores), 1)
# Raw log-probs over a vocabulary subset do not sum to 1
self.assertNotAlmostEqual(sum(scores[0]), 1.0, places=3)
def test_apply_softmax_true_normalizes(self):
"""With apply_softmax=True, scores form a valid probability distribution."""
resp = self._post(
{
"query": "The capital of France is",
"items": ["Paris", "Berlin", "Rome"],
"label_token_ids": [1, 2, 3],
"apply_softmax": True,
"model": self.model,
}
)
self.assertEqual(resp.status_code, 200)
for row in resp.json()["scores"]:
self.assertAlmostEqual(sum(row), 1.0, places=6)
for v in row:
self.assertGreaterEqual(v, 0.0)
def test_schema_rejection(self):
"""Malformed payloads must be rejected with HTTP 4xx."""
bad_payloads = [
# label_token_ids must be List[int]
{
"query": "Q",
"items": ["X"],
"label_token_ids": "bad",
"model": self.model,
},
# items must be str / List[str] / List[List[int]], not int
{
"query": "Q",
"items": 42,
"label_token_ids": [1, 2],
"model": self.model,
},
]
for payload in bad_payloads:
with self.subTest(payload=list(payload.keys())):
self.assertGreaterEqual(self._post(payload).status_code, 400)
# ---------------------------------------------------------------------------
# MIS scoring (with --multi-item-scoring-delimiter)
# ---------------------------------------------------------------------------
class TestCausalLMMISScoringHTTP(CustomTestCase):
"""Validates /v1/score with --multi-item-scoring-delimiter.
Confirms that the CLI flag is correctly wired into ServerArgs and that the
endpoint returns one probability vector per item when items are
delimiter-packed into a single forward pass.
"""
@classmethod
def setUpClass(cls):
cls.model = _MODEL
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=[
"--disable-radix-cache",
"--chunked-prefill-size",
"-1",
"--multi-item-scoring-delimiter",
str(_LLAMA3_EOT_TOKEN_ID),
],
)
@classmethod
def tearDownClass(cls):
if hasattr(cls, "process") and cls.process:
kill_process_tree(cls.process.pid)
def _score(self, query, items, label_token_ids):
resp = requests.post(
self.base_url + "/v1/score",
json={
"query": query,
"items": items,
"label_token_ids": label_token_ids,
"apply_softmax": True,
"model": self.model,
},
)
self.assertEqual(resp.status_code, 200)
return resp.json()
def test_one_probability_vector_per_item(self):
"""Each item yields one softmax-normalised score vector."""
items = ["Sacramento", "San Jose", "San Francisco"]
label_token_ids = [9454, 2753]
scores = self._score("Is each the capital?", items, label_token_ids)["scores"]
self.assertEqual(len(scores), len(items))
for i, row in enumerate(scores):
self.assertEqual(len(row), len(label_token_ids))
self.assertAlmostEqual(sum(row), 1.0, places=6, msg=f"Item {i}")
for v in row:
self.assertGreaterEqual(v, 0.0)
def test_empty_items_returns_empty_scores(self):
result = self._score("Test query", [], [1, 2])
self.assertEqual(len(result["scores"]), 0)
def test_varying_item_counts(self):
"""1, 2, 4, and 6 items all return the correct number of score vectors."""
label_token_ids = [1, 2, 3, 4, 5]
for items in (
["Single item"],
["Item 1", "Item 2"],
["A", "B", "C", "D"],
["X", "Y", "Z", "W", "V", "U"],
):
with self.subTest(n=len(items)):
scores = self._score("Rate each:", items, label_token_ids)["scores"]
self.assertEqual(len(scores), len(items))
for row in scores:
self.assertEqual(len(row), len(label_token_ids))
self.assertAlmostEqual(sum(row), 1.0, places=6)
def test_deterministic(self):
"""Back-to-back identical requests return identical scores."""
items, label_token_ids = ["Option A", "Option B", "Option C"], [1, 2, 3]
s1 = self._score("Choose:", items, label_token_ids)["scores"]
s2 = self._score("Choose:", items, label_token_ids)["scores"]
self.assertEqual(len(s1), len(s2))
for r1, r2 in zip(s1, s2):
for v1, v2 in zip(r1, r2):
self.assertAlmostEqual(v1, v2, places=6)
if __name__ == "__main__":
unittest.main(verbosity=3)
test/registered/prefill_only/test_score_engine.py
+522
View File
@@ -0,0 +1,522 @@
"""Engine API tests for the /v1/score scoring pipeline.
Two model types, two scoring modes:
TestCausalLMScoring — CausalLM, single-item and batched multi-item
TestSeqClsScoring — SequenceClassification, single-item mode
TestSeqClsMISScoring — SequenceClassification, MIS delimiter mode
The Engine (Python API) is the right layer for correctness testing: it
exercises tokenization, forward pass, pooling, and score extraction without
the HTTP serialization overhead. HTTP-layer tests live in test_score_api.py.
"""
import json
import unittest
from unittest.mock import patch
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from sglang.srt.entrypoints.engine import Engine
from sglang.test.ci.ci_register import register_cuda_ci
from sglang.test.test_utils import DEFAULT_SMALL_MODEL_NAME_FOR_TEST, CustomTestCase
register_cuda_ci(est_time=200, suite="stage-b-test-1-gpu-small")
_CAUSAL_LM_MODEL = DEFAULT_SMALL_MODEL_NAME_FOR_TEST # Llama-3.2-1B-Instruct
_SEQCLS_MODEL = "Qwen/Qwen3-0.6B" # backbone; arch overridden to SeqCls below
# <|endoftext|> for Qwen3 tokenizer — used as MIS delimiter
_QWEN3_EOT_TOKEN_ID = 151643
# ---------------------------------------------------------------------------
# CausalLM
# ---------------------------------------------------------------------------
class TestCausalLMScoring(CustomTestCase):
"""CausalLM scoring via Engine — correctness, batching, and edge cases.
A single Engine instance is shared across all test methods (class-level
setup) so model loading happens once, not once per test.
"""
@classmethod
def setUpClass(cls):
cls.engine = Engine(model_path=_CAUSAL_LM_MODEL)
@classmethod
def tearDownClass(cls):
if hasattr(cls, "engine") and cls.engine:
cls.engine.shutdown()
torch.cuda.empty_cache()
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _hf_scores(self, query, items, label_token_ids, item_first=False):
"""Reference scores computed directly with HuggingFace (CPU inference)."""
tokenizer = AutoTokenizer.from_pretrained(
_CAUSAL_LM_MODEL, trust_remote_code=True
)
model = AutoModelForCausalLM.from_pretrained(
_CAUSAL_LM_MODEL, trust_remote_code=True
)
try:
scores = []
for item in items:
text = f"{item}{query}" if item_first else f"{query}{item}"
inputs = tokenizer(text, return_tensors="pt").to(model.device)
with torch.no_grad():
last_logits = model(**inputs).logits[0, -1]
target_probs = torch.softmax(last_logits[label_token_ids], dim=-1)
scores.append([p.item() for p in target_probs])
return scores
finally:
model.cpu()
del model, tokenizer
torch.cuda.empty_cache()
def _assert_scores_close(self, hf, sgl, tol=0.01):
self.assertEqual(len(hf), len(sgl))
for hf_row, sgl_row in zip(hf, sgl):
self.assertEqual(len(hf_row), len(sgl_row))
for h, s in zip(hf_row, sgl_row):
self.assertLessEqual(abs(h - s), tol, f"HF={h:.6f} SGLang={s:.6f}")
self.assertAlmostEqual(sum(sgl_row), 1.0, places=6)
# ------------------------------------------------------------------
# Correctness
# ------------------------------------------------------------------
def test_scores_match_hf_reference(self):
"""SGLang scores agree with HuggingFace within 1% tolerance."""
label_token_ids = []
tokenizer = AutoTokenizer.from_pretrained(
_CAUSAL_LM_MODEL, trust_remote_code=True
)
for token in [" to", " the"]:
label_token_ids.append(
tokenizer(token, add_special_tokens=False)["input_ids"][0]
)
del tokenizer
for query, items, item_first in [
("I pledge allegiance", ["", " to"], False),
(" is a city", ["Tokyo", "Japan"], True),
]:
with self.subTest(query=query):
sgl = self.engine.score(
query=query,
items=items,
label_token_ids=label_token_ids,
apply_softmax=True,
item_first=item_first,
).scores
hf = self._hf_scores(query, items, label_token_ids, item_first)
self._assert_scores_close(hf, sgl)
def test_request_avoids_decode_phase(self):
"""Internal request must have max_new_tokens=0, logprob=True, stream=False."""
captured = []
original = self.engine.tokenizer_manager.generate_request
async def capturing_gen(req, request=None):
captured.append(req)
async for result in original(req, request):
yield result
with patch.object(
self.engine.tokenizer_manager,
"generate_request",
side_effect=capturing_gen,
):
self.engine.score(
query="What is the capital of",
items=["France", "Germany"],
label_token_ids=[1, 2, 3],
apply_softmax=True,
)
self.assertEqual(len(captured), 1)
req = captured[0]
if isinstance(req.sampling_params, list):
max_new_tokens = req.sampling_params[0].get("max_new_tokens", 0)
elif isinstance(req.sampling_params, dict):
max_new_tokens = req.sampling_params.get("max_new_tokens", 0)
else:
max_new_tokens = getattr(req.sampling_params, "max_new_tokens", 0)
self.assertEqual(max_new_tokens, 0)
self.assertTrue(req.return_logprob)
self.assertFalse(req.stream)
# ------------------------------------------------------------------
# Multi-item / batching
# ------------------------------------------------------------------
def test_score_batch_sizes(self):
"""Correct output count and shape for batch sizes 1, 2, 4, 8."""
label_token_ids = [1, 2, 3]
for n in [1, 2, 4, 8]:
with self.subTest(n=n):
scores = self.engine.score(
query="The test was",
items=[f"test {i}" for i in range(n)],
label_token_ids=label_token_ids,
apply_softmax=True,
).scores
self.assertEqual(len(scores), n)
for row in scores:
self.assertEqual(len(row), len(label_token_ids))
self.assertTrue(all(isinstance(v, float) for v in row))
self.assertAlmostEqual(sum(row), 1.0, places=6)
def test_score_empty_items(self):
"""Empty items list → empty scores and zero prompt_tokens."""
result = self.engine.score(
query="Test query", items=[], label_token_ids=[1, 2], apply_softmax=True
)
self.assertEqual(len(result.scores), 0)
self.assertEqual(result.prompt_tokens, 0)
def test_score_without_softmax(self):
"""apply_softmax=False returns raw logits (not probability-constrained)."""
scores = self.engine.score(
query="Rate each:",
items=["Good", "Bad", "Neutral"],
label_token_ids=[1, 2, 3],
apply_softmax=False,
).scores
self.assertEqual(len(scores), 3)
for row in scores:
self.assertEqual(len(row), 3)
for v in row:
self.assertIsInstance(v, (int, float))
def test_score_varying_label_token_sets(self):
"""Different label_token_ids lengths all produce correct-shaped output."""
for n_labels in [1, 2, 4, 8]:
with self.subTest(n_labels=n_labels):
scores = self.engine.score(
query="Choose:",
items=["Option A", "Option B"],
label_token_ids=list(range(1, n_labels + 1)),
apply_softmax=True,
).scores
self.assertEqual(len(scores), 2)
for row in scores:
self.assertEqual(len(row), n_labels)
self.assertAlmostEqual(sum(row), 1.0, places=6)
def test_score_unicode(self):
"""Unicode query and items do not crash and produce valid scores."""
scores = self.engine.score(
query="选择最佳选项：",
items=["选项A", "选项B", "选项C"],
label_token_ids=[1, 2, 3],
apply_softmax=True,
).scores
self.assertEqual(len(scores), 3)
for row in scores:
self.assertAlmostEqual(sum(row), 1.0, places=6)
def test_score_deterministic(self):
"""Identical calls return numerically equivalent scores (within GPU float tolerance)."""
kwargs = dict(query="Choose:", items=["A", "B", "C"], label_token_ids=[1, 2, 3])
scores_a = self.engine.score(**kwargs).scores
scores_b = self.engine.score(**kwargs).scores
self.assertEqual(len(scores_a), len(scores_b))
for row_a, row_b in zip(scores_a, scores_b):
self.assertEqual(len(row_a), len(row_b))
for a, b in zip(row_a, row_b):
self.assertAlmostEqual(a, b, places=5)
def test_score_error_handling(self):
"""Invalid argument types raise ValueError or TypeError."""
with self.assertRaises((ValueError, TypeError)):
self.engine.score(
query="Q", items=["X"], label_token_ids="bad", apply_softmax=True
)
with self.assertRaises((ValueError, TypeError)):
self.engine.score(
query="Q", items=None, label_token_ids=[1, 2], apply_softmax=True
)
# ---------------------------------------------------------------------------
# SequenceClassification — single-item mode
# ---------------------------------------------------------------------------
class TestSeqClsScoring(CustomTestCase):
"""SequenceClassification scoring via Engine — no MIS delimiter.
Uses json_model_override_args to load Qwen3-0.6B backbone weights into
Qwen3ForSequenceClassification. The classification head is randomly
initialised; shape/pipeline correctness is what matters here.
"""
NUM_LABELS = 2
@classmethod
def setUpClass(cls):
cls.engine = Engine(
model_path=_SEQCLS_MODEL,
disable_radix_cache=True,
json_model_override_args=json.dumps(
{
"architectures": ["Qwen3ForSequenceClassification"],
"num_labels": cls.NUM_LABELS,
}
),
mem_fraction_static=0.15,
)
@classmethod
def tearDownClass(cls):
if hasattr(cls, "engine") and cls.engine:
cls.engine.shutdown()
torch.cuda.empty_cache()
def test_score_shape(self):
"""Each item gets a score vector of length num_labels."""
scores = self.engine.score(
query="Rate each option:",
items=["Option A", "Option B"],
apply_softmax=True,
).scores
self.assertEqual(len(scores), 2)
for i, row in enumerate(scores):
self.assertEqual(len(row), self.NUM_LABELS)
self.assertAlmostEqual(sum(row), 1.0, places=5)
for v in row:
self.assertGreaterEqual(v, 0.0)
self.assertLessEqual(v, 1.0)
def test_score_single_item_edge_case(self):
"""Single item in the list."""
scores = self.engine.score(
query="Evaluate:", items=["Only item"], apply_softmax=True
).scores
self.assertEqual(len(scores), 1)
self.assertEqual(len(scores[0]), self.NUM_LABELS)
self.assertAlmostEqual(sum(scores[0]), 1.0, places=5)
def test_score_without_softmax(self):
"""Without softmax, returns raw logits (no probability constraints)."""
scores = self.engine.score(
query="Evaluate:", items=["Alpha", "Beta"], apply_softmax=False
).scores
self.assertEqual(len(scores), 2)
for row in scores:
self.assertEqual(len(row), self.NUM_LABELS)
for v in row:
self.assertIsInstance(v, (int, float))
def test_score_deterministic(self):
"""Identical inputs yield near-identical scores (fp16 tolerance)."""
kwargs = dict(query="Evaluate:", items=["alpha", "beta", "gamma"])
scores1 = self.engine.score(**kwargs).scores
scores2 = self.engine.score(**kwargs).scores
self.assertEqual(len(scores1), len(scores2))
for s1, s2 in zip(scores1, scores2):
for v1, v2 in zip(s1, s2):
self.assertAlmostEqual(v1, v2, places=1)
def test_score_tokenized_inputs(self):
"""Pre-tokenized query/items match text input scores."""
from transformers import AutoTokenizer
tok = AutoTokenizer.from_pretrained(_SEQCLS_MODEL)
query, items = "Rate this:", ["Good", "Bad"]
text_scores = self.engine.score(
query=query, items=items, apply_softmax=True
).scores
token_scores = self.engine.score(
query=tok.encode(query),
items=[tok.encode(i) for i in items],
apply_softmax=True,
).scores
self.assertEqual(len(text_scores), len(token_scores))
for ts, ks in zip(text_scores, token_scores):
for t, k in zip(ts, ks):
self.assertAlmostEqual(t, k, places=4)
def test_label_token_ids_ignored(self):
"""SeqCls models ignore label_token_ids — output width is always num_labels."""
scores = self.engine.score(
query="Evaluate:",
items=["Test item"],
label_token_ids=[1, 2, 3],
apply_softmax=True,
).scores
self.assertEqual(len(scores), 1)
self.assertEqual(len(scores[0]), self.NUM_LABELS)
# ---------------------------------------------------------------------------
# SequenceClassification — MIS (delimiter) mode
# ---------------------------------------------------------------------------
class TestSeqClsMISScoring(CustomTestCase):
"""SeqCls MIS: all items packed into one sequence separated by delimiter token.
score_and_pool() extracts per-item scores at delimiter positions.
Two sub-cases are tested:
- NUM_LABELS=2 — standard binary classification head
- NUM_LABELS=12 — stress-tests 2-D tensor indexing in score_and_pool()
"""
NUM_LABELS = 2
@classmethod
def setUpClass(cls):
cls.engine = Engine(
model_path=_SEQCLS_MODEL,
disable_radix_cache=True,
chunked_prefill_size=-1,
multi_item_scoring_delimiter=_QWEN3_EOT_TOKEN_ID,
json_model_override_args=json.dumps(
{
"architectures": ["Qwen3ForSequenceClassification"],
"num_labels": cls.NUM_LABELS,
}
),
mem_fraction_static=0.15,
)
@classmethod
def tearDownClass(cls):
if hasattr(cls, "engine") and cls.engine:
cls.engine.shutdown()
torch.cuda.empty_cache()
def test_mis_one_vector_per_item(self):
"""MIS produces exactly one score vector per item."""
items = ["Option A", "Option B", "Option C"]
scores = self.engine.score(
query="Rate each option:", items=items, apply_softmax=True
).scores
self.assertEqual(len(scores), len(items))
for i, row in enumerate(scores):
self.assertEqual(len(row), self.NUM_LABELS)
self.assertAlmostEqual(sum(row), 1.0, places=5)
for v in row:
self.assertGreaterEqual(v, 0.0)
self.assertLessEqual(v, 1.0)
def test_mis_single_item_edge_case(self):
"""Single item through MIS path."""
scores = self.engine.score(
query="Evaluate:", items=["Single item"], apply_softmax=True
).scores
self.assertEqual(len(scores), 1)
self.assertEqual(len(scores[0]), self.NUM_LABELS)
self.assertAlmostEqual(sum(scores[0]), 1.0, places=5)
def test_mis_many_items(self):
"""10 items all return valid probability vectors."""
items = [f"Item {i}" for i in range(10)]
scores = self.engine.score(
query="Classify each:", items=items, apply_softmax=True
).scores
self.assertEqual(len(scores), len(items))
for row in scores:
self.assertEqual(len(row), self.NUM_LABELS)
self.assertAlmostEqual(sum(row), 1.0, places=5)
def test_mis_items_produce_distinct_scores(self):
"""Different items must yield different score vectors.
Catches bugs where all delimiter positions share the same pooled
hidden state (e.g. off-by-one in score_and_pool indexing).
"""
items = [
"Option A is about cats",
"Option B is about dogs",
"Option C is about fish",
]
scores = self.engine.score(query="Rate each option:", items=items).scores
self.assertEqual(len(scores), len(items))
self.assertFalse(
all(scores[0] == s for s in scores[1:]),
f"All items returned identical scores — delimiter indexing is likely broken. "
f"Scores: {scores[0]}",
)
def test_mis_deterministic(self):
"""Identical MIS requests return identical scores."""
kwargs = dict(query="Evaluate:", items=["alpha", "beta", "gamma"])
self.assertEqual(
self.engine.score(**kwargs).scores, self.engine.score(**kwargs).scores
)
# ---------------------------------------------------------------------------
# SequenceClassification — MIS with many labels (tensor shape stress test)
# ---------------------------------------------------------------------------
class TestSeqClsMISAdvancedScoring(CustomTestCase):
"""SeqCls MIS with 12 labels — stresses the 2-D tensor path in score_and_pool.
Kept in a separate class (own Engine instance) so it doesn't fight the
2-label class-level engine for GPU memory.
"""
NUM_LABELS = 12
@classmethod
def setUpClass(cls):
cls.engine = Engine(
model_path=_SEQCLS_MODEL,
disable_radix_cache=True,
chunked_prefill_size=-1,
multi_item_scoring_delimiter=_QWEN3_EOT_TOKEN_ID,
json_model_override_args=json.dumps(
{
"architectures": ["Qwen3ForSequenceClassification"],
"num_labels": cls.NUM_LABELS,
}
),
mem_fraction_static=0.15,
)
@classmethod
def tearDownClass(cls):
if hasattr(cls, "engine") and cls.engine:
cls.engine.shutdown()
torch.cuda.empty_cache()
def test_many_labels_correct_shape(self):
"""5 items × 12 labels — each score vector has the right length."""
items = [f"Item {i}" for i in range(5)]
scores = self.engine.score(
query="Classify:", items=items, apply_softmax=True
).scores
self.assertEqual(len(scores), len(items))
for row in scores:
self.assertEqual(len(row), self.NUM_LABELS)
self.assertAlmostEqual(sum(row), 1.0, places=5)
def test_many_items_produce_distinct_scores(self):
"""15 items should not all return identical score vectors."""
items = [f"City {i}" for i in range(15)]
scores = self.engine.score(query="Classify each city:", items=items).scores
self.assertEqual(len(scores), len(items))
self.assertGreater(
len({tuple(s) for s in scores}),
1,
"All 15 items returned identical scores",
)
if __name__ == "__main__":
unittest.main(verbosity=3)
test/registered/openai_server/basic/test_serving_rerank.py → test/registered/prefill_only/test_serving_rerank.py
View File
test/registered/unit/test_pooler_score_and_pool.py → test/registered/unit/layers/test_pooler_score_and_pool.py
View File
test/srt/test_multi_item_scoring.py
-442
View File
@@ -1,442 +0,0 @@
import os
import signal
import subprocess
import tempfile
import time
import unittest
import requests
from sglang.test.test_utils import DEFAULT_SMALL_MODEL_NAME_FOR_TEST, CustomTestCase
class TestMultiItemScoringServer(CustomTestCase):
"""Test multi-item scoring functionality through the server API."""
def setUp(self):
"""Set up each test case."""
self.model_path = DEFAULT_SMALL_MODEL_NAME_FOR_TEST
self.port = 30001 # Use different port to avoid conflicts
self.host = "localhost"
self.base_url = f"http://{self.host}:{self.port}"
self.server_process = None
self.server_log_file = None
def tearDown(self):
"""Clean up after each test case."""
self.stop_server()
def start_server(self, multi_item_scoring_delimiter=None):
"""Start the SGLang server with multi-item scoring enabled."""
if self.server_process is not None:
self.stop_server()
# Create a temporary log file
self.server_log_file = tempfile.NamedTemporaryFile(mode="w+", delete=False)
cmd = [
"python",
"-m",
"sglang.launch_server",
"--model-path",
self.model_path,
"--port",
str(self.port),
"--host",
self.host,
"--chunked-prefill-size",
"-1",
"--dtype",
"float16",
"--max-prefill-tokens",
"30000",
"--mem-fraction-static",
"0.3",
"--disable-radix-cache",
"--disable-cuda-graph",
"--attention-backend",
"flashinfer",
]
if multi_item_scoring_delimiter is not None:
cmd.extend(
["--multi-item-scoring-delimiter", str(multi_item_scoring_delimiter)]
)
# Start server process
self.server_process = subprocess.Popen(
cmd,
stdout=self.server_log_file,
stderr=subprocess.STDOUT,
preexec_fn=os.setsid if hasattr(os, "setsid") else None,
)
# Wait for server to start
max_wait_time = 60 # seconds
start_time = time.time()
while time.time() - start_time < max_wait_time:
try:
response = requests.get(f"{self.base_url}/get_model_info", timeout=5)
if response.status_code == 200:
return True
except requests.exceptions.RequestException:
pass
time.sleep(1)
# If we get here, server didn't start properly
self.stop_server()
raise RuntimeError("Failed to start SGLang server within timeout period")
def stop_server(self):
"""Stop the SGLang server."""
if self.server_process is not None:
try:
# Kill the process group
if hasattr(os, "killpg"):
os.killpg(os.getpgid(self.server_process.pid), signal.SIGTERM)
else:
self.server_process.terminate()
# Wait for graceful shutdown
try:
self.server_process.wait(timeout=10)
except subprocess.TimeoutExpired:
# Force kill if graceful shutdown fails
if hasattr(os, "killpg"):
os.killpg(os.getpgid(self.server_process.pid), signal.SIGKILL)
else:
self.server_process.kill()
self.server_process.wait()
except (ProcessLookupError, OSError):
# Process already terminated
pass
finally:
self.server_process = None
if self.server_log_file is not None:
try:
self.server_log_file.close()
os.unlink(self.server_log_file.name)
except (OSError, FileNotFoundError):
pass
finally:
self.server_log_file = None
def get_server_logs(self):
"""Get server logs for debugging."""
if self.server_log_file is not None:
try:
with open(self.server_log_file.name, "r") as f:
return f.read()
except (OSError, FileNotFoundError):
pass
return "No logs available"
def test_multi_item_scoring_server_basic(self):
"""Test basic multi-item scoring through server API."""
# Start server with multi-item scoring enabled
delimiter_token_id = 151655 # Example delimiter token ID
self.start_server(multi_item_scoring_delimiter=delimiter_token_id)
# Test data
query = "What is the capital of California? Answer Yes or No for each of the following options:"
items = ["Sacramento", "San Jose", "San Francisco"]
label_token_ids = [9454, 2753] # "Yes" and "No" tokens
# Make scoring request
payload = {
"query": query,
"items": items,
"label_token_ids": label_token_ids,
"model": self.model_path,
}
response = requests.post(
f"{self.base_url}/v1/score",
headers={"Content-Type": "application/json"},
json=payload,
timeout=30,
)
# Check response
self.assertEqual(
response.status_code,
200,
f"Server returned {response.status_code}. Logs: {self.get_server_logs()}",
)
result = response.json()
# Verify response structure
self.assertIn("scores", result)
self.assertIn("model", result)
self.assertIn("object", result)
self.assertEqual(result["object"], "scoring")
self.assertEqual(result["model"], self.model_path)
# Verify scores
scores = result["scores"]
self.assertEqual(len(scores), len(items), "Should get one score list per item")
for i, score_list in enumerate(scores):
self.assertEqual(
len(score_list),
len(label_token_ids),
f"Item {i} should have {len(label_token_ids)} scores",
)
# Verify scores are probabilities (sum to 1)
self.assertAlmostEqual(
sum(score_list),
1.0,
places=6,
msg=f"Scores for item {i} should sum to 1",
)
# Verify all scores are non-negative
for j, score in enumerate(score_list):
self.assertGreaterEqual(
score, 0, f"Score {j} for item {i} should be non-negative"
)
def test_multi_item_scoring_server_different_sizes(self):
"""Test multi-item scoring with different numbers of items through server."""
delimiter_token_id = 151655
self.start_server(multi_item_scoring_delimiter=delimiter_token_id)
query = "Rate each option:"
label_token_ids = [1, 2, 3, 4, 5]
test_cases = [
["Single item"],
["Item 1", "Item 2"],
["A", "B", "C", "D"],
["X", "Y", "Z", "W", "V", "U"],
]
for items in test_cases:
with self.subTest(items=items):
payload = {
"query": query,
"items": items,
"label_token_ids": label_token_ids,
"model": self.model_path,
}
response = requests.post(
f"{self.base_url}/v1/score",
headers={"Content-Type": "application/json"},
json=payload,
timeout=30,
)
self.assertEqual(
response.status_code,
200,
f"Failed for items {items}. Logs: {self.get_server_logs()}",
)
result = response.json()
scores = result["scores"]
self.assertEqual(
len(scores), len(items), f"Should get {len(items)} score lists"
)
for i, score_list in enumerate(scores):
self.assertEqual(len(score_list), len(label_token_ids))
self.assertAlmostEqual(sum(score_list), 1.0, places=6)
def test_multi_item_scoring_server_empty_items(self):
"""Test multi-item scoring with empty items list through server."""
delimiter_token_id = 151655
self.start_server(multi_item_scoring_delimiter=delimiter_token_id)
payload = {
"query": "Test query",
"items": [],
"label_token_ids": [1, 2],
"model": self.model_path,
}
response = requests.post(
f"{self.base_url}/v1/score",
headers={"Content-Type": "application/json"},
json=payload,
timeout=30,
)
self.assertEqual(response.status_code, 200)
result = response.json()
self.assertEqual(
len(result["scores"]), 0, "Should return empty list for empty items"
)
def test_multi_item_scoring_server_without_delimiter(self):
"""Test that server works without multi-item scoring delimiter."""
# Start server without multi-item scoring delimiter
self.start_server(multi_item_scoring_delimiter=None)
payload = {
"query": "Test query",
"items": ["Item 1", "Item 2"],
"label_token_ids": [1, 2],
"model": self.model_path,
}
response = requests.post(
f"{self.base_url}/v1/score",
headers={"Content-Type": "application/json"},
json=payload,
timeout=30,
)
# Should still work (falls back to regular scoring)
self.assertEqual(response.status_code, 200)
result = response.json()
self.assertIn("scores", result)
def test_multi_item_scoring_server_error_handling(self):
"""Test error handling in multi-item scoring server API."""
delimiter_token_id = 151655
self.start_server(multi_item_scoring_delimiter=delimiter_token_id)
# Test with invalid payload
invalid_payloads = [
{
"query": "Test",
"items": "not a list",
"label_token_ids": [1, 2],
"model": self.model_path,
},
{
"query": "Test",
"items": ["Item 1"],
"label_token_ids": "not a list",
"model": self.model_path,
},
{
"query": "Test",
"items": ["Item 1"],
"label_token_ids": [1, 2],
}, # Missing model
{
"items": ["Item 1"],
"label_token_ids": [1, 2],
"model": self.model_path,
}, # Missing query
]
for i, payload in enumerate(invalid_payloads):
with self.subTest(payload=i):
response = requests.post(
f"{self.base_url}/v1/score",
headers={"Content-Type": "application/json"},
json=payload,
timeout=30,
)
# Should return error status
self.assertGreaterEqual(
response.status_code,
400,
f"Should return error for invalid payload {i}",
)
def test_multi_item_scoring_server_consistency(self):
"""Test that multi-item scoring gives consistent results through server."""
delimiter_token_id = 151655
self.start_server(multi_item_scoring_delimiter=delimiter_token_id)
query = "Choose the best option:"
items = ["Option A", "Option B", "Option C"]
label_token_ids = [1, 2, 3]
payload = {
"query": query,
"items": items,
"label_token_ids": label_token_ids,
"model": self.model_path,
}
# Run the same test multiple times
scores1 = None
scores2 = None
for attempt in range(2):
response = requests.post(
f"{self.base_url}/v1/score",
headers={"Content-Type": "application/json"},
json=payload,
timeout=30,
)
self.assertEqual(
response.status_code,
200,
f"Attempt {attempt + 1} failed. Logs: {self.get_server_logs()}",
)
result = response.json()
scores = result["scores"]
if attempt == 0:
scores1 = scores
else:
scores2 = scores
# Results should be identical (deterministic)
self.assertEqual(len(scores1), len(scores2), "Should get same number of items")
for i, (s1, s2) in enumerate(zip(scores1, scores2)):
self.assertEqual(
len(s1), len(s2), f"Item {i} should have same number of scores"
)
for j, (score1, score2) in enumerate(zip(s1, s2)):
self.assertAlmostEqual(
score1,
score2,
places=6,
msg=f"Score {j} for item {i} should be identical",
)
def test_multi_item_scoring_server_large_batch(self):
"""Test multi-item scoring with large batch through server."""
delimiter_token_id = 151655
self.start_server(multi_item_scoring_delimiter=delimiter_token_id)
query = "Classify each item:"
items = [f"Item {i}" for i in range(10)] # 10 items (smaller for test)
label_token_ids = [1, 2, 3]
payload = {
"query": query,
"items": items,
"label_token_ids": label_token_ids,
"model": self.model_path,
}
response = requests.post(
f"{self.base_url}/v1/score",
headers={"Content-Type": "application/json"},
json=payload,
timeout=60, # Longer timeout for large batch
)
self.assertEqual(
response.status_code,
200,
f"Large batch failed. Logs: {self.get_server_logs()}",
)
result = response.json()
scores = result["scores"]
self.assertEqual(len(scores), len(items), "Should handle large batches")
for i, score_list in enumerate(scores):
self.assertEqual(len(score_list), len(label_token_ids))
self.assertAlmostEqual(sum(score_list), 1.0, places=6)
if __name__ == "__main__":
unittest.main()