| from dataclasses import dataclass |
| from typing import Optional, Type |
| from abc import ABC |
| from importlib import import_module |
|
|
| from ..exceptions.exceptions import InvalidConfigException |
| from ..utils import Config |
|
|
|
|
| @dataclass |
| class BaseToolRequest(ABC): |
| input_text: Optional[str] |
|
|
|
|
| @dataclass |
| class BaseToolResponse(ABC): |
| output_text: Optional[str] |
|
|
|
|
| |
| class BaseTool(ABC): |
| _name = None |
| _description = None |
|
|
| def __init__(self, name, description, **kwargs): |
| self._name = name |
| self._description = description |
| self.setup() |
|
|
| @property |
| def name(self): |
| """Getter for name.""" |
| return self._name |
|
|
| @property |
| def description(self): |
| """Getter for description.""" |
| return self._description |
|
|
| @classmethod |
| def from_config(cls, config_input, **kwargs): |
| """Create a BaseTool instance from a config file path or a config data dictionary. |
| |
| :param config_input: Either a file path to a config file or a config data dictionary. |
| :type config_input: str or dict |
| :param kwargs: Additional keyword arguments to pass to the class constructor. |
| :return: A BaseTool instance. |
| :rtype: BaseTool |
| """ |
| if isinstance(config_input, str): |
| |
| config_data = Config.load(config_input) |
| elif isinstance(config_input, dict): |
| |
| config_data = config_input |
| else: |
| raise InvalidConfigException( |
| f"Invalid config_input type: {type(config_input)}. " |
| "Expected str (file path) or dict (config data)." |
| ) |
|
|
| module_name = config_data['module_name'] |
| class_name = config_data['class_name'] |
| module = import_module(module_name) |
| clazz = getattr(module, class_name) |
| return clazz(**config_data, **kwargs) |
|
|
| @classmethod |
| async def async_from_config(cls, config_input, **params): |
| """Asynchronously create a BaseTool instance from a config file path or a config data dictionary. |
| |
| :param config_input: Either a file path to a config file or a config data dictionary. |
| :type config_input: str or dict |
| :param params: Additional parameters to pass to the create method. |
| :return: A BaseTool instance. |
| :rtype: BaseTool |
| """ |
| |
|
|
| if isinstance(config_input, str): |
| |
| config_data = Config.load(config_input) |
| elif isinstance(config_input, dict): |
| |
| config_data = config_input |
| else: |
| raise InvalidConfigException( |
| f"Invalid config_input type: {type(config_input)}. " |
| "Expected str (file path) or dict (config data)." |
| ) |
|
|
| |
| module_name = config_data['module_name'] |
| class_name = config_data['class_name'] |
| module = import_module(module_name) |
| clazz = getattr(module, class_name) |
| |
| return await clazz.create(config_data, **params) |
|
|
| @classmethod |
| async def async_from_config_path(cls, config_path, **params): |
| return await cls.async_from_config_data(config_data=Config.load(config_path), **params) |
|
|
| @classmethod |
| async def async_from_config_data(cls, config_data, **params): |
| module_name = config_data['module_name'] |
| class_name = config_data['class_name'] |
|
|
| module = import_module(module_name) |
| clazz = getattr(module, class_name) |
|
|
| return await clazz.create(config_data, **params) |
|
|
| @classmethod |
| async def create(cls, config_data, **params): |
| """ |
| Async create tool instance. init cannot be async, so wrap async init logic here. |
| """ |
| pass |
|
|
| def setup(self): |
| pass |
|
|
| def run(self, req: BaseToolRequest): |
| pass |
|
|
| async def async_run(self, req: BaseToolRequest): |
| """ |
| Async run tool. |
| """ |
| return self.run(req) |
|
|
