| """Interface for tools.""" |
|
|
| from typing import List, Optional |
|
|
| from langchain_core.callbacks import ( |
| AsyncCallbackManagerForToolRun, |
| CallbackManagerForToolRun, |
| ) |
| from langchain_core.tools import BaseTool, tool |
|
|
|
|
| class InvalidTool(BaseTool): |
| """Tool that is run when invalid tool name is encountered by agent.""" |
|
|
| name: str = "invalid_tool" |
| """Name of the tool.""" |
| description: str = "Called when tool name is invalid. Suggests valid tool names." |
| """Description of the tool.""" |
|
|
| def _run( |
| self, |
| requested_tool_name: str, |
| available_tool_names: List[str], |
| run_manager: Optional[CallbackManagerForToolRun] = None, |
| ) -> str: |
| """Use the tool.""" |
| available_tool_names_str = ", ".join([tool for tool in available_tool_names]) |
| return ( |
| f"{requested_tool_name} is not a valid tool, " |
| f"try one of [{available_tool_names_str}]." |
| ) |
|
|
| async def _arun( |
| self, |
| requested_tool_name: str, |
| available_tool_names: List[str], |
| run_manager: Optional[AsyncCallbackManagerForToolRun] = None, |
| ) -> str: |
| """Use the tool asynchronously.""" |
| available_tool_names_str = ", ".join([tool for tool in available_tool_names]) |
| return ( |
| f"{requested_tool_name} is not a valid tool, " |
| f"try one of [{available_tool_names_str}]." |
| ) |
|
|
|
|
| __all__ = ["InvalidTool", "tool"] |
|
|
