SuperAGI

1
import os
2
import sys
3
import subprocess
4
from time import sleep
5
import shutil
6
from sys import platform
7
from multiprocessing import Process
8
from superagi.lib.logger import logger
9

10

11
def check_command(command, message):
12
    if not shutil.which(command):
13
        logger.info(message)
14
        sys.exit(1)
15

16

17
def run_npm_commands(shell=False):
18
    os.chdir("gui")
19
    try:
20
        subprocess.run(["npm", "install"], check=True, shell=shell)
21
    except subprocess.CalledProcessError:
22
        logger.error(f"Error during '{' '.join(sys.exc_info()[1].cmd)}'. Exiting.")
23
        sys.exit(1)
24
    os.chdir("..")
25

26

27
def run_server(shell=False,a_name=None,a_description=None,goals=None):
28
    tgwui_process = Process(target=subprocess.run, args=(["python", "test.py","--name",a_name,"--description",a_description,"--goals"]+goals,), kwargs={"shell": shell})
29
    api_process = Process(target=subprocess.run, args=(["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"],), kwargs={"shell": shell})
30
    celery_process = Process(target=subprocess.run, args=(["celery", "-A", "celery_app", "worker", "--loglevel=info"],), kwargs={"shell": shell})
31
    ui_process = Process(target=subprocess.run, args=(["python", "test.py","--name",a_name,"--description",a_description,"--goals"]+goals,), kwargs={"shell": shell})
32
    api_process.start()
33
    celery_process.start()
34
    ui_process.start()
35

36
    return api_process, ui_process, celery_process
37

38

39
def cleanup(api_process, ui_process, celery_process):
40
    logger.info("Shutting down processes...")
41
    api_process.terminate()
42
    ui_process.terminate()
43
    celery_process.terminate()
44
    logger.info("Processes terminated. Exiting.")
45
    sys.exit(1)
46

47

48
if __name__ == "__main__":
49
    check_command("node", "Node.js is not installed. Please install it and try again.")
50
    check_command("npm", "npm is not installed. Please install npm to proceed.")
51
    check_command("uvicorn", "uvicorn is not installed. Please install uvicorn to proceed.")
52

53
    agent_name = input("Enter an agent name: ")
54
    agent_description = input("Enter an agent description: ")
55
    goals = []
56
    while True:
57
        goal = input("Enter a goal (or 'q' to quit): ")
58
        if goal == 'q':
59
            break
60
        goals.append(goal)
61
    isWindows = False
62
    if platform == "win32" or platform == "cygwin":
63
        isWindows = True
64
    run_npm_commands(shell=isWindows)
65

66
    try:
67
        api_process, ui_process, celery_process = run_server(isWindows, agent_name, agent_description, goals)
68
        while True:
69
            try:
70
                sleep(30)
71
            except KeyboardInterrupt:
72
                cleanup(api_process, ui_process, celery_process)
73
    except Exception as e:
74
        cleanup(api_process, ui_process, celery_process)