aljaz/Auto-GPT
1
0
Fork 0
mirror of https://github.com/aljazceru/Auto-GPT.git synced 2025-12-18 06:24:20 +01:00
Code Issues Packages Projects Releases Wiki Activity

black formatting

Browse Source
This commit is contained in:
SwiftyOS
2023-09-16 18:28:26 +02:00
parent 7fda0c7884
commit 2211efc800
1 changed files with 411 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files

545
cli.py
View File

@@ -3,8 +3,9 @@ try:
import github import github
except ImportError: except ImportError:
import os import os
os.system('pip3 install click')
os.system('pip3 install PyGithub') os.system("pip3 install click")
os.system("pip3 install PyGithub")
import click import click
@@ -12,69 +13,148 @@ except ImportError:
def cli(): def cli():
pass pass
@cli.command() @cli.command()
def setup(): def setup():
"""Installs dependencies needed for your system. Works with Linux, MacOS and Windows WSL.""" """Installs dependencies needed for your system. Works with Linux, MacOS and Windows WSL."""
import os import os
import subprocess import subprocess
script_dir = os.path.dirname(os.path.realpath(__file__)) script_dir = os.path.dirname(os.path.realpath(__file__))
setup_script = os.path.join(script_dir, 'setup.sh') setup_script = os.path.join(script_dir, "setup.sh")
if os.path.exists(setup_script): if os.path.exists(setup_script):
subprocess.Popen([setup_script], cwd=script_dir) subprocess.Popen([setup_script], cwd=script_dir)
click.echo(click.style("🚀 Setup initiated", fg='green')) click.echo(click.style("🚀 Setup initiated", fg="green"))
else: else:
click.echo(click.style("❌ Error: setup.sh does not exist in the current directory.", fg='red')) click.echo(
click.style(
"❌ Error: setup.sh does not exist in the current directory.", fg="red"
)
)
try: try:
# Check if GitHub user name is configured # Check if GitHub user name is configured
user_name = subprocess.check_output(['git', 'config', 'user.name']).decode('utf-8').strip() user_name = (
user_email = subprocess.check_output(['git', 'config', 'user.email']).decode('utf-8').strip() subprocess.check_output(["git", "config", "user.name"])
.decode("utf-8")
.strip()
)
user_email = (
subprocess.check_output(["git", "config", "user.email"])
.decode("utf-8")
.strip()
)
if user_name and user_email: if user_name and user_email:
click.echo(click.style(f"✅ GitHub account is configured with username: {user_name} and email: {user_email}", fg='green')) click.echo(
click.style(
f"✅ GitHub account is configured with username: {user_name} and email: {user_email}",
fg="green",
)
)
else: else:
raise subprocess.CalledProcessError(returncode=1, cmd='git config user.name or user.email') raise subprocess.CalledProcessError(
returncode=1, cmd="git config user.name or user.email"
)
except subprocess.CalledProcessError: except subprocess.CalledProcessError:
# If the GitHub account is not configured, print instructions on how to set it up # If the GitHub account is not configured, print instructions on how to set it up
click.echo(click.style("❌ GitHub account is not configured.", fg='red')) click.echo(click.style("❌ GitHub account is not configured.", fg="red"))
click.echo(click.style("To configure your GitHub account, use the following commands:", fg='red')) click.echo(
click.echo(click.style(" git config --global user.name \"Your GitHub Username\"", fg='red')) click.style(
click.echo(click.style(" git config --global user.email \"Your GitHub Email\"", fg='red')) "To configure your GitHub account, use the following commands:",
fg="red",
)
)
click.echo(
click.style(
' git config --global user.name "Your GitHub Username"', fg="red"
)
)
click.echo(
click.style(
' git config --global user.email "Your GitHub Email"', fg="red"
)
)
# Check for the existence of the .github_access_token file # Check for the existence of the .github_access_token file
if os.path.exists('.github_access_token'): if os.path.exists(".github_access_token"):
with open('.github_access_token', 'r') as file: with open(".github_access_token", "r") as file:
github_access_token = file.read().strip() github_access_token = file.read().strip()
if github_access_token: if github_access_token:
click.echo(click.style("✅ GitHub access token loaded successfully.", fg='green')) click.echo(
click.style(
"✅ GitHub access token loaded successfully.", fg="green"
)
)
# Check if the token has the required permissions # Check if the token has the required permissions
import requests import requests
headers = {'Authorization': f'token {github_access_token}'}
response = requests.get('https://api.github.com/user', headers=headers) headers = {"Authorization": f"token {github_access_token}"}
response = requests.get("https://api.github.com/user", headers=headers)
if response.status_code == 200: if response.status_code == 200:
scopes = response.headers.get('X-OAuth-Scopes') scopes = response.headers.get("X-OAuth-Scopes")
if 'public_repo' in scopes or 'repo' in scopes: if "public_repo" in scopes or "repo" in scopes:
click.echo(click.style("✅ GitHub access token has the required permissions.", fg='green')) click.echo(
click.style(
"✅ GitHub access token has the required permissions.",
fg="green",
)
)
else: else:
click.echo(click.style("❌ GitHub access token does not have the required permissions. Please ensure it has 'public_repo' or 'repo' scope.", fg='red')) click.echo(
click.style(
"❌ GitHub access token does not have the required permissions. Please ensure it has 'public_repo' or 'repo' scope.",
fg="red",
)
)
else: else:
click.echo(click.style("❌ Failed to validate GitHub access token. Please ensure it is correct.", fg='red')) click.echo(
click.style(
"❌ Failed to validate GitHub access token. Please ensure it is correct.",
fg="red",
)
)
else: else:
click.echo(click.style("❌ GitHub access token file is empty. Please follow the instructions below to set up your GitHub access token.", fg='red')) click.echo(
click.style(
"❌ GitHub access token file is empty. Please follow the instructions below to set up your GitHub access token.",
fg="red",
)
)
else: else:
# Create the .github_access_token file if it doesn't exist # Create the .github_access_token file if it doesn't exist
with open('.github_access_token', 'w') as file: with open(".github_access_token", "w") as file:
file.write('') file.write("")
# Instructions to set up GitHub access token # Instructions to set up GitHub access token
click.echo(click.style("❌ To configure your GitHub access token, follow these steps:", fg='red')) click.echo(
click.echo(click.style("\t1. Ensure you are logged into your GitHub account", fg='red')) click.style(
click.echo(click.style("\t2. Navigate to https://github.com/settings/tokens", fg='red')) "❌ To configure your GitHub access token, follow these steps:", fg="red"
click.echo(click.style("\t6. Click on 'Generate new token'.", fg='red')) )
click.echo(click.style("\t7. Fill out the form to generate a new token. Ensure you select the 'repo' scope.", fg='red')) )
click.echo(click.style("\t8. Open the '.github_access_token' file in the same directory as this script and paste the token into this file.", fg='red')) click.echo(
click.echo(click.style("\t9. Save the file and run the setup command again.", fg='red')) click.style("\t1. Ensure you are logged into your GitHub account", fg="red")
)
click.echo(
click.style("\t2. Navigate to https://github.com/settings/tokens", fg="red")
)
click.echo(click.style("\t6. Click on 'Generate new token'.", fg="red"))
click.echo(
click.style(
"\t7. Fill out the form to generate a new token. Ensure you select the 'repo' scope.",
fg="red",
)
)
click.echo(
click.style(
"\t8. Open the '.github_access_token' file in the same directory as this script and paste the token into this file.",
fg="red",
)
)
click.echo(
click.style("\t9. Save the file and run the setup command again.", fg="red")
)
@cli.group() @cli.group()
@@ -82,54 +162,89 @@ def agent():
"""Commands to create, start and stop agents""" """Commands to create, start and stop agents"""
pass pass
@agent.command() @agent.command()
@click.argument('agent_name') @click.argument("agent_name")
def create(agent_name): def create(agent_name):
"""Create's a new agent with the agent name provieded""" """Create's a new agent with the agent name provieded"""
import os import os
import shutil
import re import re
import shutil
if not re.match("^[a-zA-Z0-9_-]*$", agent_name): if not re.match("^[a-zA-Z0-9_-]*$", agent_name):
click.echo(click.style(f"😞 Agent name '{agent_name}' is not valid. It should not contain spaces or special characters other than -_", fg='red')) click.echo(
click.style(
f"😞 Agent name '{agent_name}' is not valid. It should not contain spaces or special characters other than -_",
fg="red",
)
)
return return
try: try:
new_agent_dir = f'./autogpts/{agent_name}' new_agent_dir = f"./autogpts/{agent_name}"
agent_json_file = f'./arena/{agent_name}.json' agent_json_file = f"./arena/{agent_name}.json"
if not os.path.exists(new_agent_dir) and not os.path.exists(agent_json_file): if not os.path.exists(new_agent_dir) and not os.path.exists(agent_json_file):
shutil.copytree('./autogpts/forge', new_agent_dir) shutil.copytree("./autogpts/forge", new_agent_dir)
click.echo(click.style(f"🎉 New agent '{agent_name}' created. The code for your new agent is in: autogpts/{agent_name}", fg='green')) click.echo(
click.echo(click.style(f"🚀 If you would like to enter the arena, run './run arena enter {agent_name}'", fg='yellow')) click.style(
f"🎉 New agent '{agent_name}' created. The code for your new agent is in: autogpts/{agent_name}",
fg="green",
)
)
click.echo(
click.style(
f"🚀 If you would like to enter the arena, run './run arena enter {agent_name}'",
fg="yellow",
)
)
else: else:
click.echo(click.style(f"😞 Agent '{agent_name}' already exists. Enter a different name for your agent", fg='red')) click.echo(
click.style(
f"😞 Agent '{agent_name}' already exists. Enter a different name for your agent",
fg="red",
)
)
except Exception as e: except Exception as e:
click.echo(click.style(f"😢 An error occurred: {e}", fg='red')) click.echo(click.style(f"😢 An error occurred: {e}", fg="red"))
@agent.command() @agent.command()
@click.argument('agent_name') @click.argument("agent_name")
def start(agent_name): def start(agent_name):
"""Start agent command""" """Start agent command"""
import os import os
import subprocess import subprocess
script_dir = os.path.dirname(os.path.realpath(__file__)) script_dir = os.path.dirname(os.path.realpath(__file__))
agent_dir = os.path.join(script_dir, f'autogpts/{agent_name}') agent_dir = os.path.join(script_dir, f"autogpts/{agent_name}")
run_command = os.path.join(agent_dir, 'run') run_command = os.path.join(agent_dir, "run")
if os.path.exists(agent_dir) and os.path.isfile(run_command): if os.path.exists(agent_dir) and os.path.isfile(run_command):
os.chdir(agent_dir) os.chdir(agent_dir)
subprocess.Popen(["./run"], cwd=agent_dir) subprocess.Popen(["./run"], cwd=agent_dir)
click.echo(f"Agent '{agent_name}' started") click.echo(f"Agent '{agent_name}' started")
elif not os.path.exists(agent_dir): elif not os.path.exists(agent_dir):
click.echo(click.style(f"😞 Agent '{agent_name}' does not exist. Please create the agent first.", fg='red')) click.echo(
click.style(
f"😞 Agent '{agent_name}' does not exist. Please create the agent first.",
fg="red",
)
)
else: else:
click.echo(click.style(f"😞 Run command does not exist in the agent '{agent_name}' directory.", fg='red')) click.echo(
click.style(
f"😞 Run command does not exist in the agent '{agent_name}' directory.",
fg="red",
)
)
@agent.command() @agent.command()
def stop(): def stop():
"""Stop agent command""" """Stop agent command"""
import subprocess
import os import os
import signal import signal
import subprocess
try: try:
pid = int(subprocess.check_output(["lsof", "-t", "-i", ":8000"])) pid = int(subprocess.check_output(["lsof", "-t", "-i", ":8000"]))
os.kill(pid, signal.SIGTERM) os.kill(pid, signal.SIGTERM)
@@ -144,19 +259,24 @@ def stop():
def list(): def list():
"""List agents command""" """List agents command"""
import os import os
try: try:
agents_dir = './autogpts' agents_dir = "./autogpts"
agents_list = [d for d in os.listdir(agents_dir) if os.path.isdir(os.path.join(agents_dir, d))] agents_list = [
d
for d in os.listdir(agents_dir)
if os.path.isdir(os.path.join(agents_dir, d))
]
if agents_list: if agents_list:
click.echo(click.style('Available agents: 🤖', fg='green')) click.echo(click.style("Available agents: 🤖", fg="green"))
for agent in agents_list: for agent in agents_list:
click.echo(click.style(f"\t🐙 {agent}", fg='blue')) click.echo(click.style(f"\t🐙 {agent}", fg="blue"))
else: else:
click.echo(click.style("No agents found 😞", fg='red')) click.echo(click.style("No agents found 😞", fg="red"))
except FileNotFoundError: except FileNotFoundError:
click.echo(click.style("The autogpts directory does not exist 😢", fg='red')) click.echo(click.style("The autogpts directory does not exist 😢", fg="red"))
except Exception as e: except Exception as e:
click.echo(click.style(f"An error occurred: {e} 😢", fg='red')) click.echo(click.style(f"An error occurred: {e} 😢", fg="red"))
@cli.group() @cli.group()
@@ -164,43 +284,61 @@ def benchmark():
"""Commands to start the benchmark and list tests and categories""" """Commands to start the benchmark and list tests and categories"""
pass pass
@benchmark.command(context_settings=dict(
@benchmark.command(
context_settings=dict(
ignore_unknown_options=True, ignore_unknown_options=True,
)) )
@click.argument('agent_name') )
@click.argument('subprocess_args', nargs=-1, type=click.UNPROCESSED) @click.argument("agent_name")
@click.argument("subprocess_args", nargs=-1, type=click.UNPROCESSED)
def start(agent_name, subprocess_args): def start(agent_name, subprocess_args):
"""Starts the benchmark command""" """Starts the benchmark command"""
import os import os
import subprocess import subprocess
script_dir = os.path.dirname(os.path.realpath(__file__)) script_dir = os.path.dirname(os.path.realpath(__file__))
agent_dir = os.path.join(script_dir, f'autogpts/{agent_name}') agent_dir = os.path.join(script_dir, f"autogpts/{agent_name}")
benchmark_script = os.path.join(agent_dir, 'run_benchmark.sh') benchmark_script = os.path.join(agent_dir, "run_benchmark.sh")
if os.path.exists(agent_dir) and os.path.isfile(benchmark_script): if os.path.exists(agent_dir) and os.path.isfile(benchmark_script):
os.chdir(agent_dir) os.chdir(agent_dir)
subprocess.Popen([benchmark_script, *subprocess_args], cwd=agent_dir) subprocess.Popen([benchmark_script, *subprocess_args], cwd=agent_dir)
click.echo(click.style(f"🚀 Running benchmark for '{agent_name}' with subprocess arguments: {' '.join(subprocess_args)}", fg='green')) click.echo(
click.style(
f"🚀 Running benchmark for '{agent_name}' with subprocess arguments: {' '.join(subprocess_args)}",
fg="green",
)
)
else: else:
click.echo(click.style(f"😞 Agent '{agent_name}' does not exist. Please create the agent first.", fg='red')) click.echo(
click.style(
f"😞 Agent '{agent_name}' does not exist. Please create the agent first.",
fg="red",
)
)
@benchmark.group(name='categories') @benchmark.group(name="categories")
def benchmark_categories(): def benchmark_categories():
"""Benchmark categories group command""" """Benchmark categories group command"""
pass pass
@benchmark_categories.command(name='list')
@benchmark_categories.command(name="list")
def benchmark_categories_list(): def benchmark_categories_list():
"""List benchmark categories command""" """List benchmark categories command"""
import os
import json
import glob import glob
import json
import os
categories = set() categories = set()
# Get the directory of this file # Get the directory of this file
this_dir = os.path.dirname(os.path.abspath(__file__)) this_dir = os.path.dirname(os.path.abspath(__file__))
glob_path = os.path.join(this_dir, "./benchmark/agbenchmark/challenges/**/[!deprecated]*/data.json") glob_path = os.path.join(
this_dir, "./benchmark/agbenchmark/challenges/**/[!deprecated]*/data.json"
)
# Use it as the base for the glob pattern, excluding 'deprecated' directory # Use it as the base for the glob pattern, excluding 'deprecated' directory
for data_file in glob.glob(glob_path, recursive=True): for data_file in glob.glob(glob_path, recursive=True):
with open(data_file, "r") as f: with open(data_file, "r") as f:
@@ -215,30 +353,35 @@ def benchmark_categories_list():
continue continue
if categories: if categories:
click.echo(click.style('Available categories: 📚', fg='green')) click.echo(click.style("Available categories: 📚", fg="green"))
for category in categories: for category in categories:
click.echo(click.style(f"\t📖 {category}", fg='blue')) click.echo(click.style(f"\t📖 {category}", fg="blue"))
else: else:
click.echo(click.style("No categories found 😞", fg='red')) click.echo(click.style("No categories found 😞", fg="red"))
@benchmark.group(name='tests')
@benchmark.group(name="tests")
def benchmark_tests(): def benchmark_tests():
"""Benchmark tests group command""" """Benchmark tests group command"""
pass pass
@benchmark_tests.command(name='list')
@benchmark_tests.command(name="list")
def benchmark_tests_list(): def benchmark_tests_list():
"""List benchmark tests command""" """List benchmark tests command"""
import os
import json
import glob import glob
import json
import os
import re import re
tests = {} tests = {}
# Get the directory of this file # Get the directory of this file
this_dir = os.path.dirname(os.path.abspath(__file__)) this_dir = os.path.dirname(os.path.abspath(__file__))
glob_path = os.path.join(this_dir, "./benchmark/agbenchmark/challenges/**/[!deprecated]*/data.json") glob_path = os.path.join(
this_dir, "./benchmark/agbenchmark/challenges/**/[!deprecated]*/data.json"
)
# Use it as the base for the glob pattern, excluding 'deprecated' directory # Use it as the base for the glob pattern, excluding 'deprecated' directory
for data_file in glob.glob(glob_path, recursive=True): for data_file in glob.glob(glob_path, recursive=True):
with open(data_file, "r") as f: with open(data_file, "r") as f:
@@ -258,64 +401,133 @@ def benchmark_tests_list():
continue continue
if tests: if tests:
click.echo(click.style('Available tests: 📚', fg='green')) click.echo(click.style("Available tests: 📚", fg="green"))
for category, test_list in tests.items(): for category, test_list in tests.items():
click.echo(click.style(f"\t📖 {category}", fg='blue')) click.echo(click.style(f"\t📖 {category}", fg="blue"))
for test in sorted(test_list): for test in sorted(test_list):
test_name = ' '.join(word for word in re.split('([A-Z][a-z]*)', test) if word).replace('_', '').replace('C L I', 'CLI')[5:].replace(' ', ' ') test_name = (
" ".join(word for word in re.split("([A-Z][a-z]*)", test) if word)
.replace("_", "")
.replace("C L I", "CLI")[5:]
.replace(" ", " ")
)
test_name_padded = f"{test_name:<40}" test_name_padded = f"{test_name:<40}"
click.echo(click.style(f"\t\t🔬 {test_name_padded} - {test}", fg='cyan')) click.echo(click.style(f"\t\t🔬 {test_name_padded} - {test}", fg="cyan"))
else: else:
click.echo(click.style("No tests found 😞", fg='red')) click.echo(click.style("No tests found 😞", fg="red"))
@benchmark_tests.command(name='details')
@click.argument('test_name') @benchmark_tests.command(name="details")
@click.argument("test_name")
def benchmark_tests_details(test_name): def benchmark_tests_details(test_name):
"""Benchmark test details command""" """Benchmark test details command"""
import os
import json
import glob import glob
import json
import os
# Get the directory of this file # Get the directory of this file
this_dir = os.path.dirname(os.path.abspath(__file__)) this_dir = os.path.dirname(os.path.abspath(__file__))
glob_path = os.path.join(this_dir, "./benchmark/agbenchmark/challenges/**/[!deprecated]*/data.json") glob_path = os.path.join(
this_dir, "./benchmark/agbenchmark/challenges/**/[!deprecated]*/data.json"
)
# Use it as the base for the glob pattern, excluding 'deprecated' directory # Use it as the base for the glob pattern, excluding 'deprecated' directory
for data_file in glob.glob(glob_path, recursive=True): for data_file in glob.glob(glob_path, recursive=True):
with open(data_file, "r") as f: with open(data_file, "r") as f:
try: try:
data = json.load(f) data = json.load(f)
if data.get("name") == test_name: if data.get("name") == test_name:
click.echo(click.style(f"\n{data.get('name')}\n{'-'*len(data.get('name'))}\n", fg='blue')) click.echo(
click.echo(click.style(f"\tCategory: {', '.join(data.get('category'))}", fg='green')) click.style(
click.echo(click.style(f"\tTask: {data.get('task')}", fg='green')) f"\n{data.get('name')}\n{'-'*len(data.get('name'))}\n",
click.echo(click.style(f"\tDependencies: {', '.join(data.get('dependencies')) if data.get('dependencies') else 'None'}", fg='green')) fg="blue",
click.echo(click.style(f"\tCutoff: {data.get('cutoff')}\n", fg='green')) )
click.echo(click.style("\tTest Conditions\n\t-------", fg='magenta')) )
click.echo(click.style(f"\t\tAnswer: {data.get('ground').get('answer')}", fg='magenta')) click.echo(
click.echo(click.style(f"\t\tShould Contain: {', '.join(data.get('ground').get('should_contain'))}", fg='magenta')) click.style(
click.echo(click.style(f"\t\tShould Not Contain: {', '.join(data.get('ground').get('should_not_contain'))}", fg='magenta')) f"\tCategory: {', '.join(data.get('category'))}",
click.echo(click.style(f"\t\tFiles: {', '.join(data.get('ground').get('files'))}", fg='magenta')) fg="green",
click.echo(click.style(f"\t\tEval: {data.get('ground').get('eval').get('type')}\n", fg='magenta')) )
click.echo(click.style("\tInfo\n\t-------", fg='yellow')) )
click.echo(click.style(f"\t\tDifficulty: {data.get('info').get('difficulty')}", fg='yellow')) click.echo(click.style(f"\tTask: {data.get('task')}", fg="green"))
click.echo(click.style(f"\t\tDescription: {data.get('info').get('description')}", fg='yellow')) click.echo(
click.echo(click.style(f"\t\tSide Effects: {', '.join(data.get('info').get('side_effects'))}", fg='yellow')) click.style(
f"\tDependencies: {', '.join(data.get('dependencies')) if data.get('dependencies') else 'None'}",
fg="green",
)
)
click.echo(
click.style(f"\tCutoff: {data.get('cutoff')}\n", fg="green")
)
click.echo(
click.style("\tTest Conditions\n\t-------", fg="magenta")
)
click.echo(
click.style(
f"\t\tAnswer: {data.get('ground').get('answer')}",
fg="magenta",
)
)
click.echo(
click.style(
f"\t\tShould Contain: {', '.join(data.get('ground').get('should_contain'))}",
fg="magenta",
)
)
click.echo(
click.style(
f"\t\tShould Not Contain: {', '.join(data.get('ground').get('should_not_contain'))}",
fg="magenta",
)
)
click.echo(
click.style(
f"\t\tFiles: {', '.join(data.get('ground').get('files'))}",
fg="magenta",
)
)
click.echo(
click.style(
f"\t\tEval: {data.get('ground').get('eval').get('type')}\n",
fg="magenta",
)
)
click.echo(click.style("\tInfo\n\t-------", fg="yellow"))
click.echo(
click.style(
f"\t\tDifficulty: {data.get('info').get('difficulty')}",
fg="yellow",
)
)
click.echo(
click.style(
f"\t\tDescription: {data.get('info').get('description')}",
fg="yellow",
)
)
click.echo(
click.style(
f"\t\tSide Effects: {', '.join(data.get('info').get('side_effects'))}",
fg="yellow",
)
)
break break
except json.JSONDecodeError: except json.JSONDecodeError:
print(f"Error: {data_file} is not a valid JSON file.") print(f"Error: {data_file} is not a valid JSON file.")
continue continue
except IOError: except IOError:
print(f"IOError: file could not be read: {data_file}") print(f"IOError: file could not be read: {data_file}")
continue continue
@cli.command() @cli.command()
def frontend(): def frontend():
"""Starts the frontend""" """Starts the frontend"""
import os import os
import subprocess
import socket import socket
import subprocess
try: try:
output = subprocess.check_output(["lsof", "-t", "-i", ":8000"]) output = subprocess.check_output(["lsof", "-t", "-i", ":8000"])
if output: if output:
@@ -325,8 +537,8 @@ def frontend():
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
click.echo("Error: Unexpected error occurred.") click.echo("Error: Unexpected error occurred.")
return return
frontend_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'frontend') frontend_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), "frontend")
run_file = os.path.join(frontend_dir, 'run') run_file = os.path.join(frontend_dir, "run")
if os.path.exists(frontend_dir) and os.path.isfile(run_file): if os.path.exists(frontend_dir) and os.path.isfile(run_file):
subprocess.Popen(["./run"], cwd=frontend_dir) subprocess.Popen(["./run"], cwd=frontend_dir)
click.echo("Launching frontend") click.echo("Launching frontend")
@@ -334,49 +546,79 @@ def frontend():
click.echo("Error: Frontend directory or run file does not exist.") click.echo("Error: Frontend directory or run file does not exist.")
@cli.group() @cli.group()
def arena(): def arena():
"""Commands to enter the arena""" """Commands to enter the arena"""
pass pass
@arena.command() @arena.command()
@click.argument('agent_name') @click.argument("agent_name")
@click.option('--branch', default='master', help='Branch to use instead of master') @click.option("--branch", default="master", help="Branch to use instead of master")
def enter(agent_name, branch): def enter(agent_name, branch):
import subprocess
from github import Github
from datetime import datetime
import os
import json import json
import os
import subprocess
from datetime import datetime
from github import Github
# Check if the agent_name directory exists in the autogpts directory # Check if the agent_name directory exists in the autogpts directory
agent_dir = f'./autogpts/{agent_name}' agent_dir = f"./autogpts/{agent_name}"
if not os.path.exists(agent_dir): if not os.path.exists(agent_dir):
click.echo(click.style(f"❌ The directory for agent '{agent_name}' does not exist in the autogpts directory.", fg='red')) click.echo(
click.echo(click.style(f"🚀 Run './run agent create {agent_name}' to create the agent.", fg='yellow')) click.style(
f"❌ The directory for agent '{agent_name}' does not exist in the autogpts directory.",
fg="red",
)
)
click.echo(
click.style(
f"🚀 Run './run agent create {agent_name}' to create the agent.",
fg="yellow",
)
)
return return
else: else:
# Check if the agent has already entered the arena # Check if the agent has already entered the arena
if os.path.exists(f'arena/{agent_name}.json'): if os.path.exists(f"arena/{agent_name}.json"):
click.echo(click.style(f"⚠️ The agent '{agent_name}' has already entered the arena. Use './run arena submit' to update your submission.", fg='yellow')) click.echo(
click.style(
f"⚠️ The agent '{agent_name}' has already entered the arena. Use './run arena submit' to update your submission.",
fg="yellow",
)
)
return return
# Check if there are staged changes # Check if there are staged changes
staged_changes = [line for line in subprocess.check_output(['git', 'status', '--porcelain']).decode('utf-8').split('\n') if line and line[0] in ('A', 'M', 'D', 'R', 'C')] staged_changes = [
line
for line in subprocess.check_output(["git", "status", "--porcelain"])
.decode("utf-8")
.split("\n")
if line and line[0] in ("A", "M", "D", "R", "C")
]
if staged_changes: if staged_changes:
click.echo(click.style(f"❌ There are staged changes. Please commit or stash them and run the command again.", fg='red')) click.echo(
click.style(
f"❌ There are staged changes. Please commit or stash them and run the command again.",
fg="red",
)
)
return return
try: try:
# Load GitHub access token from file # Load GitHub access token from file
with open('.github_access_token', 'r') as file: with open(".github_access_token", "r") as file:
github_access_token = file.read().strip() github_access_token = file.read().strip()
# Get GitHub repository URL # Get GitHub repository URL
github_repo_url = subprocess.check_output(['git', 'config', '--get', 'remote.origin.url']).decode('utf-8').strip() github_repo_url = (
subprocess.check_output(["git", "config", "--get", "remote.origin.url"])
.decode("utf-8")
.strip()
)
# If --branch is passed, use it instead of master # If --branch is passed, use it instead of master
if branch: if branch:
@@ -385,12 +627,16 @@ def enter(agent_name, branch):
branch_to_use = "master" branch_to_use = "master"
# Get the commit hash of HEAD of the branch_to_use # Get the commit hash of HEAD of the branch_to_use
commit_hash_to_benchmark = subprocess.check_output(['git', 'rev-parse', branch_to_use]).decode('utf-8').strip() commit_hash_to_benchmark = (
subprocess.check_output(["git", "rev-parse", branch_to_use])
.decode("utf-8")
.strip()
)
arena_submission_branch = f'arena_submission_{agent_name}' arena_submission_branch = f"arena_submission_{agent_name}"
# Create a new branch called arena_submission_{agent_name} # Create a new branch called arena_submission_{agent_name}
# subprocess.check_call(['git', 'checkout', '-b', arena_submission_branch]) # subprocess.check_call(['git', 'checkout', '-b', arena_submission_branch])
subprocess.check_call(['git', 'checkout', arena_submission_branch]) subprocess.check_call(["git", "checkout", arena_submission_branch])
# Create a dictionary with the necessary fields # Create a dictionary with the necessary fields
data = { data = {
"github_repo_url": github_repo_url, "github_repo_url": github_repo_url,
@@ -403,50 +649,61 @@ def enter(agent_name, branch):
data["branch_to_benchmark"] = branch data["branch_to_benchmark"] = branch
# Create agent directory if it does not exist # Create agent directory if it does not exist
subprocess.check_call(['mkdir', '-p', 'arena']) subprocess.check_call(["mkdir", "-p", "arena"])
# Create a JSON file with the data # Create a JSON file with the data
with open(f'arena/{agent_name}.json', 'w') as json_file: with open(f"arena/{agent_name}.json", "w") as json_file:
json.dump(data, json_file, indent=4) json.dump(data, json_file, indent=4)
# Create a commit with the specified message # Create a commit with the specified message
subprocess.check_call(['git', 'add', f'arena/{agent_name}.json']) subprocess.check_call(["git", "add", f"arena/{agent_name}.json"])
subprocess.check_call(['git', 'commit', '-m', f'{agent_name} entering the arena']) subprocess.check_call(
["git", "commit", "-m", f"{agent_name} entering the arena"]
)
# Push the commit # Push the commit
subprocess.check_call(['git', 'push', 'origin', arena_submission_branch]) subprocess.check_call(["git", "push", "origin", arena_submission_branch])
# Create a PR into the parent repository # Create a PR into the parent repository
g = Github(github_access_token) g = Github(github_access_token)
repo = g.get_repo(github_repo_url.split(':')[-1].split('.git')[0]) repo = g.get_repo(github_repo_url.split(":")[-1].split(".git")[0])
parent_repo = repo.parent parent_repo = repo.parent
if parent_repo: if parent_repo:
pr = parent_repo.create_pull( pr = parent_repo.create_pull(
title=f'{agent_name} entering the arena', title=f"{agent_name} entering the arena",
body='''**Introduction:** body="""**Introduction:**
**Team Members:** **Team Members:**
**What we are working on:** **What we are working on:**
Please replace this text with your own introduction, the names of your team members, and a brief description of your work.''', Please replace this text with your own introduction, the names of your team members, and a brief description of your work.""",
head=f'arena_submission_{agent_name}', head=f"arena_submission_{agent_name}",
base=branch_to_use, base=branch_to_use,
) )
click.echo(click.style(f"🚀 {agent_name} has entered the arena! Please edit your PR description at the following URL: {pr.html_url}", fg='green')) click.echo(
click.style(
f"🚀 {agent_name} has entered the arena! Please edit your PR description at the following URL: {pr.html_url}",
fg="green",
)
)
else: else:
click.echo(click.style("❌ This repository does not have a parent repository to sync with.", fg='red')) click.echo(
click.style(
"❌ This repository does not have a parent repository to sync with.",
fg="red",
)
)
return return
# Switch back to the master branch # Switch back to the master branch
subprocess.check_call(['git', 'checkout', branch_to_use]) subprocess.check_call(["git", "checkout", branch_to_use])
except Exception as e: except Exception as e:
click.echo(click.style(f"❌ An error occurred: {e}", fg='red')) click.echo(click.style(f"❌ An error occurred: {e}", fg="red"))
# Switch back to the master branch # Switch back to the master branch
subprocess.check_call(['git', 'checkout', branch_to_use]) subprocess.check_call(["git", "checkout", branch_to_use])
if __name__ == '__main__': if __name__ == "__main__":
cli() cli()