tool_utils.py 11.2 KB
Newer Older
chenych's avatar
chenych committed
1
# Copyright 2025 the LlamaFactory team.
chenych's avatar
chenych committed
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass
luopl's avatar
luopl committed
19
20
from datetime import datetime
from typing import Any, Dict, List, NamedTuple, Tuple, Union
chenych's avatar
chenych committed
21

luopl's avatar
luopl committed
22
23
from typing_extensions import override

chenych's avatar
chenych committed
24

luopl's avatar
luopl committed
25
26
27
28
29
class FunctionCall(NamedTuple):
    name: str
    arguments: str


chenych's avatar
chenych committed
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
DEFAULT_TOOL_PROMPT = (
    "You have access to the following tools:\n{tool_text}"
    "Use the following format if using a tool:\n"
    "```\n"
    "Action: tool name (one of [{tool_names}])\n"
    "Action Input: the input to the tool, in a JSON format representing the kwargs "
    """(e.g. ```{{"input": "hello world", "num_beams": 5}}```)\n"""
    "```\n"
)

GLM4_TOOL_PROMPT = (
    "你是一个名为 ChatGLM 的人工智能助手。你是基于智谱AI训练的语言模型 GLM-4 模型开发的,"
    "你的任务是针对用户的问题和要求提供适当的答复和支持。# 可用工具{tool_text}"
)

luopl's avatar
luopl committed
45
46
47
48
49
50
LLAMA3_TOOL_PROMPT = (
    "Cutting Knowledge Date: December 2023\nToday Date: {date}\n\n"
    "You have access to the following functions. To call a function, please respond with JSON for a function call. "
    """Respond in the format {{"name": function name, "parameters": dictionary of argument name and its value}}. """
    "Do not use variables.\n\n{tool_text}"
)
chenych's avatar
chenych committed
51

luopl's avatar
luopl committed
52
53
54
55
56
QWEN_TOOL_PROMPT = (
    "\n\n# Tools\n\nYou may call one or more functions to assist with the user query.\n\n"
    "You are provided with function signatures within <tools></tools> XML tags:\n<tools>{tool_text}"
    "\n</tools>\n\nFor each function call, return a json object with function name and arguments within "
    """<tool_call></tool_call> XML tags:\n<tool_call>\n{{"name": <function-name>, """
chenych's avatar
chenych committed
57
    """"arguments": <args-json-object>}}\n</tool_call>"""
luopl's avatar
luopl committed
58
)
luopl's avatar
luopl committed
59
60


chenych's avatar
chenych committed
61
62
@dataclass
class ToolUtils(ABC):
luopl's avatar
luopl committed
63
64
65
66
    """
    Base class for tool utilities.
    """

chenych's avatar
chenych committed
67
68
    @staticmethod
    @abstractmethod
luopl's avatar
luopl committed
69
    def tool_formatter(tools: List[Dict[str, Any]]) -> str:
luopl's avatar
luopl committed
70
        r"""
luopl's avatar
luopl committed
71
        Generates the system message describing all the available tools.
luopl's avatar
luopl committed
72
73
        """
        ...
chenych's avatar
chenych committed
74
75
76

    @staticmethod
    @abstractmethod
chenych's avatar
chenych committed
77
    def function_formatter(functions: List["FunctionCall"]) -> str:
luopl's avatar
luopl committed
78
        r"""
luopl's avatar
luopl committed
79
        Generates the assistant message including all the tool calls.
luopl's avatar
luopl committed
80
81
        """
        ...
chenych's avatar
chenych committed
82
83
84

    @staticmethod
    @abstractmethod
luopl's avatar
luopl committed
85
86
    def tool_extractor(content: str) -> Union[str, List["FunctionCall"]]:
        r"""
luopl's avatar
luopl committed
87
88
89
        Extracts all the function calls from the assistant message.

        It should be an inverse function of `function_formatter`.
luopl's avatar
luopl committed
90
91
        """
        ...
chenych's avatar
chenych committed
92
93
94


class DefaultToolUtils(ToolUtils):
luopl's avatar
luopl committed
95
96
97
    r"""
    Default tool using template.
    """
chenych's avatar
chenych committed
98

luopl's avatar
luopl committed
99
    @override
