#!/usr/bin/env python3
import os
import argparse
import json
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'
GREEN = '\033[92m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
MAGENTA = '\033[95m'
CYAN = '\033[96m'
RESET = '\033[0m'
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
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:
def __init__(self, prompt, memory=None):
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",
"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):
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):
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}")
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:
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)
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:
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:
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 })
except Exception as e:
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:
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 ""
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:
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):
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):
logging.info(f"Attempting to read file: '{filepath}'")
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
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:
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:
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)
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:
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:
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 }
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:
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):
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"}
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}\"")
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__, prompt])
success_msg = "New bot instance launched successfully."
print(f"{Colors.GREEN}{success_msg}{Colors.RESET}")
logging.info(success_msg)
except Exception as e:
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()
Reference in New Issue View Git Blame Copy Permalink