update scheduler and add operation for dehallucination (#769)

* 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

Author: tangg555

docs/README.md

@@ -1,3 +1,3 @@
 All documentation has been moved to a separate repository: https://github.com/MemTensor/MemOS-Docs. Please edit documentation there.
 
-所有文档已迁移至独立仓库：https://github.com/MemTensor/MemOS-Docs。请在该仓库中编辑文档。
+所有文档已迁移至独立仓库 https://github.com/MemTensor/MemOS-Docs 。请在该仓库中编辑文档。

examples/mem_scheduler/task_stop_rerun.py …m_scheduler/scheduler_for_async_tasks.pyexamples/mem_scheduler/task_stop_rerun.py renamed to examples/mem_scheduler/scheduler_for_async_tasks.py examples/mem_scheduler/task_stop_rerun.py renamed to examples/mem_scheduler/scheduler_for_async_tasks.py

@@ -25,7 +25,7 @@ def my_test_handler(messages: list[ScheduleMessageItem]):
         task_id = str(msg.item_id)
         file_path = tmp_dir / f"{task_id}.txt"
         try:
-            sleep(1)
+            sleep(5)
             file_path.write_text(f"Task {task_id} processed.\n")
             print(f"writing {file_path} done")
         except Exception as e:
@@ -58,7 +58,7 @@ def submit_tasks():
 mem_scheduler.register_handlers({TEST_HANDLER_LABEL: my_test_handler})
 
 # 10s to restart
-mem_scheduler.orchestrator.tasks_min_idle_ms[TEST_HANDLER_LABEL] = 10_000
+mem_scheduler.orchestrator.tasks_min_idle_ms[TEST_HANDLER_LABEL] = 5_000
 
 tmp_dir = Path("./tmp")
 tmp_dir.mkdir(exist_ok=True)
@@ -88,6 +88,6 @@ def submit_tasks():
 print(f"[Result] Final files in tmp: {len(list(tmp_dir.glob('*.txt')))})")
 
 # 7. Stop the scheduler
+sleep(20)
 print("Stopping the scheduler...")
-sleep(5)
 mem_scheduler.stop()

src/memos/api/config.py

@@ -7,16 +7,19 @@
 import re
 import time
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import requests
 
 from dotenv import load_dotenv
 
-from memos.configs.mem_cube import GeneralMemCubeConfig
-from memos.configs.mem_os import MOSConfig
 from memos.context.context import ContextThread
-from memos.mem_cube.general import GeneralMemCube
+
+
+if TYPE_CHECKING:
+    from memos.configs.mem_cube import GeneralMemCubeConfig
+    from memos.configs.mem_os import MOSConfig
+    from memos.mem_cube.general import GeneralMemCube
 
 
 # Load environment variables
@@ -805,8 +808,12 @@ def get_start_default_config() -> dict[str, Any]:
         return config
 
     @staticmethod
-    def create_user_config(user_name: str, user_id: str) -> tuple[MOSConfig, GeneralMemCube]:
+    def create_user_config(user_name: str, user_id: str) -> tuple["MOSConfig", "GeneralMemCube"]:
         """Create configuration for a specific user."""
+        from memos.configs.mem_cube import GeneralMemCubeConfig
+        from memos.configs.mem_os import MOSConfig
+        from memos.mem_cube.general import GeneralMemCube
+
         openai_config = APIConfig.get_openai_config()
         qwen_config = APIConfig.qwen_config()
         vllm_config = APIConfig.vllm_config()
@@ -933,12 +940,14 @@ def create_user_config(user_name: str, user_id: str) -> tuple[MOSConfig, General
         return default_config, default_mem_cube
 
     @staticmethod
-    def get_default_cube_config() -> GeneralMemCubeConfig | None:
+    def get_default_cube_config() -> "GeneralMemCubeConfig | None":
         """Get default cube configuration for product initialization.
 
         Returns:
             GeneralMemCubeConfig | None: Default cube configuration if enabled, None otherwise.
         """
+        from memos.configs.mem_cube import GeneralMemCubeConfig
+
         if not APIConfig.is_default_cube_config_enabled():
             return None
 

src/memos/llms/openai.py

@@ -59,8 +59,8 @@ def generate(self, messages: MessageList, **kwargs) -> str:
         if self.config.remove_think_prefix:
             return remove_thinking_tags(response_content)
         if reasoning_content:
-            return reasoning_content + response_content
-        return response_content
+            return reasoning_content + (response_content or "")
+        return response_content or ""
 
     @timed_with_status(
         log_prefix="OpenAI LLM Stream",
@@ -151,7 +151,7 @@ def generate(self, messages: MessageList, **kwargs) -> str:
         if self.config.remove_think_prefix:
             return remove_thinking_tags(response_content)
         else:
-            return response_content
+            return response_content or ""
 
     def generate_stream(self, messages: MessageList, **kwargs) -> Generator[str, None, None]:
         """Stream response from Azure OpenAI LLM with optional reasoning support."""