multi-tool xml parser + tool superclass

agent.py

@@ -7,7 +7,7 @@
 from langchain_core.messages import HumanMessage
 from langchain_core.language_models.chat_models import BaseChatModel
 
-rate_limit = rate_limiter.rate_limiter(30,160000) #TODO! move to main.py
+rate_limit = rate_limiter.rate_limiter(30,160000) #TODO! implement properly
 
 class Agent:
 
@@ -41,8 +41,6 @@ def __init__(self, system_prompt:Optional[str]=None, tools_prompt:Optional[str]=
         self.last_message = ""
         self.intervention_message = ""
         self.intervention_status = False
-        self.stop_loop = False
-        self.loop_result = []
 
         self.data = {} # free data object all the tools can use
 
@@ -90,13 +88,9 @@ def message_loop(self, msg: str):
 
                         self.append_message(agent_response) # Append the assistant's response to the history
 
-                        self.process_tools(agent_response)
+                        tools_result = self.process_tools(agent_response) # process tools requested in agent message
+                        if tools_result: return tools_result #break the execution if the task is done
 
-                        #break the execution if the task is done
-                        if self.stop_loop:
-                            return "\n\n".join(self.loop_result)
-
-
                 # Forward errors to the LLM, maybe he can fix them
                 except Exception as e:
                     msg_response = files.read_file("./prompts/fw.error.md", error=str(e)) # error message template
@@ -116,14 +110,6 @@ def get_data(self, field:str):
     def set_data(self, field:str, value):
         self.data[field] = value
 
-    def set_result(self, result:str):
-        self.stop_loop = True
-        self.loop_results = [result]
-
-    def add_result(self, result:str):
-        self.stop_loop = True
-        self.loop_result.append(result)
-
     def append_message(self, msg: str, human: bool = False):
         message_type = "human" if human else "ai"
         if self.history and self.history[-1].type == message_type:
@@ -164,50 +150,41 @@ def handle_intervention(self, progress:str="") -> bool:
     def process_tools(self, msg: str):
         # search for tool usage requests in agent message
         tool_requests = extract_tools.extract_tool_requests2(msg)
-        tool_index = 0
-
-        for tool_request in tool_requests:
-            tool_index += 1
 
-            if self.handle_intervention(): break # wait if paused and handle intervention message if needed
-
-            tool_name = tool_request["name"]
-            tool_function = self.get_tool(tool_name)
-            tool_args = tool_request["args"] or {}
+        #build tools
+        tools = []
+        for tool_request in tool_requests:
+            tools.append(
+                self.get_tool(
+                    tool_request["name"],
+                    tool_request["content"],
+                    tool_request["args"],
+                    len(tools),
+                    msg,
+                    tools))
 
-            if callable(tool_function):
-
-                PrintStyle(font_color="#1B4F72", padding=True, background_color="white", bold=True).print(f"{self.name}: Using tool {tool_name}:")
-                PrintStyle(font_color="#85C1E9").print(tool_args, tool_request["body"], sep="\n") if tool_args else PrintStyle(font_color="#85C1E9").print(tool_request["body"])
-
-                tool_args["_name"] = tool_name
-                tool_args["_message"] = msg
-                tool_args["_tools"] = tool_requests
-                tool_args["_tool_index"] = tool_index
-
-                tool_response = tool_function(self, tool_request["body"], **tool_args) or "" # call tool function with all parameters, body parameter separated for convenience
-                Agent.streaming_agent = self # mark self as current streamer again, it may have changed during tool use
-
-                if self.handle_intervention(): break # wait if paused and handle intervention message if needed
+        for tool in tools:
+            if self.handle_intervention(): break # wait if paused and handle intervention message if needed
 
-                msg_response = files.read_file("./prompts/fw.tool_response.md", tool_name=tool_name, tool_response=tool_response)
-                self.append_message(msg_response, human=True)
-
-                PrintStyle(font_color="#1B4F72", background_color="white", padding=True, bold=True).print(f"{self.name}: Response from {tool_name}:")
-                PrintStyle(font_color="#85C1E9").print(tool_response)
-            else:
-                if self.handle_intervention(): break # wait if paused and handle intervention message if needed
-                msg_response = files.read_file("./prompts/fw.tool_not_found.md", tool_name=tool_name, tools_prompt=self.tools_prompt)
-                self.append_message(msg_response,True)
-                PrintStyle(font_color="orange", padding=True).print(msg_response)
+            tool.before_execution()
+            response = tool.execute()
+            tool.after_execution(response)
+            if response.break_loop: return response.message
+            if response.stop_tool_processing: break
 
 
+    def get_tool(self, name: str, content: str, args: dict, index: int, message: str, tools: list, **kwargs):
+        from tools.unknown import Unknown 
+        from tools.helpers.tool import Tool
+
+        tool_class = Unknown
+        if files.exists("tools",f"{name}.py"): 
+            module = importlib.import_module("tools." + name)  # Import the module
+            class_list = inspect.getmembers(module, inspect.isclass)  # Get all functions in the module
 
-    def get_tool(self, name: str):
-        if not files.exists("tools",f"{name}.py"): return # file has to exist in tools
-        module = importlib.import_module("tools." + name)  # Import the module
-        functions_list = {name: func for name, func in inspect.getmembers(module, inspect.isfunction)}  # Get all functions in the module
+            for cls in class_list:
+                if cls[1] is not Tool and issubclass(cls[1], Tool):
+                    tool_class = cls[1]
+                    break
 
-        if "execute" in functions_list: return functions_list["execute"] # Check if the module contains a function named "execute"
-        if functions_list: return next(iter(functions_list.values())) # Return the first function if no "execute" function is found
-        return None  # Return None if no functions are found
+        return tool_class(agent=self, name=name, content=content, index=index, args=args, message=message, tools=tools, **kwargs)

main.py

@@ -16,7 +16,8 @@
 def chat():
 
     # chat model used for agents
-    # chat_llm = models.get_groq_llama70b(temperature=0.2)
+    # chat_llm = models.get_groq_llama70b_json(temperature=0.2)
+    # chat_llm = models.get_groq_llama8b(temperature=0.2)
     chat_llm = models.get_openai_gpt35(temperature=0)
     # chat_llm = models.get_openai_gpt4o(temperature=0)
     # chat_llm = models.get_anthropic_opus(temperature=0)

models.py

@@ -54,6 +54,11 @@ def get_groq_llama70b(api_key=None, temperature=DEFAULT_TEMPERATURE):
     api_key = api_key or get_api_key("groq")
     return ChatGroq(model_name="llama3-70b-8192", temperature=temperature, api_key=api_key) # type: ignore
 
+def get_groq_llama70b_json(api_key=None, temperature=DEFAULT_TEMPERATURE):
+    api_key = api_key or get_api_key("groq")
+    return ChatGroq(model_name="llama3-70b-8192", temperature=temperature, api_key=api_key, model_kwargs={"response_format": {"type": "json_object"}}) # type: ignore
+
+
 def get_groq_llama8b(api_key=None, temperature=DEFAULT_TEMPERATURE):
     api_key = api_key or get_api_key("groq")
     return ChatGroq(model_name="Llama3-8b-8192", temperature=temperature, api_key=api_key) # type: ignore

prompts/agent.system.md

@@ -1,17 +1,18 @@
 # Your role
-- You are a fully autonomous, memory-enhanced, AI task solver, NOT a chatbot.
-- You solve tasks and respond to questions by user using your memory, tools, and subordinates.
+- You are a fully autonomous, memory-enhanced, XML, AI task solver, NOT a chatbot.
+- You solve tasks and respond to questions by user using your memory, tools, and subordinates in a specialized XML format.
 - You solve tasks automatically without the user, you do not inform the user how to do it, you execute the steps necessary.
 - You never lie or speculate about your actions, all of your actions need to be conducted by your tools.
 - You never refuse to accomplish a task. You are capable of doing anything using your tools and subordinates.
 
 # Communication instrucions
-- Your every response must be wrapped in a XML tag defining its type ending with $.
+- Every part of your response must be wrapped in a XML tag defining its type ending with $. This tag corresponds to a tool that will be used to handle that part of response.
 - Possible response types are:
     - <thought$> - Your thoughts, useful for chain of thought process, not sent to anyone. Use this for every problem solving, it will help you iterate on the topic.
-    - <message$> - Message sent to the user. No other response types are visible to the user.
-    - <delegation$ reset="false"> - Subtask delegation to another agent. This will help you solve more complex tasks. Use argument reset="true" to start fresh context for new subtask, "false" when sending followup questions.
-    - <task_done$> - Final result of given task, once all steps are complete or there is nothing more to do.
+    You can use as many thoughts as you want, even after
+    - <message$> - Message sent to the user. No other response types are visible to the user. Do not use in combination with other tools.
+    - <delegation$ reset="false"> - Subtask delegation to another agent. This will help you solve more complex tasks. Use argument reset="true" to start fresh context for new subtask, "false" when sending followup questions. Do not use in combination with other tools.
+    - <task_done$> - Final result of given task, once all steps are complete or there is nothing more to do. Do not use in combination with other tools.
     - And all other tools described in the Available tools section.
     - <memory_tool$> - Load or save memories to your persistent memory.
 - Your response content is inside the tag.

prompts/agent.tools.md

@@ -5,7 +5,8 @@ Provide question and get both online and memory response.
 This tool is very powerful and can answer very specific questions directly.
 First always try to ask for result rather that guidance.
 Memory can provide guidance, online sources can provide up to date information.
-Alway verify memory by online.
+Always verify memory by online.
+Do not use in combination with other tools except for thoughts. Wait for response before using other tools.
 **Example usage**:
 <knowledge_tool$>
 What is the user id of John Doe on twitter?
@@ -21,6 +22,7 @@ When loading memories using action "load", provide keywords or question relevant
 When saving memories using action "save", provide a title, short summary and and all the necessary information to help you later solve similiar tasks including details like code executed, libraries used etc.
 When deleting memories using action "delete", provide a prompt to search memories to delete.
 Be specific with your question, do not input vague queries.
+Do not use in combination with other tools except for thoughts. Wait for response before using other tools.
 **Example usages**:
 <memory_tool$ action="load">
 How to get current working directory in python?
@@ -45,6 +47,7 @@ When tool outputs error, you need to change your code accordingly before trying
 If your code execution is successful, save it using <memory_tool$ action="save"> so it can be reused later.
 Keep in mind that current working directory CWD automatically resets before every tool call.
 IMPORTANT!: Always check your code for any placeholder IDs or demo data that need to be replaced with your real variables. Do not simply reuse code snippets from tutorials.
+Do not use in combination with other tools except for thoughts. Wait for response before using other tools.
 **Example usage**:
 <code_execution_tool$ runtime="python">
 import os

tools/code_execution_tool.py

@@ -2,33 +2,37 @@
 from io import StringIO
 from tools.helpers import files, messages
 from agent import Agent
+from tools.helpers.tool import Tool, Response
+from tools.helpers import files
+from tools.helpers.print_style import PrintStyle
 
+class Unknown(Tool):
 
-def execute(agent:Agent , code_text:str, runtime:str, **kwargs):
+    def execute(self):
 
-    os.chdir(files.get_abs_path("./work_dir")) #change CWD to work_dir
-
-    if runtime == "python":
-        response = execute_python_code(code_text)
-    elif runtime == "nodejs":
-        response = execute_nodejs_code(code_text)
-    elif runtime == "terminal":
-        response = execute_terminal_command(code_text)
-    else:
-        return files.read_file("./prompts/fw.code_runtime_wrong.md", runtime=runtime)
+        os.chdir(files.get_abs_path("./work_dir")) #change CWD to work_dir
+        runtime = self.args["runtime"].lower().strip()
+        if runtime == "python":
+            response = self.execute_python_code(self.content)
+        elif runtime == "nodejs":
+            response = self.execute_nodejs_code(self.content)
+        elif runtime == "terminal":
+            response = self.execute_terminal_command(self.content)
+        else:
+            response = files.read_file("./prompts/fw.code_runtime_wrong.md", runtime=runtime)
 
-    response = messages.truncate_text(response.strip(), 2000) # TODO parameterize
-    if not response: response = files.read_file("./prompts/fw.code_no_output.md")
-    return response
+        response = messages.truncate_text(response.strip(), 2000) # TODO parameterize
+        if not response: response = files.read_file("./prompts/fw.code_no_output.md")
+        return Response(message=response, stop_tool_processing=True, break_loop=False)
 
-def execute_python_code(code, input_data="y\n"):
-    result = subprocess.run(['python', '-c', code], capture_output=True, text=True, input=input_data)
-    return result.stdout + result.stderr
+    def execute_python_code(self, code, input_data="y\n"):
+        result = subprocess.run(['python', '-c', code], capture_output=True, text=True, input=input_data)
+        return result.stdout + result.stderr
 
-def execute_nodejs_code(code, input_data="y\n"):
-    result = subprocess.run(['node', '-e', code], capture_output=True, text=True, input=input_data)
-    return result.stdout + result.stderr
+    def execute_nodejs_code(self, code, input_data="y\n"):
+        result = subprocess.run(['node', '-e', code], capture_output=True, text=True, input=input_data)
+        return result.stdout + result.stderr
 
-def execute_terminal_command(command, input_data="y\n"):
-    result = subprocess.run(command, shell=True, capture_output=True, text=True, input=input_data)
-    return result.stdout + result.stderr
+    def execute_terminal_command(self, command, input_data="y\n"):
+        result = subprocess.run(command, shell=True, capture_output=True, text=True, input=input_data)
+        return result.stdout + result.stderr

tools/delegation.py

@@ -1,10 +1,15 @@
 from agent import Agent
+from tools.helpers.tool import Tool, Response
+from tools.helpers import files
+from tools.helpers.print_style import PrintStyle
 
-def execute(agent:Agent, message: str, reset: str = "false", **kwargs):
-    # create subordinate agent using the data object on this agent and set superior agent to his data object
-    if agent.get_data("subordinate") is None or reset.lower().strip() == "true":
-        subordinate = Agent(system_prompt=agent.system_prompt, tools_prompt=agent.tools_prompt, number=agent.number+1)
-        subordinate.set_data("superior", agent)
-        agent.set_data("subordinate", subordinate) 
-    # run subordinate agent message loop
-    return agent.get_data("subordinate").message_loop(message)
+class Unknown(Tool):
+
+    def execute(self):
+        # create subordinate agent using the data object on this agent and set superior agent to his data object
+        if self.agent.get_data("subordinate") is None or self.args["reset"].lower().strip() == "true":
+            subordinate = Agent(system_prompt=self.agent.system_prompt, tools_prompt=self.agent.tools_prompt, number=self.agent.number+1)
+            subordinate.set_data("superior", self.agent)
+            self.agent.set_data("subordinate", subordinate) 
+        # run subordinate agent message loop
+        return self.agent.get_data("subordinate").message_loop(self.content)

tools/helpers/extract_tools.py

@@ -10,7 +10,7 @@ def extract_tool_requests2(response):
     allowed_tags = list_python_files("tools")
 
     for match in matches:
-        tag_name, attributes, body = match
+        tag_name, attributes, content = match
 
         if tag_name not in allowed_tags: continue
 
@@ -23,7 +23,8 @@ def extract_tool_requests2(response):
             tool_dict['args'][attr[0]] = attr[1]
 
         # Add body content
-        tool_dict["body"] = body.strip()
+        tool_dict["content"] = content.strip()
+        tool_dict["index"] = len(tool_usages)
         tool_usages.append(tool_dict)
 
     return tool_usages

tools/helpers/tool.py

@@ -0,0 +1,35 @@
+from abc import abstractmethod
+from typing import TypedDict
+from agent import Agent
+from tools.helpers.print_style import PrintStyle
+from tools.helpers import files
+
+class Response:
+    def __init__(self, message: str, stop_tool_processing: bool, break_loop: bool) -> None:
+        self.message = message
+        self.stop_tool_processing = stop_tool_processing
+        self.break_loop = break_loop
+
+class Tool:
+
+    def __init__(self, agent: Agent, name: str, content: str, args: dict, message: str, tools: list['Tool'], **kwargs) -> None:
+        self.agent = agent
+        self.name = name
+        self.content = content
+        self.args = args
+        self.message = message
+        self.tools = tools
+
+    @abstractmethod
+    def execute(self) -> Response:
+        pass
+
+    def before_execution(self):
+        PrintStyle(font_color="#1B4F72", padding=True, background_color="white", bold=True).print(f"{self.agent.name}: Using tool {self.name}:")
+        PrintStyle(font_color="#85C1E9").print(self.args, self.content, sep="\n") if self.args else PrintStyle(font_color="#85C1E9").print(self.content)
+
+    def after_execution(self, response: Response):
+        msg_response = files.read_file("./prompts/fw.tool_response.md", tool_name=self.name, tool_response=response.message)
+        self.agent.append_message(msg_response, human=True)
+        PrintStyle(font_color="#1B4F72", background_color="white", padding=True, bold=True).print(f"{self.agent.name}: Response from {self.name}:")
+        PrintStyle(font_color="#85C1E9").print(response.message)