#!/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()
Reference in New Issue View Git Blame Copy Permalink