|
|
"""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"] |
|
|
|
