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

Fix: py interpreter problems, Feat : more pretty output

Browse Source
This commit is contained in:
martin legrand
2025-02-27 16:29:44 +01:00
parent be22f0232e
commit 14fcbd2343
7 changed files with 197 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sources/agent.py
+17 -32
View File
@@ -2,7 +2,8 @@ from typing import Tuple, Callable
from abc import abstractmethod
import os
import random
from sources.history import History
from sources.utility import pretty_print
class Agent():
def __init__(self, model: str,
name: str,
@@ -12,17 +13,10 @@ class Agent():
self._current_directory = os.getcwd()
self._model = model
self._llm = provider
self._history = []
self._history = History(self.load_prompt(prompt_path),
memory_compression=False)
self._tools = {}
self.set_system_prompt(prompt_path)
def set_system_prompt(self, prompt_path: str) -> None:
self.set_history(self.load_prompt(prompt_path))
@property
def history(self):
return self._history
@property
def name(self) -> str:
return self._name
@@ -31,21 +25,6 @@ class Agent():
def get_tools(self) -> dict:
return self._tools
def set_history(self, system_prompt: str) -> None:
"""
Set the default history for the agent.
Deepseek developers recommand not using a system prompt directly.
We therefore pass the system prompt as a user message.
"""
self._history = [{'role': 'user', 'content': system_prompt},
{'role': 'assistant', 'content': f'Hello, How can I help you today ?'}]
def add_to_history(self, role: str, content: str) -> None:
self._history.append({'role': role, 'content': content})
def clear_history(self) -> None:
self._history = []
def add_tool(self, name: str, tool: Callable) -> None:
if tool is not Callable:
raise TypeError("Tool must be a callable object (a method)")
@@ -81,12 +60,13 @@ class Agent():
end_idx = text.rfind(end_tag)+8
return text[start_idx:end_idx]
def llm_request(self, history, verbose = True) -> Tuple[str, str]:
def llm_request(self, verbose = True) -> Tuple[str, str]:
history = self._history.get()
thought = self._llm.respond(history, verbose)
reasoning = self.extract_reasoning_text(thought)
answer = self.remove_reasoning_text(thought)
self.add_to_history('assistant', answer)
self._history.push('assistant', answer)
return answer, reasoning
def wait_message(self, speech_module):
@@ -96,6 +76,13 @@ class Agent():
"Hold on, I’m crunching numbers.",
"Working on it sir, please let me think."]
speech_module.speak(messages[random.randint(0, len(messages)-1)])
def print_code_blocks(self, blocks: list, name: str):
for block in blocks:
pretty_print(f"Executing {name} code...\n", color="output")
pretty_print("-"*100, color="output")
pretty_print(block, color="code")
pretty_print("-"*100, color="output")
def execute_modules(self, answer: str) -> Tuple[bool, str]:
feedback = ""
@@ -106,15 +93,13 @@ class Agent():
blocks, save_path = tool.load_exec_block(answer)
if blocks != None:
self.print_code_blocks(blocks, name)
output = tool.execute(blocks)
feedback = tool.interpreter_feedback(output)
answer = tool.remove_block(answer)
self.add_to_history('user', feedback)
self._history.push('user', feedback)
if "failure" in feedback.lower():
return False, feedback
if blocks == None:
return True, feedback
if save_path != None:
tool.save_block(blocks, save_path)
return True, feedback
return True, feedback
sources/code_agent.py
+22 -3
View File
@@ -6,24 +6,43 @@ from sources.agent import Agent
class CoderAgent(Agent):
def __init__(self, model, name, prompt_path, provider):
super().__init__(model, name, prompt_path, provider)
self.set_system_prompt(prompt_path)
self._tools = {
"bash": BashInterpreter(),
"python": PyInterpreter()
}
def remove_blocks(self, text: str) -> str:
"""
Remove all code/query blocks within a tag from the answer text.
"""
tag = f'```'
lines = text.split('\n')
post_lines = []
in_block = False
for line in lines:
if tag in line and not in_block:
in_block = True
continue
if not in_block:
post_lines.append(line)
if tag in line:
in_block = False
return "\n".join(post_lines)
def answer(self, prompt, speech_module) -> str:
answer = ""
attempt = 0
max_attempts = 3
self.add_to_history('user', prompt)
self._history.push('user', prompt)
while attempt < max_attempts:
pretty_print("Thinking...", color="status")
self.wait_message(speech_module)
answer, reasoning = self.llm_request(self.history)
answer, reasoning = self.llm_request()
exec_success, feedback = self.execute_modules(answer)
pretty_print(feedback, color="failure" if "failure" in feedback.lower() else "success")
answer = self.remove_blocks(answer)
pretty_print(answer, color="output")
if exec_success:
break
attempt += 1
sources/history.py
+120
View File
@@ -0,0 +1,120 @@
import torch
from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
class History():
"""
History is a class for managing the conversation history
It provides a method to compress the history (experimental, use with caution).
"""
def __init__(self, system_prompt: str, memory_compression: bool = True):
self._history = []
self._history = [{'role': 'user', 'content': system_prompt},
{'role': 'assistant', 'content': f'Hello, How can I help you today ?'}]
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)
def get_cuda_device(self) -> str:
if torch.backends.mps.is_available():
return "mps"
elif torch.cuda.is_available():
return "cuda"
else:
return "cpu"
def summarize(self, text: str, min_length: int = 64) -> str:
if self._tokenizer is None or self._model is None:
return text
max_length = len(text) // 2 if len(text) > min_length*2 else min_length*2
input_text = "summarize: " + text
inputs = self._tokenizer(input_text, return_tensors="pt", max_length=512, truncation=True)
summary_ids = self._model.generate(
inputs['input_ids'],
max_length=max_length, # Maximum length of the summary
min_length=min_length, # Minimum length of the summary
length_penalty=1.0, # Adjusts length preference
num_beams=4, # Beam search for better quality
early_stopping=True # Stop when all beams finish
)
summary = self._tokenizer.decode(summary_ids[0], skip_special_tokens=True)
return summary
def timer_decorator(func):
from time import time
def wrapper(*args, **kwargs):
start_time = time()
result = func(*args, **kwargs)
end_time = time()
print(f"{func.__name__} took {end_time - start_time:.2f} seconds to execute")
return result
return wrapper
@timer_decorator
def compress(self) -> str:
if not self.memory_compression:
return
for i in range(len(self._history)):
if i <= 2:
continue
if self._history[i]['role'] == 'assistant':
self._history[i]['content'] = self.summarize(self._history[i]['content'])
def reset(self, history: list) -> None:
self._history = history
def push(self, role: str, content: str) -> None:
self._history.append({'role': role, 'content': content})
# EXPERIMENTAL
if self.memory_compression and role == 'assistant':
self.compress()
def clear(self) -> None:
self._history = []
def get(self) -> list:
return self._history
if __name__ == "__main__":
history = History("You are a helpful assistant.")
sample_text = """
The error you're encountering:
Copy
cuda.cu:52:10: fatal error: helper_functions.h: No such file or directory
#include <helper_functions.h>
^~~~~~~~~~~~~~~~~~~~
compilation terminated.
indicates that the compiler cannot find the helper_functions.h file. This is because the #include <helper_functions.h> directive is looking for the file in the system's include paths, but the file is either not in those paths or is located in a different directory.
Solutions
1. Use #include "helper_functions.h" Instead of #include <helper_functions.h>
Angle brackets (< >) are used for system or standard library headers.
Quotes (" ") are used for local or project-specific headers.
If helper_functions.h is in the same directory as cuda.cu, change the include directive to:
3. Verify the File Exists
Double-check that helper_functions.h exists in the specified location. If the file is missing, you'll need to obtain or recreate it.
4. Use the Correct CUDA Samples Path (if applicable)
If helper_functions.h is part of the CUDA Samples, ensure you have the CUDA Samples installed and include the correct path. For example, on Linux, the CUDA Samples are typically located in /usr/local/cuda/samples/common/inc. You can include this path like so:
Use #include "helper_functions.h" for local files.
Use the -I flag to specify the directory containing helper_functions.h.
Ensure the file exists in the specified location.
"""
history.push('user', "why do i get this error?")
history.push('assistant', sample_text)
print("\n---\nHistory before:", history.get())
history.compress()
print("\n---\nHistory after:", history.get())
sources/tools/PyInterpreter.py
+9 -4
View File
@@ -72,15 +72,20 @@ class PyInterpreter(Tools):
if __name__ == "__main__":
text = """
Sure here is how to print in python:
For Python, let's also do a quick check:
```python
print("Hello from Python!")
```
If these work, you'll see the outputs in the next message. Let me know if you'd like me to test anything specific!
here is a save test
```python:tmp.py
def print_hello():
hello = "Hello World"
print(hello)
print_hello()
```
"""
py = PyInterpreter()
sources/tools/tools.py
+24 -14
View File
@@ -22,6 +22,7 @@ HU787
"""
import sys
import os
from abc import abstractmethod
sys.path.append('..')
@@ -57,26 +58,19 @@ class Tools():
"""
pass
def remove_block(self, text:str) -> str:
"""
Remove all code/query blocks within a tag from the answer text.
"""
assert self.tag != "undefined", "Tag not defined"
start_tag = f'```{self.tag}'
end_tag = '```'
start_idx = text.find(start_tag)
end_idx = text.rfind(end_tag)+3
if start_idx == -1 or end_idx == -1:
return text
return text[:start_idx] + text[end_idx:]
def save_block(self, blocks:[str], save_path:str) -> None:
"""
Save the code/query block to a file.
Creates the directory path if it doesn't exist.
"""
if save_path is None:
return
directory = os.path.dirname(save_path)
if directory and not os.path.exists(directory):
print(f"Creating directory: {directory}")
os.makedirs(directory)
for block in blocks:
print(f"Saving code block to: {save_path}")
with open(save_path, 'w') as f:
f.write(block)
@@ -123,4 +117,20 @@ class Tools():
content = content[content.find('\n')+1:]
code_blocks.append(content)
start_index = end_pos + len(end_tag)
return code_blocks, save_path
return code_blocks, save_path
if __name__ == "__main__":
tool = Tools()
tool.tag = "python"
rt = tool.load_exec_block("""
Got it, let me show you the Python files in the current directory using Python:
```python
import os
for file in os.listdir():
if file.endswith('.py'):
print(file)
```
""")
print(rt)