Update obtje.py

Browse Source
This commit is contained in:
retoor 2025-09-28 09:08:06 +02:00
parent 994f940140
commit 3f60abde14
1 changed files with 84 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

149
botje.py
View File

@ -1,3 +1,4 @@
#!/usr/bin/env python3
import os
import argparse
import json
@ -5,6 +6,14 @@ import subprocess
import sys
import shlex
import requests
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
filename='.botje.run.log',
filemode='w'
)
class Colors:
RED = '\033[91m'
@ -17,7 +26,9 @@ class Colors:
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
print(f"{Colors.RED}Error: OPENAI_API_KEY environment variable not set.{Colors.RESET}")
error_message = "Error: OPENAI_API_KEY environment variable not set."
print(f"{Colors.RED}{error_message}{Colors.RESET}")
logging.critical(error_message)
exit(1)
class AutonomousBot:
@ -25,6 +36,7 @@ class AutonomousBot:
self.initial_prompt = prompt
self.memory = memory if memory is not None else []
self.is_finished = False
logging.info(f"Bot initialized with prompt: '{prompt}'")
self.tools = [
{
"type": "function",
@ -162,84 +174,67 @@ Here is a summary of the steps taken so far:
def run(self):
print(f"{Colors.GREEN}Starting bot with prompt: '{self.initial_prompt}'{Colors.RESET}")
logging.info(f"Starting bot run with prompt: '{self.initial_prompt}'")
messages = [
{"role": "system", "content": self.get_system_prompt()},
{"role": "user", "content": self.initial_prompt}
]
while not self.is_finished:
try:
print(f"\n{Colors.BLUE}Thinking...{Colors.RESET}")
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"
}
payload = {
"model": "gpt-4.1-nano",
"messages": messages,
"tools": self.tools,
"tool_choice": "auto"
}
logging.info("Thinking... Sending request to OpenAI API.")
headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" }
payload = { "model": "gpt-4.1-nano", "messages": messages, "tools": self.tools, "tool_choice": "auto" }
response_raw = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)
logging.info(f"API response received. Status code: {response_raw.status_code}")
if response_raw.status_code != 200:
print(f"{Colors.RED}Error from OpenAI API: {response_raw.status_code} - {response_raw.text}{Colors.RESET}")
error_msg = f"Error from OpenAI API: {response_raw.status_code} - {response_raw.text}"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg)
break
response_json = response_raw.json()
response_message = response_json['choices'][0]['message']
messages.append(response_message)
tool_calls = response_message.get('tool_calls')
if not tool_calls:
print(f"\n{Colors.CYAN}AI Response (no tool call):{Colors.RESET}")
no_tool_call_message = response_message.get('content') or "No text content in response."
print(no_tool_call_message)
messages.append({
"role": "user",
"content": "You did not call a function. Please proceed with the next action by calling a function, or call `finish_task` if the entire goal is complete."
})
logging.info(f"AI response with no tool call: {no_tool_call_message}")
messages.append({ "role": "user", "content": "You did not call a function. Please proceed with the next action by calling a function, or call `finish_task` if the entire goal is complete."})
continue
for tool_call in tool_calls:
function_name = tool_call['function']['name']
function_to_call = self.available_functions.get(function_name)
function_args = {}
if not function_to_call:
print(f"{Colors.RED}Unknown function called by model: {function_name}{Colors.RESET}")
error_msg = f"Unknown function called by model: {function_name}"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg)
tool_result = {"status": "error", "message": f"Unknown function: {function_name}"}
else:
try:
function_args = json.loads(tool_call['function']['arguments'])
print(f"\n{Colors.GREEN}Calling function: `{function_name}` with arguments:{Colors.RESET}")
print(json.dumps(function_args, indent=2))
logging.info(f"Executing tool '{function_name}' with args: {json.dumps(function_args)}")
tool_result = function_to_call(**function_args)
logging.info(f"Tool '{function_name}' result: {json.dumps(tool_result)}")
except Exception as e:
print(f"{Colors.RED}Error calling function '{function_name}': {e}{Colors.RESET}")
error_msg = f"Error calling function '{function_name}': {e}"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg, exc_info=True)
tool_result = {"status": "error", "message": str(e)}
messages.append({
"tool_call_id": tool_call['id'],
"role": "tool",
"name": function_name,
"content": json.dumps(tool_result),
})
self.memory.append({
"function_name": function_name,
"arguments": function_args,
"result": tool_result
})
messages.append({ "tool_call_id": tool_call['id'], "role": "tool", "name": function_name, "content": json.dumps(tool_result), })
self.memory.append({ "function_name": function_name, "arguments": function_args, "result": tool_result })
except Exception as e:
print(f"\n{Colors.RED}An unexpected error occurred in the main loop: {e}{Colors.RESET}")
error_msg = f"An unexpected error occurred in the main loop: {e}"
print(f"\n{Colors.RED}{error_msg}{Colors.RESET}")
logging.critical(error_msg, exc_info=True)
break
def _write_file(self, filepath, content, purpose=""):
logging.info(f"Attempting to write file at '{filepath}' for purpose: {purpose}")
try:
dir_path = os.path.dirname(filepath)
if dir_path:
@ -247,10 +242,14 @@ Here is a summary of the steps taken so far:
with open(filepath, 'w', encoding='utf-8') as f:
f.write(content)
purpose_text = f" ({purpose})" if purpose else ""
print(f"{Colors.GREEN}File '{filepath}' created{purpose_text} successfully.{Colors.RESET}")
success_msg = f"File '{filepath}' created{purpose_text} successfully."
print(f"{Colors.GREEN}{success_msg}{Colors.RESET}")
logging.info(success_msg)
return {"status": "success", "filepath": filepath}
except Exception as e:
print(f"{Colors.RED}Error creating file '{filepath}': {e}{Colors.RESET}")
error_msg = f"Error creating file '{filepath}': {e}"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg, exc_info=True)
return {"status": "error", "message": str(e)}
def create_file(self, filepath, content):
@ -260,53 +259,67 @@ Here is a summary of the steps taken so far:
return self._write_file(filepath, content, "test code")
def read_file(self, filepath):
logging.info(f"Attempting to read file: '{filepath}'")
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
print(f"{Colors.GREEN}File '{filepath}' read successfully.{Colors.RESET}")
success_msg = f"File '{filepath}' read successfully."
print(f"{Colors.GREEN}{success_msg}{Colors.RESET}")
logging.info(success_msg)
return {"status": "success", "content": content}
except FileNotFoundError:
print(f"{Colors.RED}File not found: '{filepath}'{Colors.RESET}")
error_msg = f"File not found: '{filepath}'"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg)
return {"status": "error", "message": "File not found."}
except Exception as e:
print(f"{Colors.RED}Error reading file '{filepath}': {e}{Colors.RESET}")
error_msg = f"Error reading file '{filepath}': {e}"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg, exc_info=True)
return {"status": "error", "message": str(e)}
def delete_file(self, filepath):
logging.info(f"Attempting to delete file: '{filepath}'")
try:
os.remove(filepath)
print(f"{Colors.GREEN}File '{filepath}' deleted successfully.{Colors.RESET}")
success_msg = f"File '{filepath}' deleted successfully."
print(f"{Colors.GREEN}{success_msg}{Colors.RESET}")
logging.info(success_msg)
return {"status": "success", "filepath": filepath}
except FileNotFoundError:
print(f"{Colors.RED}File not found: '{filepath}'{Colors.RESET}")
error_msg = f"File not found: '{filepath}'"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg)
return {"status": "error", "message": "File not found."}
except Exception as e:
print(f"{Colors.RED}Error deleting file '{filepath}': {e}{Colors.RESET}")
error_msg = f"Error deleting file '{filepath}': {e}"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg, exc_info=True)
return {"status": "error", "message": str(e)}
def terminal_execute(self, command):
print(f"Executing command: `{command}`")
logging.info(f"Executing terminal command: `{command}`")
try:
command_parts = shlex.split(command)
result = subprocess.run(command_parts, capture_output=True, text=True, check=False)
output = {
"status": "success" if result.returncode == 0 else "error",
"stdout": result.stdout.strip(),
"stderr": result.stderr.strip(),
"returncode": result.returncode
}
output = { "status": "success" if result.returncode == 0 else "error", "stdout": result.stdout.strip(), "stderr": result.stderr.strip(), "returncode": result.returncode }
logging.info(f"Command executed. Return code: {result.returncode}. STDOUT: {result.stdout.strip()}. STDERR: {result.stderr.strip()}")
print(f" - STDOUT: {result.stdout.strip()}")
if result.stderr.strip():
print(f" - {Colors.YELLOW}STDERR: {result.stderr.strip()}{Colors.RESET}")
print(f" - Return Code: {result.returncode}")
return output
except Exception as e:
print(f"{Colors.RED}Error executing command '{command}': {e}{Colors.RESET}")
error_msg = f"Error executing command '{command}': {e}"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg, exc_info=True)
return {"status": "error", "message": str(e), "stdout": "", "stderr": str(e), "returncode": -1}
def finish_task(self, reason):
print(f"\n{Colors.YELLOW}BOT INSTANCE FINISHED: {reason}{Colors.RESET}")
finish_msg = f"BOT INSTANCE FINISHED: {reason}"
print(f"\n{Colors.YELLOW}{finish_msg}{Colors.RESET}")
logging.info(finish_msg)
self.is_finished = True
return {"status": "finished"}
@ -316,26 +329,32 @@ Here is a summary of the steps taken so far:
print("="*50)
print(f"{Colors.MAGENTA}Delegating remaining prompt to a new bot instance...{Colors.RESET}")
print(f" python {__file__} \"{remaining_prompt}\"")
logging.info(f"Splitting task. Delegating '{remaining_prompt}'. Current bot continues with '{current_task_prompt}'.")
prompt = f"# PREVIOUS PROMPT\n```markdown\n{current_task_prompt}\n```\n\n# PROMPT\n{remaining_prompt}"
try:
subprocess.Popen([sys.executable, __file__, remaining_prompt])
print(f"{Colors.GREEN}New bot instance launched successfully.{Colors.RESET}")
subprocess.Popen([sys.executable, __file__, prompt])
success_msg = "New bot instance launched successfully."
print(f"{Colors.GREEN}{success_msg}{Colors.RESET}")
logging.info(success_msg)
except Exception as e:
print(f"{Colors.RED}Failed to launch new bot instance: {e}{Colors.RESET}")
error_msg = f"Failed to launch new bot instance: {e}"
print(f"{Colors.RED}{error_msg}{Colors.RESET}")
logging.error(error_msg, exc_info=True)
return {"status": "error", "message": f"Failed to delegate: {e}"}
self.initial_prompt = current_task_prompt
print(f"This instance will now proceed to solve its narrowed task: '{current_task_prompt}'")
print("="*50 + "\n")
return {"status": "success", "message": "Task split. This bot will now handle the first part."}
def main():
logging.info("Main function started.")
parser = argparse.ArgumentParser(description="An autonomous bot to execute tasks.")
parser.add_argument("prompt", type=str, help="The user prompt describing the task to be done.")
args = parser.parse_args()
logging.info(f"Arguments parsed: {args}")
bot = AutonomousBot(args.prompt)
bot.run()
logging.info("Bot run finished. Exiting main.")
if __name__ == "__main__":
main()