From cc352e7312080ab75d420763a3c96fac73776529 Mon Sep 17 00:00:00 2001 From: Vivian Fang Date: Mon, 23 Oct 2023 00:07:12 -0700 Subject: [PATCH] address comments, run black on interface.py --- config.py | 35 +++++++++++-- interface.py | 143 +++++++++++++++++++++++++++++++++------------------ main.py | 11 ++-- 3 files changed, 128 insertions(+), 61 deletions(-) diff --git a/config.py b/config.py index abe86b2b..14833fb4 100644 --- a/config.py +++ b/config.py @@ -66,7 +66,8 @@ class Config: f"Would you like to recompute embeddings? Do this if your files have changed.\nFiles:{self.archival_storage_files}", default=False, ) - await self.configure_archival_storage(recompute_embeddings) + if self.archival_storage_files: + await self.configure_archival_storage(recompute_embeddings) return self @classmethod @@ -120,7 +121,7 @@ class Config: print(self.memgpt_persona) self.human_persona = await questionary.select( - "Which persona would you like to use?", + "Which user would you like to use?", Config.get_user_personas(), ).ask_async() @@ -223,14 +224,40 @@ class Config: if dir_path is None: dir_path = Config.personas_dir all_personas = Config.get_personas(dir_path) - return Config.get_persona_choices([p for p in all_personas], get_persona_text) + default_personas = [ + "sam", + "sam_pov", + "memgpt_starter", + "memgpt_doc", + "sam_simple_pov_gpt35", + ] + custom_personas = list(set(all_personas) - set(default_personas)) + return Config.get_persona_choices( + [p for p in custom_personas + default_personas], get_persona_text + ) + [ + questionary.Separator(), + questionary.Choice( + f"📝 You can create your own personas by adding .txt files to {dir_path}.", + disabled=True, + ), + ] @staticmethod def get_user_personas(dir_path=None): if dir_path is None: dir_path = Config.humans_dir all_personas = Config.get_personas(dir_path) - return Config.get_persona_choices([p for p in all_personas], get_human_text) + default_personas = ["basic", "cs_phd"] + custom_personas = list(set(all_personas) - set(default_personas)) + return Config.get_persona_choices( + [p for p in custom_personas + default_personas], get_human_text + ) + [ + questionary.Separator(), + questionary.Choice( + f"📝 You can create your own human profiles by adding .txt files to {dir_path}.", + disabled=True, + ), + ] @staticmethod def get_personas(dir_path) -> List[str]: diff --git a/interface.py b/interface.py index af5cfc46..0e66af08 100644 --- a/interface.py +++ b/interface.py @@ -10,131 +10,172 @@ init(autoreset=True) # DEBUG = True # puts full message outputs in the terminal DEBUG = False # only dumps important messages in the terminal + def important_message(msg): - print(f'{Fore.MAGENTA}{Style.BRIGHT}{msg}{Style.RESET_ALL}') + print(f"{Fore.MAGENTA}{Style.BRIGHT}{msg}{Style.RESET_ALL}") + + +def warning_message(msg): + print(f"{Fore.RED}{Style.BRIGHT}{msg}{Style.RESET_ALL}") + async def internal_monologue(msg): # ANSI escape code for italic is '\x1B[3m' - print(f'\x1B[3m{Fore.LIGHTBLACK_EX}💭 {msg}{Style.RESET_ALL}') + print(f"\x1B[3m{Fore.LIGHTBLACK_EX}💭 {msg}{Style.RESET_ALL}") + async def assistant_message(msg): - print(f'{Fore.YELLOW}{Style.BRIGHT}🤖 {Fore.YELLOW}{msg}{Style.RESET_ALL}') + print(f"{Fore.YELLOW}{Style.BRIGHT}🤖 {Fore.YELLOW}{msg}{Style.RESET_ALL}") + async def memory_message(msg): - print(f'{Fore.LIGHTMAGENTA_EX}{Style.BRIGHT}🧠 {Fore.LIGHTMAGENTA_EX}{msg}{Style.RESET_ALL}') + print( + f"{Fore.LIGHTMAGENTA_EX}{Style.BRIGHT}🧠 {Fore.LIGHTMAGENTA_EX}{msg}{Style.RESET_ALL}" + ) + async def system_message(msg): - printd(f'{Fore.MAGENTA}{Style.BRIGHT}🖥️ [system] {Fore.MAGENTA}{msg}{Style.RESET_ALL}') + printd( + f"{Fore.MAGENTA}{Style.BRIGHT}🖥️ [system] {Fore.MAGENTA}{msg}{Style.RESET_ALL}" + ) + async def user_message(msg, raw=False): if isinstance(msg, str): if raw: - printd(f'{Fore.GREEN}{Style.BRIGHT}🧑 {Fore.GREEN}{msg}{Style.RESET_ALL}') + printd(f"{Fore.GREEN}{Style.BRIGHT}🧑 {Fore.GREEN}{msg}{Style.RESET_ALL}") return else: try: msg_json = json.loads(msg) except: printd(f"Warning: failed to parse user message into json") - printd(f'{Fore.GREEN}{Style.BRIGHT}🧑 {Fore.GREEN}{msg}{Style.RESET_ALL}') + printd( + f"{Fore.GREEN}{Style.BRIGHT}🧑 {Fore.GREEN}{msg}{Style.RESET_ALL}" + ) return - if msg_json['type'] == 'user_message': - msg_json.pop('type') - printd(f'{Fore.GREEN}{Style.BRIGHT}🧑 {Fore.GREEN}{msg_json}{Style.RESET_ALL}') - elif msg_json['type'] == 'heartbeat': + if msg_json["type"] == "user_message": + msg_json.pop("type") + printd(f"{Fore.GREEN}{Style.BRIGHT}🧑 {Fore.GREEN}{msg_json}{Style.RESET_ALL}") + elif msg_json["type"] == "heartbeat": if DEBUG: - msg_json.pop('type') - printd(f'{Fore.GREEN}{Style.BRIGHT}💓 {Fore.GREEN}{msg_json}{Style.RESET_ALL}') - elif msg_json['type'] == 'system_message': - msg_json.pop('type') - printd(f'{Fore.GREEN}{Style.BRIGHT}🖥️ {Fore.GREEN}{msg_json}{Style.RESET_ALL}') + msg_json.pop("type") + printd( + f"{Fore.GREEN}{Style.BRIGHT}💓 {Fore.GREEN}{msg_json}{Style.RESET_ALL}" + ) + elif msg_json["type"] == "system_message": + msg_json.pop("type") + printd(f"{Fore.GREEN}{Style.BRIGHT}🖥️ {Fore.GREEN}{msg_json}{Style.RESET_ALL}") else: - printd(f'{Fore.GREEN}{Style.BRIGHT}🧑 {Fore.GREEN}{msg_json}{Style.RESET_ALL}') + printd(f"{Fore.GREEN}{Style.BRIGHT}🧑 {Fore.GREEN}{msg_json}{Style.RESET_ALL}") + async def function_message(msg): - if isinstance(msg, dict): - printd(f'{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}') + printd(f"{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}") return - if msg.startswith('Success: '): - printd(f'{Fore.RED}{Style.BRIGHT}⚡🟢 [function] {Fore.RED}{msg}{Style.RESET_ALL}') - elif msg.startswith('Error: '): - printd(f'{Fore.RED}{Style.BRIGHT}⚡🔴 [function] {Fore.RED}{msg}{Style.RESET_ALL}') - elif msg.startswith('Running '): + if msg.startswith("Success: "): + printd( + f"{Fore.RED}{Style.BRIGHT}⚡🟢 [function] {Fore.RED}{msg}{Style.RESET_ALL}" + ) + elif msg.startswith("Error: "): + printd( + f"{Fore.RED}{Style.BRIGHT}⚡🔴 [function] {Fore.RED}{msg}{Style.RESET_ALL}" + ) + elif msg.startswith("Running "): if DEBUG: - printd(f'{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}') + printd( + f"{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}" + ) else: - if 'memory' in msg: - match = re.search(r'Running (\w+)\((.*)\)', msg) + if "memory" in msg: + match = re.search(r"Running (\w+)\((.*)\)", msg) if match: function_name = match.group(1) function_args = match.group(2) - print(f'{Fore.RED}{Style.BRIGHT}⚡🧠 [function] {Fore.RED}updating memory with {function_name}{Style.RESET_ALL}:') + print( + f"{Fore.RED}{Style.BRIGHT}⚡🧠 [function] {Fore.RED}updating memory with {function_name}{Style.RESET_ALL}:" + ) try: msg_dict = eval(function_args) - if function_name == 'archival_memory_search': - print(f'{Fore.RED}\tquery: {msg_dict["query"]}, page: {msg_dict["page"]}') + if function_name == "archival_memory_search": + print( + f'{Fore.RED}\tquery: {msg_dict["query"]}, page: {msg_dict["page"]}' + ) else: - print(f'{Fore.RED}{Style.BRIGHT}\t{Fore.RED} {msg_dict["old_content"]}\n\t{Fore.GREEN}→ {msg_dict["new_content"]}') + print( + f'{Fore.RED}{Style.BRIGHT}\t{Fore.RED} {msg_dict["old_content"]}\n\t{Fore.GREEN}→ {msg_dict["new_content"]}' + ) except Exception as e: printd(e) printd(msg_dict) pass else: printd(f"Warning: did not recognize function message") - printd(f'{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}') - elif 'send_message' in msg: + printd( + f"{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}" + ) + elif "send_message" in msg: # ignore in debug mode pass else: - printd(f'{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}') + printd( + f"{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}" + ) else: try: msg_dict = json.loads(msg) if "status" in msg_dict and msg_dict["status"] == "OK": - printd(f'{Fore.GREEN}{Style.BRIGHT}⚡ [function] {Fore.GREEN}{msg}{Style.RESET_ALL}') + printd( + f"{Fore.GREEN}{Style.BRIGHT}⚡ [function] {Fore.GREEN}{msg}{Style.RESET_ALL}" + ) except Exception: printd(f"Warning: did not recognize function message {type(msg)} {msg}") - printd(f'{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}') + printd( + f"{Fore.RED}{Style.BRIGHT}⚡ [function] {Fore.RED}{msg}{Style.RESET_ALL}" + ) + async def print_messages(message_sequence): for msg in message_sequence: - role = msg['role'] - content = msg['content'] + role = msg["role"] + content = msg["content"] - if role == 'system': + if role == "system": await system_message(content) - elif role == 'assistant': + elif role == "assistant": # Differentiate between internal monologue, function calls, and messages - if msg.get('function_call'): + if msg.get("function_call"): if content is not None: await internal_monologue(content) - await function_message(msg['function_call']) + await function_message(msg["function_call"]) # assistant_message(content) else: await internal_monologue(content) - elif role == 'user': + elif role == "user": await user_message(content) - elif role == 'function': + elif role == "function": await function_message(content) else: - print(f'Unknown role: {content}') + print(f"Unknown role: {content}") + async def print_messages_simple(message_sequence): for msg in message_sequence: - role = msg['role'] - content = msg['content'] + role = msg["role"] + content = msg["content"] - if role == 'system': + if role == "system": await system_message(content) - elif role == 'assistant': + elif role == "assistant": await assistant_message(content) - elif role == 'user': + elif role == "user": await user_message(content, raw=True) else: - print(f'Unknown role: {content}') + print(f"Unknown role: {content}") + async def print_messages_raw(message_sequence): for msg in message_sequence: diff --git a/main.py b/main.py index ade7aa9b..c0a92941 100644 --- a/main.py +++ b/main.py @@ -235,7 +235,11 @@ async def main(): else: cfg = await Config.config_init() - print("Running... [exit by typing '/exit']") + interface.important_message("Running... [exit by typing '/exit']") + if cfg.model != constants.DEFAULT_MEMGPT_MODEL: + interface.warning_message( + f"⛔️ Warning - you are running MemGPT with {cfg.model}, which is not officially supported (yet). Expect bugs!" + ) # Azure OpenAI support if FLAGS.use_azure_openai: @@ -269,11 +273,6 @@ async def main(): ) return - if cfg.model != constants.DEFAULT_MEMGPT_MODEL: - interface.important_message( - f"Warning - you are running MemGPT with {cfg.model}, which is not officially supported (yet). Expect bugs!" - ) - if cfg.archival_storage_index: persistence_manager = InMemoryStateManagerWithFaiss( cfg.index, cfg.archival_database