defining the Solver's agents (part 1)

solver.py

@@ -10,26 +10,37 @@ from toolbox import Toolbox
 from args import Args
 
 
-class Solver:
+class Summarizer:
     def __init__(self, temperature, max_tokens):
-        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "06_math_expert.txt")
+        # Load the system prompt from a file
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "04_summarizer.txt")
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
+        # Define the LLM and agent
         llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
-        self.agent = AgentWorkflow.from_tools_or_functions(
-            [
-            Toolbox.math.symbolic_calc,
-            Toolbox.math.unit_converter,
-            ],
-            llm=llm
-        )
+        self.agent = AgentWorkflow.setup_agent(llm=llm)
         self.ctx = Context(self.agent)
     
-    def get_system_prompt(self):
+    def get_system_prompt(self) -> str:
+        """
+        Retrieves the system prompt.
+
+        Returns:
+            str: The system prompt string.
+        """
         return self.system_prompt
 
     async def query(self, question: str) -> str:
+        """
+        Asynchronously queries the agent with a given question and returns the response.
+
+        Args:
+            question (str): The question to be sent to the agent.
+
+        Returns:
+            str: The response from the agent as a string.
+        """
         response = await self.agent.run(question, ctx=self.ctx)
         response = str(response)
         return response
@@ -42,42 +53,96 @@ class Solver:
         self.ctx = Context(self.agent)
 
 
-class Summarizer:
+class Researcher:
     def __init__(self, temperature, max_tokens):
-        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "01_assistant.txt")
+        # Load the system prompt from a file
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "05_researcher.txt")
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
+        # Define the LLM and agent
         llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
-        self.agent = AgentWorkflow.setup_agent(llm=llm)
+        self.agent = AgentWorkflow.from_tools_or_functions(
+            Toolbox.web_search.duck_duck_go_tools,
+            llm=llm
+        )
         self.ctx = Context(self.agent)
+    
+    def get_system_prompt(self) -> str:
+        """
+        Retrieves the system prompt.
+
+        Returns:
+            str: The system prompt string.
+        """
+        return self.system_prompt
 
     async def query(self, question: str) -> str:
+        """
+        Asynchronously queries the agent with a given question and returns the response.
+
+        Args:
+            question (str): The question to be sent to the agent.
+
+        Returns:
+            str: The response from the agent as a string.
+        """
         response = await self.agent.run(question, ctx=self.ctx)
         response = str(response)
         return response
 
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        """
+        self.ctx = Context(self.agent)
+
 
-class MathExpert:
+class EncryptionExpert:
     def __init__(self, temperature, max_tokens):
-        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "06_math_expert.txt")
+        # Load the system prompt from a file
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "06_encryption_expert.txt")
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
+        # Define the LLM and agent
         llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
         self.agent = AgentWorkflow.from_tools_or_functions(
             [
-            Toolbox.math.symbolic_calc,
-            Toolbox.math.unit_converter,
+                Toolbox.encryption.base64_encode,
+                Toolbox.encryption.base64_decode,
+                Toolbox.encryption.caesar_cipher_encode,
+                Toolbox.encryption.caesar_cipher_decode,
+                Toolbox.encryption.reverse_string
+                # TODO: Add more encryption tools
             ],
             llm=llm
         )
         self.ctx = Context(self.agent)
+        # Initialize the tool agents
+        self.math_expert = MathExpert(temperature, max_tokens)
+        self.reasoner = Reasoner(temperature, max_tokens)
     
-    def get_system_prompt(self):
+    def get_system_prompt(self) -> str:
+        """
+        Retrieves the system prompt.
+
+        Returns:
+            str: The system prompt string.
+        """
         return self.system_prompt
 
     async def query(self, question: str) -> str:
+        """
+        Asynchronously queries the agent with a given question and returns the response.
+
+        Args:
+            question (str): The question to be sent to the agent.
+
+        Returns:
+            str: The response from the agent as a string.
+        """
         response = await self.agent.run(question, ctx=self.ctx)
         response = str(response)
         return response
@@ -86,137 +151,264 @@ class MathExpert:
         """
         Clears the current context of the agent, resetting any conversation history.
         This is useful when starting a new conversation or when the context needs to be refreshed.
