tool

Tools

ToolError

Bases: AgereUtilsError

Raised when encountering an error related to the tool.

Source code in src/agere/utils/tool.py

class ToolError(AgereUtilsError):
    """Raised when encountering an error related to the tool."""

ToolsManager

Bases: ToolsManagerInterface

A class that manages tools.

Allowing for tool registration, addition, removal, and querying, as well as the generation of tool metadata and manuals.

Source code in src/agere/utils/tool.py

class ToolsManager(ToolsManagerInterface):
    """A class that manages tools.

    Allowing for tool registration, addition, removal, and querying,
    as well as the generation of tool metadata and manuals.
    """
    def __init__(self, tool_model: ToolModelInterface):
        self._tool_model = tool_model
        self._permanent_tools: dict[str, ToolMetadata] = {}
        self._temporary_tools: dict[str, ToolMetadata] = {}
        self._registered_tools: dict[str, ToolMetadata] = {}

    @property
    def tool_model_name(self) -> str:
        """Get the name of the tool model."""
        return self.tool_model.tool_model_name

    @property
    def tool_model_type(self) -> Literal["CUSTOM", "PROVIDED"]:
        """Get the type of the tool model.

        Returns:
            Either 'CUSTOM' or 'PROVIDED'.
            'CUSTOM' means that LLM does not provide an excuse for tool invocation,
            and a custom protocol is used to implement tool invocation.
            'PROVIDED' means using the tool invocation interface provided by LLM itself.
        """
        if isinstance(self._tool_model, CustomToolModelInterface):
            return "CUSTOM"
        elif isinstance(self._tool_model, ProvidedToolModelInterface):
            return "PROVIDED"
        else:
            assert False

    @property
    def tool_model(self) -> ToolModelInterface:
        """Get the tool model instance."""
        return self._tool_model

    @tool_model.setter
    def tool_model(self, value: ToolModelInterface):
        """Set the tool model instance."""
        self._tool_model = value

    def add_tool(
        self,
        tool: Callable | ToolKit,
        tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY",
    ) -> None:
        """Add a single tool to the tool manager.

        Args:
            tool (Callable | ToolKit): The tool or tool kit to add.
            tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
                The type of tool to add, either "PERMANENT" or "TEMPORARY".
                Defaults to "TEMPORARY".

        Raises:
            ToolError: If the provided tool is not a defined tool.
        """
        self._validate_tool(tool)
        if isinstance(tool, ToolKit):
            tools = tool.tools
        else:
            tools = [tool]
        for one_tool in tools:
            if tool_type == "PERMANENT":
                self._permanent_tools[one_tool.__name__] = self.create_tool_metadata(one_tool)
            if tool_type == "TEMPORARY":
                self._temporary_tools[one_tool.__name__] = self.create_tool_metadata(one_tool)

    def add_tools(
        self,
        tools: list[Callable | ToolKit],
        tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY",
    ) -> None:
        """Add multiple tools to the tool manager.

        Args:
            tools (list[Callable | ToolKit]): A list of tools or tool kits to add.
            tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
                The type of tools to add, either "PERMANENT" or "TEMPORARY".
                Defaults to "TEMPORARY".

        Raises:
            ToolError: If any of the provided tools is not a defined tool.
        """
        for tool in tools:
            self.add_tool(tool, tool_type=tool_type)

    def remove_tool(
        self,
        tool: Callable | str | ToolKit,
        tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY",
    ) -> None:
        """Remove a single tool from the tool manager.

        If the tool to be removed cannot be found, do nothing and will not raise an error.

        Args:
            tool (Callable | str | ToolKit):
                The tool, tool kit, or name of the tool to remove.
            tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
                The type of tool to remove from, either "PERMANENT" or "TEMPORARY".
                Defaults to "TEMPORARY".
        """
        if isinstance(tool, str):
            if tool_type == "PERMANENT":
                self._permanent_tools.pop(tool, None)
            if tool_type == "TEMPORARY":
                self._temporary_tools.pop(tool, None)
            return
        self._validate_tool(tool)
        if isinstance(tool, ToolKit):
            tools = tool.tools
        else:
            tools = [tool]
        for one_tool in tools:
            if tool_type == "PERMANENT":
                self._permanent_tools.pop(one_tool.__name__, None)
            if tool_type == "TEMPORARY":
                self._temporary_tools.pop(one_tool.__name__, None)

    def remove_tools(
        self,
        tools: list[Callable | str | ToolKit],
        tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY",
    ) -> None:
        """Remove multiple tools from the tool manager.

        If the tool to be removed cannot be found, do nothing and will not raise an error.

        Args:
            tools (list[Callable | str | ToolKit]):
                A list of tools, tool kits, or names of tools to remove.
            tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
                The type of tools to remove from, either "PERMANENT" or "TEMPORARY".
                Defaults to "TEMPORARY".
        """
        for tool in tools:
            self.remove_tool(tool, tool_type=tool_type)

    def clear_tools(self, tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY") -> None:
        """Clear all tools of a specified type from the tool manager.

        Args:
            tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
                The type of tools to clear, either "PERMANENT" or "TEMPORARY".
                Defaults to "TEMPORARY".
        """
        if tool_type == "PERMANENT":
            self._permanent_tools = {}
        if tool_type == "TEMPORARY":
            self._temporary_tools = {}

    def register_tool(self, tool: Callable | ToolKit) -> None:
        """Register a single tool with registered tool list of the tool manager.

        Args:
            tool (Callable | ToolKit): The tool or tool kit to register.

        Raises:
            ToolError: If the provided tool is not a defined tool.
        """
        self._validate_tool(tool)
        if isinstance(tool, ToolKit):
            tools = tool.tools
        else:
            tools = [tool]
        for one_tool in tools:
            self._registered_tools[one_tool.__name__] = self.create_tool_metadata(one_tool)

    def register_tools(self, tools: list[Callable | ToolKit]) -> None:
        """Register multiple tools with the registered tool list of the tool manager.

        Args:
            tools (list[Callable | ToolKit]):
                A list of tools or tool kits to register.

        Raises:
            ToolError: If any of the provided tools is not a defined tool.
        """
        for tool in tools:
            self.register_tool(tool)

    def unregister_tools(self, tools: list[Callable | str | ToolKit]) -> None:
        """Unregister multiple tools from the registered tool list of the tool manager.

        Args:
            tools (list[Callable | str | ToolKit]):
                A list of tools, tool kits, or names of tools to unregister.
        """
        for tool in tools:
            if isinstance(tool, str):
                self._registered_tools.pop(tool, None)
                continue
            self._validate_tool(tool)
            if isinstance(tool, ToolKit):
                tools_list = tool.tools
            else:
                tools_list = [tool]
            for one_tool in tools_list:
                self._registered_tools.pop(one_tool.__name__, None)

    def clear_registered_tools(self) -> None:
        """Clear all registered tools from the registered tool list of the tool manager."""
        self._registered_tools = {}

    @property
    def registered_tool_names(self) -> list[str]:
        """Get the list of names of all registered tools.

        Returns:
            list[str]: The list of tool names in _registered_tools of the tool manager.
        """
        return list(self._registered_tools.keys())

    @property
    def all_tools_names(self) -> list[str]:
        """Geta the list of names of all tools, both permanent, temporary, and registered.

        Returns:
            list[str]: The list of all tool names.
        """
        result = set()
        result.update(self._permanent_tools.keys())
        result.update(self._temporary_tools.keys())
        result.update(self._registered_tools.keys())
        return list(result)

    def create_tool_metadata(self, tool: Callable) -> ToolMetadata:
        """Create metadata for a given tool.

        Args:
            tool (Callable): The tool for which to create metadata.

        Returns:
            ToolMetadata: The metadata for the tool.

        Raises:
            ToolError: If the provided tool is not a defined tool.
        """
        self._validate_tool(tool)
        tool_metadata = ToolMetadata(
            name = tool.__name__,
            description = tool.__tool_description__,
            parameters = tool.__tool_parameters__,
            linked_tool = tool,
            tool_kit = tool.__tool_kit__,
        )
        return tool_metadata

    def get_tools_metadata(
        self,
        by_types: list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"] = "ALL",
        by_names: list[str | ToolKit] | None = None,
    ) -> list[ToolMetadata]:
        """Get metadata for tools based on type and/or name.

        Args:
            by_types (list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"], optional):
                The types of tools to get metadata for. Defaults to "ALL".
            by_names (list[str | ToolKit] | None, optional):
                The specific names or tool kits to get metadata for. Defaults to None.

        Returns:
            list[ToolMetadata]: The list of tool metadata.
        """
        tool_names = []
        by_types = ["PERMANENT", "TEMPORARY"] if by_types == "ALL" else by_types
        if "PERMANENT" in by_types:
            tool_names.extend([name for name in self._permanent_tools.keys() if name not in tool_names])
        if "TEMPORARY" in by_types:
            tool_names.extend([name for name in self._temporary_tools.keys() if name not in tool_names])
        if by_names:
            tool_names.extend([name for name in by_names if name not in tool_names])
        return self.get_tools_metadata_by_names(tool_names)

    def get_tools_metadata_by_names(self, names: list[str | ToolKit]) -> list[ToolMetadata]:
        """Get metadata for tools based on a list of names or tool kits.

        Args:
            names (list[str | ToolKit]): The list of tool names or tool kits to get metadata for.

        Returns:
            list[ToolMetadata]: The list of tool metadata.
        """
        result = []
        for name in names:
            if isinstance(name, ToolKit):
                tools = name.tools
                names_list = [tool.__name__ for tool in tools]
            else:
                names_list = [name]
            for one_name in names_list:
                if one_name in self._registered_tools:
                    result.append(self._registered_tools[one_name])
                    continue
                if one_name in self._temporary_tools:
                    result.append(self._temporary_tools[one_name])
                    continue
                if one_name in self._permanent_tools:
                    result.append(self._permanent_tools[one_name])
                    continue
        return result

    def get_tools_manual(
        self,
        by_types: list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"] = "ALL",
        by_names: list[str | ToolKit] | None = None,
    ) -> Any:
        """Get the manual for tools based on type and/or name.

        Args:
            by_types (list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"], optional):
                The types of tools to get the manual for. Defaults to "ALL".
            by_names (list[str | ToolKit] | None, optional):
                The specific names or tool kits to get the manual for. Defaults to None.

        Returns:
            Any: The manual for the tools.
        """
        tools = self.get_tools_metadata(by_types=by_types, by_names=by_names)
        return self.tool_model.get_tools_manual(tools=tools)

    def get_linked_tool(self, tool_name: str) -> Callable:
        """Get the callable linked tool by its name.

        Args:
            tool_name (str): The name of the tool.

        Returns:
            Callable: The linked tool.

        Raises:
            ValueError: If there is no corresponding tool for the given name.
        """
        if tool_name in self._registered_tools:
            linked_tool = self._registered_tools[tool_name]
            return linked_tool.linked_tool
        if tool_name in self._temporary_tools:
            linked_tool = self._temporary_tools[tool_name]
            return linked_tool.linked_tool
        if tool_name in self._permanent_tools:
            linked_tool = self._permanent_tools[tool_name]
            return linked_tool.linked_tool
        raise ValueError(f"There is no corresponding tool for '{tool_name}'.")

    def wrap_tools_to_bead(self, tools: list[ToolMetadata | Callable | ToolKit]) -> list:
        """Wrap tools into a bead format as required by the tool model.

        Args:
            tools (list[ToolMetadata | Callable | ToolKit]):
                The list of tools (ToolMetadata or Callable tool) or tool kits.

        Returns:
            list: The wrapped tools infomation in bead format.
        """
        assert isinstance(self.tool_model, CustomToolModelInterface), (
            "The method 'wrap_tools_to_bead' should only be used when utilizing "
            "a CustomToolModelInterface type of tool model."
        )
        tools_metadata_list = []
        for tool in tools:
            if isinstance(tool, ToolMetadata):
                tools_metadata_list.append(tool)
            elif isinstance(tool, ToolKit):
                for one_tool in tool.tools:
                    tools_metadata_list.append(self.create_tool_metadata(one_tool))
            elif isinstance(tool, Callable):
                tools_metadata_list.append(self.create_tool_metadata(tool))
            else:
                assert False
        description = self.tool_model.get_tools_manual(tools_metadata_list)
        tool_bead = self.tool_model.tool_manual_bead_maker(description)
        return [tool_bead]

    def tools_manual_token_num(
        self,
        by_types: list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"] = "ALL",
        by_names: list[str | ToolKit] | None = None,
    ) -> int:
        """Get the number of tokens in the manual for tools based on type and/or name.

        Args:
            by_types (list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"], optional):
                The types of tools to count tokens for. Defaults to "ALL".
            by_names (list[str | ToolKit] | None, optional):
                The specific names or tool kits to count tokens for. Defaults to None.

        Returns:
            int: The number of tokens in the manual.
        """
        assert isinstance(self.tool_model, ProvidedToolModelInterface), (
            "The method 'tools_manual_token_num' should only be used when utilizing "
            "a ProvidedToolModelInterface type of tool model."
        )
        tools_metadata = self.get_tools_metadata(by_types=by_types, by_names=by_names)
        return self.tool_model.tool_manual_token_counter(tools_metadata)

    def parse_response(
        self,
        response: AsyncIterator
    ) -> Coroutine[Any, Any, Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]]:
        """Parse the response from the LLM.

        It will return a factory function, with which you can respectively obtain an
        asynchronous generator for the messages sent to the user and an asynchronous
        generator for function calls.

        Args:
            response (AsyncIterator): The response to parse.

        Returns:
            Callable: The factory function to make asynchronous generator.
        """
        return self.tool_model.parse_response(response)

    def _validate_tool(self, tool: Callable | ToolKit) -> bool:
        """Validate whether the provided object is a tool or tool kit."""
        if getattr(tool, "__tool__", None) is True:
            return True
        if isinstance(tool, ToolKit):
            return True
        raise ToolError(f"The '{tool}' is not a defined tool.")

