From 30a35ecd90227216a72833aa29ef02dc067db827 Mon Sep 17 00:00:00 2001 From: Viacheslav Date: Wed, 18 Mar 2026 08:45:01 +0300 Subject: [PATCH] Add gigachat3.1 parser (#19886) Signed-off-by: Viacheslav Barinov Signed-off-by: Viacheslav Bv Co-authored-by: Viacheslav Barinov --- .../srt/function_call/gigachat3_detector.py | 19 +-- .../test_function_call_parser.py | 129 ++++++++++++++++++ 2 files changed, 135 insertions(+), 13 deletions(-) diff --git a/python/sglang/srt/function_call/gigachat3_detector.py b/python/sglang/srt/function_call/gigachat3_detector.py index ef94bdf9a..8e2e2d631 100644 --- a/python/sglang/srt/function_call/gigachat3_detector.py +++ b/python/sglang/srt/function_call/gigachat3_detector.py @@ -14,12 +14,12 @@ from sglang.srt.function_call.core_types import ( logger = logging.getLogger(__name__) REGEX_FUNCTION_CALL = re.compile( - r"function call<\|role_sep\|>\n(.*)", + r"(?:function call<\|role_sep\|>\n|<\|function_call\|>)(.*)", re.DOTALL, ) REGEX_CONTENT_PATTERN = re.compile( - r"^(.*?)<\|message_sep\|>", + r"^(.*?)(?:<\|message_sep\|>|<\|function_call\|>)", re.DOTALL, ) @@ -45,7 +45,7 @@ class GigaChat3Detector(BaseFormatDetector): def has_tool_call(self, text: str) -> bool: """Check if text contains a tool call marker""" - return "function call<|role_sep|>\n" in text + return "function call<|role_sep|>\n" in text or "<|function_call|>" in text def detect_and_parse( self, @@ -84,10 +84,7 @@ class GigaChat3Detector(BaseFormatDetector): if m_content: content = m_content.group(1) else: - if "<|message_sep|>" in model_output: - content = model_output.split("<|message_sep|>")[0] - else: - content = model_output + content = model_output if not function_call: return StreamingParseResult(normal_text=content, calls=[]) name = function_call["name"] @@ -121,12 +118,8 @@ class GigaChat3Detector(BaseFormatDetector): content = m_content.group(1) self.end_content = True else: - if "<|message_sep|>" in delta_text: - content = delta_text.split("<|message_sep|>")[0] - self.end_content = True - else: - if not self.end_content: - content = delta_text + if not self.end_content: + content = delta_text if m_func: self.tool_started = True logger.debug("[GigaChat3] Tool call started") diff --git a/test/registered/unit/function_call/test_function_call_parser.py b/test/registered/unit/function_call/test_function_call_parser.py index 5e8c1928b..ca67e2bf2 100644 --- a/test/registered/unit/function_call/test_function_call_parser.py +++ b/test/registered/unit/function_call/test_function_call_parser.py @@ -3274,6 +3274,7 @@ class TestGigaChat3Detector(unittest.TestCase): def test_has_tool_call(self): """Test detection of tool call markers.""" self.assertTrue(self.detector.has_tool_call("function call<|role_sep|>\n{}")) + self.assertTrue(self.detector.has_tool_call("<|function_call|>{}")) self.assertFalse(self.detector.has_tool_call("No tool call here")) def test_detect_and_parse_no_tool_call(self): @@ -3723,6 +3724,134 @@ function call<|role_sep|> params = json.loads(tool_calls_by_index[0]["parameters"]) self.assertEqual(params["city"], "Rome") + def test_detect_and_parse_function_call_marker_simple_tool_call(self): + """Test parsing a simple <|function_call|> tool call (GigaChat3.1-style).""" + text = '<|function_call|>{"name": "manage_user_memory", "arguments": {"action": "create", "id": "preferences"}}' + result = self.detector.detect_and_parse(text, self.tools) + + self.assertEqual(result.normal_text, "") + self.assertEqual(len(result.calls), 1) + self.assertEqual(result.calls[0].name, "manage_user_memory") + + params = json.loads(result.calls[0].parameters) + self.assertEqual(params["action"], "create") + self.assertEqual(params["id"], "preferences") + + def test_detect_and_parse_function_call_marker_with_content_before(self): + """Test parsing <|function_call|> tool call with prefix content.""" + text = ( + 'I\'ll check that for you.<|function_call|>{"name": "get_weather", ' + '"arguments": {"city": "Tokyo"}}' + ) + result = self.detector.detect_and_parse(text, self.tools) + + self.assertEqual(result.normal_text, "I'll check that for you.") + self.assertEqual(len(result.calls), 1) + self.assertEqual(result.calls[0].name, "get_weather") + + params = json.loads(result.calls[0].parameters) + self.assertEqual(params["city"], "Tokyo") + + def test_detect_and_parse_function_call_marker_with_eos_token(self): + """Test parsing <|function_call|> tool call with EOS token at the end.""" + text = '<|function_call|>{"name": "manage_user_memory", "arguments": {"action": "create", "id": "preferences"}}' + result = self.detector.detect_and_parse(text, self.tools) + + self.assertEqual(result.normal_text, "") + self.assertEqual(len(result.calls), 1) + self.assertEqual(result.calls[0].name, "manage_user_memory") + + params = json.loads(result.calls[0].parameters) + self.assertEqual(params["action"], "create") + self.assertEqual(params["id"], "preferences") + + def test_detect_and_parse_function_call_marker_invalid_json(self): + """Test parsing invalid JSON after <|function_call|> marker.""" + text = '<|function_call|>{"name": "manage_user_memory", "arguments": {invalid json}}' + result = self.detector.detect_and_parse(text, self.tools) + + self.assertIn("<|function_call|>", result.normal_text) + self.assertEqual(len(result.calls), 0) + + def test_streaming_function_call_marker_simple_tool_call(self): + """Test streaming parsing of the <|function_call|> marker form.""" + chunks = [ + "I'll help you.", + "<|function_call|>", + '{"name": "manage_user_memory", "arguments": ', + '{"action": "create", "id": "prefs"}}', + ] + + accumulated_text = "" + tool_calls_by_index = {} + + for chunk in chunks: + result = self.detector.parse_streaming_increment(chunk, self.tools) + accumulated_text += result.normal_text + + for call in result.calls: + if call.tool_index is not None: + if call.tool_index not in tool_calls_by_index: + tool_calls_by_index[call.tool_index] = { + "name": "", + "parameters": "", + } + + if call.name: + tool_calls_by_index[call.tool_index]["name"] = call.name + if call.parameters: + tool_calls_by_index[call.tool_index][ + "parameters" + ] += call.parameters + + self.assertEqual(accumulated_text, "I'll help you.") + self.assertEqual(len(tool_calls_by_index), 1) + self.assertEqual(tool_calls_by_index[0]["name"], "manage_user_memory") + + params = json.loads(tool_calls_by_index[0]["parameters"]) + self.assertEqual(params["action"], "create") + self.assertEqual(params["id"], "prefs") + + def test_streaming_function_call_marker_json_split_at_quotes(self): + """Test streaming when JSON is split at quote boundaries (<|function_call|>).""" + chunks = [ + "<|function_call|>", + '{"name', + '": "', + "get_weather", + '", "arguments', + '": {"city', + '": "', + "Rome", + '"}}', + ] + + tool_calls_by_index = {} + + for chunk in chunks: + result = self.detector.parse_streaming_increment(chunk, self.tools) + + for call in result.calls: + if call.tool_index is not None: + if call.tool_index not in tool_calls_by_index: + tool_calls_by_index[call.tool_index] = { + "name": "", + "parameters": "", + } + + if call.name: + tool_calls_by_index[call.tool_index]["name"] = call.name + if call.parameters: + tool_calls_by_index[call.tool_index][ + "parameters" + ] += call.parameters + + self.assertEqual(len(tool_calls_by_index), 1) + self.assertEqual(tool_calls_by_index[0]["name"], "get_weather") + + params = json.loads(tool_calls_by_index[0]["parameters"]) + self.assertEqual(params["city"], "Rome") + if __name__ == "__main__": unittest.main()