chenych's avatar
chenych committed
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
    @staticmethod
    def tool_formatter(tools: List[Dict[str, Any]]) -> str:
        tool_text = ""
        tool_names = []
        for tool in tools:
            param_text = ""
            for name, param in tool["parameters"]["properties"].items():
                required, enum, items = "", "", ""
                if name in tool["parameters"].get("required", []):
                    required = ", required"

                if param.get("enum", None):
                    enum = ", should be one of [{}]".format(", ".join(param["enum"]))

                if param.get("items", None):
                    items = ", where each item should be {}".format(param["items"].get("type", ""))

                param_text += "  - {name} ({type}{required}): {desc}{enum}{items}\n".format(
                    name=name,
                    type=param.get("type", ""),
                    required=required,
                    desc=param.get("description", ""),
                    enum=enum,
                    items=items,
                )

            tool_text += "> Tool Name: {name}\nTool Description: {desc}\nTool Args:\n{args}\n".format(
                name=tool["name"], desc=tool.get("description", ""), args=param_text
            )
            tool_names.append(tool["name"])

        return DEFAULT_TOOL_PROMPT.format(tool_text=tool_text, tool_names=", ".join(tool_names))

luopl's avatar
luopl committed
133
134
    @override
    @staticmethod
chenych's avatar
chenych committed
135
    def function_formatter(functions: List["FunctionCall"]) -> str:
luopl's avatar
luopl committed
136
137
138
139
        function_text = ""
        for name, arguments in functions:
            function_text += f"Action: {name}\nAction Input: {arguments}\n"

chenych's avatar
chenych committed
140
        return function_text
luopl's avatar
luopl committed
141

luopl's avatar
luopl committed
142
    @override
chenych's avatar
chenych committed
143
    @staticmethod
luopl's avatar
luopl committed
144
    def tool_extractor(content: str) -> Union[str, List["FunctionCall"]]:
chenych's avatar
chenych committed
145
146
147
148
149
150
151
152
153
154
155
        regex = re.compile(r"Action:\s*([a-zA-Z0-9_]+)\s*Action Input:\s*(.+?)(?=\s*Action:|\s*$)", re.DOTALL)
        action_match: List[Tuple[str, str]] = re.findall(regex, content)
        if not action_match:
            return content

        results = []
        for match in action_match:
            tool_name = match[0].strip()
            tool_input = match[1].strip().strip('"').strip("```")
            try:
                arguments = json.loads(tool_input)
luopl's avatar
luopl committed
156
                results.append(FunctionCall(tool_name, json.dumps(arguments, ensure_ascii=False)))
chenych's avatar
chenych committed
157
158
159
160
161
162
163
            except json.JSONDecodeError:
                return content

        return results


class GLM4ToolUtils(ToolUtils):
luopl's avatar
luopl committed
164
165
166
    r"""
    GLM-4 tool using template.
    """
chenych's avatar
chenych committed
167

luopl's avatar
luopl committed
168
    @override
chenych's avatar
chenych committed
169
170
171
172
173
174
175
176
177
178
    @staticmethod
    def tool_formatter(tools: List[Dict[str, Any]]) -> str:
        tool_text = ""
        for tool in tools:
            tool_text += "\n\n## {name}\n\n{body}\n在调用上述函数时,请使用 Json 格式表示调用的参数。".format(
                name=tool["name"], body=json.dumps(tool, indent=4, ensure_ascii=False)
            )

        return GLM4_TOOL_PROMPT.format(tool_text=tool_text)

luopl's avatar
luopl committed
179
180
    @override
    @staticmethod
chenych's avatar
chenych committed
181
    def function_formatter(functions: List["FunctionCall"]) -> str:
luopl's avatar
luopl committed
182
183
184
        if len(functions) > 1:
            raise ValueError("GLM-4 does not support parallel functions.")

chenych's avatar
chenych committed
185
        return f"{functions[0].name}\n{functions[0].arguments}"
luopl's avatar
luopl committed
186

luopl's avatar
luopl committed
187
    @override
chenych's avatar
chenych committed
188
    @staticmethod
luopl's avatar
luopl committed
189
    def tool_extractor(content: str) -> Union[str, List["FunctionCall"]]:
chenych's avatar
chenych committed
190
191
192
193
194
        if "\n" not in content:
            return content

        tool_name, tool_input = content.split("\n", maxsplit=1)
        try:
luopl's avatar
luopl committed
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
            arguments = json.loads(tool_input.strip())
        except json.JSONDecodeError:
            return content

        return [FunctionCall(tool_name, json.dumps(arguments, ensure_ascii=False))]


class Llama3ToolUtils(ToolUtils):
    r"""
    Llama 3.x tool using template with `tools_in_user_message=False`.

    Reference: https://www.llama.com/docs/model-cards-and-prompt-formats/llama3_1/#json-based-tool-calling
    """

    @override
    @staticmethod
    def tool_formatter(tools: List[Dict[str, Any]]) -> str:
        date = datetime.now().strftime("%d %b %Y")
        tool_text = ""
        for tool in tools:
            wrapped_tool = {"type": "function", "function": tool}
            tool_text += json.dumps(wrapped_tool, indent=4, ensure_ascii=False) + "\n\n"

        return LLAMA3_TOOL_PROMPT.format(date=date, tool_text=tool_text)

    @override
    @staticmethod