all_tools_names: list[str] property

Geta the list of names of all tools, both permanent, temporary, and registered.

Returns:

Type	Description
list[str]	list[str]: The list of all tool names.

registered_tool_names: list[str] property

Get the list of names of all registered tools.

Returns:

Type	Description
list[str]	list[str]: The list of tool names in _registered_tools of the tool manager.

tool_model: ToolModelInterface property writable

Get the tool model instance.

tool_model_name: str property

Get the name of the tool model.

tool_model_type: Literal['CUSTOM', 'PROVIDED'] property

Get the type of the tool model.

Returns:

Type	Description
Literal['CUSTOM', 'PROVIDED']	Either 'CUSTOM' or 'PROVIDED'.
Literal['CUSTOM', 'PROVIDED']	'CUSTOM' means that LLM does not provide an excuse for tool invocation,
Literal['CUSTOM', 'PROVIDED']	and a custom protocol is used to implement tool invocation.
Literal['CUSTOM', 'PROVIDED']	'PROVIDED' means using the tool invocation interface provided by LLM itself.

add_tool(tool, tool_type='TEMPORARY')

Add a single tool to the tool manager.

Parameters:

Name	Type	Description	Default
tool	Callable | ToolKit	The tool or tool kit to add.	required
tool_type	Literal['PERMANENT', 'TEMPORARY']	The type of tool to add, either "PERMANENT" or "TEMPORARY". Defaults to "TEMPORARY".	'TEMPORARY'

