main
py 434 lines 15.6 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4 import json
5 from collections.abc import Sequence
6 from typing import Any
7
8 import regex as re
9
10 from vllm.entrypoints.chat_utils import make_tool_call_id
11 from vllm.entrypoints.generate.base.protocol import (
12 DeltaFunctionCall,
13 DeltaMessage,
14 DeltaToolCall,
15 ExtractedToolCallInformation,
16 FunctionCall,
17 ToolCall,
18 )
19 from vllm.entrypoints.openai.chat_completion.protocol import (
20 ChatCompletionRequest,
21 ChatCompletionToolsParam,
22 )
23 from vllm.logger import init_logger
24 from vllm.tokenizers import TokenizerLike
25 from vllm.tool_parsers.abstract_tool_parser import Tool, ToolParser
26 from vllm.tool_parsers.utils import is_complete_json, partial_tag_overlap
27
28 logger = init_logger(__name__)
29
30
31 class DotsToolParser(ToolParser):
32 """Parse Dots tool calls in their XML wrapper format.
33
34 The canonical body contains one or more ``invoke`` elements::
35
36 <dots_function_call>
37 <invoke name="search">
38 <parameter name="query">weather in Shanghai</parameter>
39 </invoke>
40 </dots_function_call>
41
42 A JSON object with ``name`` and ``arguments`` is also accepted as a
43 fallback. Multiple wrapper blocks and multiple invokes per block are
44 supported.
45 """
46
47 supports_required_and_named = False
48
49 tool_call_start_token = "<dots_function_call>"
50 tool_call_end_token = "</dots_function_call>"
51
52 _block_regex = re.compile(
53 rf"{re.escape(tool_call_start_token)}\s*(.*?)\s*"
54 rf"{re.escape(tool_call_end_token)}",
55 re.DOTALL,
56 )
57 _invoke_regex = re.compile(
58 r"<invoke\s+name\s*=\s*(?P<name>[^>]+)>(?P<body>.*?)</invoke>",
59 re.DOTALL,
60 )
61 _parameter_regex = re.compile(
62 r"<parameter\s+name\s*=\s*(?P<name>[^>]+)>(?P<value>.*?)</parameter>",
63 re.DOTALL,
64 )
65
66 def __init__(
67 self,
68 tokenizer: TokenizerLike,
69 tools: list[Tool] | None = None,
70 ) -> None:
71 super().__init__(tokenizer, tools)
72 self._buffer = ""
73
74 @staticmethod
75 def _extract_name(value: str) -> str:
76 value = value.strip()
77 if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}:
78 return value[1:-1]
79 return value
80
81 @staticmethod
82 def _convert_param_value(value: str, param_type: Any) -> Any:
83 if value.lower() == "null":
84 return None
85
86 if isinstance(param_type, list):
87 param_type = next((item for item in param_type if item != "null"), "string")
88 if not isinstance(param_type, str):
89 param_type = str(param_type)
90 param_type = param_type.lower()
91
92 if param_type in {"string", "str", "text"}:
93 return value
94 if param_type in {"integer", "int"}:
95 try:
96 return int(value)
97 except (TypeError, ValueError):
98 return value
99 if param_type in {"number", "float"}:
100 try:
101 number = float(value)
102 return int(number) if number.is_integer() else number
103 except (TypeError, ValueError):
104 return value
105 if param_type in {"boolean", "bool"}:
106 return value.lower() in {"true", "1"}
107
108 try:
109 return json.loads(value)
110 except (json.JSONDecodeError, TypeError, ValueError):
111 return value
112
113 def _resolve_param_type(
114 self,
115 schema: Any,
116 defs: dict[str, Any],
117 depth: int = 0,
118 ) -> Any | None:
119 if not isinstance(schema, dict) or depth > 10:
120 return None
121 if "type" in schema:
122 return schema["type"]
123
124 ref = schema.get("$ref")
125 if isinstance(ref, str) and ref.startswith("#/$defs/"):
126 return self._resolve_param_type(
127 defs.get(ref.rsplit("/", 1)[-1]), defs, depth + 1
128 )
129
130 for keyword in ("anyOf", "oneOf", "allOf"):
131 alternatives = schema.get(keyword)
132 if not isinstance(alternatives, list):
133 continue
134 for alternative in alternatives:
135 if isinstance(alternative, dict) and alternative.get("type") == "null":
136 continue
137 resolved = self._resolve_param_type(alternative, defs, depth + 1)
138 if resolved is not None:
139 return resolved
140 return None
141
142 @staticmethod
143 def _tool_schema(
144 name: str,
145 tools: list[ChatCompletionToolsParam] | None,
146 ) -> tuple[dict[str, Any], dict[str, Any]]:
147 for tool in tools or []:
148 if tool.function.name != name:
149 continue
150 schema = tool.function.parameters
151 if not isinstance(schema, dict):
152 break
153 properties = schema.get("properties", {})
154 defs = schema.get("$defs", {})
155 return (
156 properties if isinstance(properties, dict) else {},
157 defs if isinstance(defs, dict) else {},
158 )
159 return {}, {}
160
161 def _parse_xml_invoke(
162 self,
163 match: re.Match[str],
164 tools: list[ChatCompletionToolsParam] | None,
165 ) -> dict[str, Any]:
166 name = self._extract_name(match.group("name"))
167 properties, defs = self._tool_schema(name, tools)
168 arguments: dict[str, Any] = {}
169
170 for parameter in self._parameter_regex.finditer(match.group("body")):
171 param_name = self._extract_name(parameter.group("name"))
172 value = parameter.group("value").strip()
173 param_type: Any = "string"
174 if param_name in properties:
175 param_type = (
176 self._resolve_param_type(properties[param_name], defs) or "string"
177 )
178 arguments[param_name] = self._convert_param_value(value, param_type)
179
180 return {"name": name, "arguments": arguments}
181
182 def _parse_block(
183 self,
184 content: str,
185 tools: list[ChatCompletionToolsParam] | None,
186 ) -> list[dict[str, Any]]:
187 content = content.strip()
188 if content.startswith("<invoke"):
189 return [
190 self._parse_xml_invoke(match, tools)
191 for match in self._invoke_regex.finditer(content)
192 ]
193
194 parsed = json.loads(content)
195 if not isinstance(parsed, dict):
196 raise TypeError("Dots JSON tool call must be an object")
197 return [parsed]
198
199 @staticmethod
200 def _known_tool_names(
201 tools: list[ChatCompletionToolsParam] | None,
202 ) -> set[str]:
203 return {tool.function.name for tool in tools or []}
204
205 def _validated_call(
206 self,
207 parsed: dict[str, Any],
208 tools: list[ChatCompletionToolsParam] | None,
209 ) -> tuple[str, dict[str, Any]] | None:
210 name = parsed.get("name")
211 if not isinstance(name, str) or name not in self._known_tool_names(tools):
212 return None
213 arguments = parsed.get("arguments", parsed.get("parameters", {})) or {}
214 if not isinstance(arguments, dict):
215 return None
216 return name, arguments
217
218 def extract_tool_calls(
219 self,
220 model_output: str,
221 request: ChatCompletionRequest,
222 ) -> ExtractedToolCallInformation:
223 marker_index = model_output.find(self.tool_call_start_token)
224 if marker_index == -1:
225 return ExtractedToolCallInformation(
226 tools_called=False,
227 tool_calls=[],
228 content=model_output,
229 )
230
231 tool_calls: list[ToolCall] = []
232 for block in self._block_regex.finditer(model_output):
233 try:
234 for parsed in self._parse_block(block.group(1), request.tools):
235 validated = self._validated_call(parsed, request.tools)
236 if validated is None:
237 continue
238 name, arguments = validated
239 tool_calls.append(
240 ToolCall(
241 function=FunctionCall(
242 name=name,
243 arguments=json.dumps(arguments, ensure_ascii=False),
244 )
245 )
246 )
247 except (json.JSONDecodeError, TypeError, ValueError) as exc:
248 logger.warning("Failed to parse Dots tool call: %s", exc)
249
250 normal_text = model_output[:marker_index].strip()
251 if not tool_calls:
252 return ExtractedToolCallInformation(
253 tools_called=False,
254 tool_calls=[],
255 content=model_output,
256 )
257 return ExtractedToolCallInformation(
258 tools_called=True,
259 tool_calls=tool_calls,
260 content=normal_text or None,
261 )
262
263 def _append_complete_stream_call(
264 self,
265 name: str,
266 arguments: dict[str, Any],
267 tool_calls: list[DeltaToolCall],
268 ) -> None:
269 self.current_tool_id += 1
270 serialized = json.dumps(arguments, ensure_ascii=False)
271 self.prev_tool_call_arr.append({"name": name, "arguments": arguments})
272 self.streamed_args_for_tool.append(serialized)
273 tool_calls.append(
274 DeltaToolCall(
275 index=self.current_tool_id,
276 id=make_tool_call_id(),
277 type="function",
278 function=DeltaFunctionCall(name=name, arguments=serialized),
279 )
280 )
281
282 def _stream_complete_json_body(
283 self,
284 tools: list[ChatCompletionToolsParam] | None,
285 tool_calls: list[DeltaToolCall],
286 ) -> None:
287 content = self._buffer[len(self.tool_call_start_token) :].strip()
288 if not content or not is_complete_json(content):
289 return
290
291 try:
292 parsed = json.loads(content)
293 except (json.JSONDecodeError, TypeError, ValueError):
294 return
295 if not isinstance(parsed, dict):
296 return
297 validated = self._validated_call(parsed, tools)
298 if validated is None:
299 return
300
301 name, arguments = validated
302 serialized = json.dumps(arguments, ensure_ascii=False)
303 if not self.current_tool_name_sent:
304 self.current_tool_id += 1
305 tool_calls.append(
306 DeltaToolCall(
307 index=self.current_tool_id,
308 id=make_tool_call_id(),
309 type="function",
310 function=DeltaFunctionCall(name=name, arguments=""),
311 )
312 )
313 self.prev_tool_call_arr.append({"name": name, "arguments": arguments})
314 self.streamed_args_for_tool.append("")
315 self.current_tool_name_sent = True
316
317 streamed = self.streamed_args_for_tool[self.current_tool_id]
318 if serialized.startswith(streamed):
319 argument_diff = serialized[len(streamed) :]
320 if argument_diff:
321 tool_calls.append(
322 DeltaToolCall(
323 index=self.current_tool_id,
324 function=DeltaFunctionCall(arguments=argument_diff),
325 )
326 )
327 self.streamed_args_for_tool[self.current_tool_id] += argument_diff
328
329 def extract_tool_calls_streaming(
330 self,
331 previous_text: str,
332 current_text: str,
333 delta_text: str,
334 previous_token_ids: Sequence[int],
335 current_token_ids: Sequence[int],
336 delta_token_ids: Sequence[int],
337 request: ChatCompletionRequest,
338 ) -> DeltaMessage | None:
339 del current_text, previous_token_ids, current_token_ids, delta_token_ids
340 if not previous_text:
341 self._buffer = ""
342 self.prev_tool_call_arr = []
343 self.current_tool_id = -1
344 self.current_tool_name_sent = False
345 self.streamed_args_for_tool = []
346
347 self._buffer += delta_text
348 normal_parts: list[str] = []
349 tool_calls: list[DeltaToolCall] = []
350
351 while self._buffer:
352 marker_index = self._buffer.find(self.tool_call_start_token)
353 if marker_index == -1:
354 partial_len = partial_tag_overlap(
355 self._buffer, self.tool_call_start_token
356 )
357 if partial_len:
358 normal_parts.append(self._buffer[:-partial_len])
359 self._buffer = self._buffer[-partial_len:]
360 else:
361 normal_parts.append(self._buffer)
362 self._buffer = ""
363 normal_parts = [
364 part.replace(self.tool_call_end_token, "") for part in normal_parts
365 ]
366 break
367
368 if marker_index > 0:
369 normal_parts.append(self._buffer[:marker_index])
370 self._buffer = self._buffer[marker_index:]
371
372 end_index = self._buffer.find(
373 self.tool_call_end_token, len(self.tool_call_start_token)
374 )
375 if end_index == -1:
376 self._stream_complete_json_body(request.tools, tool_calls)
377 break
378
379 content = self._buffer[len(self.tool_call_start_token) : end_index]
380 self._buffer = self._buffer[end_index + len(self.tool_call_end_token) :]
381 try:
382 parsed_calls = self._parse_block(content, request.tools)
383 if not parsed_calls:
384 raise ValueError("Dots tool-call block contains no invoke")
385
386 block_had_streamed_call = self.current_tool_name_sent
387 valid_calls = [
388 validated
389 for parsed in parsed_calls
390 if (validated := self._validated_call(parsed, request.tools))
391 is not None
392 ]
393 if self.current_tool_name_sent and valid_calls:
394 name, arguments = valid_calls.pop(0)
395 serialized = json.dumps(arguments, ensure_ascii=False)
396 streamed = self.streamed_args_for_tool[self.current_tool_id]
397 if serialized.startswith(streamed):
398 remaining = serialized[len(streamed) :]
399 if remaining:
400 tool_calls.append(
401 DeltaToolCall(
402 index=self.current_tool_id,
403 function=DeltaFunctionCall(arguments=remaining),
404 )
405 )
406 self.prev_tool_call_arr[self.current_tool_id] = {
407 "name": name,
408 "arguments": arguments,
409 }
410 self.streamed_args_for_tool[self.current_tool_id] = serialized
411
412 for name, arguments in valid_calls:
413 self._append_complete_stream_call(name, arguments, tool_calls)
414
415 if not valid_calls and not block_had_streamed_call:
416 normal_parts.append(content.strip())
417 except (json.JSONDecodeError, TypeError, ValueError) as exc:
418 logger.warning("Failed to parse streamed Dots tool call: %s", exc)
419 normal_parts.append(content.strip())
420
421 self.current_tool_name_sent = False
422
423 content_delta = "".join(normal_parts)
424 if not content_delta and not tool_calls:
425 return None
426 return DeltaMessage(content=content_delta or None, tool_calls=tool_calls)
427
428 def flush_pending_normal_text(self) -> str:
429 """Return a partial opening marker as text when generation ends."""
430 if not self._buffer or self.tool_call_start_token in self._buffer:
431 return ""
432 normal_text = self._buffer.replace(self.tool_call_end_token, "")
433 self._buffer = ""
434 return normal_text