100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
|
|
import inspect
|
|
|
|
# Do not call methods from instances of this class, they must not get overridden
|
|
class Toolset:
|
|
# Fields: description, identifier, _tools: list[ToolDescription]
|
|
def __init__(self):
|
|
# TODO: receive app, assign _tools
|
|
pass
|
|
|
|
def get_namespace_definitions(toolset):
|
|
header = toolset.description
|
|
if toolset.identifier:
|
|
header = f"Namespace {toolset.identifier}: {toolset.description}"
|
|
lines = [header]
|
|
|
|
for tool in toolset._tools:
|
|
full_name = tool.identifier
|
|
if toolset.identifier:
|
|
full_name = toolset.identifier + "." + full_name
|
|
lines.append(f"- {tool.description}\n!!{full_name} {tool.schema}")
|
|
return "\n".join(lines)
|
|
|
|
async def execute_action(toolset, name: str, argument: dict):
|
|
method = getattr(toolset, name)
|
|
result = method(**argument)
|
|
if inspect.isawaitable(result):
|
|
result = await result
|
|
|
|
class ToolDescription:
|
|
def __init__(self, identifier, description, schema):
|
|
self.identifier=identifier
|
|
self.description=description
|
|
self.schema=schema
|
|
|
|
# An action is an object representing a command invokation
|
|
class Action:
|
|
def __init__(self, toolset_name: str, tool_name: str, args: dict, is_query: bool = False):
|
|
self.toolset_name = toolset_name
|
|
self.tool_name = tool_name
|
|
self.args = args
|
|
self.is_query = is_query
|
|
|
|
def is_special(self, name: str):
|
|
return self.toolset_name == "" and self.tool_name == name
|
|
|
|
def __str__(self):
|
|
return f"{self.is_query and 'query' or ''}{self.toolset_name}.{self.tool_name} {self.args}"
|
|
|
|
def __repr__(self):
|
|
return f"Action({self.__str__()})"
|
|
|
|
class SpecialToolset(Toolset):
|
|
def __init__(self):
|
|
self.identifier = ""
|
|
self.description = "Special commands that don't receive arguments"
|
|
self._tools = [
|
|
ToolDescription("resume", "Resume the conversation with the command outputs", ''),
|
|
ToolDescription("respond", "When used, all text after this command will be used as the actual response", ''),
|
|
]
|
|
|
|
def resume(self):
|
|
pass
|
|
|
|
def respond(self):
|
|
pass
|
|
|
|
class BasicToolset(Toolset):
|
|
def __init__(self):
|
|
self.identifier = "util"
|
|
self.description = "Basic assistant utility tools"
|
|
self._tools = [
|
|
ToolDescription("google_search", "Executes a google search", '{"query":string}'),
|
|
ToolDescription("fetch", "Fetch the contents of a url", '{"url":string}'),
|
|
ToolDescription("eval", "Evaluates a python expression", '{"expression":string}'),
|
|
]
|
|
|
|
def google_search(self, query):
|
|
return "We're sorry. Google is not accessible at this time"
|
|
|
|
def fetch(self, url):
|
|
return "No internet connection."
|
|
|
|
def eval(self, expression):
|
|
try:
|
|
return eval(expression)
|
|
except Exception as e:
|
|
return str(e)
|
|
|
|
class HubControlsToolset(Toolset):
|
|
def __init__(self):
|
|
self.identifier = "hub"
|
|
self.description = "Controls for the Hub system"
|
|
self._tools = [
|
|
ToolDescription("get_chats", "Get all the existing chats", '{}'),
|
|
ToolDescription("get_agents", "Get all the existing bots", '{}'),
|
|
ToolDescription("create_chat", "Creates a new chat", '{"name":string,"agent_ids":int[]}'),
|
|
ToolDescription("get_agent_data", "Gets the configuration data of an agent", '{"agent_id":int}'),
|
|
ToolDescription("set_agent_data", "Get all the existing bots", '{"agent_id":int,"data":object}'),
|
|
] |