Skip to content

vllm.entrypoints.openai.tool_parsers.glm4_moe_tool_parser

logger module-attribute 

logger = init_logger(__name__)

Glm4MoeModelToolParser

Bases: ToolParser

Source code in vllm/entrypoints/openai/tool_parsers/glm4_moe_tool_parser.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@ToolParserManager.register_module("glm45")
class Glm4MoeModelToolParser(ToolParser):

    def __init__(self, tokenizer: AnyTokenizer):
        super().__init__(tokenizer)
        self.current_tool_name_sent = False
        self.prev_tool_call_arr: list[dict] = []
        self.current_tool_id = -1
        self.streamed_args_for_tool: list[str] = []
        self.tool_call_start_token = "<tool_call>"
        self.tool_call_end_token = "</tool_call>"

        self.tool_calls_start_token = self.tool_call_start_token

        self.func_call_regex = re.compile(r"<tool_call>.*?</tool_call>",
                                          re.DOTALL)
        self.func_detail_regex = re.compile(
            r"<tool_call>([^\n]*)\n(.*)</tool_call>", re.DOTALL)
        self.func_arg_regex = re.compile(
            r"<arg_key>(.*?)</arg_key>\s*<arg_value>(.*?)</arg_value>",
            re.DOTALL)
        if not self.model_tokenizer:
            raise ValueError(
                "The model tokenizer must be passed to the ToolParser "
                "constructor during construction.")

        self.tool_call_start_token_id = self.vocab.get(
            self.tool_call_start_token)
        self.tool_call_end_token_id = self.vocab.get(self.tool_call_end_token)
        self._buffer = ""

    def extract_tool_calls(
        self,
        model_output: str,
        request: ChatCompletionRequest,
    ) -> ExtractedToolCallInformation:

        def _is_string_type(
                tool_name: str, arg_name: str,
                tools: Optional[list[ChatCompletionToolsParam]]) -> bool:
            if tools is None:
                return False
            for tool in tools:
                if tool.function.name == tool_name:
                    if tool.function.parameters is None:
                        return False
                    arg_type = tool.function.parameters.get(
                        "properties", {}).get(arg_name, {}).get("type", None)
                    return arg_type == "string"
            logger.warning("No tool named '%s'.", tool_name)
            return False

        def _deserialize(value: str) -> Any:
            try:
                return json.loads(value)
            except Exception:
                pass

            try:
                return ast.literal_eval(value)
            except Exception:
                pass
            return value

        matched_tool_calls = self.func_call_regex.findall(model_output)
        logger.debug("model_output: %s", model_output)
        try:
            tool_calls = []
            for match in matched_tool_calls:
                tc_detail = self.func_detail_regex.search(match)
                tc_name = tc_detail.group(1)
                tc_args = tc_detail.group(2)
                pairs = self.func_arg_regex.findall(tc_args)
                arg_dct = {}
                for key, value in pairs:
                    arg_key = key.strip()
                    arg_val = value.strip()
                    if not _is_string_type(tc_name, arg_key, request.tools):
                        arg_val = _deserialize(arg_val)
                    logger.debug("arg_key = %s, arg_val = %s", arg_key,
                                 arg_val)
                    arg_dct[arg_key] = arg_val
                tool_calls.append(
                    ToolCall(type="function",
                             function=FunctionCall(
                                 name=tc_name, arguments=json.dumps(arg_dct))))
        except Exception:
            logger.exception("Failed to extract tool call spec")
            return ExtractedToolCallInformation(tools_called=False,
                                                tool_calls=[],
                                                content=model_output)
        else:
            if len(tool_calls) > 0:
                content = model_output[:model_output.
                                       find(self.tool_calls_start_token)]
                return ExtractedToolCallInformation(tools_called=True,
                                                    tool_calls=tool_calls,
                                                    content=content)
            return ExtractedToolCallInformation(tools_called=False,
                                                tool_calls=[],
                                                content=model_output)

    def extract_tool_calls_streaming(
        self,
        previous_text: str,
        current_text: str,
        delta_text: str,
        previous_token_ids: Sequence[int],
        current_token_ids: Sequence[int],
        delta_token_ids: Sequence[int],
        request: ChatCompletionRequest,
    ) -> Union[DeltaMessage, None]:
        self._buffer += delta_text
        cur_text = self._buffer
        start_idx = cur_text.find(self.tool_call_start_token)
        if start_idx == -1:
            self._buffer = ""
            if self.current_tool_id > 0:
                cur_text = ""
            return DeltaMessage(content=cur_text)
        logger.debug("cur_text = %s", cur_text)
        end_idx = cur_text.find(self.tool_call_end_token)
        if end_idx != -1:
            if self.current_tool_id == -1:
                self.current_tool_id = 0
                self.prev_tool_call_arr = []
                self.streamed_args_for_tool = []
            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("")

            extracted_tool_calls = self.extract_tool_calls(
                cur_text[:end_idx + len(self.tool_call_end_token)], request)

            if len(extracted_tool_calls.tool_calls) == 0:
                logger.warning("Failed to extract any tool calls.")
                return None
            tool_call = extracted_tool_calls.tool_calls[0]
            self.prev_tool_call_arr[self.current_tool_id] = {
                "name": tool_call.function.name,
                "arguments": json.loads(tool_call.function.arguments)
            }
            self.streamed_args_for_tool[
                self.current_tool_id] = tool_call.function.arguments
            delta = DeltaMessage(
                content=extracted_tool_calls.content,
                tool_calls=[
                    DeltaToolCall(index=self.current_tool_id,
                                  id=tool_call.id,
                                  type=tool_call.type,
                                  function=DeltaFunctionCall(
                                      name=tool_call.function.name,
                                      arguments=tool_call.function.arguments))
                ])
            self.current_tool_id += 1
            self._buffer = cur_text[end_idx + len(self.tool_call_end_token):]
            return delta

        self._buffer = cur_text[start_idx:]
        return DeltaMessage(content=cur_text[:start_idx])

