addressed a range of bugs to make scheduler running correctly

Author: tangg555

src/memos/api/config.py

@@ -174,7 +174,7 @@ def start_config_watch(cls):
     @classmethod
     def start_watch_if_enabled(cls) -> None:
         enable = os.getenv("NACOS_ENABLE_WATCH", "false").lower() == "true"
-        print("enable:", enable)
+        logger.info(f"NACOS_ENABLE_WATCH: {enable}")
         if not enable:
             return
         interval = int(os.getenv("NACOS_WATCH_INTERVAL", "60"))

src/memos/log.py

@@ -187,7 +187,7 @@ def close(self):
     },
     "handlers": {
         "console": {
-            "level": "DEBUG",
+            "level": "WARNING",
             "class": "logging.StreamHandler",
             "stream": stdout,
             "formatter": "no_datetime",

src/memos/mem_scheduler/base_scheduler.py

@@ -539,8 +539,8 @@ def submit_messages(self, messages: ScheduleMessageItem | list[ScheduleMessageIt
             if self.disable_handlers and message.label in self.disable_handlers:
                 logger.info(f"Skipping disabled handler: {message.label} - {message.content}")
                 continue
-        self.memos_message_queue.put(message)
-        logger.info(f"Submitted message to local queue: {message.label} - {message.content}")
+            self.memos_message_queue.put(message)
+            logger.info(f"Submitted message to local queue: {message.label} - {message.content}")
 
         with contextlib.suppress(Exception):
             if messages:
@@ -609,7 +609,6 @@ def _message_consumer(self) -> None:
 
                 if messages:
                     try:
-                        print(f"dispatch {len(messages)} messages")
                         self.dispatcher.dispatch(messages)
                     except Exception as e:
                         logger.error(f"Error dispatching messages: {e!s}")

src/memos/mem_scheduler/general_modules/dispatcher.py

@@ -418,9 +418,6 @@ def dispatch(self, msg_list: list[ScheduleMessageItem]):
                         logger.info(
                             f"Dispatch {len(msgs)} message(s) to {label} handler for user {user_id} and mem_cube {mem_cube_id}."
                         )
-                        print(
-                            f"Dispatch {len(msgs)} message(s) to {label} handler for user {user_id} and mem_cube {mem_cube_id}."
-                        )
                     else:
                         wrapped_handler(msgs)
 

src/memos/mem_scheduler/general_modules/redis_queue.py

@@ -169,9 +169,6 @@ def get(
             raise ConnectionError("Not connected to Redis. Redis connection not available.")
 
         try:
-            # Ensure the consumer group and stream exist before reading
-            self._ensure_consumer_group()
-
             # Calculate timeout for Redis
             redis_timeout = None
             if block and timeout is not None:
@@ -195,7 +192,6 @@ def get(
                     logger.warning(
                         f"Consumer group or stream missing for '{self.stream_name}/{self.consumer_group}'. Attempting to create and retry."
                     )
-                    self._ensure_consumer_group()
                     messages = self._redis_conn.xreadgroup(
                         self.consumer_group,
                         self.consumer_name,
@@ -263,9 +259,6 @@ def qsize(self) -> int:
             return 0
 
         try:
-            # Ensure consumer group exists
-            self._ensure_consumer_group()
-
             # Get pending messages info for the consumer group
             # XPENDING returns info about pending messages that haven't been acknowledged
             pending_info = self._redis_conn.xpending(self.stream_name, self.consumer_group)
@@ -432,7 +425,6 @@ def connect(self) -> None:
                 # Test the connection
                 self._redis_conn.ping()
                 self._is_connected = True
-                self._ensure_consumer_group()
                 logger.debug("Redis connection established successfully")
             except Exception as e:
                 logger.error(f"Failed to connect to Redis: {e}")

src/memos/mem_scheduler/memory_manage_modules/retriever.py

@@ -21,6 +21,7 @@
 from memos.memories.textual.item import TextualMemoryMetadata
 from memos.memories.textual.tree import TextualMemoryItem, TreeTextMemory
 
+# Extract JSON response
 from .memory_filter import MemoryFilter
 
 
@@ -63,9 +64,6 @@ def evaluate_memory_answer_ability(
         response = self.process_llm.generate([{"role": "user", "content": prompt}])
 
         try:
-            # Extract JSON response
-            from memos.mem_scheduler.utils.misc_utils import extract_json_obj
-
             result = extract_json_obj(response)
 
             # Validate response structure
@@ -116,12 +114,12 @@ def _process_enhancement_batch(
                 )
                 logger.debug(
                     f"[Enhance][batch={batch_index}] Prompt (first 200 chars, len={len(prompt)}): "
-                    f"{prompt[:200]}..."
+                    f"{prompt[:200]}]..."
                 )
 
                 response = self.process_llm.generate([{"role": "user", "content": prompt}])
                 logger.debug(
-                    f"[Enhance][batch={batch_index}] Response (first 200 chars): {response[:200]}..."
+                    f"[Enhance][batch={batch_index}] Response (first 200 chars): {response}..."
                 )
 
                 processed_text_memories = extract_list_items_in_answer(response)

src/memos/mem_scheduler/schemas/message_schemas.py

@@ -33,17 +33,17 @@
 
 class ScheduleMessageItem(BaseModel, DictConversionMixin):
     item_id: str = Field(description="uuid", default_factory=lambda: str(uuid4()))
-    redis_message_id: str = Field(description="the message get from redis stream", default="")
+    redis_message_id: str = Field(default="", description="the message get from redis stream")
     user_id: str = Field(..., description="user id")
     mem_cube_id: str = Field(..., description="memcube id")
-    session_id: str | None = Field(None, description="Session ID for soft-filtering memories")
+    session_id: str = Field(default="", description="Session ID for soft-filtering memories")
     label: str = Field(..., description="Label of the schedule message")
     content: str = Field(..., description="Content of the schedule message")
     timestamp: datetime = Field(
         default_factory=get_utc_now, description="submit time for schedule_messages"
     )
-    user_name: str | None = Field(
-        default=None,
+    user_name: str = Field(
+        default="",
         description="user name / display name (optional)",
     )
 

src/memos/mem_scheduler/utils/misc_utils.py

@@ -119,22 +119,10 @@ def extract_list_items(text: str, bullet_prefixes: tuple[str, ...] = ("- ",)) ->
         if matched:
             continue
 
-        # Removed loose fallback for "• " to strictly comply with "- " prefix format
-
     if items:
         return items
-
-    # Fallback: try parsing as a JSON array (e.g., ["item1", "item2", ...])
-    try:
-        data = extract_json_obj(normalized)
-        if isinstance(data, list):
-            result: list[str] = []
-            for x in data:
-                result.append(x if isinstance(x, str) else str(x))
-            return result
-    except Exception:
-        # Swallow and return empty list below
-        pass
+    else:
+        logger.error(f"Fail to parse {text}")
 
     return []
 

src/memos/templates/mem_scheduler_prompts.py

@@ -390,7 +390,6 @@
 - Focus on whether the memories can fully answer the query without additional information
 """
 
-
 MEMORY_ENHANCEMENT_PROMPT = """
 You are a knowledgeable and precise AI assistant.
 
@@ -430,7 +429,6 @@
 Answer:
 """
 
-
 PROMPT_MAPPING = {
     "intent_recognizing": INTENT_RECOGNIZING_PROMPT,
     "memory_reranking": MEMORY_RERANKING_PROMPT,