refactor&fix: fix a range of bugs in scheduler and revise fine add apis (MemTensor#840)

* fix bugs: try to fix bugs in _submit_web_logs

* fix bugs: try to address bugs

* fix bugs

* refactor: modify examples

* revise add operation and fix an unbelievable bug

* address the bug issues

* the doc file has a format problem which has been fixed in this commit

* add a range of new feats for the add operation

* address the incompatible issue of local scheduler

* feat(scheduler): optimize redis queue consumer group management

- Proactively ensure consumer groups exist in '_refresh_stream_keys' for newly discovered streams.
- Remove redundant consumer group checks in '_read_new_messages_batch' to improve read performance.
- Clean up 'seen_streams' cache when streams are deleted to ensure correct group recreation.
- This change reduces unnecessary Redis calls during high-frequency polling.

* fix(tests): resolve AttributeError in SimpleStructMemReader tests

- Import 'parse_json_result' from 'memos.mem_reader.utils' instead of accessing it as an instance attribute.
- Fixes 'AttributeError: 'SimpleStructMemReader' object has no attribute 'parse_json_result'' in 'test_parse_json_result_success' and 'test_parse_json_result_failure'.
- Remove incorrect mock assignment of 'parse_json_result' in 'test_process_chat_data'.

* fix(mem_reader): pass info dict to add_before_search for correct user_id usage

- Update 'add_before_search' signature in 'SimpleStructMemReader' to accept 'info' dict.
- Pass 'info' (containing 'user_id' and 'session_id') to 'self.searcher.search' instead of using empty strings.
- Add 'test_add_before_search' to 'TestSimpleStructMemReader' to verify the fix and ensure 'searcher.search' receives the correct 'info'.
- This ensures that memory searches are scoped to the correct user and session.

* refactor add_before_search from mem_reader to SingleCubeView

* address bugs

* fix: fix the qsize bug of task queue, and accept change from hotfix/scheduler

* fix: address some issues to run old scheduler example and kv cache example

* fix: address the issue of Top-level import of unavailable module 'torch'

* fix: resolve linting errors and make optional dependencies lazy loaded

- Fix ambiguous characters and commented-out code in examples/mem_scheduler/quick_start_examples.py
- Fix nested if statements in src/memos/mem_os/core.py
- Move torch and transformers imports to method scope in src/memos/llms/hf.py to support optional dependencies
- Update tests/llms/test_hf.py to patch transformers module directly

* refactor: revise the rewrite prompt to make it better

* refactor: update examples

* refactor: update examples for scheduler

* fix bugs: address the unsupported xautoclaim command when redis version larger than 6.2.0 via adding a new feature of manul auto claim with the combination of xpending + xclaim

* refactor: review settings

* refactor: adjust examples to make it run better for code debugging

* refactor: review slow add apis to get a better performance on Halumen

* fix bugs: address the issue when set user_redis_queue to false, the status_tracker is still using

* refactor: allow the code to run without rabbitmq

* refactor: create a _parse_pending_entry for redis queue

* refactor: add a try/catch for status_tracker

Author: tangg555

.gitignore

@@ -204,6 +204,7 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
+.trae
 
 # VSCode
 .vscode*

docker/.env.example

@@ -123,6 +123,7 @@ API_SCHEDULER_ON=true
 API_SEARCH_WINDOW_SIZE=5
 # Specify how many rounds of previous conversations (history) to retrieve and consider during the 'hybrid search' (fast search+asynchronous fine search). This helps provide context aware search results
 API_SEARCH_HISTORY_TURNS=5
+MEMSCHEDULER_USE_REDIS_QUEUE=false
 
 ## Graph / vector stores
 # Neo4j database selection mode

examples/mem_scheduler/quick_start_examples.py

@@ -146,7 +146,9 @@ def kv_cache_only():
 
 def run_scheduler_example():
     # 使用 MemScheduler 加载主 MOS(Memory-Oriented System)配置文件
-    config = parse_yaml("./examples/data/config/mem_scheduler/memos_config_w_scheduler.yaml")
+    config = parse_yaml(
+        f"{BASE_DIR}/examples/data/config/mem_scheduler/memos_config_w_scheduler.yaml"
+    )
     # 将解析出的配置字典传入 MOSConfig 构造器, 构建配置对象
     mos_config = MOSConfig(**config)
     # 使用配置对象初始化 MOS 系统实例
@@ -159,12 +161,12 @@ def run_scheduler_example():
 
     # 从 YAML 文件加载 MemCube(记忆立方体)的通用配置
     config = GeneralMemCubeConfig.from_yaml_file(
-        "./examples/data/config/mem_scheduler/mem_cube_config.yaml"
+        f"{BASE_DIR}/examples/data/config/mem_scheduler/mem_cube_config.yaml"
     )
     # 定义 MemCube 的唯一标识符
     mem_cube_id = "mem_cube_5"
     # 定义 MemCube 的本地存储路径(路径中包含用户 ID 和 MemCube ID)
-    mem_cube_name_or_path = f"./outputs/mem_scheduler/{user_id}/{mem_cube_id}"
+    mem_cube_name_or_path = f"{BASE_DIR}/outputs/mem_scheduler/{user_id}/{mem_cube_id}"
 
     # 如果该路径已存在, 则先删除旧目录
     if Path(mem_cube_name_or_path).exists():

examples/mem_scheduler/scheduler_for_async_tasks.py

@@ -57,7 +57,7 @@ def submit_tasks():
 TEST_HANDLER_LABEL = "test_handler"
 mem_scheduler.register_handlers({TEST_HANDLER_LABEL: my_test_handler})
 
-# 10s to restart
+# 5s to restart
 mem_scheduler.orchestrator.tasks_min_idle_ms[TEST_HANDLER_LABEL] = 5_000
 
 tmp_dir = Path("./tmp")

src/memos/mem_reader/simple_struct.py

@@ -614,11 +614,9 @@ def _read_memory(
                     serialized_origin_memories = json.dumps(
                         [one.memory for one in original_memory_group], indent=2
                     )
-                    revised_memory_list = self.rewrite_memories(
+                    revised_memory_list = self.filter_hallucination_in_memories(
                         messages=combined_messages,
                         memory_list=original_memory_group,
-                        user_only=os.getenv("SIMPLE_STRUCT_REWRITE_USER_ONLY", "true").lower()
-                        == "false",
                     )
                     serialized_revised_memories = json.dumps(
                         [one.memory for one in revised_memory_list], indent=2

src/memos/mem_scheduler/base_scheduler.py

@@ -225,7 +225,7 @@ def initialize_modules(
             process_llm = chat_llm
 
         try:
-            if redis_client:
+            if redis_client and self.use_redis_queue:
                 self.status_tracker = TaskStatusTracker(redis_client)
                 if self.dispatcher:
                     self.dispatcher.status_tracker = self.status_tracker
@@ -305,7 +305,7 @@ def status_tracker(self) -> TaskStatusTracker | None:
         available via RedisSchedulerModule. This mirrors the lazy pattern used
         by `mem_cube` so downstream modules can safely access the tracker.
         """
-        if self._status_tracker is None:
+        if self._status_tracker is None and self.use_redis_queue:
             try:
                 self._status_tracker = TaskStatusTracker(self.redis)
                 # Propagate to submodules when created lazily
@@ -314,7 +314,8 @@ def status_tracker(self) -> TaskStatusTracker | None:
                 if self.memos_message_queue:
                     self.memos_message_queue.set_status_tracker(self._status_tracker)
             except Exception as e:
-                logger.warning(f"Failed to lazily initialize status_tracker: {e}", exc_info=True)
+                logger.warning(f"Failed to lazy-initialize status_tracker: {e}", exc_info=True)
+
         return self._status_tracker
 
     @status_tracker.setter
@@ -869,6 +870,8 @@ def _submit_web_logs(
             messages = [messages]  # transform single message to list
 
         for message in messages:
+            if self.rabbitmq_config is None:
+                return
             try:
                 # Always call publish; the publisher now caches when offline and flushes after reconnect
                 logger.info(

src/memos/mem_scheduler/task_schedule_modules/dispatcher.py

@@ -108,8 +108,6 @@ def __init__(
         )
 
         self.metrics = metrics
-        self._status_tracker: TaskStatusTracker | None = None
-        # Use setter to allow propagation and keep a single source of truth
         self.status_tracker = status_tracker
         self.submit_web_logs = submit_web_logs  # ADDED
 
@@ -118,35 +116,6 @@ def on_messages_enqueued(self, msgs: list[ScheduleMessageItem]) -> None:
             return
         # This is handled in BaseScheduler now
 
-    @property
-    def status_tracker(self) -> TaskStatusTracker | None:
-        """Lazy-initialized status tracker for the dispatcher.
-
-        If the tracker is None, attempt to initialize from the Redis-backed
-        components available to the dispatcher (queue or orchestrator).
-        """
-        if self._status_tracker is None:
-            try:
-                self._status_tracker = TaskStatusTracker(self.redis)
-                # Propagate to submodules when created lazily
-                if self.memos_message_queue:
-                    self.memos_message_queue.set_status_tracker(self._status_tracker)
-            except Exception as e:
-                logger.warning(f"Failed to lazily initialize status_tracker: {e}", exc_info=True)
-        return self._status_tracker
-
-    @status_tracker.setter
-    def status_tracker(self, value: TaskStatusTracker | None) -> None:
-        self._status_tracker = value
-        # Propagate to the queue if possible
-        try:
-            if self.memos_message_queue and hasattr(self.memos_message_queue, "status_tracker"):
-                self.memos_message_queue.status_tracker = value
-        except Exception as e:
-            logger.warning(
-                f"Failed to propagate dispatcher status_tracker to queue: {e}", exc_info=True
-            )
-
     def _create_task_wrapper(self, handler: Callable, task_item: RunningTaskItem):
         """
         Create a wrapper around the handler to track task execution and capture results.