ToolManager

ToolManager#

概述#

ToolManager 继承自 ToolManagerBase ，主要用于管理使用Python定义的工具。类型包括：

  • 工具基类 BaseTool

  • Python函数

  • Python类的成员函数

  • python文件下的全部Python函数 (支持指定 pattern 纳入和排除)

在Agent场景，负责向 AgentNode 提供工具列表，根据大语言模型(LLM)输出的工具调用指令执行对应工具，并向LLM反馈工具执行结果。

同时，还支持批量导出和重载所有工具的内部状态。

工具的添加、删除、查找和更新#

ToolManager 可通过不同的接口对内部工具进行增、删、改、查。

import asyncio
import math

from evofabric.core.tool import ToolManager, BaseTool


async def MULTIPLY(a: int, b: int):
    """MULTIPLY two numbers."""
    return a * b


def mydivide(a: float, b: float):
    """divide two numbers."""
    return a / b


def noinputfun():
    """The function has no params"""

    return "The function has no params."


class MyMath:
    def __init__(self) -> None:
        pass

    @staticmethod
    def cubic(a: float):
        """get a^3."""
        return a ** 3

    def mypow(self, a: int, power: int):
        """get any power of a."""
        return pow(a, power)

    @classmethod
    def logfun(cls, a: float):
        """my log function. a > 0."""
        return math.log(a)


async def main():
    mymath = MyMath()

    # =======================================================

    # ToolManager initialize
    tool_manager = ToolManager(
        tools=[MULTIPLY, noinputfun],
    )

    # =======================================================

    # add tools from python function
    tool_manager.add_callable_tools(
        tools=[mydivide, MyMath.logfun, mymath.mypow])

    async def add(a: float, b: float):
        "add two numbers"
        return a + b

    new_tool = BaseTool(
        name='add',
        description='add two float numbers.',
        func=add
    )
    # add tools from BaseTool instance
    tool_manager.add_callable_tools(
        tools=[new_tool])

    # add tools from python scripts
    tool_manager.add_python_file_tools(
        file_paths=["py_file_with_tools.py"],
        exclude_pattern_list=[["mysq*"]])

    # =======================================================

    # list tools, get all tool schemas in the toolmanager
    res = await tool_manager.list_tools()
    print("tools after update tools: \n", res)

    # =======================================================

    # delete tools
    tool_manager.delete_tools(['mydivide'])

    # =======================================================

    # update tools
    new_tool_v2 = BaseTool(
        name='add',
        description='add two FLOAT NUMBERS.',
        func=add
    )
    tool_manager.update_tools([new_tool_v2])

    # =======================================================

    # find tools
    ans = tool_manager.find_tools(['add', 'MULTIPLY'])
    print("found tools: \n", ans)

    # =======================================================


if __name__ == '__main__':
    asyncio.run(main())

工具调用#

调用 call_tools() 可以并行执行多个工具调用指令。

import json
import math

from evofabric.core.tool import ToolManager
from evofabric.core.typing import Function, ToolCall


async def MULTIPLY(a: int, b: int):
    """MULTIPLY two numbers."""
    return a * b


def mydivide(a: float, b: float):
    """divide two numbers."""
    return a / b


def noinputfun():
    """The function has no params."""

    return "The function has no params."


class MyMath:
    def __init__(self) -> None:
        pass

    @staticmethod
    def cubic(a: float):
        """get a^3."""
        return a ** 3

    def mypow(self, a: int, power: int):
        """get any power of a."""
        return pow(a, power)

    @classmethod
    def logfun(cls, a: float):
        """my log function. a > 0."""
        return math.log(a)


async def main():
    mymath = MyMath()

    toolcall_1 = ToolCall(
        id="1",
        function=Function(
            name="MULTIPLY",
            arguments=json.dumps(
                {
                    "a": 3,
                    "b": 4
                }
            )
        )

    )
    toolcall_2 = ToolCall(
        id="2",
        function=Function(
            name="mydivide",
            arguments=json.dumps(
                {
                    "a": 4,
                    "b": 0
                }
            )
        )

    )
    toolcall_3 = ToolCall(
        id="3",
        function=Function(
            name="noinputfun",
            arguments=json.dumps({}),
        )

    )

    tool_manager = ToolManager(
        tools=[MULTIPLY, mydivide, noinputfun],
    )

    # =======================================================

    # call tools
    res = await tool_manager.call_tools([toolcall_1, toolcall_2, toolcall_3])
    print("call tool result: \n", res)

工具状态的导出和重载#

针对带有 inner_state 的工具，ToolManager可以导出、重载工具的内部状态。

import asyncio
import json
import os

from evofabric.core.tool import ToolManager
from evofabric.core.typing import Function, ToolCall, ToolInnerState


async def mycd(path: str, inner_state: ToolInnerState):
    new_path = os.path.join(inner_state.state['current_dir'], path)
    inner_state.state['current_dir'] = new_path
    return new_path


async def main():
    inner_state = ToolInnerState(
        state={
            "current_dir": "/xxx/xxx"
        }
    )

    tool_manager = ToolManager(
        tools=[(mycd, inner_state)],
    )

    # Save the state of all tools in tool_manager to "test_tool_manager_state_start.json"
    await tool_manager.save_state(save_path="test_tool_manager_state_start.json")

    # Tool call information, only explicitly pass parameters that are not internal state;
    # parameters that belong to internal state are automatically passed from the tool's internal attributes
    toolcall_5 = ToolCall(
        id="5",
        function=Function(
            name="mycd",
            arguments=json.dumps({
                "path": "sss/",
            }),
        )

    )

    # When calling the tool, the tool's internal state is updated in real time according to its logic
    res1 = await tool_manager.call_tools([toolcall_5])

    # Reload the internal state of tool_manager to the state saved at the beginning
    await tool_manager.load_state(load_path="test_tool_manager_state_start.json")


if __name__ == '__main__':
    asyncio.run(main())
目录