From 05ab6c1476f33bda489c07c38336b074c2b2b194 Mon Sep 17 00:00:00 2001 From: retoor Date: Sun, 28 Sep 2025 08:26:32 +0200 Subject: [PATCH] Added botje.py --- botje.py | 352 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 352 insertions(+) create mode 100644 botje.py diff --git a/botje.py b/botje.py new file mode 100644 index 0000000..6049f53 --- /dev/null +++ b/botje.py @@ -0,0 +1,352 @@ +#!/usr/bin/env python3 +import os +import argparse +import json +import subprocess +from openai import OpenAI +import sys +import shlex + +# ANSI escape codes for colors +class Colors: + RED = '\033[91m' + GREEN = '\033[92m' + YELLOW = '\033[93m' + BLUE = '\033[94m' + MAGENTA = '\033[95m' + CYAN = '\033[96m' + RESET = '\033[0m' + +# Initialize the OpenAI client +# It will automatically pick up the OPENAI_API_KEY from your environment variables. +try: + client = OpenAI() +except ImportError: + print(f"{Colors.RED}OpenAI Python package not found. Please install it with 'pip install openai'{Colors.RESET}") + exit(1) +except Exception as e: + print(f"{Colors.RED}Error initializing OpenAI client: {e}{Colors.RESET}") + print(f"{Colors.YELLOW}Please ensure your OPENAI_API_KEY environment variable is set correctly.{Colors.RESET}") + exit(1) + +class AutonomousBot: + """ + An autonomous bot that executes tasks based on a user prompt. + It can call functions, manage its state, and decompose complex tasks. + """ + + def __init__(self, prompt, memory=None): + """ + Initializes the bot with a prompt and an optional memory of past actions. + """ + self.initial_prompt = prompt + self.memory = memory if memory is not None else [] + self.is_finished = False + self.tools = [ + { + "type": "function", + "function": { + "name": "create_file", + "description": "Create a new file with specified content. This is for application code. Overwrites if the file exists.", + "parameters": { + "type": "object", + "properties": { + "filepath": {"type": "string", "description": "The full path for the new file."}, + "content": {"type": "string", "description": "The full, complete content to write into the file."} + }, + "required": ["filepath", "content"] + } + } + }, + { + "type": "function", + "function": { + "name": "create_test_file", + "description": "Create a new test file with specified content to verify the application code.", + "parameters": { + "type": "object", + "properties": { + "filepath": {"type": "string", "description": "The full path for the new test file (e.g., 'test_app.py')."}, + "content": {"type": "string", "description": "The full, complete content for the test file."} + }, + "required": ["filepath", "content"] + } + } + }, + { + "type": "function", + "function": { + "name": "read_file", + "description": "Read the content of a file.", + "parameters": { + "type": "object", + "properties": { + "filepath": {"type": "string", "description": "The full path of the file to read."} + }, + "required": ["filepath"] + } + } + }, + { + "type": "function", + "function": { + "name": "delete_file", + "description": "Delete a file.", + "parameters": { + "type": "object", + "properties": { + "filepath": {"type": "string", "description": "The full path of the file to delete."} + }, + "required": ["filepath"] + } + } + }, + { + "type": "function", + "function": { + "name": "terminal_execute", + "description": "Execute a shell command in the terminal. Use this for tasks like creating directories, listing files, running tests, and installing packages.", + "parameters": { + "type": "object", + "properties": { + "command": {"type": "string", "description": "The command to execute."} + }, + "required": ["command"] + } + } + }, + { + "type": "function", + "function": { + "name": "finish_task", + "description": "Call this function when the current task is fully completed, verified, and all tests are passing.", + "parameters": { + "type": "object", + "properties": { + "reason": {"type": "string", "description": "A brief summary of why the task is considered finished."} + }, + "required": ["reason"] + } + } + }, + { + "type": "function", + "function": { + "name": "split_and_delegate", + "description": "If the prompt contains more than one distinct task, use this function to handle the first task and delegate the rest.", + "parameters": { + "type": "object", + "properties": { + "current_task_prompt": {"type": "string", "description": "The prompt for the immediate task this bot instance will handle."}, + "remaining_prompt": {"type": "string", "description": "The prompt describing all subsequent tasks."} + }, + "required": ["current_task_prompt", "remaining_prompt"] + } + } + } + ] + self.available_functions = { + "create_file": self.create_file, + "create_test_file": self.create_test_file, + "read_file": self.read_file, + "delete_file": self.delete_file, + "terminal_execute": self.terminal_execute, + "finish_task": self.finish_task, + "split_and_delegate": self.split_and_delegate, + } + + def get_system_prompt(self): + """Constructs the system prompt for the AI model.""" + return f"""You are an autonomous AI builder bot. Your goal is to accomplish the user's request by calling functions to interact with the system. + +You have complete CRUD access to the filesystem and can execute terminal commands. Be aware of your actions and their consequences. + +Current Directory: {os.getcwd()} +The user's high-level goal is: "{self.initial_prompt}" + +Here is a summary of the steps taken so far: +{json.dumps(self.memory, indent=2) if self.memory else 'No actions taken yet.'} + +**NEW, STRICT RULES:** +1. **NO PLACEHOLDERS:** All code you generate must be a full, working implementation. Do not generate placeholders, stubs, or incomplete functions. +2. **TEST-DRIVEN DEVELOPMENT:** For any application code you write, you MUST also write a corresponding test file. +3. **EXECUTE TESTS:** After creating the code and the test file, you MUST execute the tests using `terminal_execute` (e.g., `python -m pytest test_file.py`). +4. **VALIDATE & DEBUG:** Analyze the `stdout` and `stderr` from the test execution. If there are any failures, you MUST debug the issue. This means re-reading the code, fixing it with `create_file`, and then re-running the tests. This loop (fix -> re-test) must continue until all tests pass. +5. **SPECIFY VERSIONS:** When installing packages (e.g., using pip), you MUST specify an exact version. For example: `pip install pytest==8.1.1`. Do not use commands like `pip install pytest`. +6. **TASK DECOMPOSITION:** If the prompt contains multiple distinct tasks (e.g., "create an API and then build a frontend for it"), you MUST use the `split_and_delegate` function first. +7. **FINISH CONDITION:** Only call `finish_task` when the entire requested functionality is implemented AND all corresponding tests are passing. +8. **AUTONOMY:** Do not ask for permission. Execute the plan. +""" + + def run(self): + """ + Main execution loop for the bot. + """ + print(f"{Colors.GREEN}▶ Starting bot with prompt: '{self.initial_prompt}'{Colors.RESET}") + + 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}") + response = client.chat.completions.create( + model="gpt-4-turbo", + messages=messages, + tools=self.tools, + tool_choice="auto", + ) + + response_message = response.choices[0].message + messages.append(response_message) + tool_calls = response_message.tool_calls + + if not tool_calls: + print(f"\n{Colors.CYAN}🤖 AI Response (no tool call):{Colors.RESET}") + no_tool_call_message = response_message.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." + }) + 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}") + 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)) + tool_result = function_to_call(**function_args) + except Exception as e: + print(f"{Colors.RED}❌ Error calling function '{function_name}': {e}{Colors.RESET}") + 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 + }) + + except Exception as e: + print(f"\n{Colors.RED}❌ An unexpected error occurred in the main loop: {e}{Colors.RESET}") + break + + def _write_file(self, filepath, content, purpose=""): + try: + dir_path = os.path.dirname(filepath) + if dir_path: + os.makedirs(dir_path, exist_ok=True) + 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}") + return {"status": "success", "filepath": filepath} + except Exception as e: + print(f"{Colors.RED}❌ Error creating file '{filepath}': {e}{Colors.RESET}") + return {"status": "error", "message": str(e)} + + def create_file(self, filepath, content): + return self._write_file(filepath, content, "application code") + + def create_test_file(self, filepath, content): + return self._write_file(filepath, content, "test code") + + def read_file(self, filepath): + try: + with open(filepath, 'r', encoding='utf-8') as f: + content = f.read() + print(f"{Colors.GREEN}✅ File '{filepath}' read successfully.{Colors.RESET}") + return {"status": "success", "content": content} + except FileNotFoundError: + print(f"{Colors.RED}❌ File not found: '{filepath}'{Colors.RESET}") + return {"status": "error", "message": "File not found."} + except Exception as e: + print(f"{Colors.RED}❌ Error reading file '{filepath}': {e}{Colors.RESET}") + return {"status": "error", "message": str(e)} + + def delete_file(self, filepath): + try: + os.remove(filepath) + print(f"{Colors.GREEN}✅ File '{filepath}' deleted successfully.{Colors.RESET}") + return {"status": "success", "filepath": filepath} + except FileNotFoundError: + print(f"{Colors.RED}❌ File not found: '{filepath}'{Colors.RESET}") + return {"status": "error", "message": "File not found."} + except Exception as e: + print(f"{Colors.RED}❌ Error deleting file '{filepath}': {e}{Colors.RESET}") + return {"status": "error", "message": str(e)} + + def terminal_execute(self, command): + print(f"Executing 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 + } + 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}") + 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}") + self.is_finished = True + return {"status": "finished"} + + def split_and_delegate(self, current_task_prompt, remaining_prompt): + print("\n" + "="*50) + print(" PYRAMID OF BUILDERS") + print("="*50) + print(f"{Colors.MAGENTA}🚀 Delegating remaining prompt to a new bot instance...{Colors.RESET}") + print(f" python {__file__} \"{remaining_prompt}\"") + try: + subprocess.Popen([sys.executable, __file__, remaining_prompt]) + print(f"{Colors.GREEN}✅ New bot instance launched successfully.{Colors.RESET}") + except Exception as e: + print(f"{Colors.RED}❌ Failed to launch new bot instance: {e}{Colors.RESET}") + 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(): + """ + Entry point for the CLI application. + """ + 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() + + bot = AutonomousBot(args.prompt) + bot.run() + +if __name__ == "__main__": + main()