Update, API version.

Browse Source
This commit is contained in:
retoor 2025-09-28 08:31:47 +02:00
parent 05ab6c1476
commit 93bcd444d2
1 changed files with 49 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

109
botje.py Normal file → Executable file
View File

@ -1,13 +1,11 @@
#!/usr/bin/env python3
import os import os
import argparse import argparse
import json import json
import subprocess import subprocess
from openai import OpenAI
import sys import sys
import shlex import shlex
import requests
# ANSI escape codes for colors
class Colors: class Colors:
RED = '\033[91m' RED = '\033[91m'
GREEN = '\033[92m' GREEN = '\033[92m'
@ -17,28 +15,13 @@ class Colors:
CYAN = '\033[96m' CYAN = '\033[96m'
RESET = '\033[0m' RESET = '\033[0m'
# Initialize the OpenAI client api_key = os.environ.get("OPENAI_API_KEY")
# It will automatically pick up the OPENAI_API_KEY from your environment variables. if not api_key:
try: print(f"{Colors.RED}Error: OPENAI_API_KEY environment variable not set.{Colors.RESET}")
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) exit(1)
class AutonomousBot: 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): def __init__(self, prompt, memory=None):
"""
Initializes the bot with a prompt and an optional memory of past actions.
"""
self.initial_prompt = prompt self.initial_prompt = prompt
self.memory = memory if memory is not None else [] self.memory = memory if memory is not None else []
self.is_finished = False self.is_finished = False
@ -156,7 +139,6 @@ class AutonomousBot:
} }
def get_system_prompt(self): 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. 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. You have complete CRUD access to the filesystem and can execute terminal commands. Be aware of your actions and their consequences.
@ -179,10 +161,7 @@ Here is a summary of the steps taken so far:
""" """
def run(self): def run(self):
""" print(f"{Colors.GREEN}Starting bot with prompt: '{self.initial_prompt}'{Colors.RESET}")
Main execution loop for the bot.
"""
print(f"{Colors.GREEN}▶ Starting bot with prompt: '{self.initial_prompt}'{Colors.RESET}")
messages = [ messages = [
{"role": "system", "content": self.get_system_prompt()}, {"role": "system", "content": self.get_system_prompt()},
@ -191,21 +170,33 @@ Here is a summary of the steps taken so far:
while not self.is_finished: while not self.is_finished:
try: try:
print(f"\n{Colors.BLUE}🤔 Thinking...{Colors.RESET}") 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 headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}"
}
payload = {
"model": "gpt-4-turbo",
"messages": messages,
"tools": self.tools,
"tool_choice": "auto"
}
response_raw = requests.post("https://api.openai.com/v1/chat/completions", headers=headers, json=payload)
if response_raw.status_code != 200:
print(f"{Colors.RED}Error from OpenAI API: {response_raw.status_code} - {response_raw.text}{Colors.RESET}")
break
response_json = response_raw.json()
response_message = response_json['choices'][0]['message']
messages.append(response_message) messages.append(response_message)
tool_calls = response_message.tool_calls tool_calls = response_message.get('tool_calls')
if not tool_calls: if not tool_calls:
print(f"\n{Colors.CYAN}🤖 AI Response (no tool call):{Colors.RESET}") 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." no_tool_call_message = response_message.get('content') or "No text content in response."
print(no_tool_call_message) print(no_tool_call_message)
messages.append({ messages.append({
@ -215,25 +206,25 @@ Here is a summary of the steps taken so far:
continue continue
for tool_call in tool_calls: for tool_call in tool_calls:
function_name = tool_call.function.name function_name = tool_call['function']['name']
function_to_call = self.available_functions.get(function_name) function_to_call = self.available_functions.get(function_name)
function_args = {} function_args = {}
if not function_to_call: if not function_to_call:
print(f"{Colors.RED}❌ Unknown function called by model: {function_name}{Colors.RESET}") print(f"{Colors.RED}Unknown function called by model: {function_name}{Colors.RESET}")
tool_result = {"status": "error", "message": f"Unknown function: {function_name}"} tool_result = {"status": "error", "message": f"Unknown function: {function_name}"}
else: else:
try: try:
function_args = json.loads(tool_call.function.arguments) function_args = json.loads(tool_call['function']['arguments'])
print(f"\n{Colors.GREEN}▶ Calling function: `{function_name}` with arguments:{Colors.RESET}") print(f"\n{Colors.GREEN}Calling function: `{function_name}` with arguments:{Colors.RESET}")
print(json.dumps(function_args, indent=2)) print(json.dumps(function_args, indent=2))
tool_result = function_to_call(**function_args) tool_result = function_to_call(**function_args)
except Exception as e: except Exception as e:
print(f"{Colors.RED}❌ Error calling function '{function_name}': {e}{Colors.RESET}") print(f"{Colors.RED}Error calling function '{function_name}': {e}{Colors.RESET}")
tool_result = {"status": "error", "message": str(e)} tool_result = {"status": "error", "message": str(e)}
messages.append({ messages.append({
"tool_call_id": tool_call.id, "tool_call_id": tool_call['id'],
"role": "tool", "role": "tool",
"name": function_name, "name": function_name,
"content": json.dumps(tool_result), "content": json.dumps(tool_result),
@ -245,7 +236,7 @@ Here is a summary of the steps taken so far:
}) })
except Exception as e: except Exception as e:
print(f"\n{Colors.RED}❌ An unexpected error occurred in the main loop: {e}{Colors.RESET}") print(f"\n{Colors.RED}An unexpected error occurred in the main loop: {e}{Colors.RESET}")
break break
def _write_file(self, filepath, content, purpose=""): def _write_file(self, filepath, content, purpose=""):
@ -256,10 +247,10 @@ Here is a summary of the steps taken so far:
with open(filepath, 'w', encoding='utf-8') as f: with open(filepath, 'w', encoding='utf-8') as f:
f.write(content) f.write(content)
purpose_text = f" ({purpose})" if purpose else "" purpose_text = f" ({purpose})" if purpose else ""
print(f"{Colors.GREEN}✅ File '{filepath}' created{purpose_text} successfully.{Colors.RESET}") print(f"{Colors.GREEN}File '{filepath}' created{purpose_text} successfully.{Colors.RESET}")
return {"status": "success", "filepath": filepath} return {"status": "success", "filepath": filepath}
except Exception as e: except Exception as e:
print(f"{Colors.RED}❌ Error creating file '{filepath}': {e}{Colors.RESET}") print(f"{Colors.RED}Error creating file '{filepath}': {e}{Colors.RESET}")
return {"status": "error", "message": str(e)} return {"status": "error", "message": str(e)}
def create_file(self, filepath, content): def create_file(self, filepath, content):
@ -272,25 +263,25 @@ Here is a summary of the steps taken so far:
try: try:
with open(filepath, 'r', encoding='utf-8') as f: with open(filepath, 'r', encoding='utf-8') as f:
content = f.read() content = f.read()
print(f"{Colors.GREEN}✅ File '{filepath}' read successfully.{Colors.RESET}") print(f"{Colors.GREEN}File '{filepath}' read successfully.{Colors.RESET}")
return {"status": "success", "content": content} return {"status": "success", "content": content}
except FileNotFoundError: except FileNotFoundError:
print(f"{Colors.RED}❌ File not found: '{filepath}'{Colors.RESET}") print(f"{Colors.RED}File not found: '{filepath}'{Colors.RESET}")
return {"status": "error", "message": "File not found."} return {"status": "error", "message": "File not found."}
except Exception as e: except Exception as e:
print(f"{Colors.RED}❌ Error reading file '{filepath}': {e}{Colors.RESET}") print(f"{Colors.RED}Error reading file '{filepath}': {e}{Colors.RESET}")
return {"status": "error", "message": str(e)} return {"status": "error", "message": str(e)}
def delete_file(self, filepath): def delete_file(self, filepath):
try: try:
os.remove(filepath) os.remove(filepath)
print(f"{Colors.GREEN}✅ File '{filepath}' deleted successfully.{Colors.RESET}") print(f"{Colors.GREEN}File '{filepath}' deleted successfully.{Colors.RESET}")
return {"status": "success", "filepath": filepath} return {"status": "success", "filepath": filepath}
except FileNotFoundError: except FileNotFoundError:
print(f"{Colors.RED}❌ File not found: '{filepath}'{Colors.RESET}") print(f"{Colors.RED}File not found: '{filepath}'{Colors.RESET}")
return {"status": "error", "message": "File not found."} return {"status": "error", "message": "File not found."}
except Exception as e: except Exception as e:
print(f"{Colors.RED}❌ Error deleting file '{filepath}': {e}{Colors.RESET}") print(f"{Colors.RED}Error deleting file '{filepath}': {e}{Colors.RESET}")
return {"status": "error", "message": str(e)} return {"status": "error", "message": str(e)}
def terminal_execute(self, command): def terminal_execute(self, command):
@ -311,11 +302,11 @@ Here is a summary of the steps taken so far:
print(f" - Return Code: {result.returncode}") print(f" - Return Code: {result.returncode}")
return output return output
except Exception as e: except Exception as e:
print(f"{Colors.RED}❌ Error executing command '{command}': {e}{Colors.RESET}") print(f"{Colors.RED}Error executing command '{command}': {e}{Colors.RESET}")
return {"status": "error", "message": str(e), "stdout": "", "stderr": str(e), "returncode": -1} return {"status": "error", "message": str(e), "stdout": "", "stderr": str(e), "returncode": -1}
def finish_task(self, reason): def finish_task(self, reason):
print(f"\n{Colors.YELLOW}🏁 BOT INSTANCE FINISHED: {reason}{Colors.RESET}") print(f"\n{Colors.YELLOW}BOT INSTANCE FINISHED: {reason}{Colors.RESET}")
self.is_finished = True self.is_finished = True
return {"status": "finished"} return {"status": "finished"}
@ -323,13 +314,13 @@ Here is a summary of the steps taken so far:
print("\n" + "="*50) print("\n" + "="*50)
print(" PYRAMID OF BUILDERS") print(" PYRAMID OF BUILDERS")
print("="*50) print("="*50)
print(f"{Colors.MAGENTA}🚀 Delegating remaining prompt to a new bot instance...{Colors.RESET}") print(f"{Colors.MAGENTA}Delegating remaining prompt to a new bot instance...{Colors.RESET}")
print(f" python {__file__} \"{remaining_prompt}\"") print(f" python {__file__} \"{remaining_prompt}\"")
try: try:
subprocess.Popen([sys.executable, __file__, remaining_prompt]) subprocess.Popen([sys.executable, __file__, remaining_prompt])
print(f"{Colors.GREEN}✅ New bot instance launched successfully.{Colors.RESET}") print(f"{Colors.GREEN}New bot instance launched successfully.{Colors.RESET}")
except Exception as e: except Exception as e:
print(f"{Colors.RED}❌ Failed to launch new bot instance: {e}{Colors.RESET}") print(f"{Colors.RED}Failed to launch new bot instance: {e}{Colors.RESET}")
return {"status": "error", "message": f"Failed to delegate: {e}"} return {"status": "error", "message": f"Failed to delegate: {e}"}
self.initial_prompt = current_task_prompt self.initial_prompt = current_task_prompt
@ -338,9 +329,6 @@ Here is a summary of the steps taken so far:
return {"status": "success", "message": "Task split. This bot will now handle the first part."} return {"status": "success", "message": "Task split. This bot will now handle the first part."}
def main(): def main():
"""
Entry point for the CLI application.
"""
parser = argparse.ArgumentParser(description="An autonomous bot to execute tasks.") 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.") parser.add_argument("prompt", type=str, help="The user prompt describing the task to be done.")
args = parser.parse_args() args = parser.parse_args()
@ -350,3 +338,4 @@ def main():
if __name__ == "__main__": if __name__ == "__main__":
main() main()