chenych's avatar
chenych committed
222
    def function_formatter(functions: List["FunctionCall"]) -> str:
luopl's avatar
luopl committed
223
224
225
        if len(functions) > 1:
            raise ValueError("Llama-3 does not support parallel functions.")

chenych's avatar
chenych committed
226
        return f'{{"name": "{functions[0].name}", "parameters": {functions[0].arguments}}}'
luopl's avatar
luopl committed
227
228
229
230
231
232

    @override
    @staticmethod
    def tool_extractor(content: str) -> Union[str, List["FunctionCall"]]:
        try:
            tool = json.loads(content.strip())
chenych's avatar
chenych committed
233
234
235
        except json.JSONDecodeError:
            return content

luopl's avatar
luopl committed
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
        if "name" not in tool or "parameters" not in tool:
            return content

        return [FunctionCall(tool["name"], json.dumps(tool["parameters"], ensure_ascii=False))]


class MistralToolUtils(ToolUtils):
    r"""
    Mistral v0.3 tool using template.
    """

    @override
    @staticmethod
    def tool_formatter(tools: List[Dict[str, Any]]) -> str:
        wrapped_tools = []
        for tool in tools:
            wrapped_tools.append({"type": "function", "function": tool})

        return "[AVAILABLE_TOOLS] " + json.dumps(wrapped_tools, ensure_ascii=False) + "[/AVAILABLE_TOOLS]"

    @override
    @staticmethod
chenych's avatar
chenych committed
258
    def function_formatter(functions: List["FunctionCall"]) -> str:
luopl's avatar
luopl committed
259
260
261
262
        function_texts = []
        for name, arguments in functions:
            function_texts.append(f'{{"name": "{name}", "arguments": {arguments}}}')

chenych's avatar
chenych committed
263
        return "[" + ", ".join(function_texts) + "]"
luopl's avatar
luopl committed
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302

    @override
    @staticmethod
    def tool_extractor(content: str) -> Union[str, List["FunctionCall"]]:
        try:
            tools = json.loads(content.strip())
        except json.JSONDecodeError:
            return content

        if not isinstance(tools, list):
            tools = [tools]

        results = []
        for tool in tools:
            if "name" not in tool or "arguments" not in tool:
                return content

            results.append(FunctionCall(tool["name"], json.dumps(tool["arguments"], ensure_ascii=False)))

        return results


class QwenToolUtils(ToolUtils):
    r"""
    Qwen 2.5 tool using template.
    """

    @override
    @staticmethod
    def tool_formatter(tools: List[Dict[str, Any]]) -> str:
        tool_text = ""
        for tool in tools:
            wrapped_tool = {"type": "function", "function": tool}
            tool_text += "\n" + json.dumps(wrapped_tool, ensure_ascii=False)

        return QWEN_TOOL_PROMPT.format(tool_text=tool_text)

    @override
    @staticmethod
chenych's avatar
chenych committed
303
    def function_formatter(functions: List["FunctionCall"]) -> str:
luopl's avatar
luopl committed
304
305
306
307
308
309
        function_texts = []
        for name, arguments in functions:
            function_texts.append(
                "<tool_call>\n" + f'{{"name": "{name}", "arguments": {arguments}}}' + "\n</tool_call>"
            )

chenych's avatar
chenych committed
310
        return "\n".join(function_texts)
luopl's avatar
luopl committed
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332

    @override
    @staticmethod
    def tool_extractor(content: str) -> Union[str, List["FunctionCall"]]:
        regex = re.compile(r"<tool_call>(.+?)</tool_call>(?=\s*<tool_call>|\s*$)", re.DOTALL)
        tool_match: List[str] = re.findall(regex, content)
        if not tool_match:
            return content

        results = []
        for tool in tool_match:
            try:
                tool = json.loads(tool.strip())
            except json.JSONDecodeError:
                return content

            if "name" not in tool or "arguments" not in tool:
                return content

            results.append(FunctionCall(tool["name"], json.dumps(tool["arguments"], ensure_ascii=False)))

        return results
luopl's avatar
luopl committed
333
334
335
336
337


TOOLS = {
    "default": DefaultToolUtils(),
    "glm4": GLM4ToolUtils(),
luopl's avatar
luopl committed
338
339
340
    "llama3": Llama3ToolUtils(),
    "mistral": MistralToolUtils(),
    "qwen": QwenToolUtils(),
luopl's avatar
luopl committed
341
342
343
344
345
346
}


def get_tool_utils(name: str) -> "ToolUtils":
    tool_utils = TOOLS.get(name, None)
    if tool_utils is None:
luopl's avatar
luopl committed
347
        raise ValueError(f"Tool utils `{name}` not found.")
luopl's avatar
luopl committed
348
349

    return tool_utils