Implement basic agent and toolset functionality

backend/tools.py

@@ -0,0 +1,100 @@
+
+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}'),
+        ]