_buffer instance-attribute 

_buffer = ''

current_tool_id instance-attribute 

current_tool_id = -1

current_tool_name_sent instance-attribute 

current_tool_name_sent = False

func_arg_regex instance-attribute 

func_arg_regex = compile(
    "<arg_key>(.*?)</arg_key>\\s*<arg_value>(.*?)</arg_value>",
    DOTALL,
)

func_call_regex instance-attribute 

func_call_regex = compile(
    "<tool_call>.*?</tool_call>", DOTALL
)

func_detail_regex instance-attribute 

func_detail_regex = compile(
    "<tool_call>([^\\n]*)\\n(.*)</tool_call>", DOTALL
)

prev_tool_call_arr instance-attribute 

prev_tool_call_arr: list[dict] = []

streamed_args_for_tool instance-attribute 

streamed_args_for_tool: list[str] = []

tool_call_end_token instance-attribute 

tool_call_end_token = '</tool_call>'

tool_call_end_token_id instance-attribute 

tool_call_end_token_id = get(tool_call_end_token)

tool_call_start_token instance-attribute 

tool_call_start_token = '<tool_call>'

tool_call_start_token_id instance-attribute 

tool_call_start_token_id = get(tool_call_start_token)

tool_calls_start_token instance-attribute 

tool_calls_start_token = tool_call_start_token

__init__ 

__init__(tokenizer: AnyTokenizer)
Source code in vllm/entrypoints/openai/tool_parsers/glm4_moe_tool_parser.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(self, tokenizer: AnyTokenizer):
    super().__init__(tokenizer)
    self.current_tool_name_sent = False
    self.prev_tool_call_arr: list[dict] = []
    self.current_tool_id = -1
    self.streamed_args_for_tool: list[str] = []
    self.tool_call_start_token = "<tool_call>"
    self.tool_call_end_token = "</tool_call>"

    self.tool_calls_start_token = self.tool_call_start_token

    self.func_call_regex = re.compile(r"<tool_call>.*?</tool_call>",
                                      re.DOTALL)
    self.func_detail_regex = re.compile(
        r"<tool_call>([^\n]*)\n(.*)</tool_call>", re.DOTALL)
    self.func_arg_regex = re.compile(
        r"<arg_key>(.*?)</arg_key>\s*<arg_value>(.*?)</arg_value>",
        re.DOTALL)
    if not self.model_tokenizer:
        raise ValueError(
            "The model tokenizer must be passed to the ToolParser "
            "constructor during construction.")

    self.tool_call_start_token_id = self.vocab.get(
        self.tool_call_start_token)
    self.tool_call_end_token_id = self.vocab.get(self.tool_call_end_token)
    self._buffer = ""

extract_tool_calls 