+        Also clears the context of any tool agents.
         """
         self.ctx = Context(self.agent)
+        # Clear context for tool agents
+        self.math_expert.clear_context()
+        self.reasoner.clear_context()
 
 
-class Researcher:
+class MathExpert:
     def __init__(self, temperature, max_tokens):
-        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "04_researcher.txt")
+        # Load the system prompt from a file
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "07_math_expert.txt")
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
+        # Define the LLM and agent
         llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
-        
         self.agent = AgentWorkflow.from_tools_or_functions(
-            Toolbox.web_search.duck_duck_go_tools,
+            [
+            Toolbox.math.symbolic_calc,
+            Toolbox.math.unit_converter,
+            ],
             llm=llm
         )
         self.ctx = Context(self.agent)
+        # Initialize the tool agents
+        self.reasoner = Reasoner(temperature, max_tokens)
     
-    def get_system_prompt(self):
+    def get_system_prompt(self) -> str:
+        """
+        Retrieves the system prompt.
+
+        Returns:
+            str: The system prompt string.
+        """
         return self.system_prompt
 
     async def query(self, question: str) -> str:
+        """
+        Asynchronously queries the agent with a given question and returns the response.
+
+        Args:
+            question (str): The question to be sent to the agent.
+
+        Returns:
+            str: The response from the agent as a string.
+        """
         response = await self.agent.run(question, ctx=self.ctx)
         response = str(response)
         return response
 
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        Also clears the context of any tool agents.
+        """
+        self.ctx = Context(self.agent)
+        self.reasoner.clear_context()
+
 
-class EncryptionExpert:
+class Reasoner:
     def __init__(self, temperature, max_tokens):
-        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "05_encryption_expert.txt")
+        # Load the system prompt from a file
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "08_reasoner.txt")
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
+        # Define the LLM and agent
         llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
-        
-        self.agent = AgentWorkflow.from_tools_or_functions(
-            [
-                Toolbox.encryption.base64_encode,
-                Toolbox.encryption.base64_decode,
-                Toolbox.encryption.caesar_cipher_encode,
-                Toolbox.encryption.caesar_cipher_decode,
-                Toolbox.encryption.reverse_string
-            ],
-            llm=llm
-        )
+        self.agent = AgentWorkflow.setup_agent(llm=llm)
         self.ctx = Context(self.agent)
-    
-    def get_system_prompt(self):
-        return self.system_prompt
 
     async def query(self, question: str) -> str:
+        """
+        Asynchronously queries the agent with a given question and returns the response.
+
+        Args:
+            question (str): The question to be sent to the agent.
+
+        Returns:
+            str: The response from the agent as a string.
+        """
         response = await self.agent.run(question, ctx=self.ctx)
         response = str(response)
         return response
 
+    def get_system_prompt(self) -> str:
+        """
+        Retrieves the system prompt.
+
+        Returns:
+            str: The system prompt string.
+        """
+        return self.system_prompt
+    
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        """
+        self.ctx = Context(self.agent)
+
 
 class ImageHandler:
-    pass
+    def __init__(self, temperature, max_tokens):
+        # Load the system prompt from a file
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "09_image_handler.txt")
+        self.system_prompt = ""
+        with open(system_prompt_path, "r") as file:
+            self.system_prompt = file.read().strip()
+        pass
+
+    def get_system_prompt(self) -> str:
+        """
+        Retrieves the system prompt.
+
+        Returns:
+            str: The system prompt string.
+        """
+        return self.system_prompt
+
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        """
+        if hasattr(self, 'ctx') and hasattr(self, 'agent'):
+            self.ctx = Context(self.agent)
+
 
 class VideoHandler:
-    pass
+    def __init__(self, temperature, max_tokens):
+        # Load the system prompt from a file
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "10_video_handler.txt")
+        self.system_prompt = ""
+        with open(system_prompt_path, "r") as file:
+            self.system_prompt = file.read().strip()
+        # No implementation yet
+        pass
+    
+    def get_system_prompt(self) -> str:
+        """
+        Retrieves the system prompt.
 
-class RecursiveSolverAgent:
-    pass
+        Returns:
+            str: The system prompt string.
+        """
+        return self.system_prompt
 
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        """
+        if hasattr(self, 'ctx') and hasattr(self, 'agent'):
+            self.ctx = Context(self.agent)
 
-class Solver_2:
 
+class Solver:
     def __init__(self, temperature, max_tokens):
-        print("Agent initialized.")
-        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "01_assistant.txt")
+        # Load the system prompt from a file
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "03_solver.txt")
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
+        # Define the LLM and agent
         llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
         self.agent = AgentWorkflow.from_tools_or_functions(
             [
-            FunctionTool.from_defaults(self.delegate_to_math_expert),
-            FunctionTool.from_defaults(self.set_final_answer)
+            self.call_summarizer,
+            self.call_researcher,
+            self.call_encryption_expert,
+            self.call_math_expert,
+            self.call_reasoner,
+            self.call_image_handler,
+            self.call_video_handler
             ],
             llm=llm
         )
         self.ctx = Context(self.agent)
-        self.final_answer = ""
-
-    async def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        self.final_answer = ""
-        response = await self.query(question)
-        print(f"Agent processed the response: {response}")
-        if self.final_answer == "":
-            response = await self.query("I noticed the final_answer is an empty string. Have you forgot to set the final_answer ?")
-        return self.final_answer
-
-    def get_system_prompt(self):
-        return self.system_prompt
+        # Initialize the tool agents
+        self.summarizer = Summarizer(temperature, max_tokens)
+        self.researcher = Researcher(temperature, max_tokens)
+        self.encryption_expert = EncryptionExpert(temperature, max_tokens)
+        self.math_expert = MathExpert(temperature, max_tokens)
+        self.reasoner = Reasoner(temperature, max_tokens)
+        self.image_handler = ImageHandler(temperature, max_tokens)
+        self.video_handler = VideoHandler(temperature, max_tokens)
     
-    async def query(self, question: str) -> str:
-        response = await self.agent.run(question, ctx=self.ctx)
-        response = str(response)
+    def get_system_prompt(self) -> str:
+        """
+        Retrieves the system prompt.
 
