Skip to content

Tools

BaseTool

Bases: BaseComponent

Source code in libs/kotaemon/kotaemon/agents/tools/base.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class BaseTool(BaseComponent):
    name: str
    """The unique name of the tool that clearly communicates its purpose."""
    description: str
    """Description used to tell the model how/when/why to use the tool.
    You can provide few-shot examples as a part of the description. This will be
    input to the prompt of LLM.
    """
    args_schema: Optional[Type[BaseModel]] = None
    """Pydantic model class to validate and parse the tool's input arguments."""
    verbose: bool = False
    """Whether to log the tool's progress."""
    handle_tool_error: Optional[
        Union[bool, str, Callable[[ToolException], str]]
    ] = False
    """Handle the content of the ToolException thrown."""

    def _parse_input(
        self,
        tool_input: Union[str, Dict],
    ) -> Union[str, Dict[str, Any]]:
        """Convert tool input to pydantic model."""
        args_schema = self.args_schema
        if isinstance(tool_input, str):
            if args_schema is not None:
                key_ = next(iter(args_schema.model_fields.keys()))
                args_schema.validate({key_: tool_input})
            return tool_input
        else:
            if args_schema is not None:
                result = args_schema.parse_obj(tool_input)
                return {k: v for k, v in result.dict().items() if k in tool_input}
        return tool_input

    def _run_tool(
        self,
        *args: Any,
        **kwargs: Any,
    ) -> Any:
        """Call tool."""
        raise NotImplementedError(f"_run_tool is not implemented for {self.name}")

    def _to_args_and_kwargs(self, tool_input: Union[str, Dict]) -> Tuple[Tuple, Dict]:
        # For backwards compatibility, if run_input is a string,
        # pass as a positional argument.
        if isinstance(tool_input, str):
            return (tool_input,), {}
        else:
            return (), tool_input

    def _handle_tool_error(self, e: ToolException) -> Any:
        """Handle the content of the ToolException thrown."""
        observation = None
        if not self.handle_tool_error:
            raise e
        elif isinstance(self.handle_tool_error, bool):
            if e.args:
                observation = e.args[0]
            else:
                observation = "Tool execution error"
        elif isinstance(self.handle_tool_error, str):
            observation = self.handle_tool_error
        elif callable(self.handle_tool_error):
            observation = self.handle_tool_error(e)
        else:
            raise ValueError(
                f"Got unexpected type of `handle_tool_error`. Expected bool, str "
                f"or callable. Received: {self.handle_tool_error}"
            )
        return observation

    def to_langchain_format(self) -> LCTool:
        """Convert this tool to Langchain format to use with its agent"""
        return LCTool(name=self.name, description=self.description, func=self.run)

    def run(
        self,
        tool_input: Union[str, Dict],
        verbose: Optional[bool] = None,
        **kwargs: Any,
    ) -> Any:
        """Run the tool."""
        parsed_input = self._parse_input(tool_input)
        # TODO (verbose_): Add logging
        try:
            tool_args, tool_kwargs = self._to_args_and_kwargs(parsed_input)
            call_kwargs = {**kwargs, **tool_kwargs}
            observation = self._run_tool(*tool_args, **call_kwargs)
        except ToolException as e:
            observation = self._handle_tool_error(e)
            return observation
        else:
            return observation

    @classmethod
    def from_langchain_format(cls, langchain_tool: LCTool) -> "BaseTool":
        """Wrapper for Langchain Tool"""
        new_tool = BaseTool(
            name=langchain_tool.name, description=langchain_tool.description
        )
        new_tool._run_tool = langchain_tool._run  # type: ignore
        return new_tool

name instance-attribute 

1
name

The unique name of the tool that clearly communicates its purpose.

description instance-attribute 

1
description

Description used to tell the model how/when/why to use the tool. You can provide few-shot examples as a part of the description. This will be input to the prompt of LLM.

args_schema class-attribute instance-attribute 

1
args_schema = None

Pydantic model class to validate and parse the tool's input arguments.

verbose class-attribute instance-attribute 

1
verbose = False

Whether to log the tool's progress.

handle_tool_error class-attribute instance-attribute 

1
handle_tool_error = False

Handle the content of the ToolException thrown.

to_langchain_format 

1
to_langchain_format()

Convert this tool to Langchain format to use with its agent

Source code in libs/kotaemon/kotaemon/agents/tools/base.py 
90
91
92
def to_langchain_format(self) -> LCTool:
    """Convert this tool to Langchain format to use with its agent"""
    return LCTool(name=self.name, description=self.description, func=self.run)

run 

1
run(tool_input, verbose=None, **kwargs)

Run the tool.

Source code in libs/kotaemon/kotaemon/agents/tools/base.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def run(
    self,
    tool_input: Union[str, Dict],
    verbose: Optional[bool] = None,
    **kwargs: Any,
) -> Any:
    """Run the tool."""
    parsed_input = self._parse_input(tool_input)
    # TODO (verbose_): Add logging
    try:
        tool_args, tool_kwargs = self._to_args_and_kwargs(parsed_input)
        call_kwargs = {**kwargs, **tool_kwargs}
        observation = self._run_tool(*tool_args, **call_kwargs)
    except ToolException as e:
        observation = self._handle_tool_error(e)
        return observation
    else:
        return observation

from_langchain_format classmethod 

1
from_langchain_format(langchain_tool)

Wrapper for Langchain Tool

Source code in libs/kotaemon/kotaemon/agents/tools/base.py 
113
114
115
116
117
118
119
120
@classmethod
def from_langchain_format(cls, langchain_tool: LCTool) -> "BaseTool":
    """Wrapper for Langchain Tool"""
    new_tool = BaseTool(
        name=langchain_tool.name, description=langchain_tool.description
    )
    new_tool._run_tool = langchain_tool._run  # type: ignore
    return new_tool

ComponentTool

Bases: BaseTool

Wrapper around other BaseComponent to use it as a tool

Parameters:

Name Type Description Default
component

BaseComponent-based component to wrap

 required
postprocessor

Optional postprocessor for the component output

 required
Source code in libs/kotaemon/kotaemon/agents/tools/base.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class ComponentTool(BaseTool):
    """Wrapper around other BaseComponent to use it as a tool

    Args:
        component: BaseComponent-based component to wrap
        postprocessor: Optional postprocessor for the component output
    """

    component: BaseComponent
    postprocessor: Optional[Callable] = None

    def _run_tool(self, *args: Any, **kwargs: Any) -> Any:
        output = self.component(*args, **kwargs)
        if self.postprocessor:
            output = self.postprocessor(output)

        return output

WikipediaTool

Bases: BaseTool

Tool that adds the capability to query the Wikipedia API.

Source code in libs/kotaemon/kotaemon/agents/tools/wikipedia.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class WikipediaTool(BaseTool):
    """Tool that adds the capability to query the Wikipedia API."""

    name: str = "wikipedia"
    description: str = (
        "Search engine from Wikipedia, retrieving relevant wiki page. "
        "Useful when you need to get holistic knowledge about people, "
        "places, companies, historical events, or other subjects. "
        "Input should be a search query."
    )
    args_schema: Optional[Type[BaseModel]] = WikipediaArgs
    doc_store: Any = None

    def _run_tool(self, query: AnyStr) -> AnyStr:
        if not self.doc_store:
            self.doc_store = Wiki()
        tool = self.doc_store
        evidence = tool.search(query)
        return evidence