extract_tool_calls(
    model_output: str, request: ChatCompletionRequest
) -> ExtractedToolCallInformation
Source code in vllm/entrypoints/openai/tool_parsers/glm4_moe_tool_parser.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def extract_tool_calls(
    self,
    model_output: str,
    request: ChatCompletionRequest,
) -> ExtractedToolCallInformation:

    def _is_string_type(
            tool_name: str, arg_name: str,
            tools: Optional[list[ChatCompletionToolsParam]]) -> bool:
        if tools is None:
            return False
        for tool in tools:
            if tool.function.name == tool_name:
                if tool.function.parameters is None:
                    return False
                arg_type = tool.function.parameters.get(
                    "properties", {}).get(arg_name, {}).get("type", None)
                return arg_type == "string"
        logger.warning("No tool named '%s'.", tool_name)
        return False

    def _deserialize(value: str) -> Any:
        try:
            return json.loads(value)
        except Exception:
            pass

        try:
            return ast.literal_eval(value)
        except Exception:
            pass
        return value

    matched_tool_calls = self.func_call_regex.findall(model_output)
    logger.debug("model_output: %s", model_output)
    try:
        tool_calls = []
        for match in matched_tool_calls:
            tc_detail = self.func_detail_regex.search(match)
            tc_name = tc_detail.group(1)
            tc_args = tc_detail.group(2)
            pairs = self.func_arg_regex.findall(tc_args)
            arg_dct = {}
            for key, value in pairs:
                arg_key = key.strip()
                arg_val = value.strip()
                if not _is_string_type(tc_name, arg_key, request.tools):
                    arg_val = _deserialize(arg_val)
                logger.debug("arg_key = %s, arg_val = %s", arg_key,
                             arg_val)
                arg_dct[arg_key] = arg_val
            tool_calls.append(
                ToolCall(type="function",
                         function=FunctionCall(
                             name=tc_name, arguments=json.dumps(arg_dct))))
    except Exception:
        logger.exception("Failed to extract tool call spec")
        return ExtractedToolCallInformation(tools_called=False,
                                            tool_calls=[],
                                            content=model_output)
    else:
        if len(tool_calls) > 0:
            content = model_output[:model_output.
                                   find(self.tool_calls_start_token)]
            return ExtractedToolCallInformation(tools_called=True,
                                                tool_calls=tool_calls,
                                                content=content)
        return ExtractedToolCallInformation(tools_called=False,
                                            tool_calls=[],
                                            content=model_output)

extract_tool_calls_streaming 

extract_tool_calls_streaming(
    previous_text: str,
    current_text: str,
    delta_text: str,
    previous_token_ids: Sequence[int],
    current_token_ids: Sequence[int],
    delta_token_ids: Sequence[int],
    request: ChatCompletionRequest,
) -> Union[DeltaMessage, None]
Source code in vllm/entrypoints/openai/tool_parsers/glm4_moe_tool_parser.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def extract_tool_calls_streaming(
    self,
    previous_text: str,
    current_text: str,
    delta_text: str,
    previous_token_ids: Sequence[int],
    current_token_ids: Sequence[int],
    delta_token_ids: Sequence[int],
    request: ChatCompletionRequest,
) -> Union[DeltaMessage, None]:
    self._buffer += delta_text
    cur_text = self._buffer
    start_idx = cur_text.find(self.tool_call_start_token)
    if start_idx == -1:
        self._buffer = ""
        if self.current_tool_id > 0:
            cur_text = ""
        return DeltaMessage(content=cur_text)
    logger.debug("cur_text = %s", cur_text)
    end_idx = cur_text.find(self.tool_call_end_token)
    if end_idx != -1:
        if self.current_tool_id == -1:
            self.current_tool_id = 0
            self.prev_tool_call_arr = []
            self.streamed_args_for_tool = []
        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("")

        extracted_tool_calls = self.extract_tool_calls(
            cur_text[:end_idx + len(self.tool_call_end_token)], request)

        if len(extracted_tool_calls.tool_calls) == 0:
            logger.warning("Failed to extract any tool calls.")
            return None
        tool_call = extracted_tool_calls.tool_calls[0]
        self.prev_tool_call_arr[self.current_tool_id] = {
            "name": tool_call.function.name,
            "arguments": json.loads(tool_call.function.arguments)
        }
        self.streamed_args_for_tool[
            self.current_tool_id] = tool_call.function.arguments
        delta = DeltaMessage(
            content=extracted_tool_calls.content,
            tool_calls=[
                DeltaToolCall(index=self.current_tool_id,
                              id=tool_call.id,
                              type=tool_call.type,
                              function=DeltaFunctionCall(
                                  name=tool_call.function.name,
                                  arguments=tool_call.function.arguments))
            ])
        self.current_tool_id += 1
        self._buffer = cur_text[end_idx + len(self.tool_call_end_token):]
        return delta

    self._buffer = cur_text[start_idx:]
    return DeltaMessage(content=cur_text[:start_idx])