Raises:

Type	Description
ToolError	If the provided tool is not a defined tool.

Source code in src/agere/utils/tool.py

def add_tool(
    self,
    tool: Callable | ToolKit,
    tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY",
) -> None:
    """Add a single tool to the tool manager.

    Args:
        tool (Callable | ToolKit): The tool or tool kit to add.
        tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
            The type of tool to add, either "PERMANENT" or "TEMPORARY".
            Defaults to "TEMPORARY".

    Raises:
        ToolError: If the provided tool is not a defined tool.
    """
    self._validate_tool(tool)
    if isinstance(tool, ToolKit):
        tools = tool.tools
    else:
        tools = [tool]
    for one_tool in tools:
        if tool_type == "PERMANENT":
            self._permanent_tools[one_tool.__name__] = self.create_tool_metadata(one_tool)
        if tool_type == "TEMPORARY":
            self._temporary_tools[one_tool.__name__] = self.create_tool_metadata(one_tool)

add_tools(tools, tool_type='TEMPORARY')

Add multiple tools to the tool manager.

Parameters:

Name	Type	Description	Default
tools	list[Callable | ToolKit]	A list of tools or tool kits to add.	required
tool_type	Literal['PERMANENT', 'TEMPORARY']	The type of tools to add, either "PERMANENT" or "TEMPORARY". Defaults to "TEMPORARY".	'TEMPORARY'

Raises:

Type	Description
ToolError	If any of the provided tools is not a defined tool.

Source code in src/agere/utils/tool.py

def add_tools(
    self,
    tools: list[Callable | ToolKit],
    tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY",
) -> None:
    """Add multiple tools to the tool manager.

    Args:
        tools (list[Callable | ToolKit]): A list of tools or tool kits to add.
        tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
            The type of tools to add, either "PERMANENT" or "TEMPORARY".
            Defaults to "TEMPORARY".

    Raises:
        ToolError: If any of the provided tools is not a defined tool.
    """
    for tool in tools:
        self.add_tool(tool, tool_type=tool_type)

clear_registered_tools()

Clear all registered tools from the registered tool list of the tool manager.

Source code in src/agere/utils/tool.py

def clear_registered_tools(self) -> None:
    """Clear all registered tools from the registered tool list of the tool manager."""
    self._registered_tools = {}

clear_tools(tool_type='TEMPORARY')

Clear all tools of a specified type from the tool manager.

Parameters:

Name	Type	Description	Default
tool_type	Literal['PERMANENT', 'TEMPORARY']	The type of tools to clear, either "PERMANENT" or "TEMPORARY". Defaults to "TEMPORARY".	'TEMPORARY'

Source code in src/agere/utils/tool.py

def clear_tools(self, tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY") -> None:
    """Clear all tools of a specified type from the tool manager.

    Args:
        tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
            The type of tools to clear, either "PERMANENT" or "TEMPORARY".
            Defaults to "TEMPORARY".
    """
    if tool_type == "PERMANENT":
        self._permanent_tools = {}
    if tool_type == "TEMPORARY":
        self._temporary_tools = {}

create_tool_metadata(tool)

Create metadata for a given tool.

Parameters:

Name	Type	Description	Default
tool	Callable	The tool for which to create metadata.	required

Returns:

Name	Type	Description
ToolMetadata	ToolMetadata	The metadata for the tool.

Raises:

Type	Description
ToolError	If the provided tool is not a defined tool.

Source code in src/agere/utils/tool.py

def create_tool_metadata(self, tool: Callable) -> ToolMetadata:
    """Create metadata for a given tool.

    Args:
        tool (Callable): The tool for which to create metadata.

    Returns:
        ToolMetadata: The metadata for the tool.

    Raises:
        ToolError: If the provided tool is not a defined tool.
    """
    self._validate_tool(tool)
    tool_metadata = ToolMetadata(
        name = tool.__name__,
        description = tool.__tool_description__,
        parameters = tool.__tool_parameters__,
        linked_tool = tool,
        tool_kit = tool.__tool_kit__,
    )
    return tool_metadata

get_linked_tool(tool_name)

Get the callable linked tool by its name.

Parameters:

Name	Type	Description	Default
tool_name	str	The name of the tool.	required

Returns:

Name	Type	Description
Callable	Callable	The linked tool.

Raises:

Type	Description
ValueError	If there is no corresponding tool for the given name.

Source code in src/agere/utils/tool.py

def get_linked_tool(self, tool_name: str) -> Callable:
    """Get the callable linked tool by its name.

    Args:
        tool_name (str): The name of the tool.

    Returns:
        Callable: The linked tool.

    Raises:
        ValueError: If there is no corresponding tool for the given name.
    """
    if tool_name in self._registered_tools:
        linked_tool = self._registered_tools[tool_name]
        return linked_tool.linked_tool
    if tool_name in self._temporary_tools:
        linked_tool = self._temporary_tools[tool_name]
        return linked_tool.linked_tool
    if tool_name in self._permanent_tools:
        linked_tool = self._permanent_tools[tool_name]
        return linked_tool.linked_tool
    raise ValueError(f"There is no corresponding tool for '{tool_name}'.")

get_tools_manual(by_types='ALL', by_names=None)

Get the manual for tools based on type and/or name.

Parameters:

Name	Type	Description	Default
by_types	list[Literal['PERMANENT', 'TEMPORARY']] | Literal['ALL']	The types of tools to get the manual for. Defaults to "ALL".	'ALL'
by_names	list[str | ToolKit] | None	The specific names or tool kits to get the manual for. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any	The manual for the tools.

Source code in src/agere/utils/tool.py

def get_tools_manual(
    self,
    by_types: list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"] = "ALL",
    by_names: list[str | ToolKit] | None = None,
) -> Any:
    """Get the manual for tools based on type and/or name.

    Args:
        by_types (list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"], optional):
            The types of tools to get the manual for. Defaults to "ALL".
        by_names (list[str | ToolKit] | None, optional):
            The specific names or tool kits to get the manual for. Defaults to None.

    Returns:
        Any: The manual for the tools.
    """
    tools = self.get_tools_metadata(by_types=by_types, by_names=by_names)
    return self.tool_model.get_tools_manual(tools=tools)

get_tools_metadata(by_types='ALL', by_names=None)

Get metadata for tools based on type and/or name.

Parameters:

Name	Type	Description	Default
by_types	list[Literal['PERMANENT', 'TEMPORARY']] | Literal['ALL']	The types of tools to get metadata for. Defaults to "ALL".	'ALL'
by_names	list[str | ToolKit] | None	The specific names or tool kits to get metadata for. Defaults to None.	None

Returns:

Type	Description
list[ToolMetadata]	list[ToolMetadata]: The list of tool metadata.

Source code in src/agere/utils/tool.py

def get_tools_metadata(
    self,
    by_types: list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"] = "ALL",
    by_names: list[str | ToolKit] | None = None,
) -> list[ToolMetadata]:
    """Get metadata for tools based on type and/or name.

    Args:
        by_types (list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"], optional):
            The types of tools to get metadata for. Defaults to "ALL".
        by_names (list[str | ToolKit] | None, optional):
            The specific names or tool kits to get metadata for. Defaults to None.

    Returns:
        list[ToolMetadata]: The list of tool metadata.
    """
    tool_names = []
    by_types = ["PERMANENT", "TEMPORARY"] if by_types == "ALL" else by_types
    if "PERMANENT" in by_types:
        tool_names.extend([name for name in self._permanent_tools.keys() if name not in tool_names])
    if "TEMPORARY" in by_types:
        tool_names.extend([name for name in self._temporary_tools.keys() if name not in tool_names])
    if by_names:
        tool_names.extend([name for name in by_names if name not in tool_names])
    return self.get_tools_metadata_by_names(tool_names)

get_tools_metadata_by_names(names)

Get metadata for tools based on a list of names or tool kits.

Parameters:

Name	Type	Description	Default
names	list[str | ToolKit]	The list of tool names or tool kits to get metadata for.	required

Returns:

Type	Description
list[ToolMetadata]	list[ToolMetadata]: The list of tool metadata.

Source code in src/agere/utils/tool.py

def get_tools_metadata_by_names(self, names: list[str | ToolKit]) -> list[ToolMetadata]:
    """Get metadata for tools based on a list of names or tool kits.

    Args:
        names (list[str | ToolKit]): The list of tool names or tool kits to get metadata for.

    Returns:
        list[ToolMetadata]: The list of tool metadata.
    """
    result = []
    for name in names:
        if isinstance(name, ToolKit):
            tools = name.tools
            names_list = [tool.__name__ for tool in tools]
        else:
            names_list = [name]
        for one_name in names_list:
            if one_name in self._registered_tools:
                result.append(self._registered_tools[one_name])
                continue
            if one_name in self._temporary_tools:
                result.append(self._temporary_tools[one_name])
                continue
            if one_name in self._permanent_tools:
                result.append(self._permanent_tools[one_name])
                continue
    return result

parse_response(response)

Parse the response from the LLM.

It will return a factory function, with which you can respectively obtain an asynchronous generator for the messages sent to the user and an asynchronous generator for function calls.

Parameters:

Name	Type	Description	Default
response	AsyncIterator	The response to parse.	required

Returns:

Name	Type	Description
Callable	Coroutine[Any, Any, Callable[[Literal['to_user', 'tool_call']], AsyncGenerator]]	The factory function to make asynchronous generator.

Source code in src/agere/utils/tool.py

def parse_response(
    self,
    response: AsyncIterator
) -> Coroutine[Any, Any, Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]]:
    """Parse the response from the LLM.

    It will return a factory function, with which you can respectively obtain an
    asynchronous generator for the messages sent to the user and an asynchronous
    generator for function calls.

    Args:
        response (AsyncIterator): The response to parse.

    Returns:
        Callable: The factory function to make asynchronous generator.
    """
    return self.tool_model.parse_response(response)