-        final_answer = response
+        Returns:
+            str: The system prompt string.
+        """
+        return self.system_prompt
 
-        self.set_final_answer(final_answer)
-        return response
-    
-    def set_final_answer(self, final_answer: str) -> str:
+    async def query(self, question: str) -> str:
         """
-        Sets the final answer for the current querry.
+        Asynchronously queries the agent with a given question and returns the response.
 
         Args:
-            final_answer (str): The final answer to be set for the agent.
+            question (str): The question to be sent to the agent.
 
         Returns:
-            str: The final answer that was set.
+            str: The response from the agent as a string.
         """
-        print("-> set_final_answer !")
-        self.final_answer = final_answer
-    
-    def delegate_to_math_expert(self, question: str) -> str:
-        print("-> delegated to math agent !")
-        math_agent = MathExpert(temperature=0.7, max_tokens=100)
-        return math_agent.query(question)
-
-
-if __name__ == "__main__":
-    encryption_agent = EncryptionExpert(temperature=0.7, max_tokens=2000)
-    # encryption_query = "Descifer this: 'Bmfy bfx ymj wjxzqy gjybjjs z-hqzo fsi zsnajwxnyfyjf-hwfntaf ns fuwnq 2025 ?'"
-    encryption_query = ".rewsna eht sa ""tfel"" drow eht fo etisoppo eht etirw ,ecnetnes siht dnatsrednu uoy fI"
-    # print(encryption_agent.get_system_prompt())
-    # encoding = encryption_agent.caesar_cipher_encode(encryption_query, 5)
-    # print(encoding)
-    # print(encryption_agent.caesar_cipher_decode(encoding, 5))
-    print(asyncio.run(encryption_agent.query(encryption_query)))
+        response = await self.agent.run(question, ctx=self.ctx)
+        response = str(response)
+        return response
+
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        Also clears the context of all tool agents.
+        """
+        self.ctx = Context(self.agent)
+        # Clear context for all tool agents
+        self.summarizer.clear_context()
+        self.researcher.clear_context()
+        self.encryption_expert.clear_context()
+        self.math_expert.clear_context()
+        self.reasoner.clear_context()
+        self.image_handler.clear_context()
+        self.video_handler.clear_context()
+
+    async def call_summarizer(self, question: str) -> str:
+        return await self.summarizer.query(question)
+
+    async def call_researcher(self, question: str) -> str:
+        return await self.researcher.query(question)
+
+    async def call_encryption_expert(self, question: str) -> str:
+        return await self.encryption_expert.query(question)
+
+    async def call_math_expert(self, question: str) -> str:
+        return await self.math_expert.query(question)
+
+    async def call_reasoner(self, question: str) -> str:
+        return await self.reasoner.query(question)
+
+    async def call_image_handler(self, question: str) -> str:
+        # ImageHandler may not have a query method yet, but following the pattern
+        if hasattr(self.image_handler, 'query'):
+            return await self.image_handler.query(question)
+        return "Image handling is not implemented yet."
+        # TODO
+
+    async def call_video_handler(self, question: str) -> str:
+        # VideoHandler may not have a query method yet, but following the pattern
+        if hasattr(self.video_handler, 'query'):
+            return await self.video_handler.query(question)
+        return "Video handling is not implemented yet."
+        # TODO
+
+
+# if __name__ == "__main__":
+#     pass