From 3f60abde14ed50a8d4a03c90e73b9ae774fcc44d Mon Sep 17 00:00:00 2001 From: retoor Date: Sun, 28 Sep 2025 09:08:06 +0200 Subject: [PATCH] Update obtje.py --- botje.py | 149 +++++++++++++++++++++++++++++++------------------------ 1 file changed, 84 insertions(+), 65 deletions(-) diff --git a/botje.py b/botje.py index 616d1ff..53d7dd3 100755 --- a/botje.py +++ b/botje.py @@ -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() -