Spaces:
Runtime error
Runtime error
File size: 4,288 Bytes
ba2ab69 269ee37 ba2ab69 269ee37 ba2ab69 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 |
import os, sys
import utils
import uuid
import json
import subprocess, threading
FILE_DIR = os.path.dirname(os.path.abspath(__file__))
REPO_DIR = os.path.dirname(FILE_DIR)
STATE_DIR = os.path.join(FILE_DIR, "state")
sys.path.append(REPO_DIR)
if not os.path.exists(STATE_DIR):
os.mkdir(STATE_DIR)
import time
def get_openai_api_key():
return os.getenv("OPENAI_API_KEY")
running_apis = []
def get_state(state_file):
with open(state_file, "r") as f:
state = json.load(f)
return state
def set_state(state_file, state):
with open(state_file, "w") as f:
json.dump(state, f)
class AutoAPI:
def __init__(self, openai_key, ai_name, ai_role, top_13_goals):
self.openai_key = openai_key
hex = uuid.uuid4().hex
print(hex)
self.state_file = os.path.join(STATE_DIR, f"state_{hex}.json")
self.log_file = os.path.join(STATE_DIR, f"log_{hex}.json")
newline = "\n"
with open(os.path.join(REPO_DIR, "ai_settings.yaml"), "w") as f:
f.write(
f"""ai_goals:
{newline.join([f'- {goal[0]}' for goal in top_13_goals if goal[0]])}
ai_name: {ai_name}
ai_role: {ai_role}
"""
)
state = {
"pending_input": None,
"awaiting_input": False,
"messages": [],
"last_message_read_index": -1,
}
set_state(self.state_file, state)
with open(self.log_file, "w") as f:
subprocess.Popen(
[
"python",
os.path.join(REPO_DIR, "ui", "api.py"),
openai_key,
self.state_file,
],
cwd=REPO_DIR,
stdout=f,
stderr=f,
)
def send_message(self, message="Y"):
state = get_state(self.state_file)
state["pending_input"] = message
state["awaiting_input"] = False
set_state(self.state_file, state)
def get_chatbot_response(self):
while True:
state = get_state(self.state_file)
if (
state["awaiting_input"]
and state["last_message_read_index"] >= len(state["messages"]) - 1
):
break
if state["last_message_read_index"] >= len(state["messages"]) - 1:
time.sleep(1)
else:
state["last_message_read_index"] += 1
title, content = state["messages"][state["last_message_read_index"]]
yield (f"**{title.strip()}** " if title else "") + utils.remove_color(
content
).replace("\n", "<br />")
set_state(self.state_file, state)
if __name__ == "__main__":
print(sys.argv)
_, openai_key, state_file = sys.argv
os.environ["OPENAI_API_KEY"] = openai_key
import autogpt.config.config
from autogpt.logs import logger
from autogpt.cli import main
import autogpt.utils
from autogpt.spinner import Spinner
def add_message(title, content):
state = get_state(state_file)
state["messages"].append((title, content))
set_state(state_file, state)
def typewriter_log(title="", title_color="", content="", *args, **kwargs):
add_message(title, content)
def warn(message, title="", *args, **kwargs):
add_message(title, message)
def error(title, message="", *args, **kwargs):
add_message(title, message)
def clean_input(prompt=""):
add_message(None, prompt)
state = get_state(state_file)
state["awaiting_input"] = True
set_state(state_file, state)
while state["pending_input"] is None:
state = get_state(state_file)
print("Waiting for input...")
time.sleep(1)
print("Got input")
pending_input = state["pending_input"]
state["pending_input"] = None
set_state(state_file, state)
return pending_input
def spinner_start():
add_message(None, "Thinking...")
logger.typewriter_log = typewriter_log
logger.warn = warn
logger.error = error
autogpt.utils.clean_input = clean_input
Spinner.spin = spinner_start
sys.argv = sys.argv[:1]
main()
|