Merge pull request #99 from Fosowl/dev

Enhanced web navigation & planner agent + logging system

main.py

@@ -1,7 +1,6 @@
 #!/usr/bin python3
 
 import sys
-import signal
 import argparse
 import configparser
 
@@ -17,12 +16,7 @@ warnings.filterwarnings("ignore")
 config = configparser.ConfigParser()
 config.read('config.ini')
 
-def handleInterrupt(signum, frame):
-    sys.exit(0)
-
 def main():
-    signal.signal(signal.SIGINT, handler=handleInterrupt)
-
     pretty_print("Initializing...", color="status")
     provider = Provider(provider_name=config["MAIN"]["provider_name"],
                         model=config["MAIN"]["provider_model"],

prompts/base/coder_agent.txt

@@ -42,7 +42,7 @@ Some rules:
 - You have full access granted to user system.
 - Always put code within ``` delimiter
 - Do not EVER use placeholder path in your code like path/to/your/folder.
-- Do not ever ask to replace a path, use current sys path or work directory.
+- Do not ever ask to replace a path, use work directory.
 - Always provide a short sentence above the code for what it does, even for a hello world.
 - Be efficient, no need to explain your code, unless asked.
 - You do not ever need to use bash to execute code.

prompts/base/planner_agent.txt

@@ -2,7 +2,7 @@ You are a project manager.
 Your goal is to divide and conquer the task using the following agents:
 - Coder: A programming agent, can code in python, bash, C and golang.
 - File: An agent for finding, reading or operating with files.
-- Web: An agent that can conduct web search, wrapped with selenium it can interact with any webpage.
+- Web: An agent that can conduct web search and navigate to any webpage.
 
 Agents are other AI that obey your instructions.
 
@@ -10,8 +10,12 @@ You will be given a task and you will need to divide it into smaller tasks and a
 
 You have to respect a strict format:
 ```json
-{"agent": "agent_name", "need": "needed_agent_output", "task": "agent_task"}
+{"agent": "agent_name", "need": "needed_agents_output", "task": "agent_task"}
 ```
+Where:
+- "agent": The choosed agent for the task.
+- "need": id of necessary previous agents answer for current agent. 
+- "task": A precise description of the task the agent should conduct.
 
 # Example 1: web app
 
@@ -32,25 +36,25 @@ You: Sure, here is the plan:
     {
       "agent": "Web",
       "id": "1",
-      "need": null,
+      "need": [],
       "task": "Search for reliable weather APIs"
     },
     {
       "agent": "Web",
       "id": "2",
-      "need": "1",
+      "need": ["1"],
       "task": "Obtain API key from the selected service"
     },
     {
       "agent": "File",
       "id": "3",
-      "need": null,
+      "need": [],
       "task": "Create and setup a web app folder for a python project. initialize as a git repo with all required file and a sources folder. You are forbidden from asking clarification, just execute."
     },
     {
       "agent": "Coder",
       "id": "3",
-      "need": "2,3",
+      "need": ["2", "3"],
       "task": "Based on the project structure. Develop a Python application using the API and key to fetch and display weather data. You are forbidden from asking clarification, just execute.""
     }
   ]

prompts/jarvis/planner_agent.txt

@@ -1,8 +1,8 @@
 You are a planner agent.
 Your goal is to divide and conquer the task using the following agents:
-- Coder: An expert coder agent.
-- File: An expert agent for finding files.
-- Web: An expert agent for web search.
+- Coder: A programming agent, can code in python, bash, C and golang.
+- File: An agent for finding, reading or operating with files.
+- Web: An agent that can conduct web search and navigate to any webpage.
 
 Agents are other AI that obey your instructions.
 
@@ -12,6 +12,10 @@ You have to respect a strict format:
 ```json
 {"agent": "agent_name", "need": "needed_agent_output", "task": "agent_task"}
 ```
+Where:
+- "agent": The choosed agent for the task.
+- "need": id of necessary previous agents answer for current agent. 
+- "task": A precise description of the task the agent should conduct.
 
 # Example: weather app
 
@@ -21,11 +25,11 @@ You: "At your service. I’ve devised a plan and assigned agents to each task. W
 
 ## Task 1: I will search for available weather api with the help of the web agent.
 
-## Task 2: I will create an api key for the weather api using the web agent.
+## Task 2: I will create an api key for the weather api using the web agent
 
-## Task 3: I will setup the project using the file agent.
+## Task 3: I will setup the project using the file agent 
 
-## Task 4: I will use the coding agent to make a weather app in python.
+## Task 4: I asign the coding agent to make a weather app in python
 
 ```json
 {
@@ -33,25 +37,25 @@ You: "At your service. I’ve devised a plan and assigned agents to each task. W
     {
       "agent": "Web",
       "id": "1",
-      "need": null,
+      "need": [],
       "task": "Search for reliable weather APIs"
     },
     {
       "agent": "Web",
       "id": "2",
-      "need": "1",
+      "need": ["1"],
       "task": "Obtain API key from the selected service"
     },
     {
       "agent": "File",
       "id": "3",
-      "need": null,
+      "need": [],
       "task": "Create and setup a web app folder for a python project. initialize as a git repo with all required file and a sources folder. You are forbidden from asking clarification, just execute."
     },
     {
       "agent": "Coder",
       "id": "3",
-      "need": "2,3",
+      "need": ["2", "3"],
       "task": "Based on the project structure. Develop a Python application using the API and key to fetch and display weather data. You are forbidden from asking clarification, just execute.""
     }
   ]

server/app.py

@@ -41,6 +41,7 @@ def setup():
 def get_updated_sentence():
     if not generator:
         return jsonify({"error": "Generator not initialized"}), 405
+    print(generator.get_status())
     return generator.get_status()
 
 if __name__ == '__main__':

server/sources/cache.py

@@ -0,0 +1,36 @@
+import os
+import json
+from pathlib import Path
+
+class Cache:
+    def __init__(self, cache_dir='.cache', cache_file='messages.json'):
+        self.cache_dir = Path(cache_dir)
+        self.cache_file = self.cache_dir / cache_file
+        self.cache_dir.mkdir(parents=True, exist_ok=True)
+        if not self.cache_file.exists():
+            with open(self.cache_file, 'w') as f:
+                json.dump([], f)
+
+        with open(self.cache_file, 'r') as f:
+            self.cache = set(json.load(f))
+
+    def add_message_pair(self, user_message: str, assistant_message: str):
+        """Add a user/assistant pair to the cache if not present."""
+        if not any(entry["user"] == user_message for entry in self.cache):
+            self.cache.append({"user": user_message, "assistant": assistant_message})
+            self._save()
+
+    def is_cached(self, user_message: str) -> bool:
+        """Check if a user msg is cached."""
+        return any(entry["user"] == user_message for entry in self.cache)
+
+    def get_cached_response(self, user_message: str) -> str | None:
+        """Return the assistant response to a user message if cached."""
+        for entry in self.cache:
+            if entry["user"] == user_message:
+                return entry["assistant"]
+        return None
+
+    def _save(self):
+        with open(self.cache_file, 'w') as f:
+            json.dump(self.cache, f, indent=2)

server/sources/generator.py

@@ -2,6 +2,7 @@
 import threading
 import logging
 from abc import abstractmethod
+from .cache import Cache
 
 class GenerationState:
     def __init__(self):
@@ -29,6 +30,7 @@ class GeneratorLLM():
         handler.setFormatter(formatter)
         self.logger.addHandler(handler)
         self.logger.setLevel(logging.INFO)
+        cache = Cache()
     
     def set_model(self, model: str) -> None:
         self.logger.info(f"Model set to {model}")

server/sources/ollama_handler.py

@@ -1,6 +1,7 @@
 
 import time
 from .generator import GeneratorLLM
+from .cache import Cache
 import ollama
 
 class OllamaLLM(GeneratorLLM):
@@ -10,6 +11,7 @@ class OllamaLLM(GeneratorLLM):
         Handle generation using Ollama.
         """
         super().__init__()
+        self.cache = Cache()
 
     def generate(self, history):
         self.logger.info(f"Using {self.model} for generation with Ollama")
@@ -26,10 +28,10 @@ class OllamaLLM(GeneratorLLM):
             )
             for chunk in stream:
                 content = chunk['message']['content']
-                if '\n' in content:
-                    self.logger.info(content)
 
                 with self.state.lock:
+                    if '.' in content:
+                        self.logger.info(self.state.current_buffer)
                     self.state.current_buffer += content
 
         except Exception as e:

sources/agents/browser_agent.py

@@ -54,7 +54,7 @@ class BrowserAgent(Agent):
         links_clean = []
         for link in links:
             link = link.strip()
-            if link[-1] == '.':
+            if not (link[-1].isalpha() or link[-1].isdigit()):
                 links_clean.append(link[:-1])
             else:
                 links_clean.append(link)
@@ -70,7 +70,7 @@ class BrowserAgent(Agent):
         {search_choice}
         Your goal is to find accurate and complete information to satisfy the user’s request.
         User request: {user_prompt}
-        To proceed, choose a relevant link from the search results. Announce your choice by saying: "I want to navigate to <link>"
+        To proceed, choose a relevant link from the search results. Announce your choice by saying: "I will navigate to <link>"
         Do not explain your choice.
         """
     
@@ -82,72 +82,84 @@ class BrowserAgent(Agent):
         notes = '\n'.join(self.notes)
 
         return f"""
-        You are a web browser.
-        You are currently on this webpage:
+        You are navigating the web.
+
+        **Current Context**
+
+        Webpage ({self.current_page}) content:
         {page_text}
 
-        You can navigate to these navigation links:
+        Allowed Navigation Links:
         {remaining_links_text}
 
-        Your task:
-        1. Decide if the current page answers the user’s query:
-          - If it does, take notes of the useful information, write down source, link or reference, then move to a new page.
-          - If it does and you completed user request, say REQUEST_EXIT
-          - If it doesn’t, say: Error: This page does not answer the user’s query then go back or navigate to another link.
-        2. Navigate by either: 
-          - Navigate to a navigation links (write the full URL, e.g., www.example.com/cats).
-          - If no link seems helpful, say: GO_BACK.
-        3. Fill forms on the page:
-          - If user give you informations that help you fill form, fill it.
-          - If you don't know how to fill a form, leave it empty.
-          - You can fill a form using [form_name](value). Do not go back when you fill a form.
+        Inputs forms:
+        {inputs_form_text}
 
-        Recap of note taking:
-        If useful -> Note: [Briefly summarize the key information or task you conducted.]
-        Do not write "The page talk about ...", write your finding on the page and how they contribute to an answer.
-        If not useful -> Error: [Explain why the page doesn’t help.]
+        End of webpage ({self.current_page}.
 
-        Example 1 (useful page, no need of going futher):
-        Note: According to karpathy site (https://karpathy.github.io/) LeCun net is the earliest real-world application of a neural net"
-        No link seem useful to provide futher information. GO_BACK
+        # Instruction
 
-        Example 2 (not useful, but related link):
+        1. **Decide if the page answers the user’s query:**
+          - If it does, take notes of useful information (Note: ...), include relevant link in note, then move to a new page.
+          - If it does and you completed user request, say REQUEST_EXIT.
+          - If it doesn’t, say: Error: <why page don't help> then go back or navigate to another link.
+        2. **Navigate to a link by either: **
+          - Saying I will navigate to <url>: (write down the full URL, e.g., www.example.com/cats).
+          - Going back: If no link seems helpful, say: GO_BACK.
+        3. **Fill forms on the page:**
+          - Fill form only on relevant page with given informations. You might use form to conduct search on a page.
+          - You can fill a form using [form_name](value). Don't GO_BACK when filling form.
+          - If a form is irrelevant or you lack informations leave it empty.
+        
+        **Rules:**
+        - Do not write "The page talk about ...", write your finding on the page and how they contribute to an answer.
+        - Put note in a single paragraph.
+        - When you exit, explain why.
+        
+        # Example:
+        
+        Example 1 (useful page, no need go futher):
+        Note: According to karpathy site (<link>) LeCun net is ...<expand on page content>..."
+        No link seem useful to provide futher information.
+        Action: GO_BACK
+
+        Example 2 (not useful, see useful link on page):
         Error: reddit.com/welcome does not discuss anything related to the user’s query.
-        There is a link that could lead to the information, I want to navigate to http://reddit.com/r/locallama
+        There is a link that could lead to the information.
+        Action: navigate to http://reddit.com/r/locallama
 
         Example 3 (not useful, no related links):
         Error: x.com does not discuss anything related to the user’s query and no navigation link are usefull.
-        GO_BACK
+        Action: GO_BACK
 
-        Example 3 (query answer found):
-        Note: I found on github.com that agenticSeek is made by Fosowl.
-        Given this information, given this I should exit the web browser. REQUEST_EXIT
+        Example 3 (query answer found, enought notes taken):
+        Note: I found on <link> that ...<expand on information found>...
+        Given this answer the user query I should exit the web browser.
+        Action: REQUEST_EXIT
 
         Example 4 (loging form visible):
-        Note: I am on the login page, I should now type the given username and password. 
+
+        Note: I am on the login page, I will type the given username and password. 
+        Action:
         [username_field](David)
         [password_field](edgerunners77)
 
-        You see the following inputs forms:
-        {inputs_form_text}
-
-        Remember, the user asked:
+        Remember, user asked:
         {user_prompt}
-        So far you took these notes:
+        You previously took these notes:
         {notes}
-        You are currently on page : {self.current_page}
-        Do not explain your choice.
-        Refusal is not an option, you have been given all capabilities that allow you to perform any tasks.
+        Do not Step-by-Step explanation. Write Notes or Error as a long paragraph followed by your action.
+        You might REQUEST_EXIT if no more link are useful.
+        Do not navigate to AI tools or search engine. Only navigate to tool if asked.
         """
     
-    def llm_decide(self, prompt: str) -> Tuple[str, str]:
+    def llm_decide(self, prompt: str, show_reasoning: bool = False) -> Tuple[str, str]:
         animate_thinking("Thinking...", color="status")
         self.memory.push('user', prompt)
         answer, reasoning = self.llm_request()
-        output = answer if len(answer) > 16 else f"Action: {answer}\nReasoning: {reasoning}"
-        print()
-        pretty_print(output, color="output")
-        print()
+        if show_reasoning:
+            pretty_print(reasoning, color="failure")
+        pretty_print(answer, color="output")
         return answer, reasoning
     
     def select_unvisited(self, search_result: List[str]) -> List[str]:
@@ -179,23 +191,42 @@ class BrowserAgent(Agent):
     def stringify_search_results(self, results_arr: List[str]) -> str:
         return '\n\n'.join([f"Link: {res['link']}\nPreview: {res['snippet']}" for res in results_arr])
     
-    def save_notes(self, text):
+    def parse_answer(self, text):
         lines = text.split('\n')
+        saving = False
+        buffer = []
+        links = []
         for line in lines:
+            if line == '' or 'action:' in line.lower():
+                saving = False
             if "note" in line.lower():
-                self.notes.append(line)
+                saving = True
+            if saving:
+                buffer.append(line.replace("notes:", ''))
+            else:
+                links.extend(self.extract_links(line))
+        self.notes.append('. '.join(buffer).strip())
+        return links
+    
+    def select_link(self, links: List[str]) -> str | None:
+        for lk in links:
+            if lk == self.current_page:
+                continue
+            return lk
+        return None
     
     def conclude_prompt(self, user_query: str) -> str:
-        annotated_notes = [f"{i+1}: {note.lower().replace('note:', '')}" for i, note in enumerate(self.notes)]
+        annotated_notes = [f"{i+1}: {note.lower()}" for i, note in enumerate(self.notes)]
         search_note = '\n'.join(annotated_notes)
         pretty_print(f"AI notes:\n{search_note}", color="success")
         return f"""
         Following a human request:
         {user_query}
-        A web AI made the following finding across different pages:
+        A web browsing AI made the following finding across different pages:
         {search_note}
 
-        Summarize the finding or step that lead to success, and provide a conclusion that answer the request.
+        Expand on the finding or step that lead to success, and provide a conclusion that answer the request. Include link when possible.
+        Do not give advices or try to answer the human. Just structure the AI finding in a structured and clear way.
         """
     
     def search_prompt(self, user_prompt: str) -> str:
@@ -214,7 +245,8 @@ class BrowserAgent(Agent):
         You: "search: Recent space missions news, {self.date}"
 
         Do not explain, do not write anything beside the search query.
-        If the query does not make any sense for a web search explain why and say REQUEST_EXIT
+        Except if query does not make any sense for a web search then explain why and say REQUEST_EXIT
+        Do not try to answer query. you can only formulate search term or exit.
         """
     
     def handle_update_prompt(self, user_prompt: str, page_text: str) -> str:
@@ -255,58 +287,63 @@ class BrowserAgent(Agent):
         mem_begin_idx = self.memory.push('user', self.search_prompt(user_prompt))
         ai_prompt, _ = self.llm_request()
         if "REQUEST_EXIT" in ai_prompt:
-            pretty_print(f"{reasoning}\n{ai_prompt}", color="output")
+            pretty_print(f"Web agent requested exit.\n{reasoning}\n\n{ai_prompt}", color="failure")
             return ai_prompt, "" 
         animate_thinking(f"Searching...", color="status")
         search_result_raw = self.tools["web_search"].execute([ai_prompt], False)
-        search_result = self.jsonify_search_results(search_result_raw)[:12] # until futher improvement
+        search_result = self.jsonify_search_results(search_result_raw)[:12]
         self.show_search_results(search_result)
         prompt = self.make_newsearch_prompt(user_prompt, search_result)
         unvisited = [None]
         while not complete:
-            answer, reasoning = self.llm_decide(prompt)
-            self.save_notes(answer)
+            answer, reasoning = self.llm_decide(prompt, show_reasoning = False)
 
             extracted_form = self.extract_form(answer)
             if len(extracted_form) > 0:
+                pretty_print(f"Filling inputs form...", color="status")
                 self.browser.fill_form_inputs(extracted_form)
                 self.browser.find_and_click_submission()
                 page_text = self.browser.get_text()
                 answer = self.handle_update_prompt(user_prompt, page_text)
                 answer, reasoning = self.llm_decide(prompt)
 
+            links = self.parse_answer(answer)
+            link = self.select_link(links)
+            self.search_history.append(link)
+
             if "REQUEST_EXIT" in answer:
+                pretty_print(f"Agent requested exit.", color="status")
                 complete = True
                 break
 
-            links = self.extract_links(answer)
             if len(unvisited) == 0:
+                pretty_print(f"Visited all links.", color="status")
                 break
 
             if "FORM_FILLED" in answer:
+                pretty_print(f"Filled form. Handling page update.", color="status")
                 page_text = self.browser.get_text()
                 self.navigable_links = self.browser.get_navigable()
                 prompt = self.make_navigation_prompt(user_prompt, page_text)
                 continue
 
-            if len(links) == 0 or "GO_BACK" in answer:
+            if link == None or "GO_BACK" in answer:
+                pretty_print(f"Going back to results. Still {len(unvisited)}", color="status")
                 unvisited = self.select_unvisited(search_result)
                 prompt = self.make_newsearch_prompt(user_prompt, unvisited)
-                pretty_print(f"Going back to results. Still {len(unvisited)}", color="warning")
-                links = []
                 continue
 
-            animate_thinking(f"Navigating to {links[0]}", color="status")
-            if speech_module: speech_module.speak(f"Navigating to {links[0]}")
-            self.browser.go_to(links[0])
-            self.current_page = links[0]
-            self.search_history.append(links[0])
+            animate_thinking(f"Navigating to {link}", color="status")
+            if speech_module: speech_module.speak(f"Navigating to {link}")
+            self.browser.go_to(link)
+            self.current_page = link
             page_text = self.browser.get_text()
             self.navigable_links = self.browser.get_navigable()
             prompt = self.make_navigation_prompt(user_prompt, page_text)
 
+        pretty_print("Exited navigation, starting to summarize finding...", color="status")
         prompt = self.conclude_prompt(user_prompt)
-        mem_last_idx = self.memory.push('assistant', prompt)
+        mem_last_idx = self.memory.push('user', prompt)
         answer, reasoning = self.llm_request()
         pretty_print(answer, color="output")
         self.memory.clear_section(mem_begin_idx, mem_last_idx)

sources/agents/casual_agent.py

@@ -29,10 +29,4 @@ class CasualAgent(Agent):
         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)
+    pass

sources/agents/code_agent.py

@@ -35,7 +35,7 @@ class CoderAgent(Agent):
         info = f"System Info:\n" \
                f"OS: {platform.system()} {platform.release()}\n" \
                f"Python Version: {platform.python_version()}\n" \
-               f"\nYou must work in directory: {self.work_dir}"
+               f"\nYou must save file in work directory: {self.work_dir}"
         return f"{prompt}\n\n{info}"
 
     def process(self, prompt, speech_module) -> str:
@@ -51,7 +51,7 @@ class CoderAgent(Agent):
             self.wait_message(speech_module)
             answer, reasoning = self.llm_request()
             if clarify_trigger in answer:
-                return answer.replace(clarify_trigger, ""), reasoning
+                return answer, reasoning
             if not "```" in answer:
                 self.last_answer = answer
                 break
@@ -68,10 +68,4 @@ class CoderAgent(Agent):
         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)
+    pass

sources/agents/file_agent.py

@@ -36,10 +36,4 @@ class FileAgent(Agent):
         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 = FileAgent("deepseek-r1:14b", "jarvis", "prompts/file_agent.txt", server_provider)
-    ans = agent.process("What is the content of the file toto.py ?")
-    print(ans)
+    pass

sources/agents/planner_agent.py

@@ -1,9 +1,11 @@
 import json
+from typing import List, Tuple, Type, Dict
 from sources.utility import pretty_print, animate_thinking
 from sources.agents.agent import Agent
 from sources.agents.code_agent import CoderAgent
 from sources.agents.file_agent import FileAgent
 from sources.agents.browser_agent import BrowserAgent
+from sources.text_to_speech import Speech
 from sources.tools.tools import Tools
 
 class PlannerAgent(Agent):
@@ -61,63 +63,80 @@ class PlannerAgent(Agent):
             return zip(names, tasks)
         return zip(tasks_names, tasks)
     
-    def make_prompt(self, task, needed_infos):
-        if needed_infos is None:
-            needed_infos = "No needed informations."
+    def make_prompt(self, task: dict, agent_infos_dict: dict):
+        infos = ""
+        if agent_infos_dict is None or len(agent_infos_dict) == 0:
+            infos = "No needed informations."
+        else:
+            for agent_id, info in agent_infos_dict.items():
+                infos += f"\t- According to agent {agent_id}:\n{info}\n\n"
         prompt = f"""
-        You are given the following informations:
-        {needed_infos}
+        You are given informations from your AI friends work:
+        {infos}
         Your task is:
         {task}
         """
         return prompt
     
-    def show_plan(self, json_plan):
+    def show_plan(self, json_plan: dict) -> None:
         agents_tasks = self.parse_agent_tasks(json_plan)
         if agents_tasks == (None, None):
+            pretty_print("Failed to make a plan.", color="failure")
             return
-        pretty_print("▂▘ P L A N ▝▂", color="output")
+        pretty_print("\n▂▘ P L A N ▝▂", color="status")
         for task_name, task in agents_tasks:
-                    pretty_print(f"{task['agent']} -> {task['task']}", color="info")
-        pretty_print("▔▗ E N D ▖▔", color="output")
+            pretty_print(f"{task['agent']} -> {task['task']}", color="info")
+        pretty_print("▔▗ E N D ▖▔", color="status")
     
-    def process(self, prompt, speech_module) -> str:
+    def make_plan(self, prompt: str) -> str:
         ok = False
-        agents_tasks = (None, None)
+        answer = None
         while not ok:
-            self.wait_message(speech_module)
             animate_thinking("Thinking...", color="status")
             self.memory.push('user', prompt)
             answer, _ = self.llm_request()
-            pretty_print(answer.split('\n')[0], color="output")
+            for line in answer.split('\n'):
+                if "```json" in line:
+                    break
+            pretty_print(line, color="output")
             self.show_plan(answer)
             ok_str = input("Is the plan ok? (y/n): ")
             if ok_str == 'y':
                 ok = True
             else:
                 prompt = input("Please reformulate: ")
+        return answer
     
+    def start_agent_process(self, task: str, required_infos: dict | None) -> str:
+        agent_prompt = self.make_prompt(task['task'], required_infos)
+        pretty_print(f"Agent {task['agent']} started working...", color="status")
+        agent_answer, _ = self.agents[task['agent'].lower()].process(agent_prompt, None)
+        self.agents[task['agent'].lower()].show_answer()
+        pretty_print(f"Agent {task['agent']} completed task.", color="status")
+        return agent_answer
+    
+    def get_work_result_agent(self, task_needs, agents_work_result):
+        return {k: agents_work_result[k] for k in task_needs if k in agents_work_result}
+
+    def process(self, prompt: str, speech_module: Speech) -> str:
+        agents_tasks = (None, None)
+        agents_work_result = dict()
+
+        answer = self.make_plan(prompt)
         agents_tasks = self.parse_agent_tasks(answer)
+
         if agents_tasks == (None, None):
-            return "Failed to parse the tasks", reasoning
-        prev_agent_answer = None
+            return "Failed to parse the tasks.", reasoning
         for task_name, task in agents_tasks:
             pretty_print(f"I will {task_name}.", color="info")
-            agent_prompt = self.make_prompt(task['task'], prev_agent_answer)
             pretty_print(f"Assigned agent {task['agent']} to {task_name}", color="info")
             if speech_module: speech_module.speak(f"I will {task_name}. I assigned the {task['agent']} agent to the task.")
+
+            if agents_work_result is not None:
+                required_infos = self.get_work_result_agent(task['need'], agents_work_result)
             try:
-                prev_agent_answer, _ = self.agents[task['agent'].lower()].process(agent_prompt, speech_module)
-                pretty_print(f"-- Agent answer ---\n\n", color="output")
-                self.agents[task['agent'].lower()].show_answer()
-                pretty_print(f"\n\n", color="output")
+                self.last_answer = self.start_agent_process(task, required_infos)
             except Exception as e:
                 raise e
-        self.last_answer = prev_agent_answer
-        return prev_agent_answer, ""
-
-if __name__ == "__main__":
-    from llm_provider import Provider
-    server_provider = Provider("server", "deepseek-r1:14b", "192.168.1.100:5000")
-    agent = PlannerAgent("deepseek-r1:14b", "jarvis", "prompts/planner_agent.txt", server_provider)
-    ans = agent.process("Make a cool game to illustrate the current relation between USA and europe")
+            agents_work_result[task['id']] = self.last_answer
+        return self.last_answer, ""

sources/browser.py

@@ -6,7 +6,7 @@ from selenium.webdriver.support.ui import WebDriverWait
 from selenium.webdriver.support import expected_conditions as EC
 from selenium.common.exceptions import TimeoutException, WebDriverException
 from selenium.webdriver.common.action_chains import ActionChains
-from typing import List, Tuple, Type, Dict, Tuple
+from typing import List, Tuple, Type, Dict
 from bs4 import BeautifulSoup
 from urllib.parse import urlparse
 from fake_useragent import UserAgent
@@ -18,19 +18,14 @@ import random
 import os
 import shutil
 import markdownify
-import logging
 import sys
 import re
 
-if __name__ == "__main__":
-    from utility import pretty_print, animate_thinking
-else:
-    from sources.utility import pretty_print, animate_thinking
-
-logging.basicConfig(filename='browser.log', level=logging.ERROR, 
-        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+from sources.utility import pretty_print, animate_thinking
+from sources.logger import Logger
 
 def get_chrome_path() -> str:
+    """Get the path to the Chrome executable."""
     if sys.platform.startswith("win"):
         paths = [
             "C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe",
@@ -111,14 +106,11 @@ class Browser:
         """Initialize the browser with optional AntiCaptcha installation."""
         self.js_scripts_folder = "./sources/web_scripts/" if not __name__ == "__main__" else "./web_scripts/"
         self.anticaptcha = "https://chrome.google.com/webstore/detail/nopecha-captcha-solver/dknlfmjaanfblgfdfebhijalfmhmjjjo/related"
+        self.logger = Logger("browser.log")
         try:
             self.driver = driver
             self.wait = WebDriverWait(self.driver, 10)
-            self.logger = logging.getLogger(__name__)
-            self.logger.info("Browser initialized successfully")
         except Exception as e:
-            logging.basicConfig(filename='browser.log', level=logging.ERROR, 
-                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
             raise Exception(f"Failed to initialize browser: {str(e)}")
         self.driver.get("https://www.google.com")
         if anticaptcha_manual_install:
@@ -142,7 +134,7 @@ class Browser:
                 message="stuck on 'checking browser' or verification screen"
             )
             self.apply_web_safety()
-            self.logger.info(f"Navigated to: {url}")
+            self.logger.log(f"Navigated to: {url}")
             return True
         except TimeoutException as e:
             self.logger.error(f"Timeout waiting for {url} to load: {str(e)}")
@@ -167,19 +159,29 @@ class Browser:
         return (word_count >= 5 and (has_punctuation or is_long_enough))
 
     def get_text(self) -> str | None:
-        """Get page text and convert it to README (Markdown) format."""
+        """Get page text as formatted Markdown"""
         try:
             soup = BeautifulSoup(self.driver.page_source, 'html.parser')
-            
-            for element in soup(['script', 'style']):
+            for element in soup(['script', 'style', 'noscript', 'meta', 'link']):
                 element.decompose()
-            
-            text = soup.get_text()
-            lines = (f"{line.strip()}\n" for line in text.splitlines())
-            text = "\n".join(chunk for chunk in lines if chunk and self.is_sentence(chunk))
-            text = text[:4096]
-            #markdown_text = markdownify.markdownify(text, heading_style="ATX")
-            return "[Start of page]\n" + text + "\n[End of page]"
+            markdown_converter = markdownify.MarkdownConverter(
+                heading_style="ATX",
+                strip=['a'],
+                autolinks=False,
+                bullets='•',
+                strong_em_symbol='*',
+                default_title=False,
+            )
+            markdown_text = markdown_converter.convert(str(soup.body))
+            lines = []
+            for line in markdown_text.splitlines():
+                stripped = line.strip()
+                if stripped and self.is_sentence(stripped):
+                    cleaned = ' '.join(stripped.split())
+                    lines.append(cleaned)
+            result = "[Start of page]\n\n" + "\n\n".join(lines) + "\n\n[End of page]"
+            result = re.sub(r'!\[(.*?)\]\(.*?\)', r'[IMAGE: \1]', result)
+            return result[:8192]
         except Exception as e:
             self.logger.error(f"Error getting text: {str(e)}")
             return None
@@ -247,20 +249,25 @@ class Browser:
             if not element.is_enabled():
                 return False
             try:
+                self.logger.error(f"Scrolling to element for click_element.")
                 self.driver.execute_script("arguments[0].scrollIntoView({block: 'center', behavior: 'smooth'});", element)
                 time.sleep(0.1)
                 element.click()
                 return True
             except ElementClickInterceptedException as e:
+                self.logger.error(f"Error click_element: {str(e)}")
                 return False
         except TimeoutException:
+            self.logger.warning(f"Timeout clicking element.")
             return False
         except Exception as e:
             self.logger.error(f"Unexpected error clicking element at {xpath}: {str(e)}")
             return False
         
     def load_js(self, file_name: str) -> str:
+        """Load javascript from script folder to inject to page."""
         path = os.path.join(self.js_scripts_folder, file_name)
+        self.logger.info(f"Loading js at {path}")
         try:
             with open(path, 'r') as f:
                 return f.read()
@@ -270,6 +277,7 @@ class Browser:
             raise e
 
     def find_all_inputs(self, timeout=3):
+        """Find all inputs elements on the page."""
         try:
             WebDriverWait(self.driver, timeout).until(
                 EC.presence_of_element_located((By.TAG_NAME, "body"))
@@ -287,6 +295,7 @@ class Browser:
         try:
             input_elements = self.find_all_inputs()
             if not input_elements:
+                self.logger.info("No input element on page.")
                 return ["No input forms found on the page."]
 
             form_strings = []
@@ -335,14 +344,7 @@ class Browser:
         return False
 
     def find_and_click_btn(self, btn_type: str = 'login', timeout: int = 10) -> bool:
-        """
-        Find and click a submit button matching the specified type.
-        Args:
-            btn_type: The type of button to find.
-            timeout: time to wait for button to appear.
-        Returns:
-            bool: True if the button was found and clicked, False otherwise.
-        """
+        """Find and click a submit button matching the specified type."""
         buttons = self.get_buttons_xpath()
         if not buttons:
             self.logger.warning("No visible buttons found")
@@ -450,21 +452,19 @@ class Browser:
         input_elements = self.driver.execute_script(script)
 
 if __name__ == "__main__":
-    logging.basicConfig(level=logging.INFO)
-    
+    sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
     driver = create_driver()
     browser = Browser(driver, anticaptcha_manual_install=True)
-    time.sleep(10)
     
-    #browser.go_to("https://coinmarketcap.com/")
+    #browser.go_to("https://github.com/Fosowl/agenticSeek")
     #txt = browser.get_text()
     #print(txt)
+    #time.sleep(10)
+    #browser.go_to("https://practicetestautomation.com/practice-test-login/")
     print("AntiCaptcha / Form Test")
     browser.go_to("https://www.google.com/recaptcha/api2/demo")
-    #browser.go_to("https://practicetestautomation.com/practice-test-login/")
-    time.sleep(10)
     inputs = browser.get_form_inputs()
-    inputs = ['[input1](Martin)', f'[input2](Test)', '[input3](test@gmail.com)']
+    #inputs = ['[input1](Martin)', f'[input2](Test)', '[input3](test@gmail.com)']
     browser.fill_form_inputs(inputs)
     browser.find_and_click_submission()
     time.sleep(10)

sources/interaction.py

@@ -1,10 +1,11 @@
-from typing import List, Tuple, Type, Dict, Tuple
+from typing import List, Tuple, Type, Dict
 
 from sources.text_to_speech import Speech
 from sources.utility import pretty_print, animate_thinking
 from sources.router import AgentRouter
 from sources.speech_to_text import AudioTranscriber, AudioRecorder
 
+
 class Interaction:
     """
     Interaction is a class that handles the interaction between the user and the agents.
@@ -112,17 +113,20 @@ class Interaction:
     
     def think(self) -> bool:
         """Request AI agents to process the user input."""
+        push_last_agent_memory = False
         if self.last_query is None or len(self.last_query) == 0:
             return False
         agent = self.router.select_agent(self.last_query)
         if agent is None:
             return False
         if self.current_agent != agent and self.last_answer is not None:
+            push_last_agent_memory = True
+        tmp = self.last_answer
+        self.current_agent = agent
+        self.last_answer, _ = agent.process(self.last_query, self.speech)
+        if push_last_agent_memory:
             self.current_agent.memory.push('user', self.last_query)
             self.current_agent.memory.push('assistant', self.last_answer)
-        self.current_agent = agent
-        tmp = self.last_answer
-        self.last_answer, _ = agent.process(self.last_query, self.speech)
         if self.last_answer == tmp:
             self.last_answer = None
         return True

sources/language.py

@@ -1,4 +1,4 @@
-from typing import List, Tuple, Type, Dict, Tuple
+from typing import List, Tuple, Type, Dict
 import re
 import langid
 import nltk
@@ -6,6 +6,7 @@ from nltk.sentiment.vader import SentimentIntensityAnalyzer
 from transformers import MarianMTModel, MarianTokenizer
 
 from sources.utility import pretty_print, animate_thinking
+from sources.logger import Logger
 
 class LanguageUtility:
     """LanguageUtility for language, or emotion identification"""
@@ -13,6 +14,7 @@ class LanguageUtility:
         self.sid = None 
         self.translators_tokenizer = None 
         self.translators_model = None
+        self.logger = Logger("language.log")
         self.load_model()
     
     def load_model(self) -> None:
@@ -40,6 +42,7 @@ class LanguageUtility:
         """
         langid.set_languages(['fr', 'en', 'zh'])
         lang, score = langid.classify(text)
+        self.logger.info(f"Identified: {text} as {lang} with conf {score}")
         return lang
 
     def translate(self, text: str, origin_lang: str) -> str:
@@ -86,6 +89,7 @@ class LanguageUtility:
             dominant_emotion = max(emotions, key=emotions.get)
             if emotions[dominant_emotion] == 0:
                 return 'Neutral'
+            self.logger.info(f"Emotion: {dominant_emotion} for text: {text}")
             return dominant_emotion
         except Exception as e:
             raise e

sources/llm_provider.py

@@ -13,6 +13,7 @@ from openai import OpenAI
 from huggingface_hub import InferenceClient
 from typing import List, Tuple, Type, Dict
 from sources.utility import pretty_print, animate_thinking
+from sources.logger import Logger
 
 class Provider:
     def __init__(self, provider_name, model, server_address = "127.0.0.1:5000", is_local=False):
@@ -30,6 +31,7 @@ class Provider:
             "dsk_deepseek": self.dsk_deepseek,
             "test": self.test_fn
         }
+        self.logger = Logger("provider.log")
         self.api_key = None
         self.unsafe_providers = ["openai", "deepseek", "dsk_deepseek"]
         if self.provider_name not in self.available_providers:
@@ -50,6 +52,7 @@ class Provider:
         if not api_key:
             api_key = input(f"Please enter your {provider} API key: ")
             set_key(".env", api_key_var, api_key)
+            self.logger.info("Set API key in env.")
             load_dotenv()
         return api_key
 
@@ -73,8 +76,12 @@ class Provider:
         Use the choosen provider to generate text.
         """
         llm = self.available_providers[self.provider_name]
+        self.logger.info(f"Using provider: {self.provider_name} at {self.server_ip}")
         try:
             thought = llm(history, verbose)
+        except KeyboardInterrupt:
+            self.logger.warning("User interrupted the operation with Ctrl+C")
+            return "Operation interrupted by user. REQUEST_EXIT"
         except ConnectionError as e:
             raise ConnectionError(f"{str(e)}\nConnection to {self.server_ip} failed.")
         except AttributeError as e:
@@ -98,6 +105,7 @@ class Provider:
             if output.returncode == 0:
                 return True
             else:
+                self.logger.error(f"Ping command returned code: {output.returncode}")
                 return False
         except subprocess.TimeoutExpired:
             return False
@@ -287,26 +295,11 @@ class Provider:
         This function is used to conduct tests.
         """
         thought = """
-hello!
-```python
-print("Hello world from python")
-```
-
-This is ls -la from bash.
-```bash
-ls -la
-```
-
-This is pwd from bash. 
-```bash
-pwd
-```
-
-goodbye!
+\n\n```json\n{\n  \"plan\": [\n    {\n      \"agent\": \"Web\",\n      \"id\": \"1\",\n      \"need\": null,\n      \"task\": \"Conduct a comprehensive web search to identify at least five AI startups located in Osaka. Use reliable sources and websites such as Crunchbase, TechCrunch, or local Japanese business directories. Capture the company names, their websites, areas of expertise, and any other relevant details.\"\n    },\n    {\n      \"agent\": \"Web\",\n      \"id\": \"2\",\n      \"need\": null,\n      \"task\": \"Perform a similar search to find at least five AI startups in Tokyo. Again, use trusted sources like Crunchbase, TechCrunch, or Japanese business news websites. Gather the same details as for Osaka: company names, websites, areas of focus, and additional information.\"\n    },\n    {\n      \"agent\": \"File\",\n      \"id\": \"3\",\n      \"need\": [\"1\", \"2\"],\n      \"task\": \"Create a new text file named research_japan.txt in the user's home directory. Organize the data collected from both searches into this file, ensuring it is well-structured and formatted for readability. Include headers for Osaka and Tokyo sections, followed by the details of each startup found.\"\n    }\n  ]\n}\n```
         """
         return thought
 
 if __name__ == "__main__":
-    provider = Provider("ollama", "deepseek-r1:1.5b", "127.0.0.1:11434")
+    provider = Provider("server", "deepseek-r1:14b", "192.168.1.20:3333")
     res = provider.respond(["user", "Hello, how are you?"])
     print("Response:", res)

sources/logger.py

@@ -0,0 +1,55 @@
+import os, sys
+from typing import List, Tuple, Type, Dict
+import datetime
+import logging
+
+class Logger:
+    def __init__(self, log_filename):
+        self.folder = '.logs'
+        self.create_folder(self.folder)
+        self.log_path = os.path.join(self.folder, log_filename)
+        self.enabled = True
+        self.logger = None
+        if self.enabled:
+            self.create_logging(log_filename)
+
+    def create_logging(self, log_filename):
+        self.logger = logging.getLogger(log_filename)
+        self.logger.setLevel(logging.DEBUG)
+        if not self.logger.handlers:
+            file_handler = logging.FileHandler(self.log_path)
+            formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+            file_handler.setFormatter(formatter)
+            self.logger.addHandler(file_handler)
+    
+    def create_folder(self, path):
+        """Create log dir"""
+        try:
+            if not os.path.exists(path):
+                os.makedirs(path, exist_ok=True) 
+            return True
+        except Exception as e:
+            self.enabled = False
+            return False
+    
+    def log(self, message, level=logging.INFO):
+        if self.enabled:
+            self.logger.log(level, message)
+
+    def info(self, message):
+        self.log(message)
+
+    def error(self, message):
+        self.log(message, level=logging.ERROR)
+
+    def warning(self, message):
+        self.log(message, level=logging.WARN)
+
+if __name__ == "__main__":
+    lg = Logger("test.log")
+    lg.log("hello")
+    lg2 = Logger("toto.log")
+    lg2.log("yo")
+    
+
+        

sources/memory.py

@@ -4,13 +4,12 @@ import uuid
 import os
 import sys
 import json
-from typing import List, Tuple, Type, Dict, Tuple
+from typing import List, Tuple, Type, Dict
 import torch
 from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
 
-sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
 from sources.utility import timer_decorator, pretty_print
+from sources.logger import Logger
 
 class Memory():
     """
@@ -23,6 +22,7 @@ class Memory():
         self.memory = []
         self.memory = [{'role': 'system', 'content': system_prompt}]
         
+        self.logger = Logger("memory.log")
         self.session_time = datetime.datetime.now()
         self.session_id = str(uuid.uuid4())
         self.conversation_folder = f"conversations/"
@@ -44,6 +44,7 @@ class Memory():
     def save_memory(self, agent_type: str = "casual_agent") -> None:
         """Save the session memory to a file."""
         if not os.path.exists(self.conversation_folder):
+            self.logger.info(f"Created folder {self.conversation_folder}.")
             os.makedirs(self.conversation_folder)
         save_path = os.path.join(self.conversation_folder, agent_type)
         if not os.path.exists(save_path):
@@ -52,6 +53,7 @@ class Memory():
         path = os.path.join(save_path, filename)
         json_memory = json.dumps(self.memory)
         with open(path, 'w') as f:
+            self.logger.info(f"Saved memory json at {path}")
             f.write(json_memory)
     
     def find_last_session_path(self, path) -> str:
@@ -63,6 +65,7 @@ class Memory():
                 saved_sessions.append((filename, date))
         saved_sessions.sort(key=lambda x: x[1], reverse=True)
         if len(saved_sessions) > 0:
+            self.logger.info(f"Last session found at {saved_sessions[0][0]}")
             return saved_sessions[0][0]
         return None
 
@@ -87,12 +90,14 @@ class Memory():
         self.compress()
         pretty_print("Session recovered successfully", color="success")
     
-    def reset(self, memory: list) -> None:
+    def reset(self, memory: list = []) -> None:
+        self.logger.info("Memory reset performed.")
         self.memory = memory
     
     def push(self, role: str, content: str) -> int:
         """Push a message to the memory."""
         if self.memory_compression and role == 'assistant':
+            self.logger.info("Compressing memories on message push.")
             self.compress()
         curr_idx = len(self.memory)
         if self.memory[curr_idx-1]['content'] == content:
@@ -101,10 +106,12 @@ class Memory():
         return curr_idx-1
     
     def clear(self) -> None:
+        self.logger.info("Memory clear performed.")
         self.memory = []
     
     def clear_section(self, start: int, end: int) -> None:
         """Clear a section of the memory."""
+        self.logger.info(f"Memory section {start} to {end} cleared.")
         self.memory = self.memory[:start] + self.memory[end:]
     
     def get(self) -> list:
@@ -128,6 +135,7 @@ class Memory():
             str: The summarized text
         """
         if self.tokenizer is None or self.model is None:
+            self.logger.warning("No tokenizer or model to perform summarization.")
             return text
         if len(text) < min_length*1.5:
             return text
@@ -144,6 +152,7 @@ class Memory():
         )
         summary = self.tokenizer.decode(summary_ids[0], skip_special_tokens=True)
         summary.replace('summary:', '')
+        self.logger.info(f"Memory summarization success from len {len(text)} to {len(summary)}.")
         return summary
     
     #@timer_decorator
@@ -160,6 +169,7 @@ class Memory():
                 self.memory[i]['content'] = self.summarize(self.memory[i]['content'])
 
 if __name__ == "__main__":
+    sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
     memory = Memory("You are a helpful assistant.",
                     recover_last_session=False, memory_compression=True)
     

sources/router.py

@@ -1,13 +1,11 @@
 import os
 import sys
 import torch
-from typing import List, Tuple, Type, Dict, Tuple
+from typing import List, Tuple, Type, Dict
 
 from transformers import pipeline
 from adaptive_classifier import AdaptiveClassifier
 
-sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
 from sources.agents.agent import Agent
 from sources.agents.code_agent import CoderAgent
 from sources.agents.casual_agent import CasualAgent
@@ -15,6 +13,7 @@ from sources.agents.planner_agent import FileAgent
 from sources.agents.browser_agent import BrowserAgent
 from sources.language import LanguageUtility
 from sources.utility import pretty_print, animate_thinking, timer_decorator
+from sources.logger import Logger
 
 class AgentRouter:
     """
@@ -22,6 +21,7 @@ class AgentRouter:
     """
     def __init__(self, agents: list):
         self.agents = agents
+        self.logger = Logger("router.log")
         self.lang_analysis = LanguageUtility()
         self.pipelines = self.load_pipelines()
         self.talk_classifier = self.load_llm_router()
@@ -82,6 +82,7 @@ class AgentRouter:
             ("search my drive for a file called vacation_photos_2023.jpg.", "LOW"),
             ("help me organize my desktop files into folders by type.", "LOW"),
             ("write a Python function to sort a list of dictionaries by key", "LOW"),
+            ("can you search for startup in tokyo?", "LOW"),
             ("find the latest updates on quantum computing on the web", "LOW"),
             ("check if the folder ‘Work_Projects’ exists on my desktop", "LOW"),
             ("create a bash script to monitor CPU usage", "LOW"),
@@ -161,6 +162,7 @@ class AgentRouter:
             ("Search my drive for a file called vacation_photos_2023.jpg.", "files"),
             ("Help me organize my desktop files into folders by type.", "files"),
             ("What’s your favorite movie and why?", "talk"),
+            ("what directory are you in ?", "files"),
             ("Search my drive for a file named budget_2024.xlsx", "files"),
             ("Write a Python function to sort a list of dictionaries by key", "code"),
             ("Find the latest updates on quantum computing on the web", "web"),
@@ -307,6 +309,7 @@ class AgentRouter:
         llm_router, confidence_llm_router = result_llm_router[0], result_llm_router[1]
         final_score_bart = confidence_bart / (confidence_bart + confidence_llm_router)
         final_score_llm = confidence_llm_router / (confidence_bart + confidence_llm_router)
+        self.logger.info(f"Routing Vote: BART: {bart} ({final_score_bart}) LLM-router: {llm_router} ({final_score_llm})")
         if log_confidence:
             pretty_print(f"Agent choice -> BART: {bart} ({final_score_bart}) LLM-router: {llm_router} ({final_score_llm})")
         return bart if final_score_bart > final_score_llm else llm_router
@@ -328,12 +331,17 @@ class AgentRouter:
         Returns:
         str: The estimated complexity
         """
-        predictions = self.complexity_classifier.predict(text)
+        try:
+            predictions = self.complexity_classifier.predict(text)
+        except Exception as e:
+            pretty_print(f"Error in estimate_complexity: {str(e)}", color="failure")
+            return "LOW"
         predictions = sorted(predictions, key=lambda x: x[1], reverse=True)
         if len(predictions) == 0:
             return "LOW"
         complexity, confidence = predictions[0][0], predictions[0][1]
         if confidence < 0.4:
+            self.logger.info(f"Low confidence in complexity estimation: {confidence}")
             return "LOW"
         if complexity == "HIGH" and len(text) < 64:
             return None # ask for more info
@@ -341,8 +349,8 @@ class AgentRouter:
             return "HIGH"
         elif complexity == "LOW":
             return "LOW"
-        pretty_print(f"Failed to estimate the complexity of the text. Confidence: {confidence}", color="failure")
-        return None
+        pretty_print(f"Failed to estimate the complexity of the text.", color="failure")
+        return "LOW"
     
     def find_planner_agent(self) -> Agent:
         """
@@ -354,6 +362,7 @@ class AgentRouter:
             if agent.type == "planner_agent":
                 return agent
         pretty_print(f"Error finding planner agent. Please add a planner agent to the list of agents.", color="failure")
+        self.logger.error("Planner agent not found.")
         return None
     
     def select_agent(self, text: str) -> Agent:
@@ -386,9 +395,11 @@ class AgentRouter:
                 pretty_print(f"Selected agent: {agent.agent_name} (roles: {agent.role[lang]})", color="warning")
                 return agent
         pretty_print(f"Error choosing agent.", color="failure")
+        self.logger.error("No agent selected.")
         return None
 
 if __name__ == "__main__":
+    sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
     agents = [
         CasualAgent("jarvis", "../prompts/base/casual_agent.txt", None),
         BrowserAgent("browser", "../prompts/base/planner_agent.txt", None),

sources/text_to_speech.py

@@ -1,18 +1,15 @@
-import os
+import os, sys
 import re
 import platform
 import subprocess
 from sys import modules
-from typing import List, Tuple, Type, Dict, Tuple
+from typing import List, Tuple, Type, Dict
 
 from kokoro import KPipeline
 from IPython.display import display, Audio
 import soundfile as sf
 
-if __name__ == "__main__":
-    from utility import pretty_print, animate_thinking
-else:
-    from sources.utility import pretty_print, animate_thinking
+from sources.utility import pretty_print, animate_thinking
 
 class Speech():
     """
@@ -47,22 +44,22 @@ class Speech():
         if not os.path.exists(path):
             os.makedirs(path)
 
-    def speak(self, sentence: str, voice_number: int = 1):
+    def speak(self, sentence: str, voice_idx: int = 1):
         """
         Convert text to speech using an AI model and play the audio.
 
         Args:
             sentence (str): The text to convert to speech. Will be pre-processed.
-            voice_number (int, optional): Index of the voice to use from the voice map.
+            voice_idx (int, optional): Index of the voice to use from the voice map.
         """
         if not self.pipeline:
             return
-        if voice_number >= len(self.voice_map[self.language]) or voice_number < 0:
+        if voice_idx >= len(self.voice_map[self.language]):
             pretty_print("Invalid voice number, using default voice", color="error")
-            voice_number = 0
+            voice_idx = 0
         sentence = self.clean_sentence(sentence)
-        audio_file = f"{self.voice_folder}/sample_{self.voice_map[self.language][voice_number]}.wav"
-        self.voice = self.voice_map[self.language][voice_number]
+        audio_file = f"{self.voice_folder}/sample_{self.voice_map[self.language][voice_idx]}.wav"
+        self.voice = self.voice_map[self.language][voice_idx]
         generator = self.pipeline(
             sentence, voice=self.voice,
             speed=self.speed, split_pattern=r'\n+'
@@ -143,6 +140,7 @@ class Speech():
         return sentence
 
 if __name__ == "__main__":
+    sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
     speech = Speech()
     tosay_en = """
     I looked up recent news using the website https://www.theguardian.com/world
@@ -154,8 +152,8 @@ if __name__ == "__main__":
     J'ai consulté les dernières nouvelles sur le site https://www.theguardian.com/world
     """
     spk = Speech(enable=True, language="en", voice_idx=0)
-    spk.speak(tosay_en)
+    spk.speak(tosay_en, voice_idx=0)
     spk = Speech(enable=True, language="fr", voice_idx=0)
     spk.speak(tosay_fr)
-    spk = Speech(enable=True, language="zh", voice_idx=0)
-    spk.speak(tosay_zh)
+    #spk = Speech(enable=True, language="zh", voice_idx=0)
+    #spk.speak(tosay_zh)

sources/tools/tools.py

@@ -2,11 +2,13 @@
 """
 define a generic tool class, any tool can be used by the agent.
 
-A tool can be used by deepseek like so:
+A tool can be used by a llm like so:
 ```<tool name>
 <code or query to execute>
 ```
 
+we call these "blocks".
+
 For example:
 ```python
 print("Hello world")
@@ -40,9 +42,7 @@ class Tools():
         return self.current_dir
     
     def check_config_dir_validity(self):
-        """
-        Check if the config directory is valid.
-        """
+        """Check if the config directory is valid."""
         path = self.config['MAIN']['work_dir']
         if path == "":
             print("WARNING: Work directory not set in config.ini")
@@ -56,15 +56,11 @@ class Tools():
         return True
     
     def config_exists(self):
-        """
-        Check if the config file exists.
-        """
+        """Check if the config file exists."""
         return os.path.exists('./config.ini')
 
     def create_work_dir(self):
-        """
-        Create the work directory if it does not exist.
-        """
+        """Create the work directory if it does not exist."""
         default_path = os.path.dirname(os.getcwd())
         if self.config_exists():
             self.config.read('./config.ini')

tests/test_browser_agent_parsing.py

@@ -0,0 +1,69 @@
+import unittest
+import os
+import sys
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))  # Add project root to Python path
+from sources.agents.browser_agent import BrowserAgent
+
+class TestBrowserAgentParsing(unittest.TestCase):
+    def setUp(self):
+        # Initialize a basic BrowserAgent instance for testing
+        self.agent = BrowserAgent(
+            name="TestAgent",
+            prompt_path="../prompts/base/browser_agent.txt",
+            provider=None
+        )
+
+    def test_extract_links(self):
+        # Test various link formats
+        test_text = """
+        Check this out: https://example.com, and www.google.com!
+        Also try https://test.org/about?page=1.
+        """
+        expected = [
+            "https://example.com",
+            "www.google.com",
+            "https://test.org/about?page=1"
+        ]
+        result = self.agent.extract_links(test_text)
+        self.assertEqual(result, expected)
+
+    def test_extract_form(self):
+        # Test form extraction
+        test_text = """
+        Fill this: [username](john) and [password](secret123)
+        Not a form: [random]text
+        """
+        expected = ["[username](john)", "[password](secret123)"]
+        result = self.agent.extract_form(test_text)
+        self.assertEqual(result, expected)
+
+    def test_clean_links(self):
+        # Test link cleaning
+        test_links = [
+            "https://example.com.",
+            "www.test.com,",
+            "https://clean.org!",
+            "https://good.com"
+        ]
+        expected = [
+            "https://example.com",
+            "www.test.com",
+            "https://clean.org",
+            "https://good.com"
+        ]
+        result = self.agent.clean_links(test_links)
+        self.assertEqual(result, expected)
+
+    def test_parse_answer(self):
+        # Test parsing answer with notes and links
+        test_text = """
+        Here's some info
+        Note: This is important. We are doing test it's very cool.
+        action: 
+        i wanna navigate to https://test.com
+        """
+        self.agent.parse_answer(test_text)
+        self.assertEqual(self.agent.notes[0], "Note: This is important. We are doing test it's very cool.")
+
+if __name__ == "__main__":
+    unittest.main()