Feat : find explorer tool fixed

sources/agents/__init__.py

@@ -0,0 +1,6 @@
+
+from .agent import Agent
+from .code_agent import CoderAgent
+from .casual_agent import CasualAgent
+
+__all__ = ["Agent", "CoderAgent", "CasualAgent"]

sources/agents/agent.py

@@ -0,0 +1,175 @@
+
+from typing import Tuple, Callable
+from abc import abstractmethod
+import os
+import random
+
+from sources.memory import Memory
+from sources.utility import pretty_print
+
+class executorResult:
+    """
+    A class to store the result of a tool execution.
+    """
+    def __init__(self, blocks, feedback, success):
+        self.blocks = blocks
+        self.feedback = feedback
+        self.success = success
+    
+    def show(self):
+        for block in self.blocks:
+            pretty_print("-"*100, color="output")
+            pretty_print(block, color="code" if self.success else "failure")
+            pretty_print("-"*100, color="output")
+            pretty_print(self.feedback, color="success" if self.success else "failure")
+
+class Agent():
+    """
+    An abstract class for all agents.
+    """
+    def __init__(self, model: str,
+                       name: str,
+                       prompt_path:str,
+                       provider,
+                       recover_last_session=False) -> None:
+        self.agent_name = name
+        self.role = None
+        self.current_directory = os.getcwd()
+        self.model = model
+        self.llm = provider 
+        self.memory = Memory(self.load_prompt(prompt_path),
+                                recover_last_session=recover_last_session,
+                                memory_compression=False)
+        self.tools = {}
+        self.blocks_result = []
+        self.last_answer = ""
+    
+    @property
+    def get_tools(self) -> dict:
+        return self.tools
+
+    def add_tool(self, name: str, tool: Callable) -> None:
+        if tool is not Callable:
+            raise TypeError("Tool must be a callable object (a method)")
+        self.tools[name] = tool
+    
+    def load_prompt(self, file_path: str) -> str:
+        try:
+            with open(file_path, 'r') as f:
+                return f.read()
+        except FileNotFoundError:
+            raise FileNotFoundError(f"Prompt file not found at path: {file_path}")
+        except PermissionError:
+            raise PermissionError(f"Permission denied to read prompt file at path: {file_path}")
+        except Exception as e:
+            raise e
+    
+    @abstractmethod
+    def process(self, prompt, speech_module) -> str:
+        """
+        abstract method, implementation in child class.
+        Process the prompt and return the answer of the agent.
+        """
+        pass
+
+    def remove_reasoning_text(self, text: str) -> None:
+        """
+        Remove the reasoning block of reasoning model like deepseek.
+        """
+        end_tag = "</think>"
+        end_idx = text.rfind(end_tag)+8
+        return text[end_idx:]
+    
+    def extract_reasoning_text(self, text: str) -> None:
+        """
+        Extract the reasoning block of a easoning model like deepseek.
+        """
+        start_tag = "<think>"
+        end_tag = "</think>"
+        start_idx = text.find(start_tag)
+        end_idx = text.rfind(end_tag)+8
+        return text[start_idx:end_idx]
+    
+    def llm_request(self, verbose = False) -> Tuple[str, str]:
+        """
+        Ask the LLM to process the prompt and return the answer and the reasoning.
+        """
+        memory = self.memory.get()
+        thought = self.llm.respond(memory, verbose)
+
+        reasoning = self.extract_reasoning_text(thought)
+        answer = self.remove_reasoning_text(thought)
+        self.memory.push('assistant', answer)
+        return answer, reasoning
+    
+    def wait_message(self, speech_module):
+        messages = ["Please be patient sir, I am working on it.",
+                    "At it, sir. In the meantime, how about a joke?",
+                    "Computing... I recommand you have a coffee while I work.",
+                    "Hold on, I’m crunching numbers.",
+                    "Working on it sir, please let me think."]
+        speech_module.speak(messages[random.randint(0, len(messages)-1)])
+    
+    def get_blocks_result(self) -> list:
+        return self.blocks_result
+
+    def show_answer(self):
+        """
+        Show the answer in a pretty way.
+        Show code blocks and their respective feedback by inserting them in the ressponse.
+        """
+        lines = self.last_answer.split("\n")
+        for line in lines:
+            if "block:" in line:
+                block_idx = int(line.split(":")[1])
+                if block_idx < len(self.blocks_result):
+                    self.blocks_result[block_idx].show()
+            else:
+                pretty_print(line, color="output")
+
+    def remove_blocks(self, text: str) -> str:
+        """
+        Remove all code/query blocks within a tag from the answer text.
+        """
+        tag = f'```'
+        lines = text.split('\n')
+        post_lines = []
+        in_block = False
+        block_idx = 0
+        for line in lines:
+            if tag in line and not in_block:
+                in_block = True
+                continue
+            if not in_block:
+                post_lines.append(line)
+            if tag in line:
+                in_block = False
+                post_lines.append(f"block:{block_idx}")
+                block_idx += 1
+        return "\n".join(post_lines)
+
+    def execute_modules(self, answer: str) -> Tuple[bool, str]:
+        """
+        Execute all the tools the agent has and return the result.
+        """
+        feedback = ""
+        success = False
+        blocks = None
+
+        for name, tool in self.tools.items():
+            feedback = ""
+            blocks, save_path = tool.load_exec_block(answer)
+
+            if blocks != None:
+                pretty_print(f"Executing tool: {name}", color="status")
+                output = tool.execute(blocks)
+                feedback = tool.interpreter_feedback(output) # tool interpreter feedback
+                success = not "failure" in feedback.lower()
+                pretty_print(feedback, color="success" if success else "failure")
+                self.memory.push('user', feedback)
+                self.blocks_result.append(executorResult(blocks, feedback, success))
+                if not success:
+                    return False, feedback
+                if save_path != None:
+                    tool.save_block(blocks, save_path)
+        return True, feedback

