diff --git a/main.py b/main.py index 587b184..f03416b 100755 --- a/main.py +++ b/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"], diff --git a/prompts/base/coder_agent.txt b/prompts/base/coder_agent.txt index 8f997f6..764f409 100644 --- a/prompts/base/coder_agent.txt +++ b/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. diff --git a/prompts/base/planner_agent.txt b/prompts/base/planner_agent.txt index b144701..fc86fe2 100644 --- a/prompts/base/planner_agent.txt +++ b/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."" } ] diff --git a/prompts/jarvis/planner_agent.txt b/prompts/jarvis/planner_agent.txt index 047c057..de0b8c9 100644 --- a/prompts/jarvis/planner_agent.txt +++ b/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."" } ] diff --git a/server/app.py b/server/app.py index 2245c66..85e0f04 100644 --- a/server/app.py +++ b/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__': diff --git a/server/sources/cache.py b/server/sources/cache.py new file mode 100644 index 0000000..99c72ca --- /dev/null +++ b/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) diff --git a/server/sources/generator.py b/server/sources/generator.py index 133d6e9..9547d5d 100644 --- a/server/sources/generator.py +++ b/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}") diff --git a/server/sources/ollama_handler.py b/server/sources/ollama_handler.py index f9028a1..6be33a6 100644 --- a/server/sources/ollama_handler.py +++ b/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: diff --git a/sources/agents/browser_agent.py b/sources/agents/browser_agent.py index 812b812..259882b 100644 --- a/sources/agents/browser_agent.py +++ b/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 " + To proceed, choose a relevant link from the search results. Announce your choice by saying: "I will navigate to " 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. - - 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.] - - 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 + Inputs forms: + {inputs_form_text} - Example 2 (not useful, but related link): + End of webpage ({self.current_page}. + + # Instruction + + 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: then go back or navigate to another link. + 2. **Navigate to a link by either: ** + - Saying I will navigate to : (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 () LeCun net is ......" + 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 that ...... + 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) diff --git a/sources/agents/casual_agent.py b/sources/agents/casual_agent.py index 104e41e..339d578 100644 --- a/sources/agents/casual_agent.py +++ b/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) \ No newline at end of file + pass \ No newline at end of file diff --git a/sources/agents/code_agent.py b/sources/agents/code_agent.py index 80b1f72..396e21b 100644 --- a/sources/agents/code_agent.py +++ b/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) \ No newline at end of file + pass \ No newline at end of file diff --git a/sources/agents/file_agent.py b/sources/agents/file_agent.py index bcfaa6d..c50d66a 100644 --- a/sources/agents/file_agent.py +++ b/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) \ No newline at end of file + pass \ No newline at end of file diff --git a/sources/agents/planner_agent.py b/sources/agents/planner_agent.py index 21f99ab..3bc5395 100644 --- a/sources/agents/planner_agent.py +++ b/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") \ No newline at end of file + agents_work_result[task['id']] = self.last_answer + return self.last_answer, "" \ No newline at end of file diff --git a/sources/browser.py b/sources/browser.py index cd94274..76c402d 100644 --- a/sources/browser.py +++ b/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) diff --git a/sources/interaction.py b/sources/interaction.py index 2f07e81..10264e4 100644 --- a/sources/interaction.py +++ b/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 diff --git a/sources/language.py b/sources/language.py index f2bde5d..e153d93 100644 --- a/sources/language.py +++ b/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 diff --git a/sources/llm_provider.py b/sources/llm_provider.py index 65b9228..bffd8e6 100644 --- a/sources/llm_provider.py +++ b/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) diff --git a/sources/logger.py b/sources/logger.py new file mode 100644 index 0000000..e22ca16 --- /dev/null +++ b/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") + + + \ No newline at end of file diff --git a/sources/memory.py b/sources/memory.py index 9929203..ac6ff9b 100644 --- a/sources/memory.py +++ b/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) diff --git a/sources/router.py b/sources/router.py index 4aa5852..b152413 100644 --- a/sources/router.py +++ b/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), diff --git a/sources/text_to_speech.py b/sources/text_to_speech.py index 3d17a7c..e881f50 100644 --- a/sources/text_to_speech.py +++ b/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) \ No newline at end of file + #spk = Speech(enable=True, language="zh", voice_idx=0) + #spk.speak(tosay_zh) \ No newline at end of file diff --git a/sources/tools/searxSearch.py b/sources/tools/searxSearch.py index e0df8e0..83a6203 100644 --- a/sources/tools/searxSearch.py +++ b/sources/tools/searxSearch.py @@ -54,7 +54,7 @@ class searxSearch(Tools): status = self.link_valid(link) statuses.append(status) return statuses - + def execute(self, blocks: list, safety: bool = False) -> str: """Executes a search query against a SearxNG instance using POST and extracts URLs and titles.""" if not blocks: diff --git a/sources/tools/tools.py b/sources/tools/tools.py index 339992c..397fb8b 100644 --- a/sources/tools/tools.py +++ b/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: ``` ``` +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') diff --git a/tests/test_browser_agent_parsing.py b/tests/test_browser_agent_parsing.py new file mode 100644 index 0000000..6ec763c --- /dev/null +++ b/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() \ No newline at end of file