stephane.david/agenticSeek
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor : pre-backend implementation

Browse Source
This commit is contained in:
martin legrand
2025-04-14 16:55:15 +02:00
parent ecbdcaa57e
commit 592c7e6915
5 changed files with 108 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sources/agents/agent.py
+1 -16
View File
@@ -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.
sources/browser.py
+12 -3
View File
@@ -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")
sources/interaction.py
+20 -7
View File
@@ -21,26 +21,37 @@ 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."""
if self.stt_enabled:
@@ -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)
sources/memory.py
-2
View File
@@ -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
sources/schemas.py
+75
View File
@@ -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")