diff --git a/sources/agents/agent.py b/sources/agents/agent.py index 6f6f7e5..04a212f 100644 --- a/sources/agents/agent.py +++ b/sources/agents/agent.py @@ -7,25 +7,10 @@ import time from sources.memory import Memory from sources.utility import pretty_print +from sources.schemas import executorResult random.seed(time.time()) -class executorResult: - """ - A class to store the result of a tool execution. - """ - def __init__(self, block, feedback, success, tool_type): - self.block = block - self.feedback = feedback - self.success = success - self.tool_type = tool_type - - def show(self): - pretty_print('▂'*64, color="status") - pretty_print(self.block, color="code" if self.success else "failure") - pretty_print('▂'*64, color="status") - pretty_print(self.feedback, color="success" if self.success else "failure") - class Agent(): """ An abstract class for all agents. diff --git a/sources/browser.py b/sources/browser.py index 9b88332..3548389 100644 --- a/sources/browser.py +++ b/sources/browser.py @@ -131,6 +131,7 @@ class Browser: self.driver.get("https://www.google.com") if anticaptcha_manual_install: self.load_anticatpcha_manually() + self.screenshot_folder = os.path.join(os.getcwd(), ".screenshots") def load_anticatpcha_manually(self): pretty_print("You might want to install the AntiCaptcha extension for captchas.", color="warning") @@ -152,6 +153,7 @@ class Browser: ) self.apply_web_safety() self.logger.log(f"Navigated to: {url}") + self.screenshot() return True except TimeoutException as e: self.logger.error(f"Timeout waiting for {url} to load: {str(e)}") @@ -270,6 +272,8 @@ class Browser: self.driver.execute_script("arguments[0].scrollIntoView({block: 'center', behavior: 'smooth'});", element) time.sleep(0.1) element.click() + self.logger.info(f"Clicked element at {xpath}") + self.screenshot() return True except ElementClickInterceptedException as e: self.logger.error(f"Error click_element: {str(e)}") @@ -509,6 +513,7 @@ class Browser: if self.find_and_click_submission(): if self.wait_for_submission_outcome(): self.logger.info("Submission outcome detected") + self.screenshot() return True else: self.logger.warning("No submission outcome detected") @@ -532,15 +537,19 @@ class Browser: "window.scrollTo(0, document.body.scrollHeight);" ) time.sleep(1) + self.screenshot() return True except Exception as e: self.logger.error(f"Error scrolling: {str(e)}") return False - def screenshot(self, filename:str) -> bool: + def screenshot(self, filename:str = 'updated_screen.png') -> bool: """Take a screenshot of the current page.""" try: - self.driver.save_screenshot(filename) + path = os.path.join(self.screenshot_folder, filename) + if not os.path.exists(self.screenshot_folder): + os.makedirs(self.screenshot_folder) + self.driver.save_screenshot(path) self.logger.info(f"Screenshot saved as {filename}") return True except Exception as e: @@ -555,7 +564,7 @@ class Browser: input_elements = self.driver.execute_script(script) if __name__ == "__main__": - driver = create_driver() + driver = create_driver(headless=True, stealth_mode=True) browser = Browser(driver, anticaptcha_manual_install=True) #browser.go_to("https://github.com/Fosowl/agenticSeek") diff --git a/sources/interaction.py b/sources/interaction.py index a8b67db..8ce2de8 100644 --- a/sources/interaction.py +++ b/sources/interaction.py @@ -21,25 +21,36 @@ class Interaction: self.current_agent = None self.last_query = None self.last_answer = None - self.speech = None self.agents = agents self.tts_enabled = tts_enabled self.stt_enabled = stt_enabled self.recover_last_session = recover_last_session self.router = AgentRouter(self.agents, supported_language=langs) - if tts_enabled: - animate_thinking("Initializing text-to-speech...", color="status") - self.speech = Speech(enable=tts_enabled) self.ai_name = self.find_ai_name() + self.speech = None self.transcriber = None self.recorder = None + self.is_generating = False + if tts_enabled: + self.initialize_tts() if stt_enabled: - animate_thinking("Initializing speech recognition...", color="status") - self.transcriber = AudioTranscriber(self.ai_name, verbose=False) - self.recorder = AudioRecorder() + self.initialize_stt() if recover_last_session: self.load_last_session() self.emit_status() + + def initialize_tts(self): + """Initialize TTS.""" + if not self.speech: + animate_thinking("Initializing text-to-speech...", color="status") + self.speech = Speech(enable=self.tts_enabled) + + def initialize_stt(self): + """Initialize STT.""" + if not self.transcriber or not self.recorder: + animate_thinking("Initializing speech recognition...", color="status") + self.transcriber = AudioTranscriber(self.ai_name, verbose=False) + self.recorder = AudioRecorder() def emit_status(self): """Print the current status of agenticSeek.""" @@ -125,7 +136,9 @@ class Interaction: push_last_agent_memory = True tmp = self.last_answer self.current_agent = agent + self.is_generating = True self.last_answer, _ = agent.process(self.last_query, self.speech) + self.is_generating = False if push_last_agent_memory: self.current_agent.memory.push('user', self.last_query) self.current_agent.memory.push('assistant', self.last_answer) diff --git a/sources/memory.py b/sources/memory.py index 26e3a8b..2814120 100644 --- a/sources/memory.py +++ b/sources/memory.py @@ -130,9 +130,7 @@ class Memory(): self.logger.info(f"Clearing memory section {start} to {end}.") start = max(0, start) + 1 end = min(end, len(self.memory)-1) + 2 - self.logger.info(f"Memory before: {self.memory}") self.memory = self.memory[:start] + self.memory[end:] - self.logger.info(f"Memory after: {self.memory}") def get(self) -> list: return self.memory diff --git a/sources/schemas.py b/sources/schemas.py new file mode 100644 index 0000000..e3f633d --- /dev/null +++ b/sources/schemas.py @@ -0,0 +1,75 @@ + +from typing import Tuple, Callable +from pydantic import BaseModel + +class QueryRequest(BaseModel): + query: str + lang: str = "en" + tts_enabled: bool = True + stt_enabled: bool = False + + def __str__(self): + return f"Query: {self.query}, Language: {self.lang}, TTS: {self.tts_enabled}, STT: {self.stt_enabled}" + + def jsonify(self): + return { + "query": self.query, + "lang": self.lang, + "tts_enabled": self.tts_enabled, + "stt_enabled": self.stt_enabled + } + +class QueryResponse(BaseModel): + done: str + answer: str + agent_name: str + success: str + blocks: dict + + def __str__(self): + return f"Done: {self.done}, Answer: {self.answer}, Agent Name: {self.agent_name}, Success: {self.success}, Blocks: {self.blocks}" + + def jsonify(self): + return { + "done": self.done, + "answer": self.answer, + "agent_name": self.agent_name, + "success": self.success, + "blocks": self.blocks + } + +class executorResult: + """ + A class to store the result of a tool execution. + """ + def __init__(self, block: str, feedback: str, success: bool, tool_type: str): + """ + Initialize an agent with execution results. + + Args: + block: The content or code block processed by the agent. + feedback: Feedback or response information from the execution. + success: Boolean indicating whether the agent's execution was successful. + tool_type: The type of tool used by the agent for execution. + """ + self.block = block + self.feedback = feedback + self.success = success + self.tool_type = tool_type + + def __str__(self): + return f"Tool: {self.tool_type}\nBlock: {self.block}\nFeedback: {self.feedback}\nSuccess: {self.success}" + + def jsonify(self): + return { + "block": self.block, + "feedback": self.feedback, + "success": self.success, + "tool_type": self.tool_type + } + + def show(self): + pretty_print('▂'*64, color="status") + pretty_print(self.block, color="code" if self.success else "failure") + pretty_print('▂'*64, color="status") + pretty_print(self.feedback, color="success" if self.success else "failure") \ No newline at end of file