diff --git a/python/sglang/srt/function_call/mimo_detector.py b/python/sglang/srt/function_call/mimo_detector.py index 1af745c82d88..6a273f6a4bcd 100644 --- a/python/sglang/srt/function_call/mimo_detector.py +++ b/python/sglang/srt/function_call/mimo_detector.py @@ -12,7 +12,6 @@ # limitations under the License. # ============================================================================== -import html import json import logging import re @@ -24,6 +23,7 @@ from sglang.srt.function_call.core_types import StreamingParseResult, _GetInfoFunc from sglang.srt.function_call.utils import ( get_schema_properties, + infer_type_from_json_schema, safe_literal_eval, ) @@ -36,7 +36,7 @@ def _get_param_type(func_name: str, param_name: str, tools: List[Tool]) -> str: if tool.function.name == func_name: props = get_schema_properties(tool.function.parameters) if param_name in props: - return props[param_name].get("type", "string") + return infer_type_from_json_schema(props[param_name]) or "string" return "string" @@ -47,16 +47,12 @@ def _convert_param_value( Convert parameter value based on its type in the schema. Adapted from vllm-project/vllm (vllm/entrypoints/openai/tool_parsers/qwen3coder_tool_parser.py) """ - param_value = html.unescape(param_value) - - # Handle null value for any type - if param_value.lower() == "null": - return None - param_type = _get_param_type(func_name, param_name, tools) if param_type in ["string", "str", "text", "varchar", "char", "enum"]: return param_value + elif param_value == "null": + return None elif ( param_type.startswith("int") or param_type.startswith("integer") @@ -278,7 +274,88 @@ def _parse_tool_call( return {"name": func_name, "parameters": params} def supports_structural_tag(self) -> bool: - return False + return True def structure_info(self) -> _GetInfoFunc: raise NotImplementedError + + def get_structural_tag_name(self) -> str: + return "mimo" + + def get_structural_tag( + self, + tools=None, + tool_choice="auto", + thinking_mode=False, + parallel_tool_calls=True, + ): + try: + return super().get_structural_tag( + tools, tool_choice, thinking_mode, parallel_tool_calls + ) + except ValueError as error: + if not str(error).startswith("Unknown format type: mimo,"): + raise + + # XGrammar 0.2.7 has qwen_xml, but predates the compact MiMo builtin. + from xgrammar import normalize_tool_choice + from xgrammar.structural_tag import ( + AnyTextFormat, + JSONSchemaFormat, + SequenceFormat, + StructuralTag, + TagFormat, + TriggeredTagsFormat, + ) + + from sglang.srt.entrypoints.openai.protocol import ToolChoice + + functions, builtins, choice = normalize_tool_choice( + [tool.model_dump() for tool in tools or []], + tool_choice.model_dump() + if isinstance(tool_choice, ToolChoice) + else tool_choice, + ) + if builtins: + raise ValueError("MiMo does not support builtin tools.") + tags = [ + TagFormat( + begin=f"", + content=JSONSchemaFormat( + json_schema=( + tool.function.parameters + if tool.function.strict is not False + and tool.function.parameters + else True + ), + style="qwen_xml", + ), + end="", + ) + for tool in functions + ] + excludes = ["", "", "", ""], + tags=tags, + excludes=excludes, + at_least_one=choice == "required", + stop_after_first=not parallel_tool_calls, + ) + else: + suffix = AnyTextFormat(excludes=["", *excludes]) + if thinking_mode: + suffix = SequenceFormat( + elements=[ + TagFormat( + begin="", + content=AnyTextFormat(excludes=["", *excludes]), + end="", + ), + suffix, + ] + ) + return StructuralTag(format=suffix) 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 359ae726dbfc..20aa22a5f5f9 100644 --- a/test/registered/unit/function_call/test_function_call_parser.py +++ b/test/registered/unit/function_call/test_function_call_parser.py @@ -33,6 +33,7 @@ from sglang.srt.function_call.lfm2_detector import Lfm2Detector from sglang.srt.function_call.ling3_detector import Ling3Detector from sglang.srt.function_call.llama32_detector import Llama32Detector +from sglang.srt.function_call.mimo_detector import MiMoDetector from sglang.srt.function_call.mistral_detector import MistralDetector from sglang.srt.function_call.parser_names import TOOL_CALL_PARSER_NAMES from sglang.srt.function_call.pythonic_detector import PythonicDetector @@ -6024,5 +6025,137 @@ def test_matches_registry(self): ) +class TestMiMoStructuralTag(unittest.TestCase): + def setUp(self): + self.tools = [ + Tool( + type="function", + function=Function( + name="run", + strict=True, + parameters={ + "type": "object", + "properties": { + "text": {"type": "string"}, + "n": {"type": ["integer", "null"]}, + "data": {"type": "object"}, + }, + "required": ["text", "n", "data"], + "additionalProperties": False, + }, + ), + ) + ] + self.call = ( + "\n北京 &\n" + '2{"items": [true, null]}' + ) + + def test_compact_xml_schema_constraints(self): + from xgrammar.testing import _is_grammar_accept_string + + for policy in [ + "auto", + "required", + ToolChoice(type="function", function=ToolChoiceFuncName(name="run")), + ]: + for thinking in [False, True]: + with self.subTest(policy=policy, thinking=thinking): + detector = MiMoDetector() + self.assertTrue(detector.supports_structural_tag()) + tag = detector.get_structural_tag( + self.tools, policy, thinking_mode=thinking + ) + grammar = xgr.Grammar.from_structural_tag(tag) + prefix = "plan" if thinking else "" + self.assertTrue( + _is_grammar_accept_string(grammar, prefix + self.call) + ) + self.assertEqual( + _is_grammar_accept_string(grammar, prefix + "answer"), + policy == "auto", + ) + for invalid in [ + self.call.replace("function=run", "function=unknown"), + self.call.replace("2", ""), + self.call.replace("2", "bad"), + self.call.replace("", ""), + self.call.replace( + "", "3" + ), + ]: + self.assertFalse( + _is_grammar_accept_string(grammar, prefix + invalid) + ) + self.assertEqual( + _is_grammar_accept_string(grammar, prefix + self.call * 2), + isinstance(policy, str), + ) + single = xgr.Grammar.from_structural_tag( + detector.get_structural_tag( + self.tools, + policy, + thinking_mode=thinking, + parallel_tool_calls=False, + ) + ) + self.assertTrue( + _is_grammar_accept_string(single, prefix + self.call) + ) + self.assertFalse( + _is_grammar_accept_string(single, prefix + self.call * 2) + ) + + def test_non_strict_keeps_native_parameter_format(self): + from xgrammar.testing import _is_grammar_accept_string + + self.tools[0].function.strict = False + grammar = xgr.Grammar.from_structural_tag( + MiMoDetector().get_structural_tag(self.tools, "required") + ) + self.assertTrue( + _is_grammar_accept_string( + grammar, + "anything", + ) + ) + + def test_parameters_round_trip_without_html_or_string_coercion(self): + for text in ["\n北京 &\n", "null", "42", ""]: + for number in ["2", "null"]: + with self.subTest(text=text, number=number): + call = self.call.replace("\n北京 &\n", text).replace( + "2", "" + number + ) + result = MiMoDetector().detect_and_parse(call, self.tools) + self.assertEqual( + json.loads(result.calls[0].parameters), + { + "text": text, + "n": json.loads(number), + "data": {"items": [True, None]}, + }, + ) + + def test_streamed_compact_calls_preserve_arguments(self): + # MiMo's tool-call markers are atomic tokenizer tokens; parameter tags are not. + chunks = [ + "", + *self.call[len("") : -len("")], + "", + ] + detector = MiMoDetector() + calls = [] + for chunk in chunks * 2: + calls.extend(detector.parse_streaming_increment(chunk, self.tools).calls) + self.assertEqual(len(calls), 2) + self.assertEqual([call.tool_index for call in calls], [0, 1]) + for call in calls: + self.assertEqual( + json.loads(call.parameters), + {"text": "\n北京 &\n", "n": 2, "data": {"items": [True, None]}}, + ) + + if __name__ == "__main__": unittest.main()