Spaces:
Runtime error
Runtime error
File size: 3,083 Bytes
0534c31 |
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 |
import os, sys
import utils
FILE_DIR = os.path.dirname(os.path.abspath(__file__))
REPO_DIR = os.path.dirname(FILE_DIR)
sys.path.append(REPO_DIR)
import threading
from functools import partial
import time
def get_openai_api_key():
return os.getenv("OPENAI_API_KEY")
running_apis = []
class AutoAPI:
def __init__(self, openai_key, ai_name, ai_role, top_5_goals):
print(openai_key)
self.openai_key = openai_key
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_5_goals if goal[0]])}
ai_name: {ai_name}
ai_role: {ai_role}
"""
)
thread = threading.Thread(target=self.client_thread)
thread.start()
self.thread = thread
self.pending_input = None
self.awaiting_input = False
self.messages = []
self.last_message_read_index = -1
def add_message(self, title, content):
# print(f"{title}: {content}")
self.messages.append((title, content))
def client_thread(self):
os.environ["OPENAI_API_KEY"] = self.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 typewriter_log(self, title="", title_color="", content="", *args, **kwargs):
self.add_message(title, content)
def warn(self, message, title="", *args, **kwargs):
self.add_message(title, message)
def error(self, title, message="", *args, **kwargs):
self.add_message(title, message)
def clean_input(self, prompt=""):
self.add_message(None, prompt)
self.awaiting_input = True
while self.pending_input is None:
time.sleep(1)
pending_input = self.pending_input
self.pending_input = None
print("Sending message:", pending_input)
return pending_input
def spinner_start(self):
self.add_message(None, "Thinking...")
logger.typewriter_log = partial(typewriter_log, self)
logger.warn = partial(warn, self)
logger.error = partial(error, self)
autogpt.utils.clean_input = partial(clean_input, self)
Spinner.spin = partial(spinner_start, self)
main()
def send_message(self, message="Y"):
self.pending_input = message
self.awaiting_input = False
def get_chatbot_response(self):
while (not self.awaiting_input) or self.last_message_read_index < len(self.messages) - 1:
if self.last_message_read_index >= len(self.messages) - 1:
time.sleep(1)
else:
self.last_message_read_index += 1
title, content = self.messages[self.last_message_read_index]
yield (f"**{title.strip()}** " if title else "") + utils.remove_color(content).replace("\n", "<br />")
|