ToolManager

Overview

ToolManager inherits from ToolManagerBase, mainly used for managing tools defined using Python. Types include:

• Tool Base Class: BaseTool

• Python function

• Python class member functions

• All Python functions in the Python file (supports specifying a pattern for inclusion and exclusion)

In the Agent scenario, responsible for providing the tool list to AgentNode, executing the corresponding tools based on the tool call instructions output by the large language model (LLM), and providing feedback on the tool execution results to the LLM.

It also supports batch export and reload of all tools’ internal states.

Addition, deletion, search, and update of tools

ToolManager can perform add, delete, modify, and query operations on internal tools through different interfaces.

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())

Tool Call

Calling call_tools() can execute multiple tool call instructions in parallel.

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)

Export and Reload of Tool State

For tools with inner_state, ToolManager can export and reload the tool’s internal state.

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())