EngineX-Hygon/sglang
5
0
Fork 0
Code Issues Pull Requests Actions 7 Projects Releases Wiki Activity

GLM-4.5 Model Support (#8224)

Browse Source 
Co-authored-by: Lifu Huang <lifu.hlf@gmail.com>
Co-authored-by: Binyao Jiang <byjiang1996@gmail.com>
Co-authored-by: Stefan He <hebiaobuaa@gmail.com>
This commit is contained in:
Yuxuan Zhang
2025-07-28 13:54:07 +08:00
committed by GitHub
parent 2fd5c7049f
commit 6d6a8bc278
14 changed files with 1673 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

184
test/srt/test_function_call_parser.py
View File

@@ -6,6 +6,7 @@ from xgrammar import GrammarCompiler, TokenizerInfo
from sglang.srt.entrypoints.openai.protocol import Function, Tool
from sglang.srt.function_call.base_format_detector import BaseFormatDetector
from sglang.srt.function_call.deepseekv3_detector import DeepSeekV3Detector
from sglang.srt.function_call.glm4_moe_detector import Glm4MoeDetector
from sglang.srt.function_call.kimik2_detector import KimiK2Detector
from sglang.srt.function_call.llama32_detector import Llama32Detector
from sglang.srt.function_call.mistral_detector import MistralDetector
@@ -510,6 +511,7 @@ class TestEBNFGeneration(unittest.TestCase):
self.qwen25_detector = Qwen25Detector()
self.qwen3_coder_detector = Qwen3CoderDetector()
self.kimik2_detector = KimiK2Detector()
self.glm45_detector = Glm4MoeDetector()
def test_pythonic_detector_ebnf(self):
"""Test that the PythonicDetector generates valid EBNF."""
@@ -622,6 +624,29 @@ class TestEBNFGeneration(unittest.TestCase):
except RuntimeError as e:
self.fail(f"Failed to compile EBNF: {e}")
def test_glm45_detector_ebnf(self):
"""Test that the Glm4MoeDetector generates valid EBNF."""
ebnf = self.glm45_detector.build_ebnf(self.tools)
self.assertIsNotNone(ebnf)
# Check that the EBNF contains expected patterns for XML format
self.assertIn('"<tool_call>" function_call "</tool_call>"', ebnf)
self.assertIn('"get_weather" "\\n" arguments_get_weather', ebnf)
self.assertIn(
'"<arg_key>location</arg_key>" "\\n" "<arg_value>" xml_text "</arg_value>" ( "\\n" ( "<arg_key>unit</arg_key>" "\\n" "<arg_value>" ("celsius" | "fahrenheit") "</arg_value>" ) )?',
ebnf,
)
self.assertIn('"search" "\\n" arguments_search', ebnf)
self.assertIn(
'"<arg_key>query</arg_key>" "\\n" "<arg_value>" xml_text "</arg_value>"',
ebnf,
)
# Validate that the EBNF can be compiled by GrammarCompiler
try:
ctx = self.grammar_compiler.compile_grammar(ebnf)
self.assertIsNotNone(ctx, "EBNF should be valid and compile successfully")
except RuntimeError as e:
self.fail(f"Failed to compile EBNF: {e}")
def test_qwen3_coder_detector_ebnf(self):
"""Test that the Qwen3CoderDetector generates valid EBNF."""
ebnf = self.qwen3_coder_detector.build_ebnf(self.tools)
@@ -1919,5 +1944,164 @@ circle
self.assertEqual(params2["dimensions"], {"radius": 5})
class TestGlm4MoeDetector(unittest.TestCase):
def setUp(self):
self.tools = [
Tool(
type="function",
function=Function(
name="get_weather",
description="Get weather information",
parameters={
"type": "object",
"properties": {
"city": {"type": "string", "description": "City name"},
"date": {"type": "string", "description": "Date"},
},
"required": ["city", "date"],
},
),
),
]
self.detector = Glm4MoeDetector()
def test_single_tool_call(self):
text = (
"<tool_call>get_weather\n"
"<arg_key>city</arg_key>\n<arg_value>Beijing</arg_value>\n"
"<arg_key>date</arg_key>\n<arg_value>2024-06-27</arg_value>\n"
"</tool_call>"
)
result = self.detector.detect_and_parse(text, self.tools)
self.assertEqual(len(result.calls), 1)
self.assertEqual(result.calls[0].name, "get_weather")
self.assertEqual(
result.calls[0].parameters, '{"city": "Beijing", "date": "2024-06-27"}'
)
self.assertEqual(result.normal_text, "")
def test_multiple_tool_calls(self):
text = (
"<tool_call>get_weather\n"
"<arg_key>city</arg_key>\n<arg_value>Beijing</arg_value>\n"
"<arg_key>date</arg_key>\n<arg_value>2024-06-27</arg_value>\n"
"</tool_call>"
"<tool_call>get_weather\n"
"<arg_key>city</arg_key>\n<arg_value>Shanghai</arg_value>\n"
"<arg_key>date</arg_key>\n<arg_value>2024-06-28</arg_value>\n"
"</tool_call>"
)
result = self.detector.detect_and_parse(text, self.tools)
self.assertEqual(len(result.calls), 2)
self.assertEqual(result.calls[0].name, "get_weather")
self.assertEqual(
result.calls[0].parameters, '{"city": "Beijing", "date": "2024-06-27"}'
)
self.assertEqual(result.calls[1].name, "get_weather")
self.assertEqual(
result.calls[1].parameters, '{"city": "Shanghai", "date": "2024-06-28"}'
)
self.assertEqual(result.normal_text, "")
def test_streaming_tool_call(self):
"""Test streaming incremental parsing of a tool call."""
chunks = [
"<tool_call>get_weather\n",
"<arg_key>city</arg_key>\n<arg_value>Beijing</arg_value>\n",
"<arg_key>date</arg_key>\n<arg_value>2024-06-27</arg_value>\n",
"</tool_call>",
]
tool_calls = []
for chunk in chunks:
result = self.detector.parse_streaming_increment(chunk, self.tools)
for tool_call_chunk in result.calls:
if (
hasattr(tool_call_chunk, "tool_index")
and tool_call_chunk.tool_index is not None
):
while len(tool_calls) <= tool_call_chunk.tool_index:
tool_calls.append({"name": "", "parameters": {}})
tc = tool_calls[tool_call_chunk.tool_index]
if tool_call_chunk.name:
tc["name"] = tool_call_chunk.name
if tool_call_chunk.parameters:
tc["parameters"] = tool_call_chunk.parameters
self.assertEqual(len(tool_calls), 1)
self.assertEqual(tool_calls[0]["name"], "get_weather")
self.assertEqual(
tool_calls[0]["parameters"], '{"city": "Beijing", "date": "2024-06-27"}'
)
def test_streaming_multiple_tool_calls(self):
"""Test streaming incremental parsing of multiple tool calls."""
chunks = [
"<tool_call>get_weather\n",
"<arg_key>city</arg_key>\n<arg_value>Beijing</arg_value>\n",
"<arg_key>date</arg_key>\n<arg_value>2024-06-27</arg_value>\n",
"</tool_call><tool_call>get_weather\n",
"<arg_key>city</arg_key>\n<arg_value>Shanghai</arg_value>\n",
"<arg_key>date</arg_key>\n<arg_value>2024-06-28</arg_value>\n",
"</tool_call>",
]
tool_calls = []
for chunk in chunks:
result = self.detector.parse_streaming_increment(chunk, self.tools)
for tool_call_chunk in result.calls:
if (
hasattr(tool_call_chunk, "tool_index")
and tool_call_chunk.tool_index is not None
):
while len(tool_calls) <= tool_call_chunk.tool_index:
tool_calls.append({"name": "", "parameters": {}})
tc = tool_calls[tool_call_chunk.tool_index]
if tool_call_chunk.name:
tc["name"] = tool_call_chunk.name
if tool_call_chunk.parameters:
tc["parameters"] = tool_call_chunk.parameters
self.assertEqual(len(tool_calls), 2)
self.assertEqual(tool_calls[0]["name"], "get_weather")
self.assertEqual(
tool_calls[0]["parameters"], '{"city": "Beijing", "date": "2024-06-27"}'
)
self.assertEqual(tool_calls[1]["name"], "get_weather")
self.assertEqual(
tool_calls[1]["parameters"], '{"city": "Shanghai", "date": "2024-06-28"}'
)
def test_tool_call_completion(self):
"""Test that the buffer and state are reset after a tool call is completed."""
chunks = [
"<tool_call>get_weather\n",
"<arg_key>city</arg_key>\n<arg_value>Beijing</arg_value>\n",
"<arg_key>date</arg_key>\n<arg_value>2024-06-27</arg_value>\n",
"</tool_call>",
]
for chunk in chunks:
result = self.detector.parse_streaming_increment(chunk, self.tools)
self.assertEqual(self.detector.current_tool_id, 1)
def test_invalid_tool_call(self):
"""Test that invalid tool calls are handled correctly."""
text = "<tool_call>invalid_func\n<arg_key>city</arg_key>\n<arg_value>Beijing</arg_value>\n</tool_call>"
result = self.detector.detect_and_parse(text, self.tools)
self.assertEqual(len(result.calls), 0)
def test_partial_tool_call(self):
"""Test parsing a partial tool call that spans multiple chunks."""
text1 = "<tool_call>get_weather\n<arg_key>city</arg_key>\n"
result1 = self.detector.parse_streaming_increment(text1, self.tools)
self.assertEqual(result1.normal_text, "")
self.assertEqual(result1.calls, [])
self.assertEqual(self.detector._buffer, text1)
text2 = "<arg_value>Beijing</arg_value>\n<arg_key>date</arg_key>\n<arg_value>2024-06-27</arg_value>\n</tool_call>"
result2 = self.detector.parse_streaming_increment(text2, self.tools)
self.assertEqual(len(result2.calls), 1)
self.assertEqual(result2.calls[0].name, "get_weather")
self.assertEqual(
result2.calls[0].parameters, '{"city": "Beijing", "date": "2024-06-27"}'
)
self.assertEqual(self.detector._buffer, "")
if __name__ == "__main__":
unittest.main()