diff --git a/config/demo_config.yaml b/config/demo_config.yaml index d5ee7c77..fd6a54da 100644 --- a/config/demo_config.yaml +++ b/config/demo_config.yaml @@ -14,14 +14,14 @@ memory_service: contextual_msg_count: 6 memory_operations: read_message: - class: memory.operation.read_message + class: memory.operation.frontend_operation description: "read session messages of the user" read_memory: - class: memory.operation.read_memory + class: memory.operation.frontend_operation workflow: set_query,[extract_time|retrieve_memory1,semantic_rank],fuse_rerank description: "read related memories of the user" list_memory: - class: memory.operation.read_memory + class: memory.operation.frontend_operation workflow: set_query,retrieve_memory2,print_memory description: "read all memories of the user" write_memory: diff --git a/memory_scope/enumeration/memory_status_enum.py b/memory_scope/enumeration/memory_status_enum.py index de4c929d..9c508b43 100644 --- a/memory_scope/enumeration/memory_status_enum.py +++ b/memory_scope/enumeration/memory_status_enum.py @@ -17,3 +17,5 @@ class MemoryNodeStatus(str, Enum): CONTENT_MODIFIED = "content_modified" ACTIVE = "active" EXPIRED = "expired" + + DELETED = "deleted" diff --git a/memory_scope/memory/operation/clear_memory.py b/memory_scope/memory/operation/clear_memory.py new file mode 100644 index 00000000..7650f8b3 --- /dev/null +++ b/memory_scope/memory/operation/clear_memory.py @@ -0,0 +1,20 @@ +from memory_scope.memory.operation.base_operation import BaseOperation, OPERATION_TYPE +from memory_scope.memory.operation.base_workflow import BaseWorkflow + + +class ClearMemory(BaseWorkflow, BaseOperation): + operation_type: OPERATION_TYPE = "frontend" + + def __init__(self, + name: str, + description: str, + **kwargs): + super().__init__(name=name, **kwargs) + BaseOperation.__init__(self, name=name, description=description) + + def init_workflow(self, **kwargs): + self.init_workers(**kwargs) + + def run_operation(self, **kwargs): + self.context.clear() + self.run_workflow() diff --git a/memory_scope/memory/operation/read_memory.py b/memory_scope/memory/operation/frontend_operation.py similarity index 97% rename from memory_scope/memory/operation/read_memory.py rename to memory_scope/memory/operation/frontend_operation.py index d6288fed..5856f690 100644 --- a/memory_scope/memory/operation/read_memory.py +++ b/memory_scope/memory/operation/frontend_operation.py @@ -6,7 +6,7 @@ from memory_scope.memory.operation.base_workflow import BaseWorkflow from memory_scope.scheme.message import Message -class ReadMemory(BaseWorkflow, BaseOperation): +class FrontendOperation(BaseWorkflow, BaseOperation): operation_type: OPERATION_TYPE = "frontend" def __init__(self, diff --git a/memory_scope/memory/worker/base_worker.py b/memory_scope/memory/worker/base_worker.py index 5af82aa6..5a11b6db 100644 --- a/memory_scope/memory/worker/base_worker.py +++ b/memory_scope/memory/worker/base_worker.py @@ -61,8 +61,7 @@ class BaseWorker(metaclass=ABCMeta): raise NotImplementedError def run(self): - self.logger.info(f"----- worker.{self.name}.begin -----") - with Timer(self.name, log_time=False) as t: + with Timer(f"worker.{self.name}", time_log_type="wrap"): if self.raise_exception: self._run() else: @@ -71,8 +70,6 @@ class BaseWorker(metaclass=ABCMeta): except Exception as e: self.logger.exception(f"run {self.name} failed! args={e.args}") - self.logger.info(f"----- worker.{self.name}.end cost={t.cost_str}-----") - def get_context(self, key: str, default=None): return self.context.get(key, default) diff --git a/memory_scope/memory/worker/read/retrieve_memory_worker.py b/memory_scope/memory/worker/read/retrieve_memory_worker.py index bef21cef..ee8a1c1c 100644 --- a/memory_scope/memory/worker/read/retrieve_memory_worker.py +++ b/memory_scope/memory/worker/read/retrieve_memory_worker.py @@ -110,6 +110,10 @@ class RetrieveMemoryWorker(MemoryBaseWorker): memory_node_list.extend(result) self.logger.info(f"memory_node_list.size={len(memory_node_list)}") + if not memory_node_list: + self.continue_run = False + return + memory_node_list = sorted(memory_node_list, key=lambda x: x.score_similar, reverse=True) for node in memory_node_list: self.logger.info(f"recall_stage: content={node.content} score={node.score_similar} " diff --git a/memory_scope/models/base_model.py b/memory_scope/models/base_model.py index 1ea6315f..2b4f71a5 100644 --- a/memory_scope/models/base_model.py +++ b/memory_scope/models/base_model.py @@ -83,7 +83,7 @@ class BaseModel(metaclass=ABCMeta): :param kwargs: :return: """ - with Timer(self.__class__.__name__, log_time=False) as t: + with Timer(self.__class__.__name__, time_log_type="none") as t: self.before_call(stream=stream, **kwargs) for i in range(self.max_retries): if self.raise_exception: @@ -95,7 +95,7 @@ class BaseModel(metaclass=ABCMeta): model_response = ModelResponse(m_type=self.m_type, status=False, details=e.args) if isinstance(model_response, ModelResponse) and not model_response.status: - self.logger.warning(f"call model={self.model_name} failed! cost={t.cost_str} retry_cnt={i} " + self.logger.warning(f"call model={self.model_name} failed! {t.cost_str} retry_cnt={i} " f"details={model_response.details}", stacklevel=2) time.sleep(i * self.retry_interval) else: @@ -113,7 +113,7 @@ class BaseModel(metaclass=ABCMeta): :param kwargs: :return: """ - with Timer(self.__class__.__name__, log_time=False) as t: + with Timer(self.__class__.__name__, time_log_type="none") as t: self.before_call(**kwargs) for i in range(self.max_retries): if self.raise_exception: @@ -125,7 +125,7 @@ class BaseModel(metaclass=ABCMeta): model_response = ModelResponse(m_type=self.m_type, status=False, details=e.args) if not model_response.status: - self.logger.warning(f"async_call model={self.model_name} failed! cost={t.cost_str} retry_cnt={i} " + self.logger.warning(f"async_call model={self.model_name} failed! {t.cost_str} retry_cnt={i} " f"details={model_response.details}", stacklevel=2) time.sleep(i * self.retry_interval) else: diff --git a/memory_scope/storage/llama_index_es_memory_store_sync.py b/memory_scope/storage/llama_index_es_memory_store_sync.py index 727463c6..07797d9d 100644 --- a/memory_scope/storage/llama_index_es_memory_store_sync.py +++ b/memory_scope/storage/llama_index_es_memory_store_sync.py @@ -258,6 +258,12 @@ class LlamaIndexEsMemoryStoreSync(BaseMemoryStore): for n in expired_memories: self.delete(n) + # delete memories + deleted_memories = [n for n in nodes if n.status == MemoryNodeStatus.DELETED.value] + if deleted_memories: + for n in deleted_memories: + self.delete(n) + @staticmethod def _memory_node_2_text_node(memory_node: MemoryNode) -> TextNode: """ diff --git a/memory_scope/utils/timer.py b/memory_scope/utils/timer.py index 62071be2..56f62d59 100644 --- a/memory_scope/utils/timer.py +++ b/memory_scope/utils/timer.py @@ -1,105 +1,81 @@ import time +from typing import Literal from memory_scope.utils.logger import Logger +TIME_LOG_TYPE = Literal["end", "wrap", "none"] + class Timer(object): """ - A class used to measure the execution time of code blocks. It supports logging the elapsed time and can be customized - to display time in seconds or milliseconds. + A class used to measure the execution time of code blocks. It supports logging the elapsed time and can be + customized to display time in seconds or milliseconds. """ - def __init__(self, name: str, log_time: bool = True, use_ms: bool = True, **kwargs): - """ - Initializes the Timer object with a name, logging preference, time unit preference, and additional keyword arguments. + def __init__(self, + name: str, + time_log_type: TIME_LOG_TYPE = "end", + use_ms: bool = True, + stack_level: int = 2, + float_precision: int = 4, + **kwargs): - Args: - name (str): The name associated with this timer instance, often used in logs. - log_time (bool, optional): Determines if the elapsed time should be logged. Defaults to True. - use_ms (bool, optional): Specifies whether to use milliseconds as the time unit in logs. Defaults to True. - **kwargs: Additional keyword arguments that might be utilized by the logger or other components. - """ self.name: str = name - self.log_time: bool = log_time + self.time_log_type: TIME_LOG_TYPE = time_log_type self.use_ms: bool = use_ms + self.stack_level: int = stack_level + self.float_precision: int = float_precision self.kwargs: dict = kwargs - self.logger = Logger.get_logger() - - # time record + # time recorder self.t_start = 0 self.t_end = 0 self.cost = 0 - @classmethod - def kwargs_to_str(cls, float_precision: int = 4, **kwargs): - """ - Converts keyword arguments into a formatted string, with floats controlled by a precision setting. + self.logger = Logger.get_logger() - Args: - float_precision (int, optional): The number of decimal places for floating point numbers. Defaults to 4. - **kwargs: Arbitrary keyword arguments to be converted into strings. - - Returns: - str: A single string composed of the keyword arguments and their values, separated by spaces. - """ - line_list = [] - for k, v in kwargs.items(): - if isinstance(v, float): - float_style = f".{float_precision}f" - line = f"{k}={v:{float_style}}" # Format float value with specified precision - else: - line = f"{k}={v}" # Keep other types as is - line_list.append(line) - - return " ".join(line_list) # Join all parts into a single string with spaces - - def __enter__(self): - self.t_start = time.time() - # with Timer("XXX") as t, need return self - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - """ - Records the end time of the timed code block and calculates the elapsed time. - Logs the time cost if logging is enabled, with optional message customization. - - Args: - exc_type: The exception type (unused). - exc_val: The exception value (unused). - exc_tb: The traceback (unused). - """ + def _set_cost(self): self.t_end = time.time() self.cost = self.t_end - self.t_start if self.use_ms: self.cost *= 1000 - if self.log_time: - line = f"{self.name}.timer" - - if self.use_ms: - line = f"{line} cost={self.cost:.1f}ms" - else: - line = f"{line} cost={self.cost:.4f}s" - - if self.kwargs: - line = f"{line} {self.kwargs_to_str(**self.kwargs)}" - - self.logger.info(line, stacklevel=3) - @property def cost_str(self): - """ - Returns a string representation of the time cost, formatted as seconds or milliseconds - based on the `use_ms` attribute. - - Returns: - A string indicating the time cost in the chosen unit (seconds or milliseconds). - """ + self._set_cost() if self.use_ms: - return f"{self.cost:.1f}ms" + return f"cost={self.cost:.4f}ms" else: - return f"{self.cost:.4f}s" + return f"cost={self.cost:.4f}s" + + def __enter__(self, *args, **kwargs): + self.t_start = time.time() + if self.time_log_type == "wrap": + self.logger.info(f"----- {self.name}.begin -----") + return self + + def __exit__(self, *args, **kwargs): + if self.time_log_type == "none": + return + + lines = [] + if self.time_log_type == "wrap": + lines.append(f"----- {self.name}.end -----") + else: + lines.append(self.name) + + lines.append(self.cost_str) + + if self.kwargs: + for k, v in self.kwargs.items(): + if isinstance(v, float): + float_style = f".{self.float_precision}f" + line = f"{k}={v:{float_style}}" + else: + line = f"{k}={v}" + lines.append(line) + + self.logger.info(" ".join(lines), stacklevel=self.stack_level) def timer(func): @@ -112,6 +88,7 @@ def timer(func): Returns: Callable: The wrapper function that includes timing functionality. """ + def wrapper(*args, **kwargs): """ The wrapper function that manages the timing of the original function.