[Test] Merge all constrained decoding tests. (#12633)
This commit is contained in:
Liangsheng Yin
@@ -2,7 +2,6 @@
|
||||
Tests for JSON schema constraint functionality used by JsonArrayParser
|
||||
"""
|
||||
|
||||
import json
|
||||
import unittest
|
||||
|
||||
import jsonschema
|
||||
|
||||
@@ -1,153 +0,0 @@
|
||||
"""
|
||||
python3 -m unittest openai_server.features.test_json_constrained.TestJSONConstrainedOutlinesBackend.test_json_generate
|
||||
python3 -m unittest openai_server.features.test_json_constrained.TestJSONConstrainedXGrammarBackend.test_json_generate
|
||||
python3 -m unittest openai_server.features.test_json_constrained.TestJSONConstrainedLLGuidanceBackend.test_json_generate
|
||||
"""
|
||||
|
||||
import json
|
||||
import unittest
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
import openai
|
||||
import requests
|
||||
|
||||
from sglang.srt.utils import kill_process_tree
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
def setup_class(cls, backend: str):
|
||||
cls.model = DEFAULT_SMALL_MODEL_NAME_FOR_TEST
|
||||
cls.base_url = DEFAULT_URL_FOR_TEST
|
||||
cls.json_schema = json.dumps(
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {"type": "string", "pattern": "^[\\w]+$"},
|
||||
"population": {"type": "integer"},
|
||||
},
|
||||
"required": ["name", "population"],
|
||||
"additionalProperties": False,
|
||||
}
|
||||
)
|
||||
|
||||
other_args = [
|
||||
"--max-running-requests",
|
||||
"10",
|
||||
"--grammar-backend",
|
||||
backend,
|
||||
]
|
||||
|
||||
cls.process = popen_launch_server(
|
||||
cls.model,
|
||||
cls.base_url,
|
||||
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
||||
other_args=other_args,
|
||||
)
|
||||
|
||||
|
||||
class TestJSONConstrained(CustomTestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
setup_class(cls, backend="xgrammar")
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
kill_process_tree(cls.process.pid)
|
||||
|
||||
def run_decode(self, json_schema, return_logprob=False, top_logprobs_num=0, n=1):
|
||||
response = requests.post(
|
||||
self.base_url + "/generate",
|
||||
json={
|
||||
"text": "The capital of France is",
|
||||
"sampling_params": {
|
||||
"temperature": 0 if n == 1 else 0.5,
|
||||
"max_new_tokens": 128,
|
||||
"n": n,
|
||||
"stop_token_ids": [119690],
|
||||
"json_schema": json_schema,
|
||||
},
|
||||
"stream": False,
|
||||
"return_logprob": return_logprob,
|
||||
"top_logprobs_num": top_logprobs_num,
|
||||
"logprob_start_len": 0,
|
||||
},
|
||||
)
|
||||
ret = response.json()
|
||||
print(json.dumps(ret))
|
||||
print("=" * 100)
|
||||
|
||||
if not json_schema or json_schema == "INVALID":
|
||||
return
|
||||
|
||||
# Make sure the json output is valid
|
||||
try:
|
||||
js_obj = json.loads(ret["text"])
|
||||
except (TypeError, json.decoder.JSONDecodeError):
|
||||
raise
|
||||
|
||||
self.assertIsInstance(js_obj["name"], str)
|
||||
self.assertIsInstance(js_obj["population"], int)
|
||||
|
||||
def test_json_generate(self):
|
||||
self.run_decode(json_schema=self.json_schema)
|
||||
|
||||
def test_json_invalid(self):
|
||||
self.run_decode(json_schema="INVALID")
|
||||
|
||||
def test_json_openai(self):
|
||||
client = openai.Client(api_key="EMPTY", base_url=f"{self.base_url}/v1")
|
||||
|
||||
response = client.chat.completions.create(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{"role": "system", "content": "You are a helpful AI assistant"},
|
||||
{
|
||||
"role": "user",
|
||||
"content": "Introduce the capital of France. Return in a JSON format.",
|
||||
},
|
||||
],
|
||||
temperature=0,
|
||||
max_tokens=128,
|
||||
response_format={
|
||||
"type": "json_schema",
|
||||
"json_schema": {"name": "foo", "schema": json.loads(self.json_schema)},
|
||||
},
|
||||
)
|
||||
text = response.choices[0].message.content
|
||||
|
||||
try:
|
||||
js_obj = json.loads(text)
|
||||
except (TypeError, json.decoder.JSONDecodeError):
|
||||
print("JSONDecodeError", text)
|
||||
raise
|
||||
|
||||
self.assertIsInstance(js_obj["name"], str)
|
||||
self.assertIsInstance(js_obj["population"], int)
|
||||
|
||||
def test_mix_json_and_other(self):
|
||||
json_schemas = [None, None, self.json_schema, self.json_schema] * 10
|
||||
|
||||
with ThreadPoolExecutor(len(json_schemas)) as executor:
|
||||
list(executor.map(self.run_decode, json_schemas))
|
||||
|
||||
|
||||
class TestJSONConstrainedOutlinesBackend(TestJSONConstrained):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
setup_class(cls, backend="outlines")
|
||||
|
||||
|
||||
class TestJSONConstrainedLLGuidanceBackend(TestJSONConstrained):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
setup_class(cls, backend="llguidance")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -16,7 +16,7 @@ class TestFile:
|
||||
suites = {
|
||||
"per-commit-1-gpu": [
|
||||
TestFile("debug_utils/test_tensor_dump_forward_hook.py", 15),
|
||||
TestFile("function_call/test_json_schema_constraint.py", 30),
|
||||
TestFile("function_call/test_json_schema_constraint.py", 1),
|
||||
TestFile("hicache/test_hicache.py", 116),
|
||||
TestFile("hicache/test_hicache_eagle.py", 150),
|
||||
TestFile("hicache/test_hicache_mla.py", 127),
|
||||
@@ -46,7 +46,6 @@ suites = {
|
||||
TestFile("openai_server/basic/test_serving_completions.py", 10),
|
||||
TestFile("openai_server/basic/test_serving_embedding.py", 10),
|
||||
TestFile("openai_server/features/test_enable_thinking.py", 70),
|
||||
TestFile("openai_server/features/test_json_constrained.py", 120),
|
||||
TestFile("openai_server/features/test_json_mode.py", 120),
|
||||
TestFile("openai_server/features/test_openai_server_ebnf.py", 20),
|
||||
TestFile("openai_server/features/test_openai_server_hidden_states.py", 240),
|
||||
@@ -74,7 +73,7 @@ suites = {
|
||||
TestFile("test_eagle_infer_a.py", 370),
|
||||
TestFile("test_eagle_infer_b.py", 500),
|
||||
TestFile("test_eagle_infer_beta.py", 90),
|
||||
TestFile("test_ebnf_constrained.py", 80),
|
||||
TestFile("test_constrained_decoding.py", 120),
|
||||
TestFile("test_eval_fp8_accuracy.py", 303),
|
||||
TestFile("test_fa3.py", 420),
|
||||
TestFile("test_flashmla.py", 230),
|
||||
@@ -305,7 +304,7 @@ suites = {
|
||||
TestFile("test_int4_kernel.py"),
|
||||
TestFile("test_int8_kernel.py"),
|
||||
TestFile("test_intel_amx_attention_backend.py"),
|
||||
TestFile("test_json_constrained.py"),
|
||||
TestFile("test_constrained_decoding.py"),
|
||||
TestFile("test_json_mode.py"),
|
||||
TestFile("test_kv_events.py"),
|
||||
TestFile("test_large_max_new_tokens.py"),
|
||||
@@ -369,7 +368,7 @@ suites = {
|
||||
# NOTE: please sort the test cases alphabetically by the test file name
|
||||
suite_amd = {
|
||||
"per-commit-amd": [
|
||||
TestFile("function_call/test_json_schema_constraint.py", 30),
|
||||
TestFile("function_call/test_json_schema_constraint.py", 1),
|
||||
TestFile("hicache/test_hicache.py", 116),
|
||||
TestFile("hicache/test_hicache_mla.py", 127),
|
||||
TestFile("hicache/test_hicache_storage.py", 127),
|
||||
@@ -390,7 +389,6 @@ suite_amd = {
|
||||
TestFile("openai_server/basic/test_serving_completions.py", 10),
|
||||
TestFile("openai_server/basic/test_serving_embedding.py", 10),
|
||||
TestFile("openai_server/features/test_enable_thinking.py", 70),
|
||||
TestFile("openai_server/features/test_json_constrained.py", 120),
|
||||
TestFile("openai_server/features/test_json_mode.py", 120),
|
||||
TestFile("openai_server/features/test_openai_server_ebnf.py", 20),
|
||||
TestFile("openai_server/features/test_reasoning_content.py", 89),
|
||||
@@ -406,7 +404,6 @@ suite_amd = {
|
||||
TestFile("test_abort.py", 51),
|
||||
TestFile("test_chunked_prefill.py", 410),
|
||||
TestFile("test_create_kvindices.py", 2),
|
||||
TestFile("test_ebnf_constrained.py", 80),
|
||||
TestFile("test_eval_fp8_accuracy.py", 303),
|
||||
TestFile("test_function_call_parser.py", 10),
|
||||
TestFile("test_fused_moe.py", 80),
|
||||
@@ -423,7 +420,7 @@ suite_amd = {
|
||||
TestFile("test_pytorch_sampling_backend.py", 66),
|
||||
TestFile("test_radix_attention.py", 105),
|
||||
TestFile("test_reasoning_parser.py", 5),
|
||||
TestFile("test_regex_constrained.py", 64),
|
||||
TestFile("test_constrained_decoding.py", 120),
|
||||
TestFile("test_retract_decode.py", 450),
|
||||
TestFile("test_rope_rocm.py", 3),
|
||||
TestFile("test_server_args.py", 1),
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
import unittest
|
||||
|
||||
from sglang.srt.utils import kill_process_tree
|
||||
from sglang.test.kits.ebnf_constrained_kit import TestEBNFConstrainedMinxin
|
||||
from sglang.test.kits.json_constrained_kit import TestJSONConstrainedMixin
|
||||
from sglang.test.kits.regex_constrained_kit import TestRegexConstrainedMixin
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
class ServerWithGrammar(CustomTestCase):
|
||||
backend = "xgrammar"
|
||||
disable_overlap = False
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.model = DEFAULT_SMALL_MODEL_NAME_FOR_TEST
|
||||
cls.base_url = DEFAULT_URL_FOR_TEST
|
||||
launch_args = [
|
||||
"--max-running-requests",
|
||||
"10",
|
||||
"--grammar-backend",
|
||||
cls.backend,
|
||||
]
|
||||
|
||||
if cls.disable_overlap:
|
||||
launch_args += ["--disable-overlap-schedule"]
|
||||
|
||||
cls.process = popen_launch_server(
|
||||
cls.model,
|
||||
cls.base_url,
|
||||
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
||||
other_args=launch_args,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
kill_process_tree(cls.process.pid)
|
||||
|
||||
|
||||
class TestXGrammarBackend(
|
||||
ServerWithGrammar,
|
||||
TestJSONConstrainedMixin,
|
||||
TestEBNFConstrainedMinxin,
|
||||
TestRegexConstrainedMixin,
|
||||
):
|
||||
backend = "xgrammar"
|
||||
|
||||
|
||||
class TestOutlinesBackend(ServerWithGrammar, TestJSONConstrainedMixin):
|
||||
backend = "outlines"
|
||||
|
||||
|
||||
class TestLLGuidanceBackend(
|
||||
ServerWithGrammar,
|
||||
TestJSONConstrainedMixin,
|
||||
TestEBNFConstrainedMinxin,
|
||||
TestRegexConstrainedMixin,
|
||||
):
|
||||
backend = "llguidance"
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,282 +0,0 @@
|
||||
"""
|
||||
python3 -m unittest test_ebnf_constrained.TestEBNFConstrained.test_ebnf_generate_email
|
||||
python3 -m unittest test_ebnf_constrained.TestEBNFConstrained.test_ebnf_generate_greeting
|
||||
python3 -m unittest test_ebnf_constrained.TestEBNFConstrained.test_ebnf_generate_all_optional_function_params
|
||||
python3 -m unittest test_ebnf_constrained.TestEBNFConstrainedLLGuidance.test_ebnf_generate_email
|
||||
python3 -m unittest test_ebnf_constrained.TestEBNFConstrainedLLGuidance.test_ebnf_generate_greeting
|
||||
python3 -m unittest test_ebnf_constrained.TestEBNFConstrainedLLGuidance.test_ebnf_generate_all_optional_function_params
|
||||
"""
|
||||
|
||||
import json
|
||||
import unittest
|
||||
|
||||
import requests
|
||||
|
||||
from sglang.srt.utils import kill_process_tree
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
def setup_class(cls, backend: str, disable_overlap: bool):
|
||||
cls.model = DEFAULT_SMALL_MODEL_NAME_FOR_TEST
|
||||
cls.base_url = DEFAULT_URL_FOR_TEST
|
||||
cls.ebnf_grammar = 'root ::= "test"' # Default grammar
|
||||
|
||||
other_args = [
|
||||
"--max-running-requests",
|
||||
"10",
|
||||
"--grammar-backend",
|
||||
backend,
|
||||
]
|
||||
|
||||
if disable_overlap:
|
||||
other_args += ["--disable-overlap-schedule"]
|
||||
|
||||
cls.process = popen_launch_server(
|
||||
cls.model,
|
||||
cls.base_url,
|
||||
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
||||
other_args=other_args,
|
||||
)
|
||||
|
||||
|
||||
class TestEBNFConstrained(CustomTestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
setup_class(cls, "xgrammar", disable_overlap=False)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
kill_process_tree(cls.process.pid)
|
||||
|
||||
def run_decode(
|
||||
self,
|
||||
ebnf,
|
||||
expected_patterns,
|
||||
prompt,
|
||||
return_logprob=False,
|
||||
top_logprobs_num=0,
|
||||
n=1,
|
||||
):
|
||||
response = requests.post(
|
||||
self.base_url + "/generate",
|
||||
json={
|
||||
"text": prompt,
|
||||
"sampling_params": {
|
||||
"temperature": 0 if n == 1 else 0.5,
|
||||
"max_new_tokens": 128,
|
||||
"n": n,
|
||||
"ebnf": ebnf,
|
||||
},
|
||||
"stream": False,
|
||||
"return_logprob": return_logprob,
|
||||
"top_logprobs_num": top_logprobs_num,
|
||||
"logprob_start_len": 0,
|
||||
},
|
||||
)
|
||||
|
||||
ret = response.json()
|
||||
print(json.dumps(ret, indent=2))
|
||||
print("=" * 100)
|
||||
|
||||
if not isinstance(ret, list):
|
||||
self.fail(f"Expected response to be a list, but got {type(ret)}")
|
||||
|
||||
for item in ret:
|
||||
text = item.get("text", "").strip()
|
||||
if not text:
|
||||
self.fail("Generated text is empty.")
|
||||
|
||||
match = False
|
||||
for pattern in expected_patterns:
|
||||
if self.regex_match(text, pattern):
|
||||
match = True
|
||||
break
|
||||
if not match:
|
||||
self.fail(f"Text '{text}' does not match any of the allowed patterns.")
|
||||
|
||||
def regex_match(self, text, pattern):
|
||||
import re
|
||||
|
||||
return re.match(pattern, text) is not None
|
||||
|
||||
def test_ebnf_generate_email(self):
|
||||
self.__class__.ebnf_grammar = 'root ::= "user@example.com"'
|
||||
allowed_patterns = [r"^user@example\.com$"]
|
||||
prompt = "Generate an email address:"
|
||||
|
||||
self.run_decode(
|
||||
ebnf=self.__class__.ebnf_grammar,
|
||||
expected_patterns=allowed_patterns,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_ebnf_generate_greeting(self):
|
||||
self.__class__.ebnf_grammar = 'root ::= "Hello" | "Hi" | "Hey"'
|
||||
allowed_patterns = [r"^(Hello|Hi|Hey)$"]
|
||||
prompt = "Generate a greeting:"
|
||||
|
||||
self.run_decode(
|
||||
ebnf=self.__class__.ebnf_grammar,
|
||||
expected_patterns=allowed_patterns,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_ebnf_generate_number(self):
|
||||
self.__class__.ebnf_grammar = """
|
||||
root ::= digit digit digit
|
||||
digit ::= [0-9]
|
||||
"""
|
||||
allowed_patterns = [r"^\d{3}$"]
|
||||
prompt = "Generate a three-digit number:"
|
||||
|
||||
self.run_decode(
|
||||
ebnf=self.__class__.ebnf_grammar,
|
||||
expected_patterns=allowed_patterns,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_ebnf_generate_phone(self):
|
||||
self.__class__.ebnf_grammar = """
|
||||
root ::= "(" area ")" " " prefix "-" line
|
||||
area ::= [0-9] [0-9] [0-9]
|
||||
prefix ::= [0-9] [0-9] [0-9]
|
||||
line ::= [0-9] [0-9] [0-9] [0-9]
|
||||
"""
|
||||
allowed_patterns = [r"^\(\d{3}\) \d{3}-\d{4}$"]
|
||||
prompt = "Generate a phone number:"
|
||||
|
||||
self.run_decode(
|
||||
ebnf=self.__class__.ebnf_grammar,
|
||||
expected_patterns=allowed_patterns,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_ebnf_generate_date(self):
|
||||
self.__class__.ebnf_grammar = """
|
||||
root ::= year "-" month "-" day
|
||||
year ::= "2024"
|
||||
month ::= "01" | "02" | "03" | "04" | "05" | "06" | "07" | "08" | "09" | "10" | "11" | "12"
|
||||
day ::= "01" | "02" | "03" | "04" | "05" | "06" | "07" | "08" | "09" | "10" |
|
||||
"11" | "12" | "13" | "14" | "15" | "16" | "17" | "18" | "19" | "20" |
|
||||
"21" | "22" | "23" | "24" | "25" | "26" | "27" | "28" | "29" | "30" | "31"
|
||||
"""
|
||||
allowed_patterns = [r"^2024-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])$"]
|
||||
prompt = "Generate a date in YYYY-MM-DD format:"
|
||||
|
||||
self.run_decode(
|
||||
ebnf=self.__class__.ebnf_grammar,
|
||||
expected_patterns=allowed_patterns,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_ebnf_generate_hex_color(self):
|
||||
self.__class__.ebnf_grammar = """
|
||||
root ::= "#" hex hex hex hex hex hex
|
||||
hex ::= [0-9] | [A-F]
|
||||
"""
|
||||
allowed_patterns = [r"^#[0-9A-F]{6}$"]
|
||||
prompt = "Generate a hex color code:"
|
||||
|
||||
self.run_decode(
|
||||
ebnf=self.__class__.ebnf_grammar,
|
||||
expected_patterns=allowed_patterns,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_ebnf_generate_complex_json(self):
|
||||
self.__class__.ebnf_grammar = """
|
||||
root ::= object
|
||||
object ::= "{" ws pair (ws "," ws pair)* ws "}"
|
||||
pair ::= "\\"name\\"" ws ":" ws value |
|
||||
"\\"age\\"" ws ":" ws number |
|
||||
"\\"city\\"" ws ":" ws string
|
||||
value ::= string | number
|
||||
string ::= "\\"" [a-zA-Z0-9 ]+ "\\""
|
||||
number ::= [1-9] [0-9]*
|
||||
ws ::= [ ]*
|
||||
"""
|
||||
allowed_patterns = [
|
||||
r'^{\s*"name"\s*:\s*"[a-zA-Z0-9 ]+"\s*,\s*"age"\s*:\s*[1-9][0-9]*\s*,\s*"city"\s*:\s*"[a-zA-Z0-9 ]+"\s*}$',
|
||||
]
|
||||
prompt = "Generate a simple JSON with name, age, and city:"
|
||||
|
||||
self.run_decode(
|
||||
ebnf=self.__class__.ebnf_grammar,
|
||||
expected_patterns=allowed_patterns,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_ebnf_generate_custom_log_format(self):
|
||||
self.__class__.ebnf_grammar = """
|
||||
root ::= logentry
|
||||
logentry ::= "[" datetime "] " level ": System.process - " message
|
||||
datetime ::= "2024-01-01T12:00:00Z"
|
||||
level ::= "INFO"
|
||||
message ::= "Operation " [a-z]+ " successfully"
|
||||
"""
|
||||
allowed_patterns = [
|
||||
r"^\[2024-01-01T12:00:00Z\] INFO: System\.process - Operation [a-z]+ successfully$"
|
||||
]
|
||||
prompt = "Generate a log entry:"
|
||||
|
||||
self.run_decode(
|
||||
ebnf=self.__class__.ebnf_grammar,
|
||||
expected_patterns=allowed_patterns,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_ebnf_generate_all_optional_function_params(self):
|
||||
"""Test function call with all optional parameters - verifies flexible ordering."""
|
||||
self.__class__.ebnf_grammar = """
|
||||
root ::= function_call
|
||||
function_call ::= call_config_service
|
||||
call_config_service ::= "{" "\\"name\\"" ":" "\\"config_service\\"" ", " "\\"arguments\\"" ":" arguments_config_service "}"
|
||||
arguments_config_service ::= "{" ( "\\"theme\\"" ":" ("\\"light\\"" | "\\"dark\\"") ( "," "\\"language\\"" ":" ("\\"en\\"" | "\\"es\\"" | "\\"fr\\"") )? ( "," "\\"notifications\\"" ":" ("true" | "false") )? | "\\"language\\"" ":" ("\\"en\\"" | "\\"es\\"" | "\\"fr\\"") ( "," "\\"notifications\\"" ":" ("true" | "false") )? | "\\"notifications\\"" ":" ("true" | "false") )? "}"
|
||||
"""
|
||||
# Test patterns that should match - flexible ordering of optional parameters
|
||||
allowed_patterns = [
|
||||
# Empty arguments
|
||||
r'^\{"name":"config_service",\s*"arguments":\{\}\}$',
|
||||
# Single optional parameters (any can appear first)
|
||||
r'^\{"name":"config_service",\s*"arguments":\{"theme":"(light|dark)"\}\}$',
|
||||
r'^\{"name":"config_service",\s*"arguments":\{"language":"(en|es|fr)"\}\}$',
|
||||
r'^\{"name":"config_service",\s*"arguments":\{"notifications":(true|false)\}\}$',
|
||||
# Two optional parameters (in any order)
|
||||
r'^\{"name":"config_service",\s*"arguments":\{"theme":"(light|dark)",\s*"language":"(en|es|fr)"\}\}$',
|
||||
r'^\{"name":"config_service",\s*"arguments":\{"theme":"(light|dark)",\s*"notifications":(true|false)\}\}$',
|
||||
r'^\{"name":"config_service",\s*"arguments":\{"language":"(en|es|fr)",\s*"notifications":(true|false)\}\}$',
|
||||
# All three optional parameters
|
||||
r'^\{"name":"config_service",\s*"arguments":\{"theme":"(light|dark)",\s*"language":"(en|es|fr)",\s*"notifications":(true|false)\}\}$',
|
||||
]
|
||||
prompt = "Configure the service with optional settings:"
|
||||
|
||||
self.run_decode(
|
||||
ebnf=self.__class__.ebnf_grammar,
|
||||
expected_patterns=allowed_patterns,
|
||||
prompt=prompt,
|
||||
n=5,
|
||||
)
|
||||
|
||||
|
||||
class TestEBNFConstrainedLLGuidance(TestEBNFConstrained):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
setup_class(cls, "llguidance", disable_overlap=False)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,187 +0,0 @@
|
||||
"""
|
||||
python3 -m unittest test_regex_constrained.TestRegexConstrained.test_regex_generate_email
|
||||
python3 -m unittest test_regex_constrained.TestRegexConstrained.test_regex_generate_greeting
|
||||
python3 -m unittest test_regex_constrained.TestRegexConstrainedLLGuidance.test_regex_generate_email
|
||||
python3 -m unittest test_regex_constrained.TestRegexConstrainedLLGuidance.test_regex_generate_greeting
|
||||
"""
|
||||
|
||||
import json
|
||||
import unittest
|
||||
|
||||
import requests
|
||||
|
||||
from sglang.srt.utils import kill_process_tree
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
def setup_class(cls, backend: str, disable_overlap: bool):
|
||||
cls.model = DEFAULT_SMALL_MODEL_NAME_FOR_TEST
|
||||
cls.base_url = DEFAULT_URL_FOR_TEST
|
||||
|
||||
other_args = [
|
||||
"--max-running-requests",
|
||||
"10",
|
||||
"--grammar-backend",
|
||||
backend,
|
||||
]
|
||||
|
||||
if disable_overlap:
|
||||
other_args += ["--disable-overlap-schedule"]
|
||||
|
||||
cls.process = popen_launch_server(
|
||||
cls.model,
|
||||
cls.base_url,
|
||||
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
||||
other_args=other_args,
|
||||
)
|
||||
|
||||
|
||||
class TestRegexConstrained(CustomTestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
setup_class(cls, "xgrammar", disable_overlap=False)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
kill_process_tree(cls.process.pid)
|
||||
|
||||
def run_decode(
|
||||
self,
|
||||
regex,
|
||||
prompt,
|
||||
return_logprob=False,
|
||||
top_logprobs_num=0,
|
||||
n=1,
|
||||
):
|
||||
response = requests.post(
|
||||
self.base_url + "/generate",
|
||||
json={
|
||||
"text": prompt,
|
||||
"sampling_params": {
|
||||
"temperature": 0 if n == 1 else 0.5,
|
||||
"max_new_tokens": 128,
|
||||
"n": n,
|
||||
"regex": regex,
|
||||
},
|
||||
"stream": False,
|
||||
"return_logprob": return_logprob,
|
||||
"top_logprobs_num": top_logprobs_num,
|
||||
"logprob_start_len": 0,
|
||||
},
|
||||
)
|
||||
|
||||
ret = response.json()
|
||||
print(json.dumps(ret, indent=2))
|
||||
print("=" * 100)
|
||||
|
||||
if not isinstance(ret, list):
|
||||
self.fail(f"Expected response to be a list, but got {type(ret)}")
|
||||
|
||||
for item in ret:
|
||||
text = item.get("text", "").strip()
|
||||
if not text:
|
||||
self.fail("Generated text is empty.")
|
||||
|
||||
if not self.regex_match(text, regex):
|
||||
self.fail(f"Text '{text}' does not match regex pattern.")
|
||||
|
||||
def regex_match(self, text, pattern):
|
||||
import re
|
||||
|
||||
return re.match(pattern, text) is not None
|
||||
|
||||
def test_regex_generate_email(self):
|
||||
pattern = r"^user@example\.com$"
|
||||
prompt = "Generate an email address:"
|
||||
|
||||
self.run_decode(
|
||||
regex=pattern,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_regex_generate_greeting(self):
|
||||
pattern = r"^(Hello|Hi|Hey)$"
|
||||
prompt = "Generate a greeting:"
|
||||
|
||||
self.run_decode(
|
||||
regex=pattern,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_regex_generate_number(self):
|
||||
pattern = r"^\d{3}$"
|
||||
prompt = "Generate a three-digit number:"
|
||||
|
||||
self.run_decode(
|
||||
regex=pattern,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_regex_generate_phone(self):
|
||||
pattern = r"^\(\d{3}\) \d{3}-\d{4}$"
|
||||
prompt = "Generate a phone number:"
|
||||
|
||||
self.run_decode(
|
||||
regex=pattern,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_regex_generate_date(self):
|
||||
pattern = r"^2024-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])$"
|
||||
prompt = "Generate a date in YYYY-MM-DD format:"
|
||||
|
||||
self.run_decode(
|
||||
regex=pattern,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_regex_generate_hex_color(self):
|
||||
pattern = r"^#[0-9A-F]{6}$"
|
||||
prompt = "Generate a hex color code:"
|
||||
|
||||
self.run_decode(
|
||||
regex=pattern,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_regex_generate_complex_json(self):
|
||||
pattern = r'^\{\s*"name"\s*:\s*"[a-zA-Z0-9 ]+"\s*,\s*"age"\s*:\s*[1-9][0-9]*\s*,\s*"city"\s*:\s*"[a-zA-Z0-9 ]+"\s*\}$'
|
||||
prompt = "Generate a simple JSON with name, age, and city:"
|
||||
|
||||
self.run_decode(
|
||||
regex=pattern,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
def test_regex_generate_custom_log_format(self):
|
||||
pattern = r"^\[2024-01-01T12:00:00Z\] INFO: System\.process - Operation [a-z]+ successfully$"
|
||||
prompt = "Generate a log entry:"
|
||||
|
||||
self.run_decode(
|
||||
regex=pattern,
|
||||
prompt=prompt,
|
||||
n=3,
|
||||
)
|
||||
|
||||
|
||||
class TestRegexConstrainedLLGuidance(TestRegexConstrained):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
setup_class(cls, "llguidance", disable_overlap=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user