register_tool(tool)

Register a single tool with registered tool list of the tool manager.

Parameters:

Name	Type	Description	Default
tool	Callable | ToolKit	The tool or tool kit to register.	required

Raises:

Type	Description
ToolError	If the provided tool is not a defined tool.

Source code in src/agere/utils/tool.py

def register_tool(self, tool: Callable | ToolKit) -> None:
    """Register a single tool with registered tool list of the tool manager.

    Args:
        tool (Callable | ToolKit): The tool or tool kit to register.

    Raises:
        ToolError: If the provided tool is not a defined tool.
    """
    self._validate_tool(tool)
    if isinstance(tool, ToolKit):
        tools = tool.tools
    else:
        tools = [tool]
    for one_tool in tools:
        self._registered_tools[one_tool.__name__] = self.create_tool_metadata(one_tool)

register_tools(tools)

Register multiple tools with the registered tool list of the tool manager.

Parameters:

Name	Type	Description	Default
tools	list[Callable | ToolKit]	A list of tools or tool kits to register.	required

Raises:

Type	Description
ToolError	If any of the provided tools is not a defined tool.

Source code in src/agere/utils/tool.py

def register_tools(self, tools: list[Callable | ToolKit]) -> None:
    """Register multiple tools with the registered tool list of the tool manager.

    Args:
        tools (list[Callable | ToolKit]):
            A list of tools or tool kits to register.

    Raises:
        ToolError: If any of the provided tools is not a defined tool.
    """
    for tool in tools:
        self.register_tool(tool)

remove_tool(tool, tool_type='TEMPORARY')

Remove a single tool from the tool manager.

If the tool to be removed cannot be found, do nothing and will not raise an error.

Parameters:

Name	Type	Description	Default
tool	Callable | str | ToolKit	The tool, tool kit, or name of the tool to remove.	required
tool_type	Literal['PERMANENT', 'TEMPORARY']	The type of tool to remove from, either "PERMANENT" or "TEMPORARY". Defaults to "TEMPORARY".	'TEMPORARY'

Source code in src/agere/utils/tool.py

def remove_tool(
    self,
    tool: Callable | str | ToolKit,
    tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY",
) -> None:
    """Remove a single tool from the tool manager.

    If the tool to be removed cannot be found, do nothing and will not raise an error.

    Args:
        tool (Callable | str | ToolKit):
            The tool, tool kit, or name of the tool to remove.
        tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
            The type of tool to remove from, either "PERMANENT" or "TEMPORARY".
            Defaults to "TEMPORARY".
    """
    if isinstance(tool, str):
        if tool_type == "PERMANENT":
            self._permanent_tools.pop(tool, None)
        if tool_type == "TEMPORARY":
            self._temporary_tools.pop(tool, None)
        return
    self._validate_tool(tool)
    if isinstance(tool, ToolKit):
        tools = tool.tools
    else:
        tools = [tool]
    for one_tool in tools:
        if tool_type == "PERMANENT":
            self._permanent_tools.pop(one_tool.__name__, None)
        if tool_type == "TEMPORARY":
            self._temporary_tools.pop(one_tool.__name__, None)

remove_tools(tools, tool_type='TEMPORARY')

Remove multiple tools from the tool manager.

If the tool to be removed cannot be found, do nothing and will not raise an error.

Parameters:

Name	Type	Description	Default
tools	list[Callable | str | ToolKit]	A list of tools, tool kits, or names of tools to remove.	required
tool_type	Literal['PERMANENT', 'TEMPORARY']	The type of tools to remove from, either "PERMANENT" or "TEMPORARY". Defaults to "TEMPORARY".	'TEMPORARY'

Source code in src/agere/utils/tool.py

