feat: finish a complete version of deep search

Author: tangg555

src/memos/api/handlers/chat_handler.py

@@ -32,14 +32,13 @@
 from memos.mem_scheduler.schemas.general_schemas import (
     ANSWER_LABEL,
     QUERY_LABEL,
-    SearchMode,
 )
 from memos.mem_scheduler.schemas.message_schemas import ScheduleMessageItem
 from memos.templates.mos_prompts import (
     FURTHER_SUGGESTION_PROMPT,
     get_memos_prompt,
 )
-from memos.types import MessageList
+from memos.types import MessageList, SearchMode
 
 
 class ChatHandler(BaseHandler):

src/memos/api/handlers/component_init.py

@@ -215,6 +215,7 @@ def init_server() -> dict[str, Any]:
     searcher: Searcher = tree_mem.get_searcher(
         manual_close_internet=os.getenv("ENABLE_INTERNET", "true").lower() == "false",
         moscube=False,
+        process_llm=mem_reader.llm,
     )
     logger.debug("Searcher created")
 
@@ -236,6 +237,9 @@ def init_server() -> dict[str, Any]:
     # Initialize SchedulerAPIModule
     api_module = mem_scheduler.api_module
 
+    # TODO: must remove!
+    mem_scheduler.memos_message_queue.debug_mode_on()
+
     # Start scheduler if enabled
     if os.getenv("API_SCHEDULER_ON", "true").lower() == "true":
         mem_scheduler.start()

src/memos/api/handlers/search_handler.py

@@ -251,9 +251,7 @@ def _fine_search(
             List of enhanced search results
         """
         if FINE_STRATEGY == FineStrategy.DEEP_SEARCH:
-            return self._deep_search(
-                search_req=search_req, user_context=user_context, max_thinking_depth=3
-            )
+            return self._deep_search(search_req=search_req, user_context=user_context)
 
         target_session_id = search_req.session_id or "default_session"
         search_filter = {"session_id": search_req.session_id} if search_req.session_id else None

src/memos/mem_scheduler/analyzer/api_analyzer.py

@@ -14,7 +14,7 @@
 import requests
 
 from memos.log import get_logger
-from memos.mem_scheduler.schemas.general_schemas import SearchMode
+from memos.types import SearchMode
 
 
 logger = get_logger(__name__)
@@ -681,7 +681,7 @@ def run_all_tests(self, mode=SearchMode.MIXTURE):
         print("Using direct test mode")
         try:
             direct_analyzer = DirectSearchMemoriesAnalyzer()
-            direct_analyzer.run_all_tests(mode=SearchMode.MIXTURE)
+            direct_analyzer.run_all_tests(mode=SearchMode.FINE)
         except Exception as e:
             print(f"Direct test mode failed: {e}")
             import traceback

src/memos/mem_scheduler/base_scheduler.py

@@ -155,6 +155,7 @@ def init_mem_cube(
             self.searcher: Searcher = self.text_mem.get_searcher(
                 manual_close_internet=os.getenv("ENABLE_INTERNET", "true").lower() == "false",
                 moscube=False,
+                process_llm=self.process_llm,
             )
         else:
             self.searcher = searcher
@@ -303,6 +304,26 @@ def replace_working_memory(
             query_db_manager.sync_with_orm()
 
             query_history = query_db_manager.obj.get_queries_with_timesort()
+
+            original_count = len(original_memory)
+            # Filter out memories tagged with "mode:fast"
+            filtered_original_memory = []
+            for origin_mem in original_memory:
+                if "mode:fast" not in origin_mem.metadata.tags:
+                    filtered_original_memory.append(origin_mem)
+                else:
+                    logger.debug(
+                        f"Filtered out memory - ID: {getattr(origin_mem, 'id', 'unknown')}, Tags: {origin_mem.metadata.tags}"
+                    )
+            # Calculate statistics
+            filtered_count = original_count - len(filtered_original_memory)
+            remaining_count = len(filtered_original_memory)
+
+            logger.info(
+                f"Filtering complete. Removed {filtered_count} memories with tag 'mode:fast'. Remaining memories: {remaining_count}"
+            )
+            original_memory = filtered_original_memory
+
             memories_with_new_order, rerank_success_flag = (
                 self.retriever.process_and_rerank_memories(
                     queries=query_history,

src/memos/mem_scheduler/monitors/dispatcher_monitor.py

@@ -135,7 +135,6 @@ def _check_pools_health(self) -> None:
                 pool_info=pool_info,
                 stuck_max_interval=4,
             )
-            logger.info(f"Pool '{name}'. is_healthy: {is_healthy}. pool_info: {pool_info}")
             with self._pool_lock:
                 if is_healthy:
                     pool_info["failure_count"] = 0
@@ -235,23 +234,6 @@ def _check_pool_health(
         # If we got here, pool appears healthy
         pool_info["last_active"] = get_utc_now()
 
-        # Log health status with comprehensive information
-        if self.dispatcher:
-            # Check thread activity
-            active_threads = sum(
-                1
-                for t in threading.enumerate()
-                if t.name.startswith(executor._thread_name_prefix)  # pylint: disable=protected-access
-            )
-
-            task_count = self.dispatcher.get_running_task_count()
-            max_workers = pool_info.get("max_workers", 0)
-            stuck_count = len(stuck_tasks)
-            logger.info(
-                f"Pool health check passed - {active_threads} active threads, "
-                f"{task_count} running tasks, pool size: {max_workers}, stuck tasks: {stuck_count}"
-            )
-
         return True, ""
 
     def _restart_pool(self, name: str, pool_info: dict) -> None:

src/memos/memories/textual/tree.py

@@ -129,9 +129,7 @@ def get_current_memory_size(self, user_name: str | None = None) -> dict[str, int
         return self.memory_manager.get_current_memory_size(user_name=user_name)
 
     def get_searcher(
-        self,
-        manual_close_internet: bool = False,
-        moscube: bool = False,
+        self, manual_close_internet: bool = False, moscube: bool = False, process_llm=None
     ):
         if (self.internet_retriever is not None) and manual_close_internet:
             logger.warning(
@@ -144,6 +142,7 @@ def get_searcher(
                 self.reranker,
                 internet_retriever=None,
                 moscube=moscube,
+                process_llm=process_llm,
             )
         else:
             searcher = Searcher(
@@ -153,6 +152,7 @@ def get_searcher(
                 self.reranker,
                 internet_retriever=self.internet_retriever,
                 moscube=moscube,
+                process_llm=process_llm,
             )
         return searcher