mirror of
https://github.com/cpacker/MemGPT.git
synced 2025-06-03 04:30:22 +00:00
address comments, run black on interface.py
This commit is contained in:
Vivian Fang
parent
8b3740018d
commit
92c08e0e8b
35
config.py
35
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]:
|
||||
|
||||
143
interface.py
143
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:
|
||||
|
||||
11
main.py
11
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
|
||||
|
||||
Loading…
Reference in New Issue
Block a user