def remove_tools(
    self,
    tools: list[Callable | str | ToolKit],
    tool_type: Literal["PERMANENT", "TEMPORARY"] = "TEMPORARY",
) -> None:
    """Remove multiple tools from the tool manager.

    If the tool to be removed cannot be found, do nothing and will not raise an error.

    Args:
        tools (list[Callable | str | ToolKit]):
            A list of tools, tool kits, or names of tools to remove.
        tool_type (Literal["PERMANENT", "TEMPORARY"], optional):
            The type of tools to remove from, either "PERMANENT" or "TEMPORARY".
            Defaults to "TEMPORARY".
    """
    for tool in tools:
        self.remove_tool(tool, tool_type=tool_type)

tools_manual_token_num(by_types='ALL', by_names=None)

Get the number of tokens in the manual for tools based on type and/or name.

Parameters:

Name	Type	Description	Default
by_types	list[Literal['PERMANENT', 'TEMPORARY']] | Literal['ALL']	The types of tools to count tokens for. Defaults to "ALL".	'ALL'
by_names	list[str | ToolKit] | None	The specific names or tool kits to count tokens for. Defaults to None.	None

Returns:

Name	Type	Description
int	int	The number of tokens in the manual.

Source code in src/agere/utils/tool.py

def tools_manual_token_num(
    self,
    by_types: list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"] = "ALL",
    by_names: list[str | ToolKit] | None = None,
) -> int:
    """Get the number of tokens in the manual for tools based on type and/or name.

    Args:
        by_types (list[Literal["PERMANENT", "TEMPORARY"]] | Literal["ALL"], optional):
            The types of tools to count tokens for. Defaults to "ALL".
        by_names (list[str | ToolKit] | None, optional):
            The specific names or tool kits to count tokens for. Defaults to None.

    Returns:
        int: The number of tokens in the manual.
    """
    assert isinstance(self.tool_model, ProvidedToolModelInterface), (
        "The method 'tools_manual_token_num' should only be used when utilizing "
        "a ProvidedToolModelInterface type of tool model."
    )
    tools_metadata = self.get_tools_metadata(by_types=by_types, by_names=by_names)
    return self.tool_model.tool_manual_token_counter(tools_metadata)

unregister_tools(tools)

Unregister multiple tools from the registered tool list of the tool manager.

Parameters:

Name	Type	Description	Default
tools	list[Callable | str | ToolKit]	A list of tools, tool kits, or names of tools to unregister.	required

Source code in src/agere/utils/tool.py

def unregister_tools(self, tools: list[Callable | str | ToolKit]) -> None:
    """Unregister multiple tools from the registered tool list of the tool manager.

    Args:
        tools (list[Callable | str | ToolKit]):
            A list of tools, tool kits, or names of tools to unregister.
    """
    for tool in tools:
        if isinstance(tool, str):
            self._registered_tools.pop(tool, None)
            continue
        self._validate_tool(tool)
        if isinstance(tool, ToolKit):
            tools_list = tool.tools
        else:
            tools_list = [tool]
        for one_tool in tools_list:
            self._registered_tools.pop(one_tool.__name__, None)

wrap_tools_to_bead(tools)

Wrap tools into a bead format as required by the tool model.

Parameters:

Name	Type	Description	Default
tools	list[ToolMetadata | Callable | ToolKit]	The list of tools (ToolMetadata or Callable tool) or tool kits.	required

Returns:

Name	Type	Description
list	list	The wrapped tools infomation in bead format.

Source code in src/agere/utils/tool.py

def wrap_tools_to_bead(self, tools: list[ToolMetadata | Callable | ToolKit]) -> list:
    """Wrap tools into a bead format as required by the tool model.

    Args:
        tools (list[ToolMetadata | Callable | ToolKit]):
            The list of tools (ToolMetadata or Callable tool) or tool kits.

    Returns:
        list: The wrapped tools infomation in bead format.
    """
    assert isinstance(self.tool_model, CustomToolModelInterface), (
        "The method 'wrap_tools_to_bead' should only be used when utilizing "
        "a CustomToolModelInterface type of tool model."
    )
    tools_metadata_list = []
    for tool in tools:
        if isinstance(tool, ToolMetadata):
            tools_metadata_list.append(tool)
        elif isinstance(tool, ToolKit):
            for one_tool in tool.tools:
                tools_metadata_list.append(self.create_tool_metadata(one_tool))
        elif isinstance(tool, Callable):
            tools_metadata_list.append(self.create_tool_metadata(tool))
        else:
            assert False
    description = self.tool_model.get_tools_manual(tools_metadata_list)
    tool_bead = self.tool_model.tool_manual_bead_maker(description)
    return [tool_bead]

tool(__func=None, *, description='')

Decorator for a tool.

It can be used with or without arguments. When used with arguments, the description of the tool can be manually specified. When used without arguments, the tool's metadata will automatically be extracted from the docstring of the decorated function, including the tool's description, and the name, type, and corresponding description of each parameter of the tool. Additionally, the tool_parameter decorator can be used to set more or more specific metadata for a particular parameter, which will override all metadata for that parameter.

Source code in src/agere/utils/tool.py

def tool(__func: Callable[..., Any] | None = None, *, description: str = ''):
    """Decorator for a tool.

    It can be used with or without arguments. When used with arguments,
    the description of the tool can be manually specified.
    When used without arguments, the tool's metadata will automatically
    be extracted from the docstring of the decorated function,
    including the tool's description, and the name, type, and corresponding
    description of each parameter of the tool.
    Additionally, the tool_parameter decorator can be used to set more or
    more specific metadata for a particular parameter,
    which will override all metadata for that parameter.
    """

    from_docstring = True if __func is not None else False

    def decorator(func):
        func.__tool__ = True
        func.__tool_description__ = description
        if not hasattr(func, "__tool_parameters__"):
            func.__tool_parameters__ = []
        if not hasattr(func, "__tool_kit__"):
            func.__tool_kit__ = None

        if from_docstring:
            metadata = _parse_docstring(func)

            description_from_docstring = metadata["description"]
            if not func.__tool_description__:
                func.__tool_description__ = description_from_docstring

            param_from_docstring = metadata["parameters"]
            param_name_list = [param_info["name"] for param_info in func.__tool_parameters__]
            for param_docs in param_from_docstring:
                if param_docs["name"] not in param_name_list:
                    func.__tool_parameters__.append(param_docs)

        return func

    if __func is not None:
        return decorator(__func)
    else:
        return decorator

tool_kit(kit, description=None)

Decorator for a tool.

This decorator is used to add a tool into a tool kit. It also allows setting a description for the tool kit.

Parameters:

Name	Type	Description	Default
kit	str | ToolKit	The name of the tool kit or the ToolKit object to add the tool into.	required
description	str | None	A description for the tool kit. When it is not None, it will overwrite the tool kit description. When it is None, it will retain the original tool kit description, if it exists.	None

Source code in src/agere/utils/tool.py

def tool_kit(kit: str | ToolKit, description: str | None = None):
    """Decorator for a tool.

    This decorator is used to add a tool into a tool kit.
    It also allows setting a description for the tool kit.

    Args:
        kit (str | ToolKit): The name of the tool kit or the ToolKit object to add the tool into.
        description (str | None, optional):
            A description for the tool kit.
            When it is not None, it will overwrite the tool kit description.
            When it is None, it will retain the original tool kit description,
            if it exists.
    """

    def decorator(func):
        if isinstance(kit, ToolKit):
            the_kit = kit
        elif isinstance(kit, str):
            global_name_space = globals()
            if kit not in global_name_space:
                global_name_space[kit] = ToolKit(kit)
            the_kit = global_name_space[kit]
        else:
            assert False

        func.__tool_kit__ = the_kit

        if description is not None:
            the_kit.description = description
        the_kit.tools.append(func)

        return func

    return decorator

