File size: 4,173 Bytes
77320e4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 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 121 122 123 124 125 126 127 128 129 130 131 132 |
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]
# BaseTool
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):
# If config_input is a string, assume it's a file path.
config_data = Config.load(config_input)
elif isinstance(config_input, dict):
# If config_input is a dict, use it directly as config_data.
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):
# If config_input is a string, assume it's a file path.
config_data = Config.load(config_input)
elif isinstance(config_input, dict):
# If config_input is a dict, use it directly as config_data.
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)
|