diff --git a/main.py b/main.py index bc236a0..eac861b 100644 --- a/main.py +++ b/main.py @@ -3,9 +3,11 @@ import openai from langchain.embeddings import LocalAIEmbeddings import uuid import requests +import sys from loguru import logger from ascii_magic import AsciiArt +logger.add(sys.stderr, format="{time} {level} {message}", filter="miniAGI", level="INFO") # these three lines swap the stdlib sqlite3 lib with the pysqlite3 package for chroma __import__('pysqlite3') import sys @@ -32,7 +34,7 @@ chroma_client = Chroma(collection_name="memories", persist_directory="db", embed # Function to create images with OpenAI -def create_image(input_text=DEFAULT_SD_PROMPT, model=DEFAULT_SD_MODEL): +def display_avatar(input_text=DEFAULT_SD_PROMPT, model=DEFAULT_SD_MODEL): response = openai.Image.create( prompt=input_text, n=1, @@ -71,74 +73,6 @@ def tts(input_text, model=VOICE_MODEL): # remove the audio file os.remove(output_file_path) -def calculate_plan(user_input, agent_actions={}): - logger.info("--> Calculating plan: {description}", description=res["description"]) - messages = [ - {"role": "user", - "content": f"""Transcript of AI assistant responding to user requests. -Replies with a plan to achieve the user's goal with a list of subtasks with logical steps. - -Request: {res["description"]} -Function call: """ - } - ] - # get list of plannable actions - plannable_actions = [] - for action in agent_actions: - if agent_actions[action]["plannable"]: - # append the key of the dict to plannable_actions - plannable_actions.append(action) - - functions = [ - { - "name": "plan", - "description": """Decide to do an action.""", - "parameters": { - "type": "object", - "properties": { - "subtasks": { - "type": "array", - "items": { - "type": "object", - "properties": { - "reasoning": { - "type": "string", - "description": "subtask list", - }, - "function": { - "type": "string", - "enum": plannable_actions, - }, - }, - }, - }, - }, - "required": ["subtasks"] - } - }, - ] - response = openai.ChatCompletion.create( - #model="gpt-3.5-turbo", - model=FUNCTIONS_MODEL, - messages=messages, - functions=functions, - max_tokens=200, - stop=None, - temperature=0.5, - #function_call="auto" - function_call={"name": "plan"}, - ) - response_message = response["choices"][0]["message"] - if response_message.get("function_call"): - function_name = response.choices[0].message["function_call"].name - function_parameters = response.choices[0].message["function_call"].arguments - # read the json from the string - res = json.loads(function_parameters) - print(">>> function name: "+function_name) - print(">>> function parameters: "+function_parameters) - return res - return {"action": "none"} - def needs_to_do_action(user_input,agent_actions={}): # Get the descriptions and the actions name (the keys) @@ -202,22 +136,6 @@ Function call: """ return res return {"action": REPLY_ACTION} -### Agent capabilities -def save(memory, agent_actions={}): - print(">>> saving to memories: ") - print(memory) - chroma_client.add_texts([memory],[{"id": str(uuid.uuid4())}]) - chroma_client.persist() - return f"The object was saved permanently to memory." - -def search(query, agent_actions={}): - res = chroma_client.similarity_search(query) - print(">>> query: ") - print(query) - print(">>> retrieved memories: ") - print(res) - return res - def process_functions(user_input, action="", agent_actions={}): descriptions="" @@ -330,7 +248,7 @@ def evaluate(user_input, conversation_history = [],re_evaluate=False, agent_acti # if there are no subtasks, we can just reply, # otherwise we execute the subtasks # First we check if it's an object - if isinstance(function_results, dict) and len(function_results["subtasks"]) != 0: + if isinstance(function_results, dict) and function_results.get("subtasks") and len(function_results["subtasks"]) > 0: # cycle subtasks and execute functions for subtask in function_results["subtasks"]: print("==> subtask: ") @@ -384,9 +302,156 @@ def evaluate(user_input, conversation_history = [],re_evaluate=False, agent_acti tts(conversation_history[-1]["content"]) return conversation_history -#create_image() + + +### Agent capabilities + +def save(memory, agent_actions={}): + print(">>> saving to memories: ") + print(memory) + chroma_client.add_texts([memory],[{"id": str(uuid.uuid4())}]) + chroma_client.persist() + return f"The object was saved permanently to memory." + +def search(query, agent_actions={}): + res = chroma_client.similarity_search(query) + print(">>> query: ") + print(query) + print(">>> retrieved memories: ") + print(res) + return res + +def calculate_plan(user_input, agent_actions={}): + res = json.loads(user_input) + logger.info("--> Calculating plan: {description}", description=res["description"]) + messages = [ + {"role": "user", + "content": f"""Transcript of AI assistant responding to user requests. +Replies with a plan to achieve the user's goal with a list of subtasks with logical steps. + +Request: {res["description"]} +Function call: """ + } + ] + # get list of plannable actions + plannable_actions = [] + for action in agent_actions: + if agent_actions[action]["plannable"]: + # append the key of the dict to plannable_actions + plannable_actions.append(action) + + functions = [ + { + "name": "plan", + "description": """Decide to do an action.""", + "parameters": { + "type": "object", + "properties": { + "subtasks": { + "type": "array", + "items": { + "type": "object", + "properties": { + "reasoning": { + "type": "string", + "description": "subtask list", + }, + "function": { + "type": "string", + "enum": plannable_actions, + }, + }, + }, + }, + }, + "required": ["subtasks"] + } + }, + ] + response = openai.ChatCompletion.create( + #model="gpt-3.5-turbo", + model=FUNCTIONS_MODEL, + messages=messages, + functions=functions, + max_tokens=200, + stop=None, + temperature=0.5, + #function_call="auto" + function_call={"name": "plan"}, + ) + response_message = response["choices"][0]["message"] + if response_message.get("function_call"): + function_name = response.choices[0].message["function_call"].name + function_parameters = response.choices[0].message["function_call"].arguments + # read the json from the string + res = json.loads(function_parameters) + logger.info("<<< function name: {function_name} >>>> parameters: {parameters}", function_name=function_name,parameters=function_parameters) + return res + return {"action": REPLY_ACTION} + +# write file to disk with content +def write_file(arg, agent_actions={}): + arg = json.loads(arg) + filename = arg["filename"] + content = arg["content"] + with open(filename, 'w') as f: + f.write(content) + return f"File {filename} saved successfully." + +## Search on duckduckgo +def search_duckduckgo(args, agent_actions={}): + args = json.loads(args) + url = "https://api.duckduckgo.com/?q="+args["query"]+"&format=json&pretty=1" + response = requests.get(url) + if response.status_code == 200: + return response.json() + else: + return {"error": "No results found."} + +### End Agent capabilities + agent_actions = { + "search_internet": { + "function": search_duckduckgo, + "plannable": True, + "description": 'For searching the internet with a query, the assistant replies with the action "search_internet" and the query to search.', + "signature": { + "name": "search_internet", + "description": """For searching internet.""", + "parameters": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "information to save" + }, + }, + } + }, + }, + "write_file": { + "function": write_file, + "plannable": True, + "description": 'For writing a file to disk with content, the assistant replies with the action "write_file" and the filename and content to save.', + "signature": { + "name": "write_file", + "description": """For saving a file to disk with content.""", + "parameters": { + "type": "object", + "properties": { + "filename": { + "type": "string", + "description": "information to save" + }, + "content": { + "type": "string", + "description": "information to save" + }, + }, + } + }, + }, "save_memory": { "function": save, "plannable": True, @@ -456,6 +521,9 @@ agent_actions = { } conversation_history = [] + +display_avatar() + # TODO: process functions also considering the conversation history? conversation history + input while True: user_input = input("> ")