tool_parameter(*, name, description, default_value='', param_type='string', choices=None, required=True)

Decorator for tool parameters.

It will set the full infomation of the specified parameter.

Parameters:

Name	Type	Description	Default
name	str	The name of the tool parameter.	required
description	str	The description of the tool parameter.	required
default_value	Any	The default value of the tool parameter.	''
type		The type of the tool parameter.	required
required	bool	Whether the tool parameter is required.	True

Source code in src/agere/utils/tool.py

def tool_parameter(
    *,
    name: str,
    description: str,
    default_value: Any = "",
    param_type: str = "string",
    choices: list | None = None,
    required: bool = True,
):
    """Decorator for tool parameters.

    It will set the full infomation of the specified parameter.

    Args:
        name: The name of the tool parameter.
        description: The description of the tool parameter.
        default_value: The default value of the tool parameter.
        type: The type of the tool parameter.
        required: Whether the tool parameter is required.
    """
    def decorator(func):
        if not hasattr(func, "__tool_parameters__"):
            func.__tool_parameters__ = []

        matching_param = next(
            (
                param_info
                for param_info in func.__tool_parameters__
                if param_info.get("name") == name
            ),
            None,
        )

        if not matching_param:
            func.__tool_parameters__.append(
                {
                    "name": name,
                    "description": description,
                    "default_value": default_value,
                    "param_type": param_type,
                    "choices": choices,
                    "required": required,
                }
            )
        else:
            matching_param = {
                "name": name,
                "description": description,
                "default_value": default_value,
                "param_type": param_type,
                "choices": choices,
                "required": required,
            }

        return func

    return decorator

Custom Tool Model

CustomToolModel

Bases: CustomToolModelInterface

CustomToolModel is a class that implements the CustomToolModelInterface.

Attributes:

Name	Type	Description
tool_bead_maker	Callable[[str], Any]	A callable that takes a string and returns any type, used to create tool beads from a manual.
custom_tool_manual_template	str | None	Optional custom template for the tool manual. If None, a default template is used.

Methods:

Name	Description
tool_model_name	Property to get the name of the tool model.
custom_tool_manual_template	Property to get or set the custom tool manual template.
tool_manual_bead_maker	Method to create a tool manual bead from a manual string.
get_tools_manual	Method to generate a tools manual from a list of ToolMetadata.
parse_response	Asynchronous method to parse responses from LLM.

Source code in src/agere/utils/tool_models/custom_tool_model.py

class CustomToolModel(CustomToolModelInterface):
    """CustomToolModel is a class that implements the CustomToolModelInterface.

    Attributes:
        tool_bead_maker (Callable[[str], Any]): A callable that takes a string and returns any type,
            used to create tool beads from a manual.
        custom_tool_manual_template (str | None): Optional custom template for the tool manual.
            If None, a default template is used.

    Methods:
        tool_model_name: Property to get the name of the tool model.
        custom_tool_manual_template: Property to get or set the custom tool manual template.
        tool_manual_bead_maker: Method to create a tool manual bead from a manual string.
        get_tools_manual: Method to generate a tools manual from a list of ToolMetadata.
        parse_response: Asynchronous method to parse responses from LLM.
    """

    def __init__(
        self,
        tool_bead_maker: Callable[[str], Any],
        custom_tool_manual_template: str | None = None,
    ):
        """
        Initializes a new instance of the CustomToolModel class with a tool bead maker callable
        and an optional custom tool manual template.

        Args:
            tool_bead_maker (Callable[[str], Any]):
                The callable used to create tool beads. It takes a string as input and
                return the bead created from it.
            custom_tool_manual_template (str | None, optional): The custom template for the tool manual.
        """
        self.tool_bead_maker: Callable[[str], Any] = tool_bead_maker
        self._custom_tool_manual_template = custom_tool_manual_template

    @property
    def tool_model_name(self):
        """Gets the name of the tool model."""
        return "CUSTOM"

    @property
    def custom_tool_manual_template(self) -> str:
        """
        Gets the custom tool manual template.

        Returns:
            str: The custom tool manual template or a default template if none is set.
        """
        return self._custom_tool_manual_template or CUSTOM_TOOL_MANUAL_TEMPLATE

    @custom_tool_manual_template.setter
    def custom_tool_manual_template(self, value: str) -> None:
        """
        Sets the custom tool manual template.

        Args:
            value (str): The new custom tool manual template which has a 'tools' variable.
        """
        self._custom_tool_manual_template = value

    def tool_manual_bead_maker(self, manual: str) -> Any:
        """
        Creates a tool manual bead using the tool bead maker callable.

        Args:
            manual (str): The manual string used to create the tool manual bead.

        Returns:
            Any: The result of the tool bead maker callable.
        """
        return self.tool_bead_maker(manual)

    def get_tools_manual(self, tools: list[ToolMetadata]) -> Any:
        """
        Generates a tools manual from a list of ToolMetadata.

        Args:
            tools (list[ToolMetadata]): The list of tool metadata objects.

        Returns:
            Any: The generated tools manual.
        """
        tools_instruction = []
        for tool in tools:
            tool_dict = {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.parameters,
            }
            tools_instruction.append(tool_dict)
        tools_instruction_str = json.dumps(tools_instruction, indent=4)
        return render_prompt(
            self.custom_tool_manual_template,
            tools=tools_instruction_str,
        )

    async def parse_response(
        self,
        source: AsyncIterator,
    ) -> Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
        """
        Parses asynchronous responses from LLM.

        Args:
            source (AsyncIterator): The asynchronous iterator source for the responses.

        Returns:
            Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
            A callable that generates an asynchronous generator for either "to_user"
            or "tool_call" content.
        """
        parser = self.ParseResponse(source)
        return await parser.parse()


    class ParseResponse:

        State = Literal["closed", "in_to_user", "check_tool_start", "in_tool"]

        def __init__(self, source: AsyncIterator):
            self.source = source
            self.state: CustomToolModel.ParseResponse.State = "closed"
            self.to_user_queue = asyncio.Queue()
            self.tool_queue = asyncio.Queue()
            self._check_tag_buffer = ""
            self._tool_call_buffer = ""

        async def run(self) -> None:
            async for chunk in self.source:
                await self.state_method(chunk)
            if self.state == "check_tool_start":
                await self.to_user_queue.put(self._check_tag_buffer)
                self._check_tag_buffer = ""
            await self.to_user_queue.put(None)
            await self.tool_queue.put(None)
            if self.state == "in_tool":
                raise ParseResponseError(f"The parser exits in state {self.state}.")
            self.state = "closed"

        async def parse(self):
            def make_generator(which: Literal["to_user", "tool_call"]):
                if which == "to_user":
                    queue = self.to_user_queue
                elif which == "tool_call":
                    queue = self.tool_queue
                else:
                    assert False
                async def generator():
                    while True:
                        value = await queue.get()
                        if value is None:
                            break
                        yield value
                return generator()
            asyncio.create_task(self.run())
            return make_generator

        @property
        def state_method(self) -> Callable[[str], Coroutine]:
            if self.state == "closed":
                return self._in_closed
            if self.state == "in_to_user":
                return self._in_to_user
            if self.state == "check_tool_start":
                return self._in_check_tool_start
            if self.state == "in_tool":
                return self._in_tool
            else:
                assert False

        async def _in_closed(self, chunk: str) -> None:
            assert self.state == "closed"
            self._check_tag_buffer = ""
            self._tool_call_buffer = ""
            self.state = "in_to_user"
            await self._in_to_user(chunk)

        async def _in_to_user(self, chunk: str) -> None:
            assert self.state == "in_to_user"
            tag_start = chunk.find("<")
            if tag_start == -1:
                await self.to_user_queue.put(chunk)
                self.state = "in_to_user"
                return

            await self.to_user_queue.put(chunk[:tag_start])
            self._check_tag_buffer = chunk[tag_start:]

            if len(self._check_tag_buffer) < 6:
                self.state = "check_tool_start"
                return

            if self._check_tag_buffer.startswith("<tool>"):
                self.state = "in_tool"
                self._tool_call_buffer = self._check_tag_buffer[6:]
                self._check_tag_buffer = ""
                await self._in_tool("")
                return

            await self.to_user_queue.put(self._check_tag_buffer)
            self._check_tag_buffer = ""
            self.state = "in_to_user"
            return

        async def _in_check_tool_start(self, chunk: str) -> None:
            assert self.state == "check_tool_start"
            self._check_tag_buffer += chunk

            if len(self._check_tag_buffer) < 6:
                self.state = "check_tool_start"
                return

            if self._check_tag_buffer.startswith("<tool>"):
                self.state = "in_tool"
                self._tool_call_buffer = self._check_tag_buffer[6:]
                self._check_tag_buffer = ""
                return

            await self.to_user_queue.put(self._check_tag_buffer)
            self._check_tag_buffer = ""
            self.state = "in_to_user"
            return

        async def _in_tool(self, chunk: str) -> None:
            assert self.state == "in_tool"
            self._tool_call_buffer += chunk

            tool_tag_end = self._tool_call_buffer.find("</tool>")
            if tool_tag_end == -1:
                self.state = "in_tool"
                return

            tool_call = self._tool_call_buffer[:tool_tag_end]
            await self.tool_queue.put(tool_call)
            self._tool_call_buffer = ""
            self.state = "in_to_user"
            await self._in_to_user(self._tool_call_buffer[tool_tag_end+7:])
            return

