diff --git a/docs/advanced_features/server_arguments.md b/docs/advanced_features/server_arguments.md
index 587a92d44..faac0ab49 100644
--- a/docs/advanced_features/server_arguments.md
+++ b/docs/advanced_features/server_arguments.md
@@ -196,7 +196,7 @@ Please consult the documentation below and [server_args.py](https://github.com/s
| `--file-storage-path` | The path of the file storage in backend. | `sglang_storage` | Type: str |
| `--enable-cache-report` | Return number of cached tokens in usage.prompt_tokens_details for each openai request. | `False` | bool flag (set to enable) |
| `--reasoning-parser` | Specify the parser for reasoning models. Supported parsers: [deepseek-r1, deepseek-v3, glm45, gpt-oss, kimi, qwen3, qwen3-thinking, step3]. | `None` | `deepseek-r1`, `deepseek-v3`, `glm45`, `gpt-oss`, `kimi`, `qwen3`, `qwen3-thinking`, `step3` |
-| `--tool-call-parser` | Specify the parser for handling tool-call interactions. Supported parsers: [deepseekv3, deepseekv31, glm, glm45, gpt-oss, kimi_k2, llama3, mistral, pythonic, qwen, qwen25, qwen3_coder, step3]. | `None` | `deepseekv3`, `deepseekv31`, `glm`, `glm45`, `gpt-oss`, `kimi_k2`, `llama3`, `mistral`, `pythonic`, `qwen`, `qwen25`, `qwen3_coder`, `step3` |
+| `--tool-call-parser` | Specify the parser for handling tool-call interactions. Supported parsers: [deepseekv3, deepseekv31, glm, glm45, glm47, gpt-oss, kimi_k2, llama3, mistral, pythonic, qwen, qwen25, qwen3_coder, step3]. | `None` | `deepseekv3`, `deepseekv31`, `glm`, `glm45`, `glm47`, `gpt-oss`, `kimi_k2`, `llama3`, `mistral`, `pythonic`, `qwen`, `qwen25`, `qwen3_coder`, `step3` |
| `--sampling-defaults` | Where to get default sampling parameters. 'openai' uses SGLang/OpenAI defaults (temperature=1.0, top_p=1.0, etc.). 'model' uses the model's generation_config.json to get the recommended sampling parameters if available. Default is 'model'. | `model` | `openai`, `model` |
| `--tool-server` | Either 'demo' or a comma-separated list of tool server urls to use for the model. If not specified, no tool server will be used. | `None` | Type: str |
diff --git a/docs/basic_usage/glm45.md b/docs/basic_usage/glm45.md
index d18b0a68d..b68f984f9 100644
--- a/docs/basic_usage/glm45.md
+++ b/docs/basic_usage/glm45.md
@@ -1,4 +1,4 @@
-## Launch GLM-4.5 / GLM-4.6 with SGLang
+## Launch GLM-4.5 / GLM-4.6 / GLM-4.7 with SGLang
To serve GLM-4.5 / GLM-4.6 FP8 models on 8xH100/H200 GPUs:
@@ -35,7 +35,9 @@ python3 -m sglang.launch_server \
--enable-custom-logit-processor
```
-### Thinking Budget for GLM-4.5 / GLM-4.6
+**Note**: For GLM-4.7, `--tool-call-parser` should be set to `glm47`, for GLM-4.5 and GLM-4.6, it should be set to `glm45`.
+
+### Thinking Budget
In SGLang, we can implement thinking budget with `CustomLogitProcessor`.
diff --git a/python/sglang/srt/function_call/function_call_parser.py b/python/sglang/srt/function_call/function_call_parser.py
index 24287842b..8df0a2401 100644
--- a/python/sglang/srt/function_call/function_call_parser.py
+++ b/python/sglang/srt/function_call/function_call_parser.py
@@ -15,6 +15,7 @@ from sglang.srt.function_call.deepseekv3_detector import DeepSeekV3Detector
from sglang.srt.function_call.deepseekv31_detector import DeepSeekV31Detector
from sglang.srt.function_call.deepseekv32_detector import DeepSeekV32Detector
from sglang.srt.function_call.glm4_moe_detector import Glm4MoeDetector
+from sglang.srt.function_call.glm47_moe_detector import Glm47MoeDetector
from sglang.srt.function_call.gpt_oss_detector import GptOssDetector
from sglang.srt.function_call.internlm_detector import InternlmDetector
from sglang.srt.function_call.kimik2_detector import KimiK2Detector
@@ -46,6 +47,7 @@ class FunctionCallParser:
"deepseekv32": DeepSeekV32Detector,
"glm": Glm4MoeDetector,
"glm45": Glm4MoeDetector,
+ "glm47": Glm47MoeDetector,
"gpt-oss": GptOssDetector,
"kimi_k2": KimiK2Detector,
"llama3": Llama32Detector,
diff --git a/python/sglang/srt/function_call/glm47_moe_detector.py b/python/sglang/srt/function_call/glm47_moe_detector.py
new file mode 100644
index 000000000..726737c57
--- /dev/null
+++ b/python/sglang/srt/function_call/glm47_moe_detector.py
@@ -0,0 +1,584 @@
+import ast
+import json
+import logging
+import re
+from enum import Enum
+from typing import Any, Dict, List, Optional, Tuple
+
+from sglang.srt.entrypoints.openai.protocol import Tool
+from sglang.srt.function_call.base_format_detector import BaseFormatDetector
+from sglang.srt.function_call.core_types import (
+ StreamingParseResult,
+ ToolCallItem,
+ _GetInfoFunc,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class StreamState(str, Enum):
+ """State machine states for XML to JSON streaming conversion."""
+
+ INIT = "INIT"
+ BETWEEN = "BETWEEN"
+ IN_KEY = "IN_KEY"
+ WAITING_VALUE = "WAITING_VALUE"
+ IN_VALUE = "IN_VALUE"
+
+
+def get_argument_type(
+ func_name: str, arg_key: str, defined_tools: List[Tool]
+) -> Optional[str]:
+ """Get the expected type of a function argument from tool definitions.
+
+ Args:
+ func_name: Name of the function/tool
+ arg_key: Name of the argument
+ defined_tools: List of available tools
+
+ Returns:
+ The type string (e.g., 'string', 'number', 'object') or None if not found
+ """
+ name2tool = {tool.function.name: tool for tool in defined_tools}
+ if func_name not in name2tool:
+ return None
+ tool = name2tool[func_name]
+ properties = (tool.function.parameters or {}).get("properties", {})
+ if not isinstance(properties, dict):
+ properties = {}
+ if arg_key not in properties:
+ return None
+ return properties[arg_key].get("type", None)
+
+
+def _convert_to_number(value: str) -> Any:
+ """Convert string to appropriate number type (int or float).
+
+ Args:
+ value: String value to convert
+
+ Returns:
+ Converted number or original string if conversion fails
+ """
+ try:
+ if "." in value or "e" in value.lower():
+ return float(value)
+ else:
+ return int(value)
+ except (ValueError, AttributeError):
+ return value
+
+
+def parse_arguments(
+ json_value: str, arg_type: Optional[str] = None
+) -> Tuple[Any, bool]:
+ """Parse argument value with multiple fallback strategies.
+
+ Args:
+ json_value: Raw string value to parse
+ arg_type: Expected type hint ('string', 'number', 'object', etc.)
+
+ Returns:
+ Tuple of (parsed_value, is_valid_json)
+ """
+ # Strategy 1: Direct JSON parsing
+ try:
+ parsed_value = json.loads(json_value)
+
+ # Type coercion for number type
+ if arg_type == "number" and isinstance(parsed_value, str):
+ parsed_value = _convert_to_number(parsed_value)
+
+ return parsed_value, True
+ except (json.JSONDecodeError, ValueError):
+ pass
+
+ # Strategy 2: Unescape and parse
+ try:
+ wrapped = json.loads('{"tmp": "' + json_value + '"}')
+ parsed_value = json.loads(wrapped["tmp"])
+
+ if arg_type == "number" and isinstance(parsed_value, str):
+ parsed_value = _convert_to_number(parsed_value)
+
+ return parsed_value, True
+ except (json.JSONDecodeError, ValueError, KeyError):
+ pass
+
+ # Strategy 3: ast.literal_eval
+ try:
+ parsed_value = ast.literal_eval(json_value)
+ return parsed_value, True
+ except (ValueError, SyntaxError):
+ pass
+
+ # Strategy 4: Treat as string
+ try:
+ quoted_value = json.dumps(str(json_value))
+ return json.loads(quoted_value), True
+ except (json.JSONDecodeError, ValueError):
+ return json_value, False
+
+
+class Glm47MoeDetector(BaseFormatDetector):
+ """
+ Detector for GLM-4.7 and GLM-5 models.
+ Assumes function call format:
+ get_weathercity北京date2024-06-27get_weathercity上海date2024-06-27
+ """
+
+ def __init__(self):
+ super().__init__()
+ self.bot_token = ""
+ self.eot_token = ""
+ self.func_call_regex = r".*?"
+ self.func_detail_regex = re.compile(
+ r"(.*?)(.*?)?", re.DOTALL
+ )
+ self.func_arg_regex = re.compile(
+ r"(.*?)(?:\\n|\s)*(.*?)",
+ re.DOTALL,
+ )
+ self._last_arguments = ""
+ self.current_tool_id = -1
+ self.current_tool_name_sent = False
+ self._streamed_raw_length = 0
+ self._reset_streaming_state()
+
+ def _reset_streaming_state(self) -> None:
+ """Reset the streaming state machine for a new tool call."""
+ self._stream_state = StreamState.INIT
+ self._current_key = ""
+ self._current_value = ""
+ self._xml_tag_buffer = ""
+ self._is_first_param = True
+ self._value_started = False
+ self._cached_value_type: Optional[str] = (
+ None # Cache the value type for consistency
+ )
+
+ def has_tool_call(self, text: str) -> bool:
+ """Check if the text contains a glm-4.5 / glm-4.6 format tool call."""
+ return self.bot_token in text
+
+ def detect_and_parse(self, text: str, tools: List[Tool]) -> StreamingParseResult:
+ """
+ One-time parsing: Detects and parses tool calls in the provided text.
+
+ :param text: The complete text to parse.
+ :param tools: List of available tools.
+ :return: ParseResult indicating success or failure, consumed text, leftover text, and parsed calls.
+ """
+ idx = text.find(self.bot_token)
+ normal_text = text[:idx].strip() if idx != -1 else text
+ if self.bot_token not in text:
+ return StreamingParseResult(normal_text=normal_text, calls=[])
+ match_result_list = re.findall(self.func_call_regex, text, re.DOTALL)
+ calls = []
+ try:
+ for match_result in match_result_list:
+ # Get function name
+ func_detail = self.func_detail_regex.search(match_result)
+ func_name = func_detail.group(1)
+ func_args = func_detail.group(2)
+ arguments = {}
+ if func_args:
+ pairs = self.func_arg_regex.findall(func_args)
+ # Parse arguments using shared method
+ arguments = self._parse_argument_pairs(pairs, func_name, tools)
+
+ # construct match_result for parse_base_json
+ match_result = {"name": func_name, "parameters": arguments}
+ calls.extend(self.parse_base_json(match_result, tools))
+ return StreamingParseResult(normal_text=normal_text, calls=calls)
+ except Exception as e:
+ logger.error(f"Error in detect_and_parse: {e}", exc_info=True)
+ # return the normal text if parsing fails
+ return StreamingParseResult(normal_text=text)
+
+ def _get_value_type(self, func_name: str, key: str, tools: List[Tool]) -> str:
+ """Get parameter type from tool definition, with fallback to auto-detection.
+
+ Args:
+ func_name: Name of the function
+ key: Parameter name
+ tools: List of available tools
+
+ Returns:
+ Type string: 'string', 'number', or 'object'
+ """
+ arg_type = get_argument_type(func_name, key, tools)
+ if arg_type:
+ return arg_type
+
+ # Auto-detect type from value (best effort)
+ first_chars = self._current_value.strip()[:10] if self._current_value else ""
+ if first_chars:
+ first_char = first_chars[0]
+ if first_char.isdigit() or first_char in ["-", "."]:
+ return "number"
+ elif first_char in ["{", "["]:
+ return "object"
+
+ return "string"
+
+ def _format_value_complete(self, value: str, value_type: str) -> str:
+ """Format complete value based on type.
+
+ Args:
+ value: Raw value string
+ value_type: Expected type ('string', 'number', 'object')
+
+ Returns:
+ Properly formatted JSON value string
+ """
+ if value_type == "string":
+ # Ensure proper JSON string formatting with quotes
+ return json.dumps(value, ensure_ascii=False)
+ elif value_type == "number":
+ try:
+ num = _convert_to_number(value.strip())
+ return str(num)
+ except (ValueError, AttributeError):
+ # Fallback to string if not a valid number
+ logger.warning(
+ f"Failed to parse '{value}' as number, treating as string"
+ )
+ return json.dumps(str(value), ensure_ascii=False)
+ else:
+ # For object/array types, return as-is (should already be valid JSON)
+ return value
+
+ def _process_xml_to_json_streaming(
+ self, raw_increment: str, func_name: str, tools: List[Tool]
+ ) -> str:
+ """Convert XML increment to JSON streaming output using state machine.
+
+ This method processes XML fragments character by character and converts them
+ to JSON format incrementally. It maintains state across calls to handle
+ partial XML tags and values.
+
+ Args:
+ raw_increment: New XML content to process
+ func_name: Name of the function being called
+ tools: List of available tools for type inference
+
+ Returns:
+ JSON string increment to append to the output
+ """
+ json_output = ""
+
+ for char in raw_increment:
+ self._xml_tag_buffer += char
+
+ if self._stream_state in [StreamState.INIT, StreamState.BETWEEN]:
+ if self._xml_tag_buffer.endswith(""):
+ self._stream_state = StreamState.IN_KEY
+ self._current_key = ""
+ self._xml_tag_buffer = ""
+ json_output += "{" if self._is_first_param else ", "
+ self._is_first_param = False
+
+ elif self._stream_state == StreamState.IN_KEY:
+ if self._xml_tag_buffer.endswith(""):
+ self._current_key = self._xml_tag_buffer[:-10].strip()
+ self._xml_tag_buffer = ""
+ self._stream_state = StreamState.WAITING_VALUE
+ json_output += (
+ json.dumps(self._current_key, ensure_ascii=False) + ": "
+ )
+
+ elif self._stream_state == StreamState.WAITING_VALUE:
+ if self._xml_tag_buffer.endswith(""):
+ self._stream_state = StreamState.IN_VALUE
+ self._current_value = ""
+ self._xml_tag_buffer = ""
+ self._value_started = False
+ # Determine and cache the value type at the start
+ self._cached_value_type = self._get_value_type(
+ func_name, self._current_key, tools
+ )
+
+ elif self._stream_state == StreamState.IN_VALUE:
+ if self._xml_tag_buffer.endswith(""):
+ final_value = self._xml_tag_buffer[:-12]
+ self._current_value += final_value
+
+ # Use cached value type for consistency
+ value_type = self._cached_value_type or "string"
+
+ if self._value_started:
+ # Output any remaining content
+ if final_value:
+ if value_type == "string":
+ json_output += json.dumps(
+ final_value, ensure_ascii=False
+ )[1:-1]
+ else:
+ json_output += final_value
+ # Always output closing quote for string type when value was started
+ if value_type == "string":
+ json_output += '"'
+ else:
+ # Value was never started (empty or complete in one chunk)
+ json_output += self._format_value_complete(
+ self._current_value, value_type
+ )
+
+ self._xml_tag_buffer = ""
+ self._stream_state = StreamState.BETWEEN
+ self._current_value = ""
+ self._value_started = False
+ self._cached_value_type = None # Reset cached type
+ else:
+ closing_tag = ""
+ is_potential_closing = len(self._xml_tag_buffer) <= len(
+ closing_tag
+ ) and closing_tag.startswith(self._xml_tag_buffer)
+
+ if not is_potential_closing:
+ content = self._xml_tag_buffer
+ # Use cached value type for consistency
+ value_type = self._cached_value_type or "string"
+
+ if value_type == "string":
+ if not self._value_started:
+ json_output += '"'
+ self._value_started = True
+ if content:
+ json_output += json.dumps(content, ensure_ascii=False)[
+ 1:-1
+ ]
+ self._current_value += content
+ self._xml_tag_buffer = ""
+ elif value_type == "number":
+ if content:
+ if not self._value_started:
+ self._value_started = True
+ json_output += content
+ self._current_value += content
+ self._xml_tag_buffer = ""
+ else:
+ # For object/array types, output as-is
+ if content:
+ if not self._value_started:
+ self._value_started = True
+ json_output += content
+ self._current_value += content
+ self._xml_tag_buffer = ""
+
+ return json_output
+
+ def parse_streaming_increment(
+ self, new_text: str, tools: List[Tool]
+ ) -> StreamingParseResult:
+ """
+ Streaming incremental parsing tool calls for GLM-4.5 and GLM-4.6 format.
+ Uses a state machine to convert XML to JSON incrementally for true character-by-character streaming.
+ Outputs JSON increments immediately as XML data arrives.
+ """
+ self._buffer += new_text
+ current_text = self._buffer
+
+ # Check if we have a tool call
+ has_tool_call = self.bot_token in current_text
+
+ if not has_tool_call:
+ # Check if buffer could be the start of a tool call
+ # Keep buffer if it could be a partial match of bot_token
+ is_potential_start = any(
+ self.bot_token.startswith(current_text[-i:])
+ for i in range(1, min(len(current_text), len(self.bot_token)) + 1)
+ )
+
+ if not is_potential_start:
+ # Not a potential tool call, return as normal text
+ # Must return the entire buffer (current_text), not just new_text,
+ # because buffer may contain previously accumulated characters like '<'
+ # that turned out not to be part of a tool call
+ output_text = current_text
+ self._buffer = ""
+ if self.eot_token in output_text:
+ output_text = output_text.replace(self.eot_token, "")
+ return StreamingParseResult(normal_text=output_text)
+ else:
+ # Could be start of tool call, keep buffering
+ return StreamingParseResult(normal_text="", calls=[])
+
+ if not hasattr(self, "_tool_indices"):
+ self._tool_indices = self._get_tool_indices(tools)
+
+ calls: list[ToolCallItem] = []
+ try:
+ # Try to match a partial or complete tool call
+ partial_match = re.search(
+ pattern=r"(.*?)(.*?)?(|$)",
+ string=current_text,
+ flags=re.DOTALL,
+ )
+ if partial_match:
+ func_name = partial_match.group(1).strip()
+ func_args_raw = partial_match.group(2).strip()
+ is_tool_end = partial_match.group(3)
+
+ # Initialize state if this is the first tool call
+ if self.current_tool_id == -1:
+ self.current_tool_id = 0
+ self.prev_tool_call_arr = []
+ self.streamed_args_for_tool = [""]
+ self._streamed_raw_length = 0
+ self.current_tool_name_sent = False
+ self._reset_streaming_state()
+
+ # Ensure we have enough entries in our tracking arrays
+ while len(self.prev_tool_call_arr) <= self.current_tool_id:
+ self.prev_tool_call_arr.append({})
+ while len(self.streamed_args_for_tool) <= self.current_tool_id:
+ self.streamed_args_for_tool.append("")
+
+ # Send tool name first if not sent yet
+ if not self.current_tool_name_sent:
+ assert func_name, "func_name should not be empty"
+ calls.append(
+ ToolCallItem(
+ tool_index=self.current_tool_id,
+ name=func_name,
+ parameters="",
+ )
+ )
+ self.current_tool_name_sent = True
+ self._streamed_raw_length = 0
+ self._reset_streaming_state()
+ # Store the tool call info
+ self.prev_tool_call_arr[self.current_tool_id] = {
+ "name": func_name,
+ "arguments": {},
+ }
+ else:
+ # Process XML to JSON streaming
+ current_raw_length = len(func_args_raw)
+
+ if current_raw_length > self._streamed_raw_length:
+ # Get the new raw XML content
+ raw_increment = func_args_raw[self._streamed_raw_length :]
+
+ # Convert XML increment to JSON increment using state machine
+ json_increment = self._process_xml_to_json_streaming(
+ raw_increment, func_name, tools
+ )
+
+ if json_increment:
+ calls.append(
+ ToolCallItem(
+ tool_index=self.current_tool_id,
+ name=None,
+ parameters=json_increment,
+ )
+ )
+ self._last_arguments += json_increment
+ self.streamed_args_for_tool[
+ self.current_tool_id
+ ] += json_increment
+
+ # Update the streamed length
+ self._streamed_raw_length = current_raw_length
+
+ if is_tool_end == self.eot_token:
+ if self._is_first_param:
+ empty_object = "{}"
+ calls.append(
+ ToolCallItem(
+ tool_index=self.current_tool_id,
+ name=None,
+ parameters=empty_object,
+ )
+ )
+ self._last_arguments += empty_object
+ elif not self._last_arguments.endswith("}"):
+ closing_brace = "}"
+ calls.append(
+ ToolCallItem(
+ tool_index=self.current_tool_id,
+ name=None,
+ parameters=closing_brace,
+ )
+ )
+ self._last_arguments += closing_brace
+ self.streamed_args_for_tool[
+ self.current_tool_id
+ ] += closing_brace
+
+ try:
+ pairs = self.func_arg_regex.findall(func_args_raw)
+ if pairs:
+ arguments = self._parse_argument_pairs(
+ pairs, func_name, tools
+ )
+ self.prev_tool_call_arr[self.current_tool_id][
+ "arguments"
+ ] = arguments
+ except Exception as e:
+ logger.debug(
+ f"Failed to parse arguments: {e}", exc_info=True
+ )
+
+ # Remove the completed tool call from buffer
+ self._buffer = current_text[partial_match.end(3) :]
+
+ result = StreamingParseResult(normal_text="", calls=calls)
+ self.current_tool_id += 1
+ self._last_arguments = ""
+ self.current_tool_name_sent = False
+ self._streamed_raw_length = 0
+ self._reset_streaming_state()
+ return result
+
+ return StreamingParseResult(normal_text="", calls=calls)
+
+ except Exception as e:
+ logger.error(f"Error in parse_streaming_increment: {e}", exc_info=True)
+ return StreamingParseResult(normal_text=current_text)
+
+ def _parse_argument_pairs(
+ self, pairs: List[Tuple[str, str]], func_name: str, tools: List[Tool]
+ ) -> Dict[str, Any]:
+ """Parse argument key-value pairs with type coercion.
+
+ Args:
+ pairs: List of (key, value) tuples from regex matching
+ func_name: Name of the function
+ tools: List of available tools
+
+ Returns:
+ Dictionary of parsed arguments
+ """
+ arguments = {}
+ for arg_key, arg_value in pairs:
+ arg_key = arg_key.strip()
+ arg_value = arg_value.strip()
+ arg_type = get_argument_type(func_name, arg_key, tools)
+ parsed_value, is_good_json = parse_arguments(arg_value, arg_type)
+
+ if arg_type == "string":
+ # Only convert to string if explicitly defined as string type
+ if isinstance(parsed_value, str):
+ arguments[arg_key] = parsed_value
+ elif isinstance(parsed_value, (dict, list)):
+ # If parsed as dict/list but schema says string, convert to JSON string
+ arguments[arg_key] = json.dumps(parsed_value, ensure_ascii=False)
+ else:
+ arguments[arg_key] = str(parsed_value)
+ elif arg_type is None:
+ # If type is not defined, keep the parsed value as-is
+ arguments[arg_key] = parsed_value if is_good_json else arg_value
+ else:
+ # For other types (number, object, array, etc.), use parsed value
+ arguments[arg_key] = parsed_value if is_good_json else arg_value
+
+ return arguments
+
+ def supports_structural_tag(self) -> bool:
+ return False
+
+ def structure_info(self) -> _GetInfoFunc:
+ raise NotImplementedError()
diff --git a/python/sglang/srt/function_call/glm4_moe_detector.py b/python/sglang/srt/function_call/glm4_moe_detector.py
index 95423b414..cf2201a5a 100644
--- a/python/sglang/srt/function_call/glm4_moe_detector.py
+++ b/python/sglang/srt/function_call/glm4_moe_detector.py
@@ -43,9 +43,12 @@ def get_argument_type(
if func_name not in name2tool:
return None
tool = name2tool[func_name]
- if arg_key not in tool.function.parameters["properties"]:
+ properties = (tool.function.parameters or {}).get("properties", {})
+ if not isinstance(properties, dict):
+ properties = {}
+ if arg_key not in properties:
return None
- return tool.function.parameters["properties"][arg_key].get("type", None)
+ return properties[arg_key].get("type", None)
def _convert_to_number(value: str) -> Any:
diff --git a/python/sglang/srt/models/glm4_moe.py b/python/sglang/srt/models/glm4_moe.py
index 6483e41c1..c27816c2c 100644
--- a/python/sglang/srt/models/glm4_moe.py
+++ b/python/sglang/srt/models/glm4_moe.py
@@ -12,7 +12,7 @@
# limitations under the License.
# ==============================================================================
-"""Inference-only GLM-4.5, GLM-4.6 model compatible with HuggingFace weights"""
+"""Inference-only GLM-4.5, GLM-4.6 and GLM-4.7 model compatible with HuggingFace weights"""
import logging
from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
diff --git a/test/registered/function_call/test_function_call_parser.py b/test/registered/function_call/test_function_call_parser.py
index e695259f3..99a07639b 100644
--- a/test/registered/function_call/test_function_call_parser.py
+++ b/test/registered/function_call/test_function_call_parser.py
@@ -7,10 +7,10 @@ from sglang.srt.function_call.core_types import StreamingParseResult
from sglang.srt.function_call.deepseekv3_detector import DeepSeekV3Detector
from sglang.srt.function_call.deepseekv32_detector import DeepSeekV32Detector
from sglang.srt.function_call.glm4_moe_detector import Glm4MoeDetector
+from sglang.srt.function_call.glm47_moe_detector import Glm47MoeDetector
from sglang.srt.function_call.json_array_parser import JsonArrayParser
from sglang.srt.function_call.kimik2_detector import KimiK2Detector
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.pythonic_detector import PythonicDetector
from sglang.srt.function_call.qwen3_coder_detector import Qwen3CoderDetector
@@ -1159,9 +1159,6 @@ class TestDeepSeekV32Detector(unittest.TestCase):
),
]
self.detector = DeepSeekV32Detector()
- from transformers import AutoTokenizer
-
- self.tokenizer = AutoTokenizer.from_pretrained("deepseek-ai/DeepSeek-V3.2")
def test_detect_and_parse_xml_format(self):
"""Test parsing standard XML format (DSML)"""
@@ -1239,16 +1236,12 @@ class TestDeepSeekV32Detector(unittest.TestCase):
text = """<｜DSML｜function_calls>
<｜DSML｜invoke name="get_favorite_tourist_spot">
<｜DSML｜parameter name="city" string="true">San Francisco
- <｜DSML｜parameter name="second" string="true">London
- <｜DSML｜parameter name="topn" string="false">10
- <｜DSML｜parameter name="obj" string="false">{"name": "John", "age": 30}
"""
- input_ids = self.tokenizer.encode(text, add_special_tokens=False)
- chunk_ids = [input_ids[i : i + 5] for i in range(0, len(input_ids), 5)]
- chunks = [self.tokenizer.decode(chunk_id) for chunk_id in chunk_ids]
+ chunks = [text[i : i + 5] for i in range(0, len(text), 5)]
+ accumulated_calls = []
tool_calls_by_index = {}
for chunk in chunks:
@@ -1290,9 +1283,7 @@ class TestDeepSeekV32Detector(unittest.TestCase):
"""
- input_ids = self.tokenizer.encode(text, add_special_tokens=False)
- chunk_ids = [input_ids[i : i + 5] for i in range(0, len(input_ids), 5)]
- chunks = [self.tokenizer.decode(chunk_id) for chunk_id in chunk_ids]
+ chunks = [text[i : i + 5] for i in range(0, len(text), 5)]
tool_calls_by_index = {}
@@ -1379,10 +1370,7 @@ class TestDeepSeekV32Detector(unittest.TestCase):
self.detector = DeepSeekV32Detector()
# Simulate streaming by splitting into small chunks
- # chunks = [text[i : i + 5] for i in range(0, len(text), 5)]
- input_ids = self.tokenizer.encode(text, add_special_tokens=False)
- chunk_ids = [input_ids[i : i + 5] for i in range(0, len(input_ids), 5)]
- chunks = [self.tokenizer.decode(chunk_id) for chunk_id in chunk_ids]
+ chunks = [text[i : i + 5] for i in range(0, len(text), 5)]
tool_calls_by_index = {}
@@ -2247,444 +2235,280 @@ class TestGlm4MoeDetector(unittest.TestCase):
check_single_todos(result, expected_output)
-class TestMiMoDetector(unittest.TestCase):
+class TestGlm47MoeDetector(unittest.TestCase):
def setUp(self):
- # Create sample tools for testing
self.tools = [
Tool(
type="function",
function=Function(
- name="get_current_weather",
- description="Get the current weather",
+ name="get_weather",
+ description="Get weather information",
parameters={
+ "type": "object",
"properties": {
- "city": {"type": "string", "description": "The city name"},
- "state": {
- "type": "string",
- "description": "The state code",
- },
- "unit": {
- "type": "string",
- "enum": ["fahrenheit", "celsius"],
- },
+ "city": {"type": "string", "description": "City name"},
+ "date": {"type": "string", "description": "Date"},
},
- "required": ["city", "state"],
+ "required": ["city", "date"],
},
),
),
+ ]
+ self.detector = Glm47MoeDetector()
+
+ def test_single_tool_call(self):
+ text = (
+ "get_weather"
+ "cityBeijing"
+ "date2024-06-27"
+ ""
+ )
+ 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 = (
+ "get_weather"
+ "cityBeijing"
+ "date2024-06-27"
+ ""
+ "get_weather"
+ "cityShanghai"
+ "date2024-06-28"
+ ""
+ )
+ 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 = [
+ "get_weather",
+ "cityBeijing",
+ "date2024-06-27",
+ "",
+ ]
+ 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 = [
+ "get_weather",
+ "cityBeijing",
+ "date2024-06-27",
+ "get_weather",
+ "cityShanghai",
+ "date2024-06-28",
+ "",
+ ]
+ 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_id(self):
+ """Test that the buffer and state are reset after a tool call is completed."""
+ chunks = [
+ "get_weather",
+ "cityBeijing",
+ "date2024-06-27",
+ "",
+ ]
+ 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 = "invalid_funccityBeijing"
+ 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."""
+ chunks = [
+ "get_weather",
+ "cityBeijing",
+ "date2024-06-27",
+ ]
+
+ 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_array_argument_with_escaped_json(self):
+ """Test that array arguments with escaped JSON are properly handled without double-escaping."""
+ # Add a tool with array parameter
+ tools_with_array = [
Tool(
type="function",
function=Function(
- name="calculate_area",
- description="Calculate area of a shape",
+ name="todo_write",
+ description="Write todos",
parameters={
+ "type": "object",
"properties": {
- "shape": {"type": "string"},
- "dimensions": {"type": "object"},
- "precision": {"type": "integer"},
- }
+ "todos": {
+ "type": "array",
+ "description": "The updated todo list",
+ }
+ },
+ "required": ["todos"],
},
),
),
]
- self.detector = MiMoDetector()
- def test_has_tool_call(self):
- """Test detection of tool call markers."""
- self.assertTrue(self.detector.has_tool_call("test"))
- self.assertFalse(self.detector.has_tool_call("No tool call here"))
+ def check_params(result):
+ self.assertEqual(1, len(result.calls))
+ self.assertEqual("todo_write", result.calls[0].name)
+ params = json.loads(result.calls[0].parameters)
+ self.assertIsInstance(params["todos"], list)
+ self.assertEqual(4, len(params["todos"]))
+ self.assertEqual("1", params["todos"][0]["id"])
+ self.assertEqual(
+ "Check for hard-coded issues in the backend code",
+ params["todos"][0]["task"],
+ )
+ self.assertEqual("in_progress", params["todos"][0]["status"])
+ self.assertEqual("2", params["todos"][1]["id"])
+ self.assertEqual(
+ "Check for hard-coded issues in the frontend code",
+ params["todos"][1]["task"],
+ )
+ self.assertEqual("pending", params["todos"][1]["status"])
+ self.assertEqual("3", params["todos"][2]["id"])
+ self.assertEqual(
+ "Check for code violating the Single Responsibility Principle",
+ params["todos"][2]["task"],
+ )
+ self.assertEqual("pending", params["todos"][2]["status"])
+ self.assertEqual("4", params["todos"][3]["id"])
+ self.assertEqual(
+ "Generate a rectification proposal report", params["todos"][3]["task"]
+ )
+ self.assertEqual("pending", params["todos"][3]["status"])
- def test_detect_and_parse_no_tools(self):
- """Test parsing text without tool calls."""
- model_output = "This is a test response without any tool calls"
- result = self.detector.detect_and_parse(model_output, tools=[])
- self.assertEqual(result.normal_text, model_output)
- self.assertEqual(result.calls, [])
-
- def test_detect_and_parse_single_tool(self):
- """Test parsing a single tool call."""
- model_output = """
-
-Dallas
-TX
-fahrenheit
-
-"""
-
- result = self.detector.detect_and_parse(model_output, tools=self.tools)
-
- self.assertEqual(result.normal_text, "")
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "get_current_weather")
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["city"], "Dallas")
- self.assertEqual(params["state"], "TX")
- self.assertEqual(params["unit"], "fahrenheit")
-
- def test_detect_and_parse_with_content(self):
- """Test parsing tool call with surrounding text."""
- model_output = """Sure! Let me check the weather for you.
-
-Dallas
-TX
-fahrenheit
-
-"""
-
- result = self.detector.detect_and_parse(model_output, tools=self.tools)
-
- self.assertEqual(result.normal_text, "Sure! Let me check the weather for you.")
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "get_current_weather")
-
- def test_detect_and_parse_multiline_param(self):
- """Test parsing tool call with multiline parameter values."""
- model_output = """
-
-rectangle
-{"width": 10, "height": 20}
-2
-
-"""
-
- result = self.detector.detect_and_parse(model_output, tools=self.tools)
-
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "calculate_area")
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["shape"], "rectangle")
- self.assertEqual(params["dimensions"], {"width": 10, "height": 20})
- self.assertEqual(params["precision"], 2)
-
- def test_detect_and_parse_parallel_tools(self):
- """Test parsing multiple tool calls."""
- model_output = """
-
-Dallas
-TX
-fahrenheit
-
-
-
-
-Orlando
-FL
-fahrenheit
-
-"""
-
- result = self.detector.detect_and_parse(model_output, tools=self.tools)
-
- self.assertEqual(result.normal_text, "")
- self.assertEqual(len(result.calls), 2)
-
- # First call
- self.assertEqual(result.calls[0].name, "get_current_weather")
- params1 = json.loads(result.calls[0].parameters)
- self.assertEqual(params1["city"], "Dallas")
- self.assertEqual(params1["state"], "TX")
-
- # Second call
- self.assertEqual(result.calls[1].name, "get_current_weather")
- params2 = json.loads(result.calls[1].parameters)
- self.assertEqual(params2["city"], "Orlando")
- self.assertEqual(params2["state"], "FL")
-
- def test_parse_streaming_simple(self):
- """Test basic streaming parsing."""
- chunks = [
- "Sure! ",
- "Let me check ",
- "the weather.",
- "",
- "\n",
- "\nDallas",
- "\nTX",
- "\n",
- "\n",
- ]
-
- accumulated_text = ""
- accumulated_calls = []
- tool_calls_by_index = {}
-
- for chunk in chunks:
- result = self.detector.parse_streaming_increment(chunk, tools=self.tools)
- accumulated_text += result.normal_text
-
- # Track calls by tool_index to handle streaming properly
- 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, "Sure! Let me check the weather.")
- self.assertEqual(len(tool_calls_by_index), 1)
-
- # Get the complete tool call
- tool_call = tool_calls_by_index[0]
- self.assertEqual(tool_call["name"], "get_current_weather")
-
- # Parse the accumulated parameters
- params = json.loads(tool_call["parameters"])
- self.assertEqual(params["city"], "Dallas")
- self.assertEqual(params["state"], "TX")
-
- def test_parse_streaming_incomplete(self):
- """Test streaming with incomplete tool call."""
- # Send incomplete tool call
- chunks = [
- "",
- "\n",
- "\nDallas",
- "\n",
- # Missing , ,
- ]
-
- tool_calls_by_index = {}
- for chunk in chunks:
- result = self.detector.parse_streaming_increment(chunk, tools=self.tools)
-
- # Track calls by tool_index to handle streaming properly
- 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
-
- # Should have no complete tool calls yet (buffered)
- self.assertEqual(len(tool_calls_by_index), 0)
-
- # Now complete it
- result = self.detector.parse_streaming_increment(
- "TX\n\n", tools=self.tools
+ # Simulate the raw response from GLM-4.6 model with normal and escaped JSON in XML
+ result = self.detector.detect_and_parse(
+ """todo_writetodos[{\"id\": \"1\", \"task\": \"Check for hard-coded issues in the backend code\", \"status\": \"in_progress\"}, {\"id\": \"2\", \"task\": \"Check for hard-coded issues in the frontend code\", \"status\": \"pending\"}, {\"id\": \"3\", \"task\": \"Check for code violating the Single Responsibility Principle\", \"status\": \"pending\"}, {\"id\": \"4\", \"task\": \"Generate a rectification proposal report\", \"status\": \"pending\"}]
+""",
+ tools_with_array,
)
-
- # Update the accumulated parameters
- 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
-
- # Now should have complete tool call
- self.assertEqual(len(tool_calls_by_index), 1)
- final_params = json.loads(tool_calls_by_index[0]["parameters"])
- self.assertEqual(final_params["city"], "Dallas")
- self.assertEqual(final_params["state"], "TX")
-
- def test_edge_case_no_parameters(self):
- """Test tool call without parameters."""
- model_output = """
-
-
-"""
-
- result = self.detector.detect_and_parse(model_output, tools=self.tools)
- self.assertEqual(len(result.calls), 1)
- self.assertEqual(result.calls[0].name, "get_current_weather")
- self.assertEqual(json.loads(result.calls[0].parameters), {})
-
- def test_edge_case_special_chars_in_value(self):
- """Test parameter with special characters in value."""
- model_output = """
-
-Dallas->TX
-
-"""
-
- result = self.detector.detect_and_parse(model_output, tools=self.tools)
- self.assertEqual(len(result.calls), 1)
-
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["city"], "Dallas->TX")
-
- def test_extract_tool_calls_type_conversion(self):
- """Test parameter type conversion based on tool schema."""
- test_tool = Tool(
- type="function",
- function=Function(
- name="test_types",
- parameters={
- "type": "object",
- "properties": {
- "int_param": {"type": "integer"},
- "float_param": {"type": "float"},
- "bool_param": {"type": "boolean"},
- "str_param": {"type": "string"},
- "obj_param": {"type": "object"},
- },
- },
- ),
+ check_params(result)
+ result = self.detector.detect_and_parse(
+ r"""todo_writetodos[{\"id\": \"1\", \"task\": \"Check for hard-coded issues in the backend code\", \"status\": \"in_progress\"}, {\"id\": \"2\", \"task\": \"Check for hard-coded issues in the frontend code\", \"status\": \"pending\"}, {\"id\": \"3\", \"task\": \"Check for code violating the Single Responsibility Principle\", \"status\": \"pending\"}, {\"id\": \"4\", \"task\": \"Generate a rectification proposal report\", \"status\": \"pending\"}]
+""",
+ tools_with_array,
)
+ check_params(result)
- model_output = """
-
-42
-3.14
-true
-hello world
-{"key": "value"}
-
-"""
+ def check_single_todos(tool_result, expected):
+ self.assertEqual(1, len(tool_result.calls))
+ self.assertEqual("todo_write", tool_result.calls[0].name)
+ params = json.loads(tool_result.calls[0].parameters)
+ self.assertIsInstance(params["todos"], list)
+ self.assertEqual(1, len(params["todos"]))
+ self.assertEqual("1", params["todos"][0]["id"])
+ self.assertEqual(expected, params["todos"][0]["task"])
+ self.assertEqual("pending", params["todos"][0]["status"])
- result = self.detector.detect_and_parse(model_output, tools=[test_tool])
+ # Test with escaped JSON containing backslashes in content (e.g., Windows paths)
+ expected_path = r"Check file at C:\Users\test.txt"
+ result = self.detector.detect_and_parse(
+ """todo_writetodos[{\"id\": \"1\", \"task\": \"Check file at C:\\\\Users\\\\test.txt\", \"status\": \"pending\"}]""",
+ tools_with_array,
+ )
+ check_single_todos(result, expected_path)
+ result = self.detector.detect_and_parse(
+ r"""todo_writetodos[{\"id\": \"1\", \"task\": \"Check file at C:\\\\Users\\\\test.txt\", \"status\": \"pending\"}]""",
+ tools_with_array,
+ )
+ check_single_todos(result, expected_path)
- self.assertEqual(len(result.calls), 1)
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["int_param"], 42)
- self.assertEqual(params["float_param"], 3.14)
- self.assertEqual(params["bool_param"], True)
- self.assertEqual(params["str_param"], "hello world")
- self.assertEqual(params["obj_param"], {"key": "value"})
-
- def test_parse_streaming_incremental(self):
- """Test that streaming is truly incremental with very small chunks."""
- # Simulate more realistic token-based chunks where is a single token
- chunks = [
- "I'll check the weather.",
- "",
- "\n\n",
- "",
- "Dallas",
- "\n",
- "",
- "TX",
- "\n",
- "\n",
- "",
- ]
-
- accumulated_text = ""
- tool_calls = []
- chunks_count = 0
-
- for chunk in chunks:
- result = self.detector.parse_streaming_increment(chunk, self.tools)
- accumulated_text += result.normal_text
- chunks_count += 1
- 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.assertGreater(chunks_count, 3)
-
- # Verify the accumulated results
- self.assertIn("I'll check the weather.", accumulated_text)
- self.assertEqual(len(tool_calls), 1)
- self.assertEqual(tool_calls[0]["name"], "get_current_weather")
-
- params = json.loads(tool_calls[0]["parameters"])
- self.assertEqual(params, {"city": "Dallas", "state": "TX"})
-
- def test_parse_streaming_multiple_tools(self):
- """Test streaming with multiple tool calls."""
- model_output = """
-
-Dallas
-TX
-
-
-Some text in between.
-
-
-circle
-{"radius": 5}
-
-"""
-
- # Simulate streaming by chunks
- chunk_size = 20
- chunks = [
- model_output[i : i + chunk_size]
- for i in range(0, len(model_output), chunk_size)
- ]
-
- accumulated_text = ""
- tool_calls = []
- chunks_count = 0
-
- for chunk in chunks:
- result = self.detector.parse_streaming_increment(chunk, self.tools)
- accumulated_text += result.normal_text
- chunks_count += 1
- 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.assertIn("Some text in between.", accumulated_text)
- self.assertEqual(len(tool_calls), 2)
- self.assertEqual(tool_calls[0]["name"], "get_current_weather")
- self.assertEqual(tool_calls[1]["name"], "calculate_area")
-
- # Verify parameters
- params1 = json.loads(tool_calls[0]["parameters"])
- self.assertEqual(params1, {"city": "Dallas", "state": "TX"})
-
- params2 = json.loads(tool_calls[1]["parameters"])
- self.assertEqual(params2, {"shape": "circle", "dimensions": {"radius": 5}})
-
- def test_html_entity_decoding(self):
- """Test that HTML entities in parameter values are decoded."""
- model_output = """
-
-Dallas & Fort Worth
-TX
-
-"""
-
- result = self.detector.detect_and_parse(model_output, tools=self.tools)
-
- self.assertEqual(len(result.calls), 1)
- params = json.loads(result.calls[0].parameters)
- self.assertEqual(params["city"], "Dallas & Fort Worth")
+ # Should contain literal \n, not actual newline
+ expected_output = r"Print \n to see newline"
+ result = self.detector.detect_and_parse(
+ """todo_writetodos[{\"id\": \"1\", \"task\": \"Print \\\\n to see newline\",\"status\": \"pending\"}]""",
+ tools_with_array,
+ )
+ check_single_todos(result, expected_output)
+ result = self.detector.detect_and_parse(
+ r"""todo_writetodos[{\"id\": \"1\", \"task\": \"Print \\\\n to see newline\",\"status\": \"pending\"}]""",
+ tools_with_array,
+ )
+ check_single_todos(result, expected_output)
class TestJsonArrayParser(unittest.TestCase):