mirror of
https://github.com/aljazceru/Auto-GPT.git
synced 2025-12-17 14:04:27 +01:00
Merge branch 'master' into prompt-generator
This commit is contained in:
Alrik Olson
225
scripts/main.py
225
scripts/main.py
@@ -20,16 +20,18 @@ from prompt import get_prompt
|
||||
|
||||
cfg = Config()
|
||||
|
||||
|
||||
def check_openai_api_key():
|
||||
"""Check if the OpenAI API key is set in config.py or as an environment variable."""
|
||||
if not cfg.openai_api_key:
|
||||
print(
|
||||
Fore.RED +
|
||||
"Please set your OpenAI API key in config.py or as an environment variable."
|
||||
"Please set your OpenAI API key in .env or as an environment variable."
|
||||
)
|
||||
print("You can get your key from https://beta.openai.com/account/api-keys")
|
||||
exit(1)
|
||||
|
||||
|
||||
def attempt_to_fix_json_by_finding_outermost_brackets(json_string):
|
||||
if cfg.speak_mode and cfg.debug_mode:
|
||||
speak.say_text("I have received an invalid JSON response from the OpenAI API. Trying to fix it now.")
|
||||
@@ -58,6 +60,7 @@ def attempt_to_fix_json_by_finding_outermost_brackets(json_string):
|
||||
|
||||
return json_string
|
||||
|
||||
|
||||
def print_assistant_thoughts(assistant_reply):
|
||||
"""Prints the assistant's thoughts to the console"""
|
||||
global ai_name
|
||||
@@ -262,6 +265,7 @@ def prompt_user():
|
||||
config = AIConfig(ai_name, ai_role, ai_goals)
|
||||
return config
|
||||
|
||||
|
||||
def parse_arguments():
|
||||
"""Parses the arguments passed to the script"""
|
||||
global cfg
|
||||
@@ -316,118 +320,123 @@ def parse_arguments():
|
||||
cfg.memory_backend = chosen
|
||||
|
||||
|
||||
# TODO: fill in llm values here
|
||||
check_openai_api_key()
|
||||
parse_arguments()
|
||||
logger.set_level(logging.DEBUG if cfg.debug_mode else logging.INFO)
|
||||
ai_name = ""
|
||||
prompt = construct_prompt()
|
||||
# print(prompt)
|
||||
# Initialize variables
|
||||
full_message_history = []
|
||||
result = None
|
||||
next_action_count = 0
|
||||
# Make a constant:
|
||||
user_input = "Determine which next command to use, and respond using the format specified above:"
|
||||
def main():
|
||||
global ai_name, memory
|
||||
# TODO: fill in llm values here
|
||||
check_openai_api_key()
|
||||
parse_arguments()
|
||||
logger.set_level(logging.DEBUG if cfg.debug_mode else logging.INFO)
|
||||
ai_name = ""
|
||||
prompt = construct_prompt()
|
||||
# print(prompt)
|
||||
# Initialize variables
|
||||
full_message_history = []
|
||||
result = None
|
||||
next_action_count = 0
|
||||
# Make a constant:
|
||||
user_input = "Determine which next command to use, and respond using the format specified above:"
|
||||
# Initialize memory and make sure it is empty.
|
||||
# this is particularly important for indexing and referencing pinecone memory
|
||||
memory = get_memory(cfg, init=True)
|
||||
print('Using memory of type: ' + memory.__class__.__name__)
|
||||
# Interaction Loop
|
||||
while True:
|
||||
# Send message to AI, get response
|
||||
with Spinner("Thinking... "):
|
||||
assistant_reply = chat.chat_with_ai(
|
||||
prompt,
|
||||
user_input,
|
||||
full_message_history,
|
||||
memory,
|
||||
cfg.fast_token_limit) # TODO: This hardcodes the model to use GPT3.5. Make this an argument
|
||||
|
||||
# Initialize memory and make sure it is empty.
|
||||
# this is particularly important for indexing and referencing pinecone memory
|
||||
memory = get_memory(cfg, init=True)
|
||||
print('Using memory of type: ' + memory.__class__.__name__)
|
||||
# Print Assistant thoughts
|
||||
print_assistant_thoughts(assistant_reply)
|
||||
|
||||
# Interaction Loop
|
||||
while True:
|
||||
# Send message to AI, get response
|
||||
with Spinner("Thinking... "):
|
||||
assistant_reply = chat.chat_with_ai(
|
||||
prompt,
|
||||
user_input,
|
||||
full_message_history,
|
||||
memory,
|
||||
cfg.fast_token_limit) # TODO: This hardcodes the model to use GPT3.5. Make this an argument
|
||||
# Get command name and arguments
|
||||
try:
|
||||
command_name, arguments = cmd.get_command(
|
||||
attempt_to_fix_json_by_finding_outermost_brackets(assistant_reply))
|
||||
if cfg.speak_mode:
|
||||
speak.say_text(f"I want to execute {command_name}")
|
||||
except Exception as e:
|
||||
logger.error("Error: \n", str(e))
|
||||
|
||||
# Print Assistant thoughts
|
||||
print_assistant_thoughts(assistant_reply)
|
||||
|
||||
# Get command name and arguments
|
||||
try:
|
||||
command_name, arguments = cmd.get_command(attempt_to_fix_json_by_finding_outermost_brackets(assistant_reply))
|
||||
if cfg.speak_mode:
|
||||
speak.say_text(f"I want to execute {command_name}")
|
||||
except Exception as e:
|
||||
logger.error("Error: \n", str(e))
|
||||
|
||||
if not cfg.continuous_mode and next_action_count == 0:
|
||||
### GET USER AUTHORIZATION TO EXECUTE COMMAND ###
|
||||
# Get key press: Prompt the user to press enter to continue or escape
|
||||
# to exit
|
||||
user_input = ""
|
||||
logger.typewriter_log(
|
||||
"NEXT ACTION: ",
|
||||
Fore.CYAN,
|
||||
f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL} ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")
|
||||
print(
|
||||
f"Enter 'y' to authorise command, 'y -N' to run N continuous commands, 'n' to exit program, or enter feedback for {ai_name}...",
|
||||
flush=True)
|
||||
while True:
|
||||
console_input = utils.clean_input(Fore.MAGENTA + "Input:" + Style.RESET_ALL)
|
||||
if console_input.lower().rstrip() == "y":
|
||||
user_input = "GENERATE NEXT COMMAND JSON"
|
||||
break
|
||||
elif console_input.lower().startswith("y -"):
|
||||
try:
|
||||
next_action_count = abs(int(console_input.split(" ")[1]))
|
||||
user_input = "GENERATE NEXT COMMAND JSON"
|
||||
except ValueError:
|
||||
print("Invalid input format. Please enter 'y -n' where n is the number of continuous tasks.")
|
||||
continue
|
||||
break
|
||||
elif console_input.lower() == "n":
|
||||
user_input = "EXIT"
|
||||
break
|
||||
else:
|
||||
user_input = console_input
|
||||
command_name = "human_feedback"
|
||||
break
|
||||
|
||||
if user_input == "GENERATE NEXT COMMAND JSON":
|
||||
if not cfg.continuous_mode and next_action_count == 0:
|
||||
### GET USER AUTHORIZATION TO EXECUTE COMMAND ###
|
||||
# Get key press: Prompt the user to press enter to continue or escape
|
||||
# to exit
|
||||
user_input = ""
|
||||
logger.typewriter_log(
|
||||
"-=-=-=-=-=-=-= COMMAND AUTHORISED BY USER -=-=-=-=-=-=-=",
|
||||
Fore.MAGENTA,
|
||||
"")
|
||||
elif user_input == "EXIT":
|
||||
print("Exiting...", flush=True)
|
||||
break
|
||||
else:
|
||||
# Print command
|
||||
logger.typewriter_log(
|
||||
"NEXT ACTION: ",
|
||||
Fore.CYAN,
|
||||
f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL} ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")
|
||||
"NEXT ACTION: ",
|
||||
Fore.CYAN,
|
||||
f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL} ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")
|
||||
print(
|
||||
f"Enter 'y' to authorise command, 'y -N' to run N continuous commands, 'n' to exit program, or enter feedback for {ai_name}...",
|
||||
flush=True)
|
||||
while True:
|
||||
console_input = utils.clean_input(Fore.MAGENTA + "Input:" + Style.RESET_ALL)
|
||||
if console_input.lower().rstrip() == "y":
|
||||
user_input = "GENERATE NEXT COMMAND JSON"
|
||||
break
|
||||
elif console_input.lower().startswith("y -"):
|
||||
try:
|
||||
next_action_count = abs(int(console_input.split(" ")[1]))
|
||||
user_input = "GENERATE NEXT COMMAND JSON"
|
||||
except ValueError:
|
||||
print("Invalid input format. Please enter 'y -n' where n is the number of continuous tasks.")
|
||||
continue
|
||||
break
|
||||
elif console_input.lower() == "n":
|
||||
user_input = "EXIT"
|
||||
break
|
||||
else:
|
||||
user_input = console_input
|
||||
command_name = "human_feedback"
|
||||
break
|
||||
|
||||
# Execute command
|
||||
if command_name is not None and command_name.lower().startswith( "error" ):
|
||||
result = f"Command {command_name} threw the following error: " + arguments
|
||||
elif command_name == "human_feedback":
|
||||
result = f"Human feedback: {user_input}"
|
||||
else:
|
||||
result = f"Command {command_name} returned: {cmd.execute_command(command_name, arguments)}"
|
||||
if next_action_count > 0:
|
||||
next_action_count -= 1
|
||||
if user_input == "GENERATE NEXT COMMAND JSON":
|
||||
logger.typewriter_log(
|
||||
"-=-=-=-=-=-=-= COMMAND AUTHORISED BY USER -=-=-=-=-=-=-=",
|
||||
Fore.MAGENTA,
|
||||
"")
|
||||
elif user_input == "EXIT":
|
||||
print("Exiting...", flush=True)
|
||||
break
|
||||
else:
|
||||
# Print command
|
||||
logger.typewriter_log(
|
||||
"NEXT ACTION: ",
|
||||
Fore.CYAN,
|
||||
f"COMMAND = {Fore.CYAN}{command_name}{Style.RESET_ALL} ARGUMENTS = {Fore.CYAN}{arguments}{Style.RESET_ALL}")
|
||||
|
||||
memory_to_add = f"Assistant Reply: {assistant_reply} " \
|
||||
f"\nResult: {result} " \
|
||||
f"\nHuman Feedback: {user_input} "
|
||||
# Execute command
|
||||
if command_name is not None and command_name.lower().startswith("error"):
|
||||
result = f"Command {command_name} threw the following error: " + arguments
|
||||
elif command_name == "human_feedback":
|
||||
result = f"Human feedback: {user_input}"
|
||||
else:
|
||||
result = f"Command {command_name} returned: {cmd.execute_command(command_name, arguments)}"
|
||||
if next_action_count > 0:
|
||||
next_action_count -= 1
|
||||
|
||||
memory.add(memory_to_add)
|
||||
memory_to_add = f"Assistant Reply: {assistant_reply} " \
|
||||
f"\nResult: {result} " \
|
||||
f"\nHuman Feedback: {user_input} "
|
||||
|
||||
# Check if there's a result from the command append it to the message
|
||||
# history
|
||||
if result is not None:
|
||||
full_message_history.append(chat.create_chat_message("system", result))
|
||||
logger.typewriter_log("SYSTEM: ", Fore.YELLOW, result)
|
||||
else:
|
||||
full_message_history.append(
|
||||
chat.create_chat_message(
|
||||
"system", "Unable to execute command"))
|
||||
logger.typewriter_log("SYSTEM: ", Fore.YELLOW, "Unable to execute command")
|
||||
memory.add(memory_to_add)
|
||||
|
||||
# Check if there's a result from the command append it to the message
|
||||
# history
|
||||
if result is not None:
|
||||
full_message_history.append(chat.create_chat_message("system", result))
|
||||
logger.typewriter_log("SYSTEM: ", Fore.YELLOW, result)
|
||||
else:
|
||||
full_message_history.append(
|
||||
chat.create_chat_message(
|
||||
"system", "Unable to execute command"))
|
||||
logger.typewriter_log("SYSTEM: ", Fore.YELLOW, "Unable to execute command")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user