custom_tool_manual_template: str property writable

Gets the custom tool manual template.

Returns:

Name	Type	Description
str	str	The custom tool manual template or a default template if none is set.

tool_model_name property

Gets the name of the tool model.

__init__(tool_bead_maker, custom_tool_manual_template=None)

Initializes a new instance of the CustomToolModel class with a tool bead maker callable and an optional custom tool manual template.

Parameters:

Name	Type	Description	Default
tool_bead_maker	Callable[[str], Any]	The callable used to create tool beads. It takes a string as input and return the bead created from it.	required
custom_tool_manual_template	str | None	The custom template for the tool manual.	None

Source code in src/agere/utils/tool_models/custom_tool_model.py

def __init__(
    self,
    tool_bead_maker: Callable[[str], Any],
    custom_tool_manual_template: str | None = None,
):
    """
    Initializes a new instance of the CustomToolModel class with a tool bead maker callable
    and an optional custom tool manual template.

    Args:
        tool_bead_maker (Callable[[str], Any]):
            The callable used to create tool beads. It takes a string as input and
            return the bead created from it.
        custom_tool_manual_template (str | None, optional): The custom template for the tool manual.
    """
    self.tool_bead_maker: Callable[[str], Any] = tool_bead_maker
    self._custom_tool_manual_template = custom_tool_manual_template

get_tools_manual(tools)

Generates a tools manual from a list of ToolMetadata.

Parameters:

Name	Type	Description	Default
tools	list[ToolMetadata]	The list of tool metadata objects.	required

Returns:

Name	Type	Description
Any	Any	The generated tools manual.

Source code in src/agere/utils/tool_models/custom_tool_model.py

def get_tools_manual(self, tools: list[ToolMetadata]) -> Any:
    """
    Generates a tools manual from a list of ToolMetadata.

    Args:
        tools (list[ToolMetadata]): The list of tool metadata objects.

    Returns:
        Any: The generated tools manual.
    """
    tools_instruction = []
    for tool in tools:
        tool_dict = {
            "name": tool.name,
            "description": tool.description,
            "parameters": tool.parameters,
        }
        tools_instruction.append(tool_dict)
    tools_instruction_str = json.dumps(tools_instruction, indent=4)
    return render_prompt(
        self.custom_tool_manual_template,
        tools=tools_instruction_str,
    )

parse_response(source) async

Parses asynchronous responses from LLM.

Parameters:

Name	Type	Description	Default
source	AsyncIterator	The asynchronous iterator source for the responses.	required

Returns:

Type	Description
Callable[[Literal['to_user', 'tool_call']], AsyncGenerator]	Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
Callable[[Literal['to_user', 'tool_call']], AsyncGenerator]	A callable that generates an asynchronous generator for either "to_user"
Callable[[Literal['to_user', 'tool_call']], AsyncGenerator]	or "tool_call" content.

Source code in src/agere/utils/tool_models/custom_tool_model.py

async def parse_response(
    self,
    source: AsyncIterator,
) -> Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
    """
    Parses asynchronous responses from LLM.

    Args:
        source (AsyncIterator): The asynchronous iterator source for the responses.

    Returns:
        Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
        A callable that generates an asynchronous generator for either "to_user"
        or "tool_call" content.
    """
    parser = self.ParseResponse(source)
    return await parser.parse()

tool_manual_bead_maker(manual)

Creates a tool manual bead using the tool bead maker callable.

Parameters:

Name	Type	Description	Default
manual	str	The manual string used to create the tool manual bead.	required

Returns:

Name	Type	Description
Any	Any	The result of the tool bead maker callable.

Source code in src/agere/utils/tool_models/custom_tool_model.py

def tool_manual_bead_maker(self, manual: str) -> Any:
    """
    Creates a tool manual bead using the tool bead maker callable.

    Args:
        manual (str): The manual string used to create the tool manual bead.

    Returns:
        Any: The result of the tool bead maker callable.
    """
    return self.tool_bead_maker(manual)

Openai Tool Model

OpenaiToolModel

Bases: ProvidedToolModelInterface

OpenaiToolModel is a class that implements the ProvidedToolModelInterface for interfacing with OpenAI's tools.

Source code in src/agere/utils/tool_models/openai_tool_model.py

