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

feat : fix displaying problem + memory compression on loading + personality prompt + technical image

Browse Source
This commit is contained in:
martin legrand
2025-03-28 14:43:05 +01:00
parent 8cba1bad43
commit c862d496e3
32 changed files with 448 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sources/agents/agent.py
+15 -14
View File
@@ -14,17 +14,16 @@ class executorResult:
"""
A class to store the result of a tool execution.
"""
def __init__(self, blocks, feedback, success):
self.blocks = blocks
def __init__(self, block, feedback, success):
self.block = block
self.feedback = feedback
self.success = success
def show(self):
for block in self.blocks:
pretty_print("-"*100, color="output")
pretty_print(block, color="code" if self.success else "failure")
pretty_print("-"*100, color="output")
pretty_print(self.feedback, color="success" if self.success else "failure")
pretty_print("-"*100, color="output")
pretty_print(self.block, color="code" if self.success else "failure")
pretty_print("-"*100, color="output")
pretty_print(self.feedback, color="success" if self.success else "failure")
class Agent():
"""
@@ -178,14 +177,16 @@ class Agent():
blocks, save_path = tool.load_exec_block(answer)
if blocks != None:
output = tool.execute(blocks)
feedback = tool.interpreter_feedback(output) # tool interpreter feedback
success = not tool.execution_failure_check(output)
pretty_print(feedback, color="success" if success else "failure")
for block in blocks:
output = tool.execute([block])
feedback = tool.interpreter_feedback(output) # tool interpreter feedback
success = not tool.execution_failure_check(output)
self.blocks_result.append(executorResult(block, feedback, success))
if not success:
self.memory.push('user', feedback)
return False, feedback
self.memory.push('user', feedback)
self.blocks_result.append(executorResult(blocks, feedback, success))
if not success:
return False, feedback
if save_path != None:
tool.save_block(blocks, save_path)
self.blocks_result = list(reversed(self.blocks_result))
return True, feedback
sources/agents/browser_agent.py
-1
View File
@@ -21,7 +21,6 @@ class BrowserAgent(Agent):
"en": "web",
"fr": "web",
"zh": "网络",
"es": "web"
}
self.type = "browser_agent"
self.browser = browser
sources/agents/casual_agent.py
-1
View File
@@ -18,7 +18,6 @@ class CasualAgent(Agent):
"en": "talk",
"fr": "discuter",
"zh": "聊天",
"es": "discutir"
}
self.type = "casual_agent"
sources/agents/code_agent.py
-1
View File
@@ -24,7 +24,6 @@ class CoderAgent(Agent):
"en": "code",
"fr": "codage",
"zh": "编码",
"es": "codificación",
}
self.type = "code_agent"
sources/agents/file_agent.py
+1 -2
View File
@@ -19,13 +19,12 @@ class FileAgent(Agent):
"en": "files",
"fr": "fichiers",
"zh": "文件",
"es": "archivos",
}
self.type = "file_agent"
def process(self, prompt, speech_module) -> str:
exec_success = False
prompt += f"\nWork directory: {self.work_dir}"
prompt += f"\nYou must work in directory: {self.work_dir}"
self.memory.push('user', prompt)
while exec_success is False:
self.wait_message(speech_module)
sources/agents/planner_agent.py
+3 -4
View File
@@ -18,15 +18,14 @@ class PlannerAgent(Agent):
self.tools['json'].tag = "json"
self.browser = browser
self.agents = {
"coder": CoderAgent(name, "prompts/coder_agent.txt", provider, verbose=False),
"file": FileAgent(name, "prompts/file_agent.txt", provider, verbose=False),
"web": BrowserAgent(name, "prompts/browser_agent.txt", provider, verbose=False, browser=browser)
"coder": CoderAgent(name, "prompts/base/coder_agent.txt", provider, verbose=False),
"file": FileAgent(name, "prompts/base/file_agent.txt", provider, verbose=False),
"web": BrowserAgent(name, "prompts/base/browser_agent.txt", provider, verbose=False, browser=browser)
}
self.role = {
"en": "Research, setup and code",
"fr": "Recherche, configuration et codage",
"zh": "研究，设置和编码",
"es": "Investigación, configuración y code"
}
self.type = "planner_agent"
sources/interaction.py
+4 -3
View File
@@ -90,13 +90,13 @@ class Interaction:
self.last_query = query
return query
def think(self) -> None:
def think(self) -> bool:
"""Request AI agents to process the user input."""
if self.last_query is None or len(self.last_query) == 0:
return
return False
agent = self.router.select_agent(self.last_query)
if agent is None:
return
return False
if self.current_agent != agent and self.last_answer is not None:
## get last history from previous agent
self.current_agent.memory.push('user', self.last_query)
@@ -106,6 +106,7 @@ class Interaction:
self.last_answer, _ = agent.process(self.last_query, self.speech)
if self.last_answer == tmp:
self.last_answer = None
return True
def show_answer(self) -> None:
"""Show the answer to the user."""
sources/language.py
+1 -1
View File
@@ -19,7 +19,7 @@ class LanguageUtility:
text: string to analyze
Returns: ISO639-1 language code
"""
langid.set_languages(['fr', 'en', 'zh', 'es'])
langid.set_languages(['fr', 'en', 'zh'])
lang, score = langid.classify(text)
return lang
sources/llm_provider.py
+17 -2
View File
@@ -26,7 +26,8 @@ class Provider:
"openai": self.openai_fn,
"lm-studio": self.lm_studio_fn,
"huggingface": self.huggingface_fn,
"deepseek": self.deepseek_fn
"deepseek": self.deepseek_fn,
"test": self.test_fn
}
self.api_key = None
self.unsafe_providers = ["openai", "deepseek"]
@@ -245,13 +246,27 @@ class Provider:
This is a test response from the test provider.
Change provider to 'ollama' or 'server' to get real responses.
This is python saying hello.
```python
print("Hello world from python")
```
This is ls -la from bash.
```bash
echo "Hello world from bash"
ls -la
```
This is pwd from bash.
```bash
pwd
```
This is unsafe command.
```bash
rm does_not_exist.txt
```
goodbye
"""
return thought
sources/memory.py
+9 -7
View File
@@ -9,7 +9,7 @@ import json
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sources.utility import timer_decorator
from sources.utility import timer_decorator, pretty_print
class Memory():
"""
@@ -33,9 +33,8 @@ class Memory():
self.model = "pszemraj/led-base-book-summary"
self.device = self.get_cuda_device()
self.memory_compression = memory_compression
if memory_compression:
self.tokenizer = AutoTokenizer.from_pretrained(self.model)
self.model = AutoModelForSeq2SeqLM.from_pretrained(self.model)
self.tokenizer = AutoTokenizer.from_pretrained(self.model)
self.model = AutoModelForSeq2SeqLM.from_pretrained(self.model)
def get_filename(self) -> str:
return f"memory_{self.session_time.strftime('%Y-%m-%d_%H-%M-%S')}.txt"
@@ -78,6 +77,9 @@ class Memory():
path = os.path.join(save_path, filename)
with open(path, 'r') as f:
self.memory = json.load(f)
if self.memory[-1]['role'] == 'user':
self.memory.pop()
self.compress()
def reset(self, memory: list) -> None:
self.memory = memory
@@ -86,7 +88,9 @@ class Memory():
"""Push a message to the memory."""
if self.memory_compression and role == 'assistant':
self.compress()
# we don't compress the last message
curr_idx = len(self.memory)
if self.memory[curr_idx-1]['content'] == content:
pretty_print("Warning: same message have been pushed twice to memory", color="error")
self.memory.append({'role': role, 'content': content})
def clear(self) -> None:
@@ -134,8 +138,6 @@ class Memory():
"""
Compress the memory using the AI model.
"""
if not self.memory_compression:
return
for i in range(len(self.memory)):
if i < 3:
continue
sources/router.py
+1 -1
View File
@@ -384,7 +384,7 @@ class AgentRouter:
pretty_print(f"Complex task detected, routing to planner agent.", color="info")
return self.find_planner_agent()
for agent in self.agents:
if best_agent == agent.role[lang]:
if best_agent == agent.role["en"]:
pretty_print(f"Selected agent: {agent.agent_name} (roles: {agent.role[lang]})", color="warning")
return agent
pretty_print(f"Error choosing agent.", color="failure")
sources/tools/BashInterpreter.py
+10 -6
View File
@@ -6,8 +6,10 @@ import subprocess
if __name__ == "__main__":
from tools import Tools
from safety import is_unsafe
else:
from sources.tools.tools import Tools
from sources.tools.safety import is_unsafe
class BashInterpreter(Tools):
"""
@@ -19,16 +21,16 @@ class BashInterpreter(Tools):
def language_bash_attempt(self, command: str):
"""
detect if AI attempt to run the code using bash.
if so, return True, otherwise return False.
The philosophy is that code written by the AI will be executed, so it should not use bash to run it.
Detect if AI attempt to run the code using bash.
If so, return True, otherwise return False.
Code written by the AI will be executed automatically, so it should not use bash to run it.
"""
lang_interpreter = ["python3", "gcc", "g++", "go", "javac", "rustc", "clang", "clang++", "rustc", "rustc++", "rustc++"]
for word in command.split():
if word in lang_interpreter:
return True
return False
def execute(self, commands: str, safety=False, timeout=1000):
"""
Execute bash commands and display output in real-time.
@@ -38,6 +40,9 @@ class BashInterpreter(Tools):
concat_output = ""
for command in commands:
command = command.replace('\n', '')
if self.safe_mode and is_unsafe(commands):
return "Unsafe command detected, execution aborted."
if self.language_bash_attempt(command):
continue
try:
@@ -50,12 +55,11 @@ class BashInterpreter(Tools):
)
command_output = ""
for line in process.stdout:
print(line, end="")
command_output += line
return_code = process.wait(timeout=timeout)
if return_code != 0:
return f"Command {command} failed with return code {return_code}:\n{command_output}"
concat_output += f"Output of {command}:\n{command_output.strip()}\n\n"
concat_output += f"Output of {command}:\n{command_output.strip()}\n"
except subprocess.TimeoutExpired:
process.kill() # Kill the process if it times out
return f"Command {command} timed out. Output:\n{command_output}"
sources/tools/safety.py
+87
View File
@@ -0,0 +1,87 @@
import os
import sys
unsafe_commands_unix = [
"rm", # File/directory removal
"dd", # Low-level disk writing
"mkfs", # Filesystem formatting
"chmod", # Permission changes
"chown", # Ownership changes
"shutdown", # System shutdown
"reboot", # System reboot
"halt", # System halt
"init", # System runlevel changes
"sysctl", # Kernel parameter changes
"kill", # Process termination
"pkill", # Kill by process name
"killall", # Kill all matching processes
"wget", # Web file downloading
"curl", # Data transfer
"scp", # Secure copy
"ftp", # File transfer protocol
"bash", # Bash shell execution
"exec", # Replace process with command
"tee", # Write to files with privileges
"umount", # Unmount filesystems
"fsck", # Filesystem repair
"iptables", # Firewall rules
"ufw", # Uncomplicated firewall
"passwd", # Password changes
"useradd", # Add users
"userdel", # Delete users
"groupadd", # Add groups
"groupdel", # Delete groups
"visudo", # Edit sudoers file
"screen", # Terminal session management
"fdisk", # Disk partitioning
"parted", # Disk partitioning
"chroot", # Change root directory
"route" # Routing table management
]
unsafe_commands_windows = [
"del", # Deletes files
"erase", # Alias for del, deletes files
"rd", # Removes directories (rmdir alias)
"rmdir", # Removes directories
"format", # Formats a disk, erasing data
"diskpart", # Manages disk partitions, can wipe drives
"chkdsk /f", # Fixes filesystem, can alter data
"fsutil", # File system utilities, can modify system files
"xcopy /y", # Copies files, overwriting without prompt
"copy /y", # Copies files, overwriting without prompt
"move", # Moves files, can overwrite
"attrib", # Changes file attributes, e.g., hiding or exposing files
"icacls", # Changes file permissions (modern)
"takeown", # Takes ownership of files
"reg delete", # Deletes registry keys/values
"regedit /s", # Silently imports registry changes
"sc", # can Stops services
"net", # can Stops/break services
"shutdown", # Shuts down or restarts the system
"schtasks", # Schedules tasks, can run malicious commands
"at", # Older task scheduler (pre-Vista)
"taskkill", # Kills processes
"wmic", # Deletes processes via WMI
"bcdedit", # Modifies boot configuration
"powercfg", # Changes power settings, can disable protections
"assoc", # Changes file associations
"ftype", # Changes file type commands
"cipher /w", # Wipes free space, erasing data
"esentutl", # Database utilities, can corrupt system files
"subst", # Substitutes drive paths, can confuse system
"mklink", # Creates symbolic links, can redirect access
"bootcfg"
]
def is_unsafe(cmd):
"""
check if a bash command is unsafe.
"""
if sys.platform.startswith("win"):
if any(c in cmd for c in unsafe_commands_windows):
return True
else:
if any(c in cmd for c in unsafe_commands_unix):
return True
return False
sources/tools/tools.py
+2 -1
View File
@@ -34,6 +34,7 @@ class Tools():
self.config = configparser.ConfigParser()
self.current_dir = self.create_work_dir()
self.excutable_blocks_found = False
self.safe_mode = True
def get_work_dir(self):
return self.current_dir
@@ -186,7 +187,7 @@ class Tools():
code_blocks.append(content)
start_index = end_pos + len(end_tag)
return code_blocks, save_path
if __name__ == "__main__":
tool = Tools()
tool.tag = "python"
sources/utility.py
+5 -12
View File
@@ -53,13 +53,6 @@ def pretty_print(text, color = "info"):
print(colored(text, color_map[color]))
def animate_thinking(text, color="status", duration=2):
"""
Display an animated "thinking..." indicator in a separate thread.
Args:
text (str): Text to display
color (str): Color for the text
duration (float): How long to animate in seconds
"""
def _animate():
color_map = {
"success": (Fore.GREEN, "green"),
@@ -71,22 +64,22 @@ def animate_thinking(text, color="status", duration=2):
"default": (Fore.RESET, "black"),
"info": (Fore.CYAN, "cyan")
}
fore_color, term_color = color_map[color]
fore_color, term_color = color_map.get(color, color_map["default"])
spinner = itertools.cycle(['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'])
end_time = time.time() + duration
while time.time() < end_time:
symbol = next(spinner)
if platform.system().lower() != "windows":
print(f"\r{fore_color}{symbol} {text}{Fore.RESET}", end="", flush=True)
print(f"{fore_color}{symbol} {text}{Fore.RESET}", flush=True)
else:
print(colored(f"\r{symbol} {text}", term_color), end="", flush=True)
print(colored(f"{symbol} {text}", term_color), flush=True)
time.sleep(0.1)
print("\033[1A\033[K", end="", flush=True)
print()
animation_thread = threading.Thread(target=_animate)
animation_thread.daemon = True
animation_thread.start()
animation_thread.join()
def timer_decorator(func):
"""