From 5d6e6c0f218ee2d5b73aae60da934dd38dec7542 Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Fri, 26 Jul 2024 10:54:19 +0800 Subject: [PATCH] finish memory chat --- memoryscope/chat/api_memory_chat.py | 252 +++--------------- memoryscope/chat/base_memory_chat.py | 13 +- memoryscope/chat/cli_memory_chat.py | 30 ++- .../memory/operation/backend_operation.py | 10 +- .../memory/service/base_memory_service.py | 6 +- 5 files changed, 84 insertions(+), 227 deletions(-) diff --git a/memoryscope/chat/api_memory_chat.py b/memoryscope/chat/api_memory_chat.py index 2007ef8d..7c0ab78a 100644 --- a/memoryscope/chat/api_memory_chat.py +++ b/memoryscope/chat/api_memory_chat.py @@ -1,20 +1,14 @@ -import os -import time from typing import List -import questionary - from memoryscope.chat.base_memory_chat import BaseMemoryChat from memoryscope.constants.language_constants import DEFAULT_HUMAN_NAME from memoryscope.enumeration.message_role_enum import MessageRoleEnum from memoryscope.memory.service.base_memory_service import BaseMemoryService +from memoryscope.memoryscope_context import MemoryscopeContext from memoryscope.models.base_model import BaseModel from memoryscope.scheme.message import Message from memoryscope.scheme.model_response import ModelResponse, ModelResponseGen -from memoryscope.utils.global_context import G_CONTEXT -from memoryscope.utils.logger import Logger from memoryscope.utils.prompt_handler import PromptHandler -from memoryscope.utils.tool_functions import char_logo class ApiMemoryChat(BaseMemoryChat): @@ -22,28 +16,27 @@ class ApiMemoryChat(BaseMemoryChat): def __init__(self, memory_service: str, generation_model: str, - stream: bool = True, - human_name: str = DEFAULT_HUMAN_NAME[G_CONTEXT.language], - assistant_name: str = "AI", + context: MemoryscopeContext, + human_name: str = None, + assistant_name: str = None, **kwargs): + super().__init__(**kwargs) + self._memory_service: BaseMemoryService | str = memory_service self._generation_model: BaseModel | str = generation_model + self.context: MemoryscopeContext = context self.generation_model_kwargs: dict = kwargs.pop("generation_model_kwargs", {}) - self.stream: bool = stream self.human_name: str = human_name + if not self.human_name: + self.human_name = DEFAULT_HUMAN_NAME[self.context.language] + self.assistant_name: str = assistant_name - self.kwargs: dict = kwargs + if not self.assistant_name: + self.assistant_name = "AI" - self._logo = char_logo("MemoryScope") self._prompt_handler: PromptHandler | None = None - G_CONTEXT.meta_data.update({ - "human_name": human_name, - "assistant_name": assistant_name, - }) - - self.logger = Logger.get_logger() @property def prompt_handler(self) -> PromptHandler: @@ -57,24 +50,14 @@ class ApiMemoryChat(BaseMemoryChat): PromptHandler: An instance of the PromptHandler configured for this CLI session. """ if self._prompt_handler is None: - self._prompt_handler = PromptHandler(__file__, **self.kwargs) + self._prompt_handler = PromptHandler(__file__, prompt_file="memory_chat_prompt", **self.kwargs) return self._prompt_handler - def print_logo(self): - """ - Prints the logo of the CLI application to the console. - - The logo is composed of multiple lines, which are iterated through - and printed one by one to provide a visual identity for the chat interface. - """ - for line in self._logo: - print(line) - @property def memory_service(self) -> BaseMemoryService: """ Property to access the memory service. If the service is initially set as a string, - it will be looked up in the memory service dictionary of global context, initialized, + it will be looked up in the memory service dictionary of context, initialized, and then returned as an instance of `BaseMemoryService`. Ensures the memory service is properly started before use. @@ -82,13 +65,15 @@ class ApiMemoryChat(BaseMemoryChat): BaseMemoryService: An active memory service instance. Raises: - ValueError: If the declaration of memory service is not found in the memory service dictionary of global context. + ValueError: If the declaration of memory service is not found in the memory service dictionary of context. """ if isinstance(self._memory_service, str): - if self._memory_service not in G_CONTEXT.memory_service_conf_dict: - raise ValueError("Missing declaration of memory_service in yaml configuration: " + self._memory_service) - self._memory_service = G_CONTEXT.memory_service_conf_dict[self._memory_service] - self._memory_service.init_service() + if self._memory_service not in self.context.memory_service_dict: + raise ValueError(f"Missing declaration of memory_service in context: {self._memory_service}") + + self._memory_service: BaseMemoryService = self.context.memory_service_dict[self._memory_service] + # init service & update kwargs + self._memory_service.init_service(human_name=self.human_name, assistant_name=self.assistant_name) self._memory_service.start_backend_service() return self._memory_service @@ -99,18 +84,18 @@ class ApiMemoryChat(BaseMemoryChat): context's model dictionary. Raises: - ValueError: If the declaration of generation model is not found in the model dictionary of global context . + ValueError: If the declaration of generation model is not found in the model dictionary of context . Returns: BaseModel: An actual generation model instance. """ if isinstance(self._generation_model, str): - if self._generation_model not in G_CONTEXT.model_conf_dict: + if self._generation_model not in self.context.model_dict: raise ValueError(f"Missing declaration of generation model in yaml config: {self._generation_model}") - self._generation_model = G_CONTEXT.model_conf_dict[self._generation_model] + self._generation_model = self.context.model_dict[self._generation_model] return self._generation_model - def chat_with_memory(self, query: str, remember_response: bool = False) -> ModelResponse | ModelResponseGen: + def chat_with_memory(self, query: str, role_name: str = "") -> ModelResponse | ModelResponseGen: """ Engages in a conversation with the AI model, utilizing conversation memory. The function sends the user's query, incorporates conversation history and memory, @@ -118,8 +103,7 @@ class ApiMemoryChat(BaseMemoryChat): Args: query (str): The user's input or query for the AI. - remember_response (bool, optional): Flag indicating whether to save the AI's response to memory. - Defaults to False. + role_name (str, optional): The user's name, default value is human_name. Returns: - ModelResponse: In non-streaming mode, returns a complete AI response. @@ -129,8 +113,10 @@ class ApiMemoryChat(BaseMemoryChat): - Updates the conversation memory with the query of user and (optionally) the response of AI. - Retrieves and includes historical messages and memory content in the context of conversation. """ - new_message: Message = Message(role=MessageRoleEnum.USER.value, role_name=self.human_name, content=query) - self.memory_service.add_messages(new_message) + if not role_name: + role_name = self.human_name + new_message: Message = Message(role=MessageRoleEnum.USER.value, role_name=role_name, content=query) + self.add_messages(new_message) messages: List[Message] = [] @@ -151,172 +137,20 @@ class ApiMemoryChat(BaseMemoryChat): messages.append(new_message) self.logger.info(f"messages={messages}") - # Invoke the Language Model with the constructed message context, respecting streaming setting - generated = self.generation_model.call(messages=messages, stream=self.stream, **self.generation_model_kwargs) + result = self.generation_model.call(messages=messages, stream=self.stream, **self.generation_model_kwargs) - # In non-streaming interactions, explicitly save the AI's reply to memory if instructed - if remember_response: - assert not self.stream # Ensure we're not in streaming mode when remembering responses - generated.message.role_name = self.assistant_name - self.memory_service.add_messages(generated.message) + if self.stream: + assert isinstance(result, ModelResponseGen) + model_response: ModelResponse | None = None + for model_response in result: + yield model_response - # Return the AI's response directly or as a generator based on the streaming mode - return generated - - @staticmethod - def parse_query_command(query: str): - """ - Parses the user's input query command, separating it into the command and its associated keyword arguments. - - Args: - query (str): The raw input string from the user which includes the command and its arguments. - - Returns: - tuple: A tuple containing the command (str) as the first element and a dictionary (kwargs) of keyword - arguments as the second element. - """ - query_split = query.lstrip("/").lower().split(" ") # Split and preprocess the input command - command = query_split[0] # Extract the command - args = query_split[1:] # Extract the arguments following the command - kwargs = {} # Initialize dictionary to hold keyword arguments - - for arg in args: - # Skip if no arguments exist (unnecessary check due to prior assignment, but retained as per original) - if not args: - continue - arg_split = arg.split("=") # Split argument into key-value pair - if len(arg_split) >= 2: # Ensure there's both a key and value - k = arg_split[0] # Extract key - v = arg_split[1] # Extract value - if k and v: # Only add to kwargs if both key and value are non-empty - kwargs[k] = v - - return command, kwargs # Return the parsed command and keyword arguments - - def process_commands(self, query: str) -> bool: - """ - Parses and executes commands from user input in the CLI chat interface. - Supports operations like exiting, clearing screen, showing help, toggling stream mode, - executing predefined memory operations, and handling unknown commands. - - Args: - query (str): The user's input command string. - - Returns: - bool: Indicates whether to continue running the CLI after processing the command. - """ - continue_run = True - command, kwargs = self.parse_query_command(query) - - # Print prompt for AI's response - questionary.print("> ", end="", style="fg:yellow") - questionary.print(f"{self.assistant_name}: ", end="", style="bold") - - if command == "exit": - self.memory_service.stop_backend_service() - continue_run = False - - elif command == "clear": - os.system("clear") - - elif command == "help": - questionary.print("CLI commands", "bold") - for cmd, desc in self.USER_COMMANDS.items(): - questionary.print(text=f" /{cmd}:", style="bold") - questionary.print(text=f" {desc}") - - elif command == "stream": - self.stream = not self.stream - questionary.print(f"set stream: {self.stream}") - - elif command in self.memory_service.op_description_dict: - refresh_time = kwargs.pop("refresh_time", "") - if refresh_time and refresh_time.isdigit(): - refresh_time = int(refresh_time) - self.memory_service.stop_backend_service() - while True: - result = self.memory_service.do_operation(op_name=command, **kwargs) - os.system("clear") - self.print_logo() - if result: - if isinstance(result, list): - result = "\n".join([str(x) for x in result]) - questionary.print(result) - else: - questionary.print(f"command={command} result is empty! kwargs={kwargs}") - time.sleep(refresh_time) - - else: - result = self.memory_service.do_operation(op_name=command, **kwargs) - if result: - if isinstance(result, list): - result = "\n".join([str(x) for x in result]) - questionary.print(result) - else: - questionary.print(f"command={command} result is empty! kwargs={kwargs}") + if model_response and model_response.message: + self.add_messages(model_response.message) else: - questionary.print(f"Unknown command={command} received.") - - return continue_run - - def run(self): - """ - Runs the CLI chat loop, which handles user input, processes commands, - communicates with the AI model, manages conversation memory, and controls - the chat session including streaming responses, command execution, and error handling. - - The loop continues until the user explicitly chooses to exit. - """ - self.print_logo() - self.USER_COMMANDS.update(self.memory_service.op_description_dict) - - while True: - try: - query = questionary.text(message=f"{self.human_name}:", multiline=False, qmark=">").unsafe_ask() - if not query: - continue - - query: str = query.strip() - - # Handle special commands prefixed with '/' - if query.startswith("/"): - if self.process_commands(query=query): - continue - else: - break - - # Print prompt for AI's response - questionary.print("> ", end="", style="fg:yellow") - questionary.print(f"{self.assistant_name}: ", end="", style="bold") - - # Fetch and display AI's response, with support for streaming - self.memory_service.start_backend_service() - if self.stream: - model_response = None - for model_response in self.chat_with_memory(query=query): - questionary.print(model_response.delta, end="") - questionary.print("") - - else: - model_response = self.chat_with_memory(query=query) - questionary.print(model_response.message.content) - - # Append AI's response to the conversation memory - model_response.message.role_name = self.assistant_name - self.memory_service.add_messages(model_response.message) - - except KeyboardInterrupt: - # Handle user interruption and confirm exit - questionary.print("User interrupt occurred.") - is_exit = questionary.confirm("Continue exit?").unsafe_ask() - if is_exit: - self.memory_service.stop_backend_service() - break - - except Exception as e: - # Log and handle any unanticipated exceptions - import traceback - traceback.print_exc() - self.logger.exception(f"An exception occurred when running cli memory chat. args={e.args}.") - continue + assert isinstance(result, ModelResponse) + model_response: ModelResponse = result + if model_response and model_response.message: + self.add_messages(model_response.message) + return model_response diff --git a/memoryscope/chat/base_memory_chat.py b/memoryscope/chat/base_memory_chat.py index 72cb8a07..d67c66be 100644 --- a/memoryscope/chat/base_memory_chat.py +++ b/memoryscope/chat/base_memory_chat.py @@ -12,8 +12,8 @@ class BaseMemoryChat(metaclass=ABCMeta): It outlines the method to initiate a chat session leveraging memory data, which concrete subclasses must implement. """ - def __init__(self, generation_stream: bool = True, **kwargs): - self.generation_stream: bool = generation_stream + def __init__(self, stream: bool = True, **kwargs): + self.stream: bool = stream self.kwargs: dict = kwargs self.logger = Logger.get_logger() @@ -32,9 +32,6 @@ class BaseMemoryChat(metaclass=ABCMeta): subclass. """ - def add_message(self, messages: List[Message] | Message): - self.memory_service.add_messages(messages) - @property def memory_service(self) -> BaseMemoryService: """ @@ -45,6 +42,12 @@ class BaseMemoryChat(metaclass=ABCMeta): """ raise NotImplementedError + def add_messages(self, messages: List[Message] | Message): + self.memory_service.add_messages(messages) + + def do_memory_operation(self, op_name: str, **kwargs): + return self.memory_service.do_operation(op_name=op_name, **kwargs) + def run(self): """ Abstract method to run the chat system. diff --git a/memoryscope/chat/cli_memory_chat.py b/memoryscope/chat/cli_memory_chat.py index 895ab7b9..a4ec880a 100644 --- a/memoryscope/chat/cli_memory_chat.py +++ b/memoryscope/chat/cli_memory_chat.py @@ -25,6 +25,7 @@ class CliMemoryChat(BaseMemoryChat): "exit": "Exit the CLI.", "clear": "Clear the command history.", "help": "Display available CLI commands and their descriptions.", + "stream": "Toggle between getting streamed responses from the model." } def __init__(self, @@ -121,10 +122,27 @@ class CliMemoryChat(BaseMemoryChat): return self._generation_model def chat_with_memory(self, query: str, role_name: str = "") -> ModelResponse | ModelResponseGen: + """ + Engages in a conversation with the AI model, utilizing conversation memory. + The function sends the user's query, incorporates conversation history and memory, + and optionally remembers the AI's response based on the user's preference. + + Args: + query (str): The user's input or query for the AI. + role_name (str, optional): The user's name, default value is human_name. + + Returns: + - ModelResponse: In non-streaming mode, returns a complete AI response. + - ModelResponseGen: In streaming mode, returns a generator yielding AI response parts. + + Side Effects: + - Updates the conversation memory with the query of user and (optionally) the response of AI. + - Retrieves and includes historical messages and memory content in the context of conversation. + """ if not role_name: role_name = self.human_name new_message: Message = Message(role=MessageRoleEnum.USER.value, role_name=role_name, content=query) - self.memory_service.add_messages(new_message) + self.add_messages(new_message) messages: List[Message] = [] @@ -147,7 +165,7 @@ class CliMemoryChat(BaseMemoryChat): # Invoke the Language Model with the constructed message context, respecting streaming setting return self.generation_model.call(messages=messages, - stream=self.generation_stream, + stream=self.stream, **self.generation_model_kwargs) @staticmethod @@ -212,6 +230,10 @@ class CliMemoryChat(BaseMemoryChat): questionary.print(text=f" /{cmd}:", style="bold") questionary.print(text=f" {desc}") + elif command == "stream": + self.stream = not self.stream + questionary.print(f"set stream: {self.stream}") + elif command in self.memory_service.op_description_dict: refresh_time = kwargs.pop("refresh_time", "") if refresh_time and refresh_time.isdigit(): @@ -275,7 +297,7 @@ class CliMemoryChat(BaseMemoryChat): # Fetch and display AI's response self.memory_service.start_backend_service() - if self.generation_stream: + if self.stream: model_response = None for model_response in self.chat_with_memory(query=query): questionary.print(model_response.delta, end="") @@ -286,7 +308,7 @@ class CliMemoryChat(BaseMemoryChat): # Append AI's response to the conversation memory model_response.message.role_name = self.assistant_name - self.memory_service.add_messages(model_response.message) + self.add_messages(model_response.message) except KeyboardInterrupt: # Handle user interruption and confirm exit diff --git a/memoryscope/memory/operation/backend_operation.py b/memoryscope/memory/operation/backend_operation.py index 2fcbe398..c52cf936 100644 --- a/memoryscope/memory/operation/backend_operation.py +++ b/memoryscope/memory/operation/backend_operation.py @@ -1,11 +1,11 @@ import time from typing import List + from memoryscope.constants.common_constants import CHAT_KWARGS, RESULT, CHAT_MESSAGES from memoryscope.memory.operation.base_operation import BaseOperation, OPERATION_TYPE from memoryscope.memory.operation.base_workflow import BaseWorkflow from memoryscope.scheme.message import Message -from memoryscope.utils.global_context import G_CONTEXT from memoryscope.utils.logger import Logger @@ -30,7 +30,7 @@ class BackendOperation(BaseWorkflow, BaseOperation): self._operation_status_run: bool = False self._loop_switch: bool = False - self._run_thread = None + self._backend_task = None self.logger = Logger.get_logger() @@ -114,10 +114,12 @@ class BackendOperation(BaseWorkflow, BaseOperation): """ if not self._loop_switch: self._loop_switch = True - self._run_thread = G_CONTEXT.thread_pool.submit(self._loop_operation) + self._backend_task = G_CONTEXT.thread_pool.submit(self._loop_operation) - def stop_operation_backend(self): + def stop_operation_backend(self, wait_task_end: bool = False): """ Stops the background operation loop by setting the _loop_switch to False. """ self._loop_switch = False + if wait_task_end and self._backend_task: + self._backend_task.result() diff --git a/memoryscope/memory/service/base_memory_service.py b/memoryscope/memory/service/base_memory_service.py index 3102dc0e..d72fc4c1 100644 --- a/memoryscope/memory/service/base_memory_service.py +++ b/memoryscope/memory/service/base_memory_service.py @@ -25,15 +25,11 @@ class BaseMemoryService(metaclass=ABCMeta): """ self.memory_operations_conf: Dict[str, dict] = memory_operations self.context: MemoryscopeContext = context + self.kwargs = kwargs self._operation_dict: Dict[str, BaseOperation] = {} self._op_description_dict: Dict[str, str] = {} - self.logger = Logger.get_logger() - self.kwargs = kwargs - - def update_kwargs(self, **kwargs): - pass @abstractmethod def add_messages(self, messages: List[Message] | Message):