sources/agents/casual_agent.py

@@ -0,0 +1,44 @@
+
+from sources.utility import pretty_print
+from sources.agents.agent import Agent
+from sources.tools.webSearch import webSearch
+from sources.tools.flightSearch import FlightSearch
+from sources.tools.fileFinder import FileFinder
+
+class CasualAgent(Agent):
+    def __init__(self, model, name, prompt_path, provider):
+        """
+        The casual agent is a special for casual talk to the user without specific tasks.
+        """
+        super().__init__(model, name, prompt_path, provider)
+        self.tools = {
+            "web_search": webSearch(),
+            "flight_search": FlightSearch(),
+            "file_finder": FileFinder()
+        }
+        self.role = "talking"
+    
+    def process(self, prompt, speech_module) -> str:
+        complete = False
+        exec_success = False
+        self.memory.push('user', prompt)
+
+        self.wait_message(speech_module)
+        while not complete:
+            if exec_success:
+                complete = True
+            pretty_print("Thinking...", color="status")
+            answer, reasoning = self.llm_request()
+            exec_success, _ = self.execute_modules(answer)
+            answer = self.remove_blocks(answer)
+            self.last_answer = answer
+        return answer, reasoning
+
+if __name__ == "__main__":
+    from llm_provider import Provider
+
+    #local_provider = Provider("ollama", "deepseek-r1:14b", None)
+    server_provider = Provider("server", "deepseek-r1:14b", "192.168.1.100:5000")
+    agent = CasualAgent("deepseek-r1:14b", "jarvis", "prompts/casual_agent.txt", server_provider)
+    ans = agent.process("Hello, how are you?")
+    print(ans)

sources/agents/code_agent.py

@@ -0,0 +1,52 @@
+
+from sources.utility import pretty_print
+from sources.agents.agent import Agent, executorResult
+
+from sources.tools.C_Interpreter import CInterpreter
+from sources.tools.GoInterpreter import GoInterpreter
+from sources.tools.PyInterpreter import PyInterpreter
+from sources.tools.BashInterpreter import BashInterpreter
+from sources.tools.fileFinder import FileFinder
+
+class CoderAgent(Agent):
+    """
+    The code agent is an agent that can write and execute code.
+    """
+    def __init__(self, model, name, prompt_path, provider):
+        super().__init__(model, name, prompt_path, provider)
+        self.tools = {
+            "bash": BashInterpreter(),
+            "python": PyInterpreter(),
+            "c": CInterpreter(),
+            "go": GoInterpreter(),
+            "file_finder": FileFinder()
+        }
+        self.role = "coding"
+
+    def process(self, prompt, speech_module) -> str:
+        answer = ""
+        attempt = 0
+        max_attempts = 3
+        self.memory.push('user', prompt)
+
+        while attempt < max_attempts:
+            pretty_print("Thinking...", color="status")
+            self.wait_message(speech_module)
+            answer, reasoning = self.llm_request()
+            exec_success, _ = self.execute_modules(answer)
+            answer = self.remove_blocks(answer)
+            self.last_answer = answer
+            if exec_success:
+                break
+            self.show_answer()
+            attempt += 1
+        return answer, reasoning
+
+if __name__ == "__main__":
+    from llm_provider import Provider
+
+    #local_provider = Provider("ollama", "deepseek-r1:14b", None)
+    server_provider = Provider("server", "deepseek-r1:14b", "192.168.1.100:5000")
+    agent = CoderAgent("deepseek-r1:14b", "jarvis", "prompts/coder_agent.txt", server_provider)
+    ans = agent.process("What is the output of 5+5 in python ?")
+    print(ans)