| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import json |
| 5 | from typing import TYPE_CHECKING |
| 6 | |
| 7 | import partial_json_parser |
| 8 | import regex as re |
| 9 | from partial_json_parser.core.options import Allow |
| 10 | |
| 11 | from vllm.entrypoints.chat_utils import make_tool_call_id |
| 12 | from vllm.entrypoints.generate.base.protocol import ( |
| 13 | DeltaFunctionCall, |
| 14 | DeltaMessage, |
| 15 | DeltaToolCall, |
| 16 | ) |
| 17 | from vllm.tool_parsers.utils import partial_json_loads |
| 18 | from vllm.utils.mistral import is_mistral_tokenizer |
| 19 | |
| 20 | if TYPE_CHECKING: |
| 21 | from vllm.tokenizers import TokenizerLike |
| 22 | else: |
| 23 | TokenizerLike = object |
| 24 | |
| 25 | |
| 26 | def _bracket_level_state( |
| 27 | s: str, opening: str = "{", closing: str = "}" |
| 28 | ) -> tuple[int, bool, bool]: |
| 29 | level = 0 |
| 30 | in_string = False |
| 31 | escaped = False |
| 32 | for char in s: |
| 33 | if escaped: |
| 34 | escaped = False |
| 35 | continue |
| 36 | if in_string and char == "\\": |
| 37 | escaped = True |
| 38 | continue |
| 39 | if char == '"': |
| 40 | in_string = not in_string |
| 41 | continue |
| 42 | if not in_string: |
| 43 | if char == opening: |
| 44 | level += 1 |
| 45 | elif char == closing: |
| 46 | level -= 1 |
| 47 | return level, in_string, escaped |
| 48 | |
| 49 | |
| 50 | def filter_delta_text( |
| 51 | delta_text: str, |
| 52 | previous_text: str, |
| 53 | ) -> tuple[str, bool]: |
| 54 | """Trim trailing tool-list delimiters from required-tool streaming text.""" |
| 55 | bracket_level, in_string, escaped = _bracket_level_state(previous_text) |
| 56 | updated_delta = "" |
| 57 | passed_zero = False |
| 58 | for char in delta_text: |
| 59 | if escaped: |
| 60 | escaped = False |
| 61 | elif in_string: |
| 62 | if char == "\\": |
| 63 | escaped = True |
| 64 | elif char == '"': |
| 65 | in_string = False |
| 66 | elif char == '"': |
| 67 | in_string = True |
| 68 | elif char == "{": |
| 69 | bracket_level += 1 |
| 70 | passed_zero = bracket_level == 0 |
| 71 | elif char == "}": |
| 72 | bracket_level -= 1 |
| 73 | passed_zero = bracket_level == 0 |
| 74 | |
| 75 | if bracket_level != 0: |
| 76 | updated_delta += char |
| 77 | else: |
| 78 | if not in_string and char == ",": |
| 79 | break |
| 80 | return updated_delta, passed_zero |
| 81 | |
| 82 | |
| 83 | def extract_named_tool_call_streaming( |
| 84 | *, |
| 85 | delta_text: str, |
| 86 | function_name: str, |
| 87 | function_name_returned: bool, |
| 88 | tool_call_idx: int | None, |
| 89 | tool_call_id_type: str, |
| 90 | tokenizer: "TokenizerLike", |
| 91 | tool_call_array_index: int = 0, |
| 92 | ) -> tuple[DeltaMessage | None, bool]: |
| 93 | """Build a streaming tool-call delta for forced named tool choice.""" |
| 94 | if function_name_returned: |
| 95 | delta_tool_call = DeltaToolCall( |
| 96 | function=DeltaFunctionCall(arguments=delta_text), |
| 97 | index=tool_call_array_index, |
| 98 | ) |
| 99 | else: |
| 100 | if is_mistral_tokenizer(tokenizer): |
| 101 | # Import mistral_common only if we need it. |
| 102 | from vllm.parser.mistral import MistralToolCall |
| 103 | |
| 104 | tool_call_id = MistralToolCall.generate_random_id() |
| 105 | else: |
| 106 | tool_call_id = make_tool_call_id( |
| 107 | id_type=tool_call_id_type, |
| 108 | func_name=function_name, |
| 109 | idx=tool_call_idx, |
| 110 | ) |
| 111 | delta_tool_call = DeltaToolCall( |
| 112 | id=tool_call_id, |
| 113 | type="function", |
| 114 | function=DeltaFunctionCall( |
| 115 | name=function_name, |
| 116 | arguments=delta_text, |
| 117 | ), |
| 118 | index=tool_call_array_index, |
| 119 | ) |
| 120 | function_name_returned = True |
| 121 | return ( |
| 122 | DeltaMessage(tool_calls=[delta_tool_call]), |
| 123 | function_name_returned, |
| 124 | ) |
| 125 | |
| 126 | |
| 127 | def extract_required_tool_call_streaming( |
| 128 | *, |
| 129 | previous_text: str, |
| 130 | current_text: str | None, |
| 131 | delta_text: str, |
| 132 | function_name_returned: bool, |
| 133 | tool_call_idx: int | None, |
| 134 | tool_call_id_type: str, |
| 135 | ) -> tuple[DeltaMessage | None, bool]: |
| 136 | if current_text is None or current_text == "": |
| 137 | # if the current text is empty, we cannot parse it |
| 138 | return None, function_name_returned |
| 139 | try: |
| 140 | flags = Allow.ALL |
| 141 | obj, _ = partial_json_loads(current_text, flags) |
| 142 | except ( |
| 143 | partial_json_parser.core.exceptions.MalformedJSON, |
| 144 | json.JSONDecodeError, |
| 145 | ): |
| 146 | obj = None |
| 147 | |
| 148 | # check if the current text is a valid array |
| 149 | # containing a partial tool calling object |
| 150 | # if not repeat |
| 151 | if obj is None or not isinstance(obj, list) or not len(obj) > 0: |
| 152 | function_name_returned = False |
| 153 | delta_message = None |
| 154 | else: |
| 155 | _, finishes_previous_tool = filter_delta_text(delta_text, previous_text) |
| 156 | # take the last tool call from the generated list |
| 157 | current_tool_call = obj[-1] |
| 158 | |
| 159 | # once parameters have been generated the name is complete as well |
| 160 | if not finishes_previous_tool and ( |
| 161 | "name" not in current_tool_call or "parameters" not in current_tool_call |
| 162 | ): |
| 163 | function_name_returned = False |
| 164 | delta_message = None |
| 165 | else: |
| 166 | if not function_name_returned: |
| 167 | # get partly generated arguments from the latest tool call |
| 168 | param_match = re.search( |
| 169 | r'.*"parameters":\s*(.*)', current_text, re.DOTALL |
| 170 | ) |
| 171 | if param_match: |
| 172 | arguments = param_match.group(1) |
| 173 | arguments_prefix = current_text[: param_match.start(1)] |
| 174 | arguments, _ = filter_delta_text(arguments, arguments_prefix) |
| 175 | else: |
| 176 | arguments = "" |
| 177 | |
| 178 | # if this iteration finishes a previous tool call but a |
| 179 | # new incomplete tool is already generated, take the |
| 180 | # previous from the list |
| 181 | if finishes_previous_tool and "parameters" not in current_tool_call: |
| 182 | current_tool_call = obj[-2] |
| 183 | |
| 184 | function_name_returned = True |
| 185 | tool_call_id = make_tool_call_id( |
| 186 | id_type=tool_call_id_type, |
| 187 | func_name=current_tool_call["name"], |
| 188 | idx=tool_call_idx, |
| 189 | ) |
| 190 | delta_message = DeltaMessage( |
| 191 | tool_calls=[ |
| 192 | DeltaToolCall( |
| 193 | id=tool_call_id, |
| 194 | function=DeltaFunctionCall( |
| 195 | name=current_tool_call["name"], arguments=arguments |
| 196 | ), |
| 197 | index=len(obj) - 1, |
| 198 | type="function", |
| 199 | ) |
| 200 | ] |
| 201 | ) |
| 202 | |
| 203 | else: |
| 204 | delta_text, _ = filter_delta_text(delta_text, previous_text) |
| 205 | |
| 206 | if delta_text != "": |
| 207 | delta_message = DeltaMessage( |
| 208 | tool_calls=[ |
| 209 | DeltaToolCall( |
| 210 | function=DeltaFunctionCall( |
| 211 | # OpenAI API returns None |
| 212 | # instead of name every time |
| 213 | name=None, |
| 214 | arguments=delta_text, |
| 215 | ), |
| 216 | index=len(obj) - 1, |
| 217 | ) |
| 218 | ] |
| 219 | ) |
| 220 | else: |
| 221 | delta_message = None |
| 222 | |
| 223 | return delta_message, function_name_returned |