class OpenaiToolModel(ProvidedToolModelInterface):
    """OpenaiToolModel is a class that implements the ProvidedToolModelInterface for interfacing
    with OpenAI's tools.
    """

    def __init__(
        self,
        tool_token_counter: Callable[[str], int],
        to_user_flag: str | None = "to_user",
        logger: logging.Logger | None = None,
    ):
        """
        Initializes a new instance of the OpenaiToolModel class with a token counter
        callable, an optional to_user_flag, and an optional logger.

        Args:
            tool_token_counter (Callable[[str], int]):
                A callable that takes a string and returns an integer, used for counting
                tokens in the tool manual.
            to_user_flag (str | None, optional): The flag for user-facing content in response.
            logger (logging.Logger | None, optional): The logger for logging.
        """
        self.tool_token_counter = tool_token_counter
        if to_user_flag is None:
            to_user_flag = "__THIS_IS_A_NEVER_USED_NAME__"
        self.to_user_flag = to_user_flag
        self.logger = logger

    @property
    def tool_model_name(self) -> str:
        """Gets the name of the tool model."""
        return "OPENAI"

    def tool_manual_token_counter(self, tools: list[ToolMetadata]) -> int:
        """
        Counts the number of tokens for the manual of specified tools.

        Args:
            tools (list[ToolMetadata]): A list of ToolMetadata objects to generate the
            manual from.

        Returns:
            int: The number of tokens in the tool manual.
        """
        manual = str(self.get_tools_manual(tools=tools))
        return self.tool_token_counter(manual)

    def get_tools_manual(self, tools: list[ToolMetadata]) -> list[dict]:
        """
        Generates a tool manual for the specified tools.

        Args:
            tools (list[ToolMetadata]): The list of tools to generate the manual for.

        Returns:
            list[dict]: A list of dict representing the tool manual, which can be used for
            openai tool calls.
        """
        manual = []
        for tool in tools:
            manual.append(
                {
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": {
                            "type": "object",
                            "properties:": (parameters := {}),
                            "required": (required := []),
                        },
                    },
                },
            )
            if self.to_user_flag:
                parameters[self.to_user_flag] = {
                    "type": "string",
                    "description": (
                        "The content paraphrased for the user. "
                        "The content of this parameter can tell the user what you are about to do, "
                        "or it can be an explanation of the behavior of the function calling. "
                        "For example, 'I'm going to search the internet, please wait a moment.'"
                    ),
                }
            for param in tool.parameters:
                parameters[param["name"]] = {
                    "type": "string",
                    "description": param["description"],
                }
                if param["choices"]:
                    parameters[param["name"]]["enum"] = param["choices"]
                if param["required"]:
                    required.append(param["name"])
        return manual

    async def parse_response(
        self,
        source: AsyncIterator,
    ) -> Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
        """
        Parses asynchronous responses from LLM.

        Args:
            source (AsyncIterator): The asynchronous iterator source for the responses.

        Returns:
            Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
            A callable that generates an asynchronous generator for either "to_user"
            or "tool_call" content.
        """
        gen_maker = await async_dispatcher_tools_call_for_openai(
            source=source,
            to_user_flag=self.to_user_flag,
            logger=self.logger,
        )        
        return gen_maker

tool_model_name: str property

Gets the name of the tool model.

__init__(tool_token_counter, to_user_flag='to_user', logger=None)

Initializes a new instance of the OpenaiToolModel class with a token counter callable, an optional to_user_flag, and an optional logger.

Parameters:

Name	Type	Description	Default
tool_token_counter	Callable[[str], int]	A callable that takes a string and returns an integer, used for counting tokens in the tool manual.	required
to_user_flag	str | None	The flag for user-facing content in response.	'to_user'
logger	Logger | None	The logger for logging.	None

Source code in src/agere/utils/tool_models/openai_tool_model.py

def __init__(
    self,
    tool_token_counter: Callable[[str], int],
    to_user_flag: str | None = "to_user",
    logger: logging.Logger | None = None,
):
    """
    Initializes a new instance of the OpenaiToolModel class with a token counter
    callable, an optional to_user_flag, and an optional logger.

    Args:
        tool_token_counter (Callable[[str], int]):
            A callable that takes a string and returns an integer, used for counting
            tokens in the tool manual.
        to_user_flag (str | None, optional): The flag for user-facing content in response.
        logger (logging.Logger | None, optional): The logger for logging.
    """
    self.tool_token_counter = tool_token_counter
    if to_user_flag is None:
        to_user_flag = "__THIS_IS_A_NEVER_USED_NAME__"
    self.to_user_flag = to_user_flag
    self.logger = logger

get_tools_manual(tools)

Generates a tool manual for the specified tools.

Parameters:

Name	Type	Description	Default
tools	list[ToolMetadata]	The list of tools to generate the manual for.	required

Returns:

Type	Description
list[dict]	list[dict]: A list of dict representing the tool manual, which can be used for
list[dict]	openai tool calls.

Source code in src/agere/utils/tool_models/openai_tool_model.py

def get_tools_manual(self, tools: list[ToolMetadata]) -> list[dict]:
    """
    Generates a tool manual for the specified tools.

    Args:
        tools (list[ToolMetadata]): The list of tools to generate the manual for.

    Returns:
        list[dict]: A list of dict representing the tool manual, which can be used for
        openai tool calls.
    """
    manual = []
    for tool in tools:
        manual.append(
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": {
                        "type": "object",
                        "properties:": (parameters := {}),
                        "required": (required := []),
                    },
                },
            },
        )
        if self.to_user_flag:
            parameters[self.to_user_flag] = {
                "type": "string",
                "description": (
                    "The content paraphrased for the user. "
                    "The content of this parameter can tell the user what you are about to do, "
                    "or it can be an explanation of the behavior of the function calling. "
                    "For example, 'I'm going to search the internet, please wait a moment.'"
                ),
            }
        for param in tool.parameters:
            parameters[param["name"]] = {
                "type": "string",
                "description": param["description"],
            }
            if param["choices"]:
                parameters[param["name"]]["enum"] = param["choices"]
            if param["required"]:
                required.append(param["name"])
    return manual

parse_response(source) async

Parses asynchronous responses from LLM.

Parameters:

Name	Type	Description	Default
source	AsyncIterator	The asynchronous iterator source for the responses.	required

Returns:

Type	Description
Callable[[Literal['to_user', 'tool_call']], AsyncGenerator]	Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
Callable[[Literal['to_user', 'tool_call']], AsyncGenerator]	A callable that generates an asynchronous generator for either "to_user"
Callable[[Literal['to_user', 'tool_call']], AsyncGenerator]	or "tool_call" content.

Source code in src/agere/utils/tool_models/openai_tool_model.py

async def parse_response(
    self,
    source: AsyncIterator,
) -> Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
    """
    Parses asynchronous responses from LLM.

    Args:
        source (AsyncIterator): The asynchronous iterator source for the responses.

    Returns:
        Callable[[Literal["to_user", "tool_call"]], AsyncGenerator]:
        A callable that generates an asynchronous generator for either "to_user"
        or "tool_call" content.
    """
    gen_maker = await async_dispatcher_tools_call_for_openai(
        source=source,
        to_user_flag=self.to_user_flag,
        logger=self.logger,
    )        
    return gen_maker

tool_manual_token_counter(tools)

Counts the number of tokens for the manual of specified tools.

Parameters:

Name	Type	Description	Default
tools	list[ToolMetadata]	A list of ToolMetadata objects to generate the	required

Returns:

Name	Type	Description
int	int	The number of tokens in the tool manual.

Source code in src/agere/utils/tool_models/openai_tool_model.py

def tool_manual_token_counter(self, tools: list[ToolMetadata]) -> int:
    """
    Counts the number of tokens for the manual of specified tools.

    Args:
        tools (list[ToolMetadata]): A list of ToolMetadata objects to generate the
        manual from.

    Returns:
        int: The number of tokens in the tool manual.
    """
    manual = str(self.get_tools_manual(tools=tools))
    return self.tool_token_counter(manual)