diff --git a/3rdparty/amd/wheel/sglang/pyproject.toml b/3rdparty/amd/wheel/sglang/pyproject.toml index d039b1b7d..cbccd1008 100644 --- a/3rdparty/amd/wheel/sglang/pyproject.toml +++ b/3rdparty/amd/wheel/sglang/pyproject.toml @@ -62,7 +62,7 @@ runtime_common = [ "transformers==5.12.1", "uvicorn", "uvloop", - "xgrammar==0.2.1", + "xgrammar==0.2.7", "smg-grpc-servicer>=0.9.0", ] diff --git a/python/pyproject.toml b/python/pyproject.toml index fbf52feed..05ee34aa6 100755 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -96,7 +96,7 @@ dependencies = [ "uvicorn", "uvloop", "watchfiles", - "xgrammar==0.2.1", + "xgrammar==0.2.7", "xxhash", "zstandard", ] diff --git a/python/pyproject_cpu.toml b/python/pyproject_cpu.toml index ec9bb6a96..7ac9a0e1c 100644 --- a/python/pyproject_cpu.toml +++ b/python/pyproject_cpu.toml @@ -70,7 +70,7 @@ dependencies = [ "uvicorn", "uvloop", "xxhash", - "xgrammar==0.2.1", + "xgrammar==0.2.7", "zstandard", ] diff --git a/python/pyproject_npu.toml b/python/pyproject_npu.toml index 6796a4c0f..b48b4eb3a 100644 --- a/python/pyproject_npu.toml +++ b/python/pyproject_npu.toml @@ -67,7 +67,7 @@ dependencies = [ "uvicorn", "uvloop", "xxhash", - "xgrammar==0.2.1", + "xgrammar==0.2.7", ] [project.optional-dependencies] diff --git a/python/pyproject_other.toml b/python/pyproject_other.toml index 2822693dd..e26da61da 100755 --- a/python/pyproject_other.toml +++ b/python/pyproject_other.toml @@ -73,7 +73,7 @@ runtime_common = [ "compressed-tensors", "outlines==0.1.11", "timm==1.0.16", - "xgrammar==0.2.1", + "xgrammar==0.2.7", ] # srt_empty: device-agnostic install — pure Python packages only, no torch dependency chain. diff --git a/python/pyproject_xpu.toml b/python/pyproject_xpu.toml index ef6dfd33f..a8ccd97b2 100644 --- a/python/pyproject_xpu.toml +++ b/python/pyproject_xpu.toml @@ -69,7 +69,7 @@ dependencies = [ "uvicorn", "xxhash", "uvloop", - # "xgrammar==0.2.1", xgrammar depends on CUDA PyTorch and Triton only + # "xgrammar==0.2.7", xgrammar depends on CUDA PyTorch and Triton only ] [project.optional-dependencies] diff --git a/python/sglang/srt/arg_groups/field_order.py b/python/sglang/srt/arg_groups/field_order.py index 858860b3b..6b104eb4a 100644 --- a/python/sglang/srt/arg_groups/field_order.py +++ b/python/sglang/srt/arg_groups/field_order.py @@ -205,6 +205,7 @@ POSITIONAL_FIELD_ORDER = ( "stat_loggers", "constrained_json_whitespace_pattern", "constrained_json_disable_any_whitespace", + "constrained_json_max_whitespace_cnt", "attention_backend", "decode_attention_backend", "enable_lean_attention", diff --git a/python/sglang/srt/arg_groups/fields/serving.py b/python/sglang/srt/arg_groups/fields/serving.py index a85a3220f..11191f232 100644 --- a/python/sglang/srt/arg_groups/fields/serving.py +++ b/python/sglang/srt/arg_groups/fields/serving.py @@ -281,6 +281,10 @@ class Serving(msgspec.Struct): bool, "(xgrammar and llguidance backends only) Enforce compact representation in JSON constrained output.", ] = False + constrained_json_max_whitespace_cnt: A[ + Optional[int], + "(xgrammar backend only) Max consecutive whitespace chars allowed in JSON constrained output. None means unbounded.", + ] = None # ------------------------------------------------------------------------- # Dynamic batch tokenizer diff --git a/python/sglang/srt/constrained/base_grammar_backend.py b/python/sglang/srt/constrained/base_grammar_backend.py index 9f990755d..32a9ec2c7 100644 --- a/python/sglang/srt/constrained/base_grammar_backend.py +++ b/python/sglang/srt/constrained/base_grammar_backend.py @@ -385,6 +385,7 @@ def create_grammar_backend( vocab_size=vocab_size, model_eos_token_ids=eos_list, any_whitespace=not get_serving().constrained_json_disable_any_whitespace, + max_whitespace_cnt=get_serving().constrained_json_max_whitespace_cnt, ) except TokenizerNotSupportedError as e: if get_serving().enable_strict_thinking: diff --git a/python/sglang/srt/constrained/xgrammar_backend.py b/python/sglang/srt/constrained/xgrammar_backend.py index c4a1cf49e..86293e2d7 100644 --- a/python/sglang/srt/constrained/xgrammar_backend.py +++ b/python/sglang/srt/constrained/xgrammar_backend.py @@ -215,6 +215,7 @@ class XGrammarGrammarBackend(BaseGrammarBackend): vocab_size: int, model_eos_token_ids: Optional[List[int]] = None, any_whitespace: bool = True, + max_whitespace_cnt: Optional[int] = None, ): super().__init__() @@ -244,6 +245,7 @@ class XGrammarGrammarBackend(BaseGrammarBackend): self.vocab_size = vocab_size self.override_stop_tokens = override_stop_tokens self.any_whitespace = any_whitespace + self.max_whitespace_cnt = max_whitespace_cnt @property def is_support_token_filter(self): @@ -348,7 +350,9 @@ class XGrammarGrammarBackend(BaseGrammarBackend): schema = json.loads(key_string) validate_xgrammar_json_schema(schema) ctx = self.grammar_compiler.compile_json_schema( - schema=key_string, any_whitespace=self.any_whitespace + schema=key_string, + any_whitespace=self.any_whitespace, + max_whitespace_cnt=self.max_whitespace_cnt, ) except ( diff --git a/python/sglang/srt/function_call/base_format_detector.py b/python/sglang/srt/function_call/base_format_detector.py index e6f97c758..53c306714 100644 --- a/python/sglang/srt/function_call/base_format_detector.py +++ b/python/sglang/srt/function_call/base_format_detector.py @@ -404,10 +404,8 @@ class BaseFormatDetector(ABC): (the typical case when --reasoning-parser is configured) so only one layer constrains the reasoning section. parallel_tool_calls: Whether multiple tool calls may appear in one - assistant response. xgrammar's get_model_structural_tag does - not expose this knob, so this base implementation ignores it; - only detectors that build their own tags (e.g. Kimi K3) - honor it. + assistant response. Forwarded to XGrammar to constrain the + number of tool calls in the generated structural tag. Returns: StructuralTag if this detector supports model-native tags, otherwise None @@ -427,6 +425,7 @@ class BaseFormatDetector(ABC): tools=converted_tools, tool_choice=converted_tool_choice, reasoning=thinking_mode, + parallel_tool_calls=parallel_tool_calls, ) def get_auto_tool_call_structural_tag( diff --git a/python/sglang/srt/function_call/deepseekv32_detector.py b/python/sglang/srt/function_call/deepseekv32_detector.py index 0619cfb59..653ce7d50 100644 --- a/python/sglang/srt/function_call/deepseekv32_detector.py +++ b/python/sglang/srt/function_call/deepseekv32_detector.py @@ -75,6 +75,7 @@ class DeepSeekV32Detector(BaseFormatDetector): tool_calls_block_name = "function_calls" invoke_tag_name = "invoke" parameter_tag_name = "parameter" + strip_string_param_value: bool = True def __init__(self): super().__init__() @@ -170,7 +171,11 @@ class DeepSeekV32Detector(BaseFormatDetector): # Convert value based on type if param_type == "true": # string type - parameters[param_name] = param_value.strip() + parameters[param_name] = ( + param_value.strip() + if self.strip_string_param_value + else param_value + ) else: # Try to parse as JSON for other types try: @@ -195,7 +200,11 @@ class DeepSeekV32Detector(BaseFormatDetector): if partial_match and (param_value := partial_match.group(3)): param_name = partial_match.group(1) if partial_match.group(2) == "true": - parameters[param_name] = param_value.strip() + parameters[param_name] = ( + param_value.strip() + if self.strip_string_param_value + else param_value + ) else: try: parameters[param_name] = _partial_json_loads( diff --git a/python/sglang/srt/function_call/deepseekv41_detector.py b/python/sglang/srt/function_call/deepseekv41_detector.py index d1dc48af3..b6d7e1f59 100644 --- a/python/sglang/srt/function_call/deepseekv41_detector.py +++ b/python/sglang/srt/function_call/deepseekv41_detector.py @@ -1,18 +1,5 @@ -from typing import List, Literal, Optional, Union +from typing import Optional -from xgrammar.structural_tag import ( - AnyTextFormat, - ConstStringFormat, - JSONSchemaFormat, - OrFormat, - SequenceFormat, - TagFormat, - TagsWithSeparatorFormat, - TriggeredTagsFormat, -) - -from sglang.srt.entrypoints.openai.protocol import Tool, ToolChoice -from sglang.srt.function_call.base_format_detector import StructuralTag from sglang.srt.function_call.deepseekv32_detector import DeepSeekV32Detector @@ -25,6 +12,7 @@ class DeepSeekV41Detector(DeepSeekV32Detector): tool_calls_block_name = " calls" invoke_tag_name = " invoke" parameter_tag_name = " parameter" + strip_string_param_value: bool = False # The encoder joins an assistant turn's content and its calls block with a # blank line, and renders it even when there is no content. @@ -32,72 +20,4 @@ class DeepSeekV41Detector(DeepSeekV32Detector): think_end_token = "" def get_structural_tag_name(self) -> Optional[str]: - # xgrammar's builtin "deepseek_v4" tag hardcodes the unspaced names, - # so the V4.1 tag is assembled in get_structural_tag instead. - return None - - def get_structural_tag( - self, - tools: Union[List[Tool], None] = None, - tool_choice: Union[ToolChoice, Literal["auto", "required"]] = "auto", - thinking_mode: bool = False, - parallel_tool_calls: bool = True, - ) -> Optional[StructuralTag]: - """The builtin "deepseek_v4" shape with the spaced tag names. - - Bodies are JSON: xgrammar's "deepseek_xml" body style also hardcodes - the unspaced "parameter" name, and the V3.2-lineage parser accepts a - JSON body inside an invoke. - """ - tools = list(tools or []) - if isinstance(tool_choice, ToolChoice): - tools = [ - tool - for tool in tools - if tool.function.name == tool_choice.function.name - ] - if len(tools) != 1: - return None - if not tools: - return None - - def invoke_tag(tool: Tool) -> TagFormat: - function = tool.function - schema = function.parameters if function.strict else True - if schema is None: - schema = True - return TagFormat( - begin=f'{self.invoke_start_token} name="{function.name}">', - content=JSONSchemaFormat(json_schema=schema), - end=f"{self.invoke_end_token}\n", - ) - - tags = [invoke_tag(tool) for tool in tools] - if isinstance(tool_choice, ToolChoice): - calls = tags[0] - elif parallel_tool_calls: - calls = TagsWithSeparatorFormat(tags=tags, separator="", at_least_one=True) - else: - calls = OrFormat(elements=tags) - block_begin = f"{self.bot_token}\n" - - if tool_choice == "auto": - body = TriggeredTagsFormat( - triggers=[self.bot_token], - tags=[TagFormat(begin=block_begin, content=calls, end=self.eot_token)], - excludes=["", self.think_end_token], - ) - else: - body = SequenceFormat( - elements=[ - ConstStringFormat(value=self.tool_calls_prefix + block_begin), - calls, - ConstStringFormat(value=self.eot_token), - ] - ) - if not thinking_mode: - return StructuralTag(format=body) - reasoning = TagFormat( - begin="", content=AnyTextFormat(), end=self.think_end_token - ) - return StructuralTag(format=SequenceFormat(elements=[reasoning, body])) + return "deepseek_v4_1" diff --git a/python/sglang/srt/function_call/kimik2_detector.py b/python/sglang/srt/function_call/kimik2_detector.py index a8054b587..7263893f1 100644 --- a/python/sglang/srt/function_call/kimik2_detector.py +++ b/python/sglang/srt/function_call/kimik2_detector.py @@ -468,6 +468,7 @@ class KimiK2Detector(BaseFormatDetector): tools=converted_tools, tool_choice=converted_tool_choice, reasoning=thinking_mode, + parallel_tool_calls=parallel_tool_calls, ) def get_structural_tag_name(self) -> str: diff --git a/test/registered/dp_attn/test_dp_attention.py b/test/registered/dp_attn/test_dp_attention.py index 301e03530..7e7ea2b6f 100644 --- a/test/registered/dp_attn/test_dp_attention.py +++ b/test/registered/dp_attn/test_dp_attention.py @@ -50,6 +50,8 @@ class TestDPAttentionDP2TP2( timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH, other_args=[ "--trust-remote-code", + "--constrained-json-max-whitespace-cnt", + "4", "--tp", "2", "--enable-dp-attention", diff --git a/test/registered/unit/constrained/test_base_grammar_backend.py b/test/registered/unit/constrained/test_base_grammar_backend.py index 0c49ce598..5fe364dcc 100644 --- a/test/registered/unit/constrained/test_base_grammar_backend.py +++ b/test/registered/unit/constrained/test_base_grammar_backend.py @@ -259,6 +259,7 @@ class TestCreateGrammarBackend(unittest.TestCase): "enable_strict_thinking": enable_strict_thinking, "constrained_json_whitespace_pattern": None, "constrained_json_disable_any_whitespace": False, + "constrained_json_max_whitespace_cnt": None, } published.update(fields) self._publish(**published) @@ -335,7 +336,11 @@ class TestCreateGrammarBackend(unittest.TestCase): result = create_grammar_backend(args, "tok", 32000, {1, 2}) mock_xgrammar_cls.assert_called_once_with( - "tok", vocab_size=32000, model_eos_token_ids=[1, 2], any_whitespace=False + "tok", + vocab_size=32000, + model_eos_token_ids=[1, 2], + any_whitespace=False, + max_whitespace_cnt=None, ) self.assertIs(result, mock_backend) diff --git a/test/registered/unit/function_call/test_deepseekv41_detector.py b/test/registered/unit/function_call/test_deepseekv41_detector.py new file mode 100644 index 000000000..a7a960d95 --- /dev/null +++ b/test/registered/unit/function_call/test_deepseekv41_detector.py @@ -0,0 +1,337 @@ +"""Unit tests for DeepSeekV41Detector (spaced DSML tags) -- no server, no model loading.""" + +import json +import unittest +from typing import get_args + +import xgrammar as xgr +from xgrammar.structural_tag import JSONSchemaFormat +from xgrammar.testing import _is_grammar_accept_string + +from sglang.srt.entrypoints.openai import encoding_dsv41 +from sglang.srt.entrypoints.openai.protocol import ( + Function, + Tool, + ToolChoice, + ToolChoiceFuncName, +) +from sglang.srt.function_call.deepseekv41_detector import DeepSeekV41Detector +from sglang.srt.function_call.function_call_parser import FunctionCallParser +from sglang.srt.parser.reasoning_parser import ReasoningParser +from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase + +register_cpu_ci(est_time=2, suite="base-a-test-cpu") + +CHUNK_SIZES = [1, 2, 3, 5, 7, 11, 23, 1000] +DSML = "｜DSML｜" + + +def _tools(): + return [ + Tool( + type="function", + function=Function( + name="get_weather", + description="Get weather information", + parameters={ + "type": "object", + "properties": {"city": {"type": "string"}}, + "required": ["city"], + }, + ), + ), + Tool( + type="function", + function=Function( + name="lookup", + description="Look up a value", + parameters={ + "type": "object", + "properties": { + "query": {"type": "string"}, + "limit": {"type": "integer"}, + "flags": {"type": "array"}, + }, + }, + ), + ), + ] + + +def _assemble(calls): + """Streamed ToolCallItems -> [(name, parsed arguments)] per tool_index.""" + by_index = {} + for call in calls: + entry = by_index.setdefault(call.tool_index, {"name": None, "args": ""}) + if call.name: + entry["name"] = call.name + entry["args"] += call.parameters or "" + return [ + (entry["name"], json.loads(entry["args"])) + for _, entry in sorted(by_index.items()) + ] + + +class TestDeepSeekV41RoundTrip(CustomTestCase): + """Encoder-rendered assistant tool calls parse back to the same arguments, + in one shot and at every chunk size.""" + + ARGUMENTS = {"query": '{"a": 1}', "limit": 2, "flags": [1, True, None]} + + def setUp(self): + self.tools = _tools() + self.completion = encoding_dsv41.render_message( + 1, + [ + {"role": "user", "content": "question"}, + { + "role": "assistant", + "reasoning_content": "reason", + "content": "summary", + "tool_calls": [ + { + "type": "function", + "function": { + "name": "lookup", + "arguments": json.dumps(self.ARGUMENTS), + }, + } + ], + }, + ], + thinking_mode="thinking", + ) + self.completion = "" + self.completion + self.expected = [("lookup", self.ARGUMENTS)] + + def test_one_shot(self): + parser = FunctionCallParser(self.tools, "deepseekv41") + reasoning, content = ReasoningParser("deepseek-v41").parse_non_stream( + self.completion + ) + self.assertEqual(reasoning, "reason") + normal, calls = parser.parse_non_stream(content) + self.assertEqual(normal, "summary") + self.assertEqual( + [(c.name, json.loads(c.parameters)) for c in calls], self.expected + ) + + def test_streaming_at_every_chunk_size(self): + for chunk_size in CHUNK_SIZES: + with self.subTest(chunk_size=chunk_size): + reasoning_parser = ReasoningParser("deepseek-v41") + tool_parser = FunctionCallParser(self.tools, "deepseekv41") + reasoning, normal, calls = "", "", [] + for i in range(0, len(self.completion), chunk_size): + reason, content = reasoning_parser.parse_stream_chunk( + self.completion[i : i + chunk_size] + ) + reasoning += reason or "" + text, delta = tool_parser.parse_stream_chunk(content or "") + normal += text + calls.extend(delta) + reason, content = reasoning_parser.parse_stream_end() + reasoning += reason or "" + text, delta = tool_parser.parse_stream_chunk(content or "") + normal += text + calls.extend(delta) + text, delta = tool_parser.parse_stream_end() + normal += text + calls.extend(delta) + self.assertEqual(reasoning, "reason") + # The blank line before the block is released or trimmed depending + # on where the chunk boundary falls; the shared base behaves the + # same for V4, so only the prose itself is pinned here. + self.assertEqual(normal.strip(), "summary") + self.assertEqual(_assemble(calls), self.expected) + + +class TestDeepSeekV41ConstrainedDecoding(CustomTestCase): + """A forced call must open the calls block before the first invoke; the + per-tool legacy tag started the grammar at the invoke trigger, the model + closed a block it had not opened, and the parser dropped the call.""" + + def setUp(self): + self.tools = _tools() + self.detector = DeepSeekV41Detector() + + def test_required_tag_wraps_invokes_in_the_calls_block(self): + tag = self.detector.get_structural_tag(tools=self.tools, tool_choice="required") + opener, calls, closer = tag.format.elements + self.assertEqual(opener.value, f"\n\n<{DSML} calls>\n") + self.assertEqual(closer.value, f"") + self.assertTrue(calls.at_least_one) + self.assertEqual( + [t.begin for t in calls.tags], + [ + f'<{DSML} invoke name="get_weather">\n', + f'<{DSML} invoke name="lookup">\n', + ], + ) + self.assertEqual({t.end for t in calls.tags}, {f"\n"}) + + def test_auto_tag_triggers_on_the_calls_block(self): + tag = self.detector.get_structural_tag(tools=self.tools, tool_choice="auto") + self.assertEqual(tag.format.triggers, [f"<{DSML} calls>"]) + self.assertEqual(tag.format.tags[0].begin, f"<{DSML} calls>\n") + self.assertEqual(tag.format.tags[0].end, f"") + + def test_thinking_mode_prefixes_the_reasoning_span(self): + tag = self.detector.get_structural_tag( + tools=self.tools, tool_choice="required", thinking_mode=True + ) + reasoning, body = tag.format.elements + self.assertEqual(reasoning.end, "") + self.assertEqual(body.elements[0].value, f"\n\n<{DSML} calls>\n") + + def test_body_uses_available_xgrammar_style(self): + """Older XGrammar must keep a compilable, schema-constrained JSON fallback.""" + self.tools[0].function.strict = True + tag = self.detector.get_structural_tag(self.tools, "required") + grammar = xgr.Grammar.from_structural_tag(tag) + native_xml = tag.format.elements[1].tags[0].content.style == "deepseek_v4_1_xml" + begin = f'\n\n<{DSML} calls>\n<{DSML} invoke name="get_weather">\n' + end = f"\n" + xml = f'<{DSML} parameter name="city" string="true">Paris\n' + self.assertEqual( + _is_grammar_accept_string(grammar, begin + xml + end), native_xml + ) + self.assertEqual( + _is_grammar_accept_string(grammar, begin + '{"city":"Paris"}' + end), + not native_xml, + ) + self.assertFalse(_is_grammar_accept_string(grammar, begin + end)) + self.assertFalse(_is_grammar_accept_string(grammar, begin + "{}" + end)) + + +@unittest.skipUnless( + "deepseek_v4_1_xml" in get_args(JSONSchemaFormat.model_fields["style"].annotation), + "Requires XGrammar's DeepSeek V4.1 XML style", +) +class TestDeepSeekV41ParameterGrammar(CustomTestCase): + """The encoder emits DSML parameters; a JSON invoke body rejects valid output.""" + + def setUp(self): + self.tools = _tools() + self.tools[1].function.parameters["properties"]["flags"]["items"] = True + for tool in self.tools: + tool.function.strict = True + tool.function.parameters["additionalProperties"] = False + + @staticmethod + def _render(arguments, *, thinking=False, count=1, name="get_weather"): + return encoding_dsv41.render_message( + 1, + [ + {"role": "user", "content": "question"}, + { + "role": "assistant", + "content": "", + "reasoning_content": "reason", + "wo_eos": True, + "tool_calls": [ + { + "type": "function", + "function": { + "name": name, + "arguments": json.dumps(arguments), + }, + } + ] + * count, + }, + ], + thinking_mode="thinking" if thinking else "chat", + ) + + def _grammar(self, choice="required", thinking=False, parallel=True): + constraint = FunctionCallParser( + self.tools, "deepseekv41" + ).get_structure_constraint( + choice, thinking_mode=thinking, parallel_tool_calls=parallel + ) + self.assertIsNotNone(constraint) + self.assertEqual(constraint[0], "structural_tag") + return xgr.Grammar.from_structural_tag(constraint[1]) + + def test_encoder_output_matches_and_round_trips(self): + arguments = {"query": '{"a": 1}', "limit": 2, "flags": [True, None, 1.5]} + for thinking in (False, True): + with self.subTest(thinking=thinking): + output = self._render(arguments, thinking=thinking, name="lookup") + grammar = self._grammar(thinking=thinking) + self.assertTrue(_is_grammar_accept_string(grammar, output)) + if thinking: + output = output.split("", 1)[1] + parsed = DeepSeekV41Detector().detect_and_parse(output, self.tools) + self.assertEqual(json.loads(parsed.calls[0].parameters), arguments) + + def test_strict_schema_rejects_missing_extra_and_wrong_type(self): + for choice in ( + "auto", + "required", + ToolChoice(function=ToolChoiceFuncName(name="get_weather")), + ): + grammar = self._grammar(choice) + self.assertTrue( + _is_grammar_accept_string(grammar, self._render({"city": "杭州"})) + ) + for arguments in ({}, {"city": 42}, {"city": "Paris", "extra": True}): + with self.subTest(choice=choice, arguments=arguments): + self.assertFalse( + _is_grammar_accept_string(grammar, self._render(arguments)) + ) + self.assertFalse( + _is_grammar_accept_string( + grammar, + self._render({"city": "Paris"}).replace( + 'string="true">Paris', 'string="false">42' + ), + ) + ) + + def test_parallel_and_named_choice_limit_calls(self): + for parallel in (False, True): + grammar = self._grammar(parallel=parallel) + self.assertTrue( + _is_grammar_accept_string(grammar, self._render({"city": "Paris"})) + ) + self.assertEqual( + _is_grammar_accept_string( + grammar, self._render({"city": "Paris"}, count=2) + ), + parallel, + ) + grammar = self._grammar( + ToolChoice(function=ToolChoiceFuncName(name="get_weather")) + ) + self.assertFalse( + _is_grammar_accept_string(grammar, self._render({"city": "Paris"}, count=2)) + ) + self.assertFalse( + _is_grammar_accept_string( + grammar, self._render({"query": "Paris"}, name="lookup") + ) + ) + + def test_non_strict_still_uses_native_parameters(self): + self.tools[0].function.strict = False + grammar = self._grammar() + self.assertTrue( + _is_grammar_accept_string(grammar, self._render({"extra": [True, None, 2]})) + ) + self.assertFalse( + _is_grammar_accept_string( + grammar, + self._render({"extra": [True, None, 2]}).replace( + "[true, null, 2]", "invalid" + ), + ) + ) + + +if __name__ == "__main__": + import unittest + + unittest.main()