From f1398ad2627f12d58fca76ba87f04608bd40833d Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Tue, 30 Jul 2024 17:03:19 +0800 Subject: [PATCH] [features] memory service support multi-role_name! --- .flake8 | 2 +- examples/api/chat_example.py | 50 ++++++--- examples/cli/dash_cli_cn1.sh | 2 +- memoryscope/constants/common_constants.py | 8 +- memoryscope/core/chat/api_memory_chat.py | 27 ++--- memoryscope/core/chat/base_memory_chat.py | 10 +- memoryscope/core/config/arguments.py | 2 + memoryscope/core/config/config_manager.py | 6 +- memoryscope/core/config/demo_config.yaml | 4 +- memoryscope/core/memoryscope.py | 2 +- .../core/operation/backend_operation.py | 104 +++++------------- memoryscope/core/operation/base_operation.py | 25 +++-- .../core/operation/consolidate_memory_op.py | 62 +++++++---- .../core/operation/frontend_operation.py | 20 +++- .../core/service/base_memory_service.py | 69 +++++++----- .../core/service/memory_scope_service.py | 99 ++++++++++------- .../core/worker/backend/info_filter_worker.py | 5 +- .../worker/frontend/read_message_worker.py | 32 ++++-- .../core/worker/frontend/set_query_worker.py | 21 ---- memoryscope/core/worker/memory_base_worker.py | 69 ++++++------ 20 files changed, 339 insertions(+), 280 deletions(-) diff --git a/.flake8 b/.flake8 index 2c731db3..ec7c942f 100644 --- a/.flake8 +++ b/.flake8 @@ -1,5 +1,5 @@ [flake8] -exclude = tests/models/test_models_lli_embedding.py,tests/*,examples/*,memoryscope/core/storage/llama_index_sync_elasticsearch.py +exclude = tests/*,examples/*,memoryscope/core/storage/llama_index_sync_elasticsearch.py max-line-length = 120 inline-quotes = " avoid-escape = no diff --git a/examples/api/chat_example.py b/examples/api/chat_example.py index fef5b256..15f3e90c 100644 --- a/examples/api/chat_example.py +++ b/examples/api/chat_example.py @@ -1,7 +1,3 @@ -import sys - -sys.path.append(".") - from memoryscope import MemoryScope, Arguments arguments = Arguments( @@ -31,7 +27,6 @@ def chat_example1(): def chat_example2(): with MemoryScope(arguments=arguments) as ms: memory_chat = ms.default_memory_chat - response = memory_chat.chat_with_memory(query="我的爱好是弹琴。") print("回答：\n" + response.message.content) print("记忆：\n" + response.meta_data["memories"]) @@ -40,7 +35,6 @@ def chat_example2(): def chat_example3(): with MemoryScope(arguments=arguments) as ms: memory_chat = ms.default_memory_chat - response = memory_chat.chat_with_memory( query="你知道我消灭了几个星系吗？", temporary_memories="用户是一只哥斯拉，这是真是的，不是虚构的。\n前天用户消灭了星系0x4be？\n大前天用户消灭了星系0x5be？") @@ -50,18 +44,14 @@ def chat_example3(): def chat_example4(): with MemoryScope(arguments=arguments) as ms: memory_chat = ms.default_memory_chat - result = memory_chat.run_service_operation("delete_all") + memory_chat.run_service_operation("delete_all") response = memory_chat.chat_with_memory(query="我的爱好是弹琴。") print("回答1：\n" + response.message.content) - - # memory_chat.start_backend_service() result = memory_chat.run_service_operation("consolidate_memory") print("记忆更新变化：\n" + result) - response = memory_chat.chat_with_memory(query="你知道我的乐器爱好是什么？", - history_message_strategy=None) - + response = memory_chat.chat_with_memory(query="你知道我的乐器爱好是什么？", history_message_strategy=None) print("回答2：\n" + response.message.content) print("记忆2：\n" + response.meta_data["memories"]) @@ -72,14 +62,46 @@ def chat_example5(): memory_service.init_service() result = memory_service.list_memory() + print(f"list_memory result={result}") + result = memory_service.retrieve_memory() + print(f"retrieve_memory result={result}") + result = memory_service.consolidate_memory() - print(result) + print(f"consolidate_memory result={result}") + + +def chat_example6(): + with MemoryScope(arguments=arguments) as ms: + memory_chat = ms.default_memory_chat + memory_chat.run_service_operation("delete_all", "锦鲤") + memory_chat.run_service_operation("delete_all", "浩然") + + print("浩然=========================") + response = memory_chat.chat_with_memory(query="我的爱好是弹琴。", role_name="浩然") + print("回答1：\n" + response.message.content) + result = memory_chat.run_service_operation("consolidate_memory", role_name="浩然") + print("记忆更新变化：\n" + result) + response = memory_chat.chat_with_memory(query="你知道我的乐器爱好是什么？", role_name="浩然", + history_message_strategy=None) + print("回答2：\n" + response.message.content) + print("记忆2：\n" + response.meta_data["memories"]) + + print("锦鲤=========================") + response = memory_chat.chat_with_memory(query="我的爱好是打羽毛球。", role_name="锦鲤") + print("回答1：\n" + response.message.content) + result = memory_chat.run_service_operation("consolidate_memory", role_name="锦鲤") + print("记忆更新变化：\n" + result) + response = memory_chat.chat_with_memory(query="你知道我的运动爱好是什么？", role_name="锦鲤", + history_message_strategy=None) + print("回答2：\n" + response.message.content) + print("记忆2：\n" + response.meta_data["memories"]) if __name__ == "__main__": # chat_example1() # chat_example2() # chat_example3() - chat_example4() + # chat_example4() # chat_example5() + chat_example6() diff --git a/examples/cli/dash_cli_cn1.sh b/examples/cli/dash_cli_cn1.sh index c7698362..ab7e0e75 100644 --- a/examples/cli/dash_cli_cn1.sh +++ b/examples/cli/dash_cli_cn1.sh @@ -1 +1 @@ -python memoryscope/cli.py --config_path=memoryscope/core/config/demo_config.yaml \ No newline at end of file +memoryscope --config_path=memoryscope/core/config/demo_config.yaml \ No newline at end of file diff --git a/memoryscope/constants/common_constants.py b/memoryscope/constants/common_constants.py index cf12868f..74645416 100644 --- a/memoryscope/constants/common_constants.py +++ b/memoryscope/constants/common_constants.py @@ -13,10 +13,16 @@ MEMORIES = "memories" CHAT_MESSAGES = "chat_messages" -MEMORY_MANAGER = "memory_manager" +CHAT_MESSAGES_SCATTER = "chat_messages_scatter" CHAT_KWARGS = "chat_kwargs" +USER_NAME = "user_name" + +TARGET_NAME = "target_name" + +MEMORY_MANAGER = "memory_manager" + QUERY_WITH_TS = "query_with_ts" RETRIEVE_MEMORY_NODES = "retrieve_memory_nodes" diff --git a/memoryscope/core/chat/api_memory_chat.py b/memoryscope/core/chat/api_memory_chat.py index b6df0568..21527f13 100644 --- a/memoryscope/core/chat/api_memory_chat.py +++ b/memoryscope/core/chat/api_memory_chat.py @@ -20,8 +20,6 @@ class ApiMemoryChat(BaseMemoryChat): generation_model: str, context: MemoryscopeContext, stream: bool = False, - human_name: str = None, - assistant_name: str = None, **kwargs): super().__init__(**kwargs) @@ -32,16 +30,6 @@ class ApiMemoryChat(BaseMemoryChat): self.stream: bool = stream self.generation_model_kwargs: dict = kwargs.pop("generation_model_kwargs", {}) - self.human_name: str = human_name - if not self.human_name: - self.human_name = DEFAULT_HUMAN_NAME[self.context.language] - self.context.meta_data["human_name"] = self.human_name - - self.assistant_name: str = assistant_name - if not self.assistant_name: - self.assistant_name = "AI" - self.context.meta_data["assistant_name"] = self.assistant_name - self._prompt_handler: PromptHandler | None = None @property @@ -85,6 +73,14 @@ class ApiMemoryChat(BaseMemoryChat): self._memory_service.init_service() return self._memory_service + @property + def human_name(self): + return self.memory_service.human_name + + @property + def assistant_name(self): + return self.memory_service.assistant_name + @property def generation_model(self) -> BaseModel: """ @@ -117,7 +113,7 @@ class ApiMemoryChat(BaseMemoryChat): if model_response and model_response.message: model_response.message.role_name = self.assistant_name model_response.meta_data[MEMORIES] = memories - self.memory_service.add_messages([query_message, model_response.message]) + self.memory_service.add_messages_pair([query_message, model_response.message]) else: self.logger.warning("model_response or model_response.message is empty!") @@ -179,9 +175,10 @@ class ApiMemoryChat(BaseMemoryChat): history_messages = self.memory_service.read_message() elif isinstance(history_message_strategy, int): - history_messages = self.memory_service.chat_messages[-history_message_strategy:] + history_messages = self.memory_service.get_chat_messages_scatter(history_message_strategy) if history_messages: + assert isinstance(history_messages[0], Message) chat_messages.extend(history_messages) # Append the current user's message to the conversation context @@ -198,7 +195,7 @@ class ApiMemoryChat(BaseMemoryChat): if model_response and model_response.message: model_response.message.role_name = self.assistant_name model_response.meta_data[MEMORIES] = memories - self.memory_service.add_messages([query_message, model_response.message]) + self.memory_service.add_messages_pair([query_message, model_response.message]) else: self.logger.warning("model_response or model_response.message is empty!") return model_response diff --git a/memoryscope/core/chat/base_memory_chat.py b/memoryscope/core/chat/base_memory_chat.py index 08654edc..9558c92d 100644 --- a/memoryscope/core/chat/base_memory_chat.py +++ b/memoryscope/core/chat/base_memory_chat.py @@ -49,7 +49,7 @@ class BaseMemoryChat(metaclass=ABCMeta): - If it is set to "auto"， the history messages in the conversation will retain those that have not yet been summarized. Default to "auto". - If it is set to None， no conversation history will be saved. - - If it is set to an integer value "n", the most recent "n" messages will be retained. + - If it is set to an integer value "n", recent "n" message-pair[user, assistant] will be retained. remember_response (bool, optional): Flag indicating whether to save the AI's response to memory. Defaults to False. Returns: @@ -59,11 +59,11 @@ class BaseMemoryChat(metaclass=ABCMeta): """ raise NotImplementedError - def start_backend_service(self): - self.memory_service.start_backend_service() + def start_backend_service(self, **kwargs): + self.memory_service.start_backend_service(**kwargs) - def run_service_operation(self, name: str, **kwargs): - return self.memory_service.run_operation(name, **kwargs) + def run_service_operation(self, name: str, role_name: Optional[str] = None, **kwargs): + return self.memory_service.run_operation(name, role_name=role_name, **kwargs) def run(self): """ diff --git a/memoryscope/core/config/arguments.py b/memoryscope/core/config/arguments.py index dec565d4..0af3be80 100644 --- a/memoryscope/core/config/arguments.py +++ b/memoryscope/core/config/arguments.py @@ -17,6 +17,8 @@ class Arguments(object): memory_chat_class: str = field(default="cli_memory_chat", metadata={ "help": "cli_memory_chat(Command-line interaction), api_memory_chat(API interface interaction), etc."}) + chat_stream: bool | None = field(default=None) + human_name: str = field(default="user") assistant_name: str = field(default="AI") diff --git a/memoryscope/core/config/config_manager.py b/memoryscope/core/config/config_manager.py index 81261ec1..d6aa3e42 100644 --- a/memoryscope/core/config/config_manager.py +++ b/memoryscope/core/config/config_manager.py @@ -87,11 +87,11 @@ class ConfigManager(object): @staticmethod def update_memory_chat_by_arguments(config: dict, arguments: Arguments): memory_chat_class_split = config["class"].split(".") - stream = arguments.memory_chat_class in ["cli_memory_chat", ] + stream = arguments.chat_stream + if stream is None: + stream = arguments.memory_chat_class in ["cli_memory_chat", ] config.update({ "class": ".".join(memory_chat_class_split[:-1] + [arguments.memory_chat_class]), - "human_name": arguments.human_name if arguments.human_name else "", - "assistant_name": arguments.assistant_name if arguments.assistant_name else "", "stream": stream, }) diff --git a/memoryscope/core/config/demo_config.yaml b/memoryscope/core/config/demo_config.yaml index ad6c1334..8a2d4e6b 100644 --- a/memoryscope/core/config/demo_config.yaml +++ b/memoryscope/core/config/demo_config.yaml @@ -12,12 +12,12 @@ memory_chat: memory_service: memoryscope_service generation_model: generation_model stream: true - human_name: user - assistant_name: AI memory_service: memoryscope_service: class: core.service.memory_scope_service + human_name: user + assistant_name: AI memory_operations: read_message: class: core.operation.frontend_operation diff --git a/memoryscope/core/memoryscope.py b/memoryscope/core/memoryscope.py index d4a3da5a..f42260cb 100644 --- a/memoryscope/core/memoryscope.py +++ b/memoryscope/core/memoryscope.py @@ -59,7 +59,7 @@ class MemoryScope(ConfigManager): def close(self): # wait service to stop for _, service in self.context.memory_service_dict.items(): - service.stop_backend_service(wait_service_end=True) + service.stop_backend_service(wait_service=True) self.context.thread_pool.shutdown() diff --git a/memoryscope/core/operation/backend_operation.py b/memoryscope/core/operation/backend_operation.py index e4b4a41c..94f72b58 100644 --- a/memoryscope/core/operation/backend_operation.py +++ b/memoryscope/core/operation/backend_operation.py @@ -1,38 +1,25 @@ import time -from typing import List -from memoryscope.constants.common_constants import CHAT_KWARGS, RESULT, CHAT_MESSAGES -from memoryscope.core.operation.base_operation import BaseOperation, OPERATION_TYPE -from memoryscope.core.operation.base_workflow import BaseWorkflow -from memoryscope.core.utils.logger import Logger -from memoryscope.scheme.message import Message +from memoryscope.core.operation.base_operation import OPERATION_TYPE +from memoryscope.core.operation.frontend_operation import FrontendOperation -class BackendOperation(BaseWorkflow, BaseOperation): +class BackendOperation(FrontendOperation): """ BaseBackendOperation serves as an abstract base class for defining backend operations. It manages operation status, loop control, and integrates with a global context for thread management. """ operation_type: OPERATION_TYPE = "backend" - def __init__(self, - name: str, - description: str, - chat_messages: List[Message], - interval_time: int, - **kwargs): - super().__init__(name=name, **kwargs) - BaseOperation.__init__(self, name=name, description=description) + def __init__(self, interval_time: int, **kwargs): + super().__init__(**kwargs) - self.chat_messages: List[Message] = chat_messages - self.interval_time: int = interval_time + self._interval_time: int = interval_time self._operation_status_run: bool = False self._loop_switch: bool = False self._backend_task = None - self.logger = Logger.get_logger() - def init_workflow(self, **kwargs): """ Initializes the workflow by setting up workers with provided keyword arguments. @@ -42,84 +29,53 @@ class BackendOperation(BaseWorkflow, BaseOperation): """ self.init_workers(is_backend=True, **kwargs) - def _run_operation(self, **kwargs): - """ - Executes an operation within the workflow by clearing the context, - setting chat arguments, running the workflow, and returning the result. - - Args: - **kwargs: Keyword arguments necessary for the operation, including chat parameters. - - Returns: - Any: The result obtained after executing the workflow. - """ - # prepare kwargs - workflow_kwargs = { - CHAT_MESSAGES: self.chat_messages, - CHAT_KWARGS: {**kwargs, **self.kwargs}, - } - - # Execute the workflow with the prepared context - self.run_workflow(**workflow_kwargs) - - # Retrieve the result from the context after workflow execution - return self.context.get(RESULT) - - def run_operation(self, **kwargs): - """ - Executes the operation defined by `_run_operation` method with given keyword arguments, - while managing the operation status and exception handling. - - Args: - **kwargs: Arbitrary keyword arguments to be passed to `_run_operation`. - - Returns: - The result of the `_run_operation` method if no exception occurs, otherwise None. - """ - if self._operation_status_run: - return - - self._operation_status_run = True - result = None - try: - result = self._run_operation(**kwargs) - except Exception as e: - self.logger.exception(f"{self.name} encounter exception. args={e.args}") - - self._operation_status_run = False - return result - - def _loop_operation(self): + def _loop_operation(self, **kwargs): """ Loops until _loop_switch is False, sleeping for 1 second in each interval. At each interval, it checks if _loop_switch is still True, and if so, executes the operation. """ while self._loop_switch: - for _ in range(self.interval_time): + for _ in range(self._interval_time): if self._loop_switch: time.sleep(1) else: break - if self._loop_switch: - self.run_operation() - def start_operation_backend(self): + if self._loop_switch: + if self._operation_status_run: + continue + + self._operation_status_run = True + + if len(self.target_names) > 1: + self.logger.warning("current version is not stable under target_names.size > 1!") + + for target_name in self.target_names: + try: + self.run_operation(target_name=target_name, **kwargs) + except Exception as e: + self.logger.exception(f"op_name={self.name} target_name={target_name} encounter exception. " + f"args={e.args}") + + self._operation_status_run = False + + def start_operation_backend(self, **kwargs): """ Initiates the background operation loop if it's not already running. Sets the _loop_switch to True and submits the _loop_operation to a thread from the global thread pool. """ if not self._loop_switch: self._loop_switch = True - self._backend_task = self.thread_pool.submit(self._loop_operation) + self._backend_task = self.thread_pool.submit(self._loop_operation, **kwargs) self.logger.info(f"start operation={self.name}...") - def stop_operation_backend(self, wait_task_end: bool = False): + def stop_operation_backend(self, wait_operation: bool = False): """ Stops the background operation loop by setting the _loop_switch to False. """ self._loop_switch = False if self._backend_task: - if wait_task_end: + if wait_operation: self._backend_task.result() self.logger.info(f"stop operation={self.name}...") else: diff --git a/memoryscope/core/operation/base_operation.py b/memoryscope/core/operation/base_operation.py index 2a28edcd..2661c5f2 100644 --- a/memoryscope/core/operation/base_operation.py +++ b/memoryscope/core/operation/base_operation.py @@ -1,5 +1,7 @@ from abc import ABCMeta, abstractmethod -from typing import Literal +from typing import Literal, List + +from memoryscope.scheme.message import Message OPERATION_TYPE = Literal["frontend", "backend"] @@ -16,15 +18,19 @@ class BaseOperation(metaclass=ABCMeta): operation_type: OPERATION_TYPE = "frontend" - def __init__(self, name: str, description: str = ""): + def __init__(self, + name: str, + user_name: str, + target_names: List[str], + chat_messages: List[List[Message]], + description: str): """ Initializes a new instance of the BaseOperation. - - Args: - name (str): The name identifying the operation. - description (str): An optional description detailing the operation's purpose or behavior. """ self.name: str = name + self.user_name: str = user_name + self.target_names: List[str] = target_names + self.chat_messages: List[List[Message]] = chat_messages self.description: str = description def init_workflow(self, **kwargs): @@ -37,12 +43,13 @@ class BaseOperation(metaclass=ABCMeta): pass @abstractmethod - def run_operation(self, **kwargs): + def run_operation(self, target_name: str, **kwargs): """ Abstract method to define the operation to be run. Subclasses must implement this method. Args: + target_name (str): target_name(human name). **kwargs: Keyword arguments for running the operation. Raises: @@ -50,14 +57,14 @@ class BaseOperation(metaclass=ABCMeta): """ raise NotImplementedError - def start_operation_backend(self): + def start_operation_backend(self, **kwargs): """ Placeholder method for running an operation specific to the backend. Intended to be overridden by subclasses if backend operations are required. """ pass - def stop_operation_backend(self, wait_task_end: bool = False): + def stop_operation_backend(self, wait_operation: bool = False): """ Placeholder method to stop any ongoing backend operations. Should be implemented in subclasses where backend operations are managed. diff --git a/memoryscope/core/operation/consolidate_memory_op.py b/memoryscope/core/operation/consolidate_memory_op.py index fec60e12..5fb691f5 100644 --- a/memoryscope/core/operation/consolidate_memory_op.py +++ b/memoryscope/core/operation/consolidate_memory_op.py @@ -1,17 +1,21 @@ -from memoryscope.constants.common_constants import CHAT_KWARGS, CHAT_MESSAGES, RESULT +from typing import List + +from memoryscope.constants.common_constants import CHAT_KWARGS, CHAT_MESSAGES, RESULT, TARGET_NAME, USER_NAME from memoryscope.core.operation.backend_operation import BackendOperation -from memoryscope.enumeration.message_role_enum import MessageRoleEnum +from memoryscope.scheme.message import Message class ConsolidateMemoryOp(BackendOperation): - def __init__(self, **kwargs): - super(ConsolidateMemoryOp, self).__init__(**kwargs) + def __init__(self, + message_lock, + contextual_msg_min_count: int = 0, + **kwargs): + super().__init__(**kwargs) + self.message_lock = message_lock + self.contextual_msg_min_count: int = contextual_msg_min_count - self.message_lock = kwargs.get("message_lock", None) - self.contextual_msg_min_count: int = kwargs.get("contextual_msg_min_count", 0) - - def _run_operation(self, **kwargs): + def run_operation(self, target_name: str, **kwargs): """ Executes an operation after preparing the chat context, checking message memory status, and updating workflow status accordingly. @@ -21,32 +25,45 @@ class ConsolidateMemoryOp(BackendOperation): captures the result, and updates the memory status. Args: + target_name (str): target_name(human name). **kwargs: Keyword arguments for chat operation configuration. Returns: Any: The result obtained from running the workflow. """ - if not self.chat_messages: + chat_messages: List[List[Message]] = [] + for messages in self.chat_messages: + if not messages: + continue + + if messages[0].memorized: + continue + + contain_flag = False + + for msg in messages: + if msg.role_name == target_name: + contain_flag = True + break + + if contain_flag: + chat_messages.append(messages) + + if not chat_messages: + self.logger.info(f"empty not_memorized chat_messages for target_name={target_name}.") return - # Use shallow copy to prevent adding new messages. - chat_messages = self.chat_messages.copy() - - # filter for user-assistant pair - if chat_messages[-1].role == MessageRoleEnum.USER.value: - chat_messages = chat_messages[:-1] - - not_memorized_size = sum([not x.memorized for x in chat_messages]) - if not_memorized_size < self.contextual_msg_min_count: - self.logger.info(f"not_memorized_size({not_memorized_size}) < " - f"contextual_msg_min_count({self.contextual_msg_min_count}), skip.") + if len(chat_messages) < self.contextual_msg_min_count: + self.logger.info(f"not_memorized_size={len(chat_messages)} < {self.contextual_msg_min_count}, skip.") return # prepare kwargs workflow_kwargs = { CHAT_MESSAGES: chat_messages, CHAT_KWARGS: {**kwargs, **self.kwargs}, + TARGET_NAME: target_name, + USER_NAME: self.user_name, } # Execute the workflow with the prepared context @@ -57,7 +74,8 @@ class ConsolidateMemoryOp(BackendOperation): # set message memorized with self.message_lock: - for message in chat_messages: - message.memorized = True + for messages in chat_messages: + for msg in messages: + msg.memorized = True return result diff --git a/memoryscope/core/operation/frontend_operation.py b/memoryscope/core/operation/frontend_operation.py index 2562184b..d74212e5 100644 --- a/memoryscope/core/operation/frontend_operation.py +++ b/memoryscope/core/operation/frontend_operation.py @@ -1,6 +1,6 @@ from typing import List -from memoryscope.constants.common_constants import RESULT, CHAT_MESSAGES, CHAT_KWARGS +from memoryscope.constants.common_constants import RESULT, CHAT_MESSAGES, CHAT_KWARGS, TARGET_NAME, USER_NAME from memoryscope.core.operation.base_operation import BaseOperation, OPERATION_TYPE from memoryscope.core.operation.base_workflow import BaseWorkflow from memoryscope.scheme.message import Message @@ -11,13 +11,18 @@ class FrontendOperation(BaseWorkflow, BaseOperation): def __init__(self, name: str, + user_name: str, + target_names: List[str], + chat_messages: List[List[Message]], description: str, - chat_messages: List[Message], **kwargs): super().__init__(name=name, **kwargs) - BaseOperation.__init__(self, name=name, description=description) - - self.chat_messages: List[Message] = chat_messages + BaseOperation.__init__(self, + name=name, + user_name=user_name, + target_names=target_names, + chat_messages=chat_messages, + description=description) def init_workflow(self, **kwargs): """ @@ -28,12 +33,13 @@ class FrontendOperation(BaseWorkflow, BaseOperation): """ self.init_workers(**kwargs) - def run_operation(self, **kwargs): + def run_operation(self, target_name: str, **kwargs): """ Executes the main operation of reading recent chat messages, initializing workflow, and returning the result of the workflow execution. Args: + target_name (str): target_name(human name). **kwargs: Additional keyword arguments used in the operation context. Returns: @@ -44,6 +50,8 @@ class FrontendOperation(BaseWorkflow, BaseOperation): workflow_kwargs = { CHAT_MESSAGES: self.chat_messages, CHAT_KWARGS: {**kwargs, **self.kwargs}, + TARGET_NAME: target_name, + USER_NAME: self.user_name, } # Execute the workflow with the prepared context diff --git a/memoryscope/core/service/base_memory_service.py b/memoryscope/core/service/base_memory_service.py index b89f9d08..72dbfb1d 100644 --- a/memoryscope/core/service/base_memory_service.py +++ b/memoryscope/core/service/base_memory_service.py @@ -1,6 +1,7 @@ from abc import ABCMeta, abstractmethod from typing import List, Dict +from memoryscope.constants.language_constants import DEFAULT_HUMAN_NAME from memoryscope.core.memoryscope_context import MemoryscopeContext from memoryscope.core.operation.base_operation import BaseOperation from memoryscope.core.utils.logger import Logger @@ -14,23 +15,54 @@ class BaseMemoryService(metaclass=ABCMeta): along with logging capabilities and customizable configurations. """ - def __init__(self, memory_operations: Dict[str, dict], context: MemoryscopeContext, **kwargs): + def __init__(self, + memory_operations: Dict[str, dict], + context: MemoryscopeContext, + assistant_name: str = None, + human_name: str = None, + **kwargs): """ Initializes the BaseMemoryService with operation definitions, keys for memory access, and additional keyword arguments for flexibility. Args: memory_operations (Dict[str, dict]): A dictionary defining available memory operations. + context (MemoryscopeContext): runtime context. + human_name (str): human name. + assistant_name (str): assistant name. **kwargs: Additional parameters to customize service behavior. """ - self.memory_operations_conf: Dict[str, dict] = memory_operations - self.context: MemoryscopeContext = context - self.kwargs = kwargs + self._operations_conf: Dict[str, dict] = memory_operations + self._context: MemoryscopeContext = context + self._human_name: str = human_name + self._assistant_name: str = assistant_name + self._kwargs = kwargs + + if not self._human_name: + self._human_name = DEFAULT_HUMAN_NAME[self._context.language] + if not self._assistant_name: + self._assistant_name = "AI" self._operation_dict: Dict[str, BaseOperation] = {} - self.chat_messages: List[Message] = [] + self._chat_messages: List[List[Message]] = [] + self._role_names: List[str] = [] + self.logger = Logger.get_logger() + @property + def human_name(self) -> str: + return self._human_name + + @property + def assistant_name(self) -> str: + return self._assistant_name + + def get_chat_messages_scatter(self, recent_n_pair: int) -> List[Message]: + chat_messages_scatter: List[Message] = [] + for messages in self._chat_messages[-recent_n_pair:]: + chat_messages_scatter.extend(messages) + return chat_messages_scatter + @property def op_description_dict(self) -> Dict[str, str]: """ @@ -41,7 +73,7 @@ class BaseMemoryService(metaclass=ABCMeta): return {k: v.description for k, v in self._operation_dict.items()} @abstractmethod - def add_messages(self, messages: List[Message] | Message): + def add_messages_pair(self, messages: List[Message]): raise NotImplementedError @abstractmethod @@ -52,30 +84,15 @@ class BaseMemoryService(metaclass=ABCMeta): def init_service(self, **kwargs): raise NotImplementedError - def start_backend_service(self, name: str = None): + def start_backend_service(self, name: str = None, **kwargs): pass - def stop_backend_service(self, wait_service_end: bool = False): + def stop_backend_service(self, wait_service: bool = False): pass - def run_operation(self, name: str, **kwargs): - """ - Executes a specific operation by its name with provided keyword arguments. - - Args: - name (str): The name of the operation to execute. - **kwargs: Keyword arguments for the operation's execution. - - Returns: - The result of the operation execution, if any. Otherwise, None. - - Raises: - Warning: If the operation name is not initialized in `_operation_dict`. - """ - if name not in self._operation_dict: - self.logger.warning(f"operation={name} is not registered!") - return - return self._operation_dict[name].run_operation(**kwargs) + @abstractmethod + def run_operation(self, name: str, role_name: str = "", **kwargs): + raise NotImplementedError def __getattr__(self, name: str): assert name in self._operation_dict, f"operation={name} is not registered!" diff --git a/memoryscope/core/service/memory_scope_service.py b/memoryscope/core/service/memory_scope_service.py index 1a9b9ecc..ec6c4a79 100644 --- a/memoryscope/core/service/memory_scope_service.py +++ b/memoryscope/core/service/memory_scope_service.py @@ -10,10 +10,8 @@ from memoryscope.scheme.message import Message class MemoryScopeService(BaseMemoryService): def __init__(self, history_msg_count: int = 100, - contextual_msg_max_count: int = 20, + contextual_msg_max_count: int = 10, contextual_msg_min_count: int = 0, - human_name: str = None, - assistant_name: str = None, **kwargs): """ init function. @@ -23,47 +21,45 @@ class MemoryScopeService(BaseMemoryService): it will not be included in the context to prevent token overflow. contextual_msg_min_count (int): The minimum context length in a conversation. If it is shorter than this length, no conversation summary will be made and no long-term memory will be generated. - human_name (str): human name. - assistant_name (str): assistant name. - kwargs (dict): other kwargs. + kwargs (dict): Additional parameters to customize service behavior. """ super().__init__(**kwargs) - self.history_msg_count: int = history_msg_count - self.contextual_msg_max_count: int = contextual_msg_max_count - self.contextual_msg_min_count: int = contextual_msg_min_count + assert history_msg_count >= contextual_msg_max_count >= contextual_msg_min_count - if human_name: - self.context.meta_data["human_name"] = human_name - if assistant_name: - self.context.meta_data["assistant_name"] = assistant_name + self._history_msg_count: int = history_msg_count + self._contextual_msg_max_count: int = contextual_msg_max_count + self._contextual_msg_min_count: int = contextual_msg_min_count - self.message_lock = threading.Lock() + self._message_lock = threading.Lock() - def add_messages(self, messages: List[Message] | Message): + def add_messages_pair(self, messages: List[Message]): """ - Adds a single message or a list of messages to the chat history, ensuring the message list - remains sorted by creation time and does not exceed the maximum history message count. + Adds a list of messages to the chat history, it can be a pair [user_message, assistant_message]. + Ensuring the message list remains sorted by creation time and does not exceed the maximum history message count. Args: messages (List[Message] | Message): A single message instance or a list of message instances to be added to the chat history. """ - # If a single message is provided, convert it into a list for uniform processing - if isinstance(messages, Message): - messages = [messages] + assert messages, "messages should not be empty!" - with self.message_lock: + with self._message_lock: # Append the sorted messages to the chat history - self.chat_messages.extend(messages) + self._chat_messages.append(messages) # Sort the messages by their creation time to maintain chronological order - self.chat_messages.sort(key=lambda x: x.time_created) + self._chat_messages.sort(key=lambda x: x[0].time_created) # If the chat history exceeds the allowed message count, remove the oldest messages - if len(self.chat_messages) > self.history_msg_count: - gap_size = len(self.chat_messages) - self.history_msg_count + if len(self._chat_messages) > self._history_msg_count: + gap_size = len(self._chat_messages) - self._history_msg_count for _ in range(gap_size): - self.chat_messages.pop(0) + self._chat_messages.pop(0) + + for message in messages: + if message.role_name and message.role_name != self.assistant_name \ + and message.role_name not in self._role_names: + self._role_names.append(message.role_name) def register_operation(self, name: str, operation_config: dict, **kwargs): if name in self._operation_dict: @@ -73,11 +69,13 @@ class MemoryScopeService(BaseMemoryService): operation: BaseOperation = init_instance_by_config( config=operation_config, name=name, - chat_messages=self.chat_messages, - message_lock=self.message_lock, - memoryscope_context=self.context, - contextual_msg_max_count=self.contextual_msg_max_count, - contextual_msg_min_count=self.contextual_msg_min_count) + user_name=self._assistant_name, + target_names=self._role_names, + chat_messages=self._chat_messages, + message_lock=self._message_lock, + memoryscope_context=self._context, + contextual_msg_max_count=self._contextual_msg_max_count, + contextual_msg_min_count=self._contextual_msg_min_count) # Initialize workflow for each operation operation.init_workflow(**kwargs) @@ -85,25 +83,52 @@ class MemoryScopeService(BaseMemoryService): self.logger.info(f"service={self.__class__.__name__} init operation={name}") def init_service(self, **kwargs): - for name, operation_config in self.memory_operations_conf.items(): + for name, operation_config in self._operations_conf.items(): self.register_operation(name, operation_config, **kwargs) - def start_backend_service(self, name: str = None): + def run_operation(self, name: str, role_name: str = "", **kwargs): + """ + Executes a specific operation by its name with provided keyword arguments. + + Args: + name (str): The name of the operation to execute. + role_name (str): The name of the operation to execute. + **kwargs: Keyword arguments for the operation's execution. + + Returns: + The result of the operation execution, if any. Otherwise, None. + + Raises: + Warning: If the operation name is not initialized in `_operation_dict`. + """ + if name not in self._operation_dict: + self.logger.warning(f"operation={name} is not registered!") + return + + target_name = self._human_name + if role_name: + target_name = role_name + if role_name not in self._role_names: + self._role_names.append(role_name) + + return self._operation_dict[name].run_operation(target_name=target_name, **kwargs) + + def start_backend_service(self, name: str = None, **kwargs): """ Start all backend operations. """ for op_name, operation in self._operation_dict.items(): if name: if op_name == name: - operation.start_operation_backend() + operation.start_operation_backend(**kwargs) else: if operation.operation_type == "backend": - operation.start_operation_backend() + operation.start_operation_backend(**kwargs) - def stop_backend_service(self, wait_service_end: bool = False): + def stop_backend_service(self, wait_service: bool = False): """ Stops all backend operations that are currently running. """ for _, operation in self._operation_dict.items(): if operation.operation_type == "backend": - operation.stop_operation_backend(wait_task_end=wait_service_end) + operation.stop_operation_backend(wait_operation=wait_service) diff --git a/memoryscope/core/worker/backend/info_filter_worker.py b/memoryscope/core/worker/backend/info_filter_worker.py index 8bb41e3f..13028208 100644 --- a/memoryscope/core/worker/backend/info_filter_worker.py +++ b/memoryscope/core/worker/backend/info_filter_worker.py @@ -3,7 +3,6 @@ from typing import List from memoryscope.constants.language_constants import COLON_WORD from memoryscope.core.utils.response_text_parser import ResponseTextParser from memoryscope.core.worker.memory_base_worker import MemoryBaseWorker -from memoryscope.enumeration.message_role_enum import MessageRoleEnum from memoryscope.scheme.message import Message @@ -40,8 +39,8 @@ class InfoFilterWorker(MemoryBaseWorker): if msg.memorized: continue - # TODO: add memory for assistant - if msg.role != MessageRoleEnum.USER.value: + # TODO: add memory for all messages + if msg.role_name != self.target_name: continue if len(msg.content) >= self.info_filter_msg_max_size: diff --git a/memoryscope/core/worker/frontend/read_message_worker.py b/memoryscope/core/worker/frontend/read_message_worker.py index 32f3380b..78ca44ef 100644 --- a/memoryscope/core/worker/frontend/read_message_worker.py +++ b/memoryscope/core/worker/frontend/read_message_worker.py @@ -1,6 +1,8 @@ +from typing import List + from memoryscope.constants.common_constants import RESULT from memoryscope.core.worker.memory_base_worker import MemoryBaseWorker -from memoryscope.enumeration.message_role_enum import MessageRoleEnum +from memoryscope.scheme.message import Message class ReadMessageWorker(MemoryBaseWorker): @@ -12,11 +14,27 @@ class ReadMessageWorker(MemoryBaseWorker): """ Executes the primary function to fetch unmemorized chat messages. """ - chat_messages = [x for x in self.chat_messages if not x.memorized] - if len(chat_messages) > 0 and chat_messages[-1].role == MessageRoleEnum.USER.value: - chat_messages = chat_messages[:-1] + chat_messages_not_memorized: List[List[Message]] = [] + for messages in self.chat_messages_origin: + if not messages: + continue + + if messages[0].memorized: + continue + + contain_flag = False + + for msg in messages: + if msg.role_name == self.target_name: + contain_flag = True + break + + if contain_flag: + chat_messages_not_memorized.append(messages) contextual_msg_max_count: int = self.chat_kwargs["contextual_msg_max_count"] - chat_messages = chat_messages[-contextual_msg_max_count:] - - self.set_context(RESULT, chat_messages) + chat_message_scatter = [] + for messages in chat_messages_not_memorized[-contextual_msg_max_count:]: + chat_message_scatter.extend(messages) + chat_message_scatter.sort(key=lambda _: _.time_created) + self.set_context(RESULT, chat_message_scatter) diff --git a/memoryscope/core/worker/frontend/set_query_worker.py b/memoryscope/core/worker/frontend/set_query_worker.py index 9ab08509..0a586110 100644 --- a/memoryscope/core/worker/frontend/set_query_worker.py +++ b/memoryscope/core/worker/frontend/set_query_worker.py @@ -2,7 +2,6 @@ import datetime from memoryscope.constants.common_constants import QUERY_WITH_TS from memoryscope.core.worker.memory_base_worker import MemoryBaseWorker -from memoryscope.enumeration.message_role_enum import MessageRoleEnum class SetQueryWorker(MemoryBaseWorker): @@ -36,25 +35,5 @@ class SetQueryWorker(MemoryBaseWorker): if _timestamp and isinstance(_timestamp, int): timestamp = _timestamp - # check role_name - role_name = self.chat_kwargs.get("role_name") - if role_name: - assert role_name == self.target_name, (f"role_name={role_name} <> target_name={self.target_name} " - f"is not supported in human/assistant memory workflow!") - - elif self.chat_messages: - # If no explicit query is given, use the content of the latest chat message - chat_messages = [msg for msg in self.chat_messages if msg.role == MessageRoleEnum.USER.value] - if chat_messages: - message = chat_messages[-1] - query = message.content - timestamp = message.time_created - - # check role_name - role_name = message.role_name - if role_name: - assert role_name == self.target_name, \ - f"role_name={role_name} is not supported in human/assistant memory workflow!" - # Store the determined query and its timestamp in the context self.set_context(QUERY_WITH_TS, (query, timestamp)) diff --git a/memoryscope/core/worker/memory_base_worker.py b/memoryscope/core/worker/memory_base_worker.py index cb927626..8a007549 100644 --- a/memoryscope/core/worker/memory_base_worker.py +++ b/memoryscope/core/worker/memory_base_worker.py @@ -2,7 +2,7 @@ from abc import ABCMeta from typing import List, Dict, Any from memoryscope.constants.common_constants import CHAT_MESSAGES, CHAT_KWARGS, MEMORYSCOPE_CONTEXT, \ - WORKFLOW_NAME, MEMORY_MANAGER + WORKFLOW_NAME, MEMORY_MANAGER, USER_NAME, TARGET_NAME, CHAT_MESSAGES_SCATTER from memoryscope.constants.language_constants import DEFAULT_HUMAN_NAME, USER_NAME_EXPRESSION from memoryscope.core.memoryscope_context import MemoryscopeContext from memoryscope.core.models.base_model import BaseModel @@ -45,11 +45,18 @@ class MemoryBaseWorker(BaseWorker, metaclass=ABCMeta): self._memory_store: BaseMemoryStore | None = None self._monitor: BaseMonitor | None = None - - self._user_name: str | None = None - self._target_name: str | None = None self._prompt_handler: PromptHandler | None = None + @property + def chat_messages_origin(self) -> List[List[Message]]: + """ + Property to get the chat messages. + + Returns: + List[Message]: List of chat messages. + """ + return self.get_context(CHAT_MESSAGES) + @property def chat_messages(self) -> List[Message]: """ @@ -58,15 +65,30 @@ class MemoryBaseWorker(BaseWorker, metaclass=ABCMeta): Returns: List[Message]: List of chat messages. """ - return self.get_context(CHAT_MESSAGES) + result = self.get_context(CHAT_MESSAGES_SCATTER) + + if not result: + if isinstance(self.chat_messages_origin[0], list): + chat_messages: List[Message] = [] + for messages in self.chat_messages_origin: + if messages: + chat_messages.extend(messages) + chat_messages.sort(key=lambda _: _.time_created) + self.set_context(CHAT_MESSAGES_SCATTER, chat_messages) + + else: + assert isinstance(self.chat_messages_origin[0], Message) + self.set_context(CHAT_MESSAGES_SCATTER, self.chat_messages_origin) + + return self.get_context(CHAT_MESSAGES_SCATTER) @chat_messages.setter - def chat_messages(self, value): + def chat_messages(self, value: List[Message]): """ Set the chat messages with the new value. """ - self.set_context(CHAT_MESSAGES, value) + self.set_context(CHAT_MESSAGES_SCATTER, value) @property def chat_kwargs(self) -> Dict[str, Any]: @@ -81,6 +103,14 @@ class MemoryBaseWorker(BaseWorker, metaclass=ABCMeta): """ return self.get_context(CHAT_KWARGS) + @property + def user_name(self) -> str: + return self.get_context(USER_NAME) + + @property + def target_name(self) -> str: + return self.get_context(TARGET_NAME) + @property def workflow_name(self) -> str: return self.get_context(WORKFLOW_NAME) @@ -158,31 +188,6 @@ class MemoryBaseWorker(BaseWorker, metaclass=ABCMeta): self._monitor = self.memoryscope_context.monitor return self._monitor - @property - def user_name(self) -> str: - """ - Property to get the username from the meta_data of the global context. - If not set initially, it retrieves the 'assistant_name' as the username. - - Returns: - str: The name of the assistant. - """ - if self._user_name is None: - self._user_name = self.memoryscope_context.meta_data["assistant_name"] - return self._user_name - - @property - def target_name(self) -> str: - """ - Retrieves the target name, initializing it from meta_data if not set. - - Returns: - str: The readable name of the human. - """ - if self._target_name is None: - self._target_name = self.memoryscope_context.meta_data["human_name"] - return self._target_name - @property def prompt_handler(self) -> PromptHandler: """