Merge branch 'dev' into feat/deep-search

Author: fridayL

docker/requirements.txt

@@ -160,4 +160,3 @@ xlrd==2.0.2
 xlsxwriter==3.2.5
 prometheus-client==0.23.1
 pymilvus==2.5.12
-langchain-text-splitters==1.0.0

examples/mem_scheduler/memos_w_scheduler.py

@@ -13,15 +13,15 @@
 from memos.mem_cube.general import GeneralMemCube
 from memos.mem_os.main import MOS
 from memos.mem_scheduler.general_scheduler import GeneralScheduler
-from memos.mem_scheduler.schemas.general_schemas import (
-    ADD_LABEL,
-    ANSWER_LABEL,
-    MEM_ARCHIVE_LABEL,
-    MEM_ORGANIZE_LABEL,
-    MEM_UPDATE_LABEL,
-    QUERY_LABEL,
-)
 from memos.mem_scheduler.schemas.message_schemas import ScheduleLogForWebItem
+from memos.mem_scheduler.schemas.task_schemas import (
+    ADD_TASK_LABEL,
+    ANSWER_TASK_LABEL,
+    MEM_ARCHIVE_TASK_LABEL,
+    MEM_ORGANIZE_TASK_LABEL,
+    MEM_UPDATE_TASK_LABEL,
+    QUERY_TASK_LABEL,
+)
 from memos.mem_scheduler.utils.filter_utils import transform_name_to_key
 
 
@@ -118,24 +118,24 @@ def _first_content() -> str:
             return memcube_content[0].get("content", "") or content
         return content
 
-    if label in ("addMessage", QUERY_LABEL, ANSWER_LABEL):
+    if label in ("addMessage", QUERY_TASK_LABEL, ANSWER_TASK_LABEL):
         target_cube = cube_display.replace("MemCube", "")
         title = _format_title(item.timestamp, f"addMessages to {target_cube} MemCube")
         return title, _truncate_with_rules(_first_content())
 
-    if label in ("addMemory", ADD_LABEL):
+    if label in ("addMemory", ADD_TASK_LABEL):
         title = _format_title(item.timestamp, f"{cube_display} added {memory_len} memories")
         return title, _truncate_with_rules(_first_content())
 
-    if label in ("updateMemory", MEM_UPDATE_LABEL):
+    if label in ("updateMemory", MEM_UPDATE_TASK_LABEL):
         title = _format_title(item.timestamp, f"{cube_display} updated {memory_len} memories")
         return title, _truncate_with_rules(_first_content())
 
-    if label in ("archiveMemory", MEM_ARCHIVE_LABEL):
+    if label in ("archiveMemory", MEM_ARCHIVE_TASK_LABEL):
         title = _format_title(item.timestamp, f"{cube_display} archived {memory_len} memories")
         return title, _truncate_with_rules(_first_content())
 
-    if label in ("mergeMemory", MEM_ORGANIZE_LABEL):
+    if label in ("mergeMemory", MEM_ORGANIZE_TASK_LABEL):
         title = _format_title(item.timestamp, f"{cube_display} merged {memory_len} memories")
         merged = [c for c in memcube_content if c.get("type") == "merged"]
         post = [c for c in memcube_content if c.get("type") == "postMerge"]

examples/mem_scheduler/redis_example.py

@@ -9,8 +9,8 @@
 from memos.configs.mem_scheduler import SchedulerConfigFactory
 from memos.mem_cube.general import GeneralMemCube
 from memos.mem_scheduler.scheduler_factory import SchedulerFactory
-from memos.mem_scheduler.schemas.general_schemas import QUERY_LABEL
 from memos.mem_scheduler.schemas.message_schemas import ScheduleMessageItem
+from memos.mem_scheduler.schemas.task_schemas import QUERY_TASK_LABEL
 
 
 if TYPE_CHECKING:
@@ -55,7 +55,7 @@ def service_run():
         message_item = ScheduleMessageItem(
             user_id=user_id,
             mem_cube_id="mem_cube_2",
-            label=QUERY_LABEL,
+            label=QUERY_TASK_LABEL,
             mem_cube=mem_cube,
             content=query,
             timestamp=datetime.now(),

examples/mem_scheduler/try_schedule_modules.py

@@ -14,7 +14,7 @@
 from memos.mem_cube.general import GeneralMemCube
 from memos.mem_scheduler.analyzer.mos_for_test_scheduler import MOSForTestScheduler
 from memos.mem_scheduler.general_scheduler import GeneralScheduler
-from memos.mem_scheduler.schemas.general_schemas import (
+from memos.mem_scheduler.schemas.task_schemas import (
     NOT_APPLICABLE_TYPE,
 )
 

src/memos/api/handlers/chat_handler.py

@@ -30,11 +30,11 @@
     prepare_reference_data,
     process_streaming_references_complete,
 )
-from memos.mem_scheduler.schemas.general_schemas import (
-    ANSWER_LABEL,
-    QUERY_LABEL,
-)
 from memos.mem_scheduler.schemas.message_schemas import ScheduleMessageItem
+from memos.mem_scheduler.schemas.task_schemas import (
+    ANSWER_TASK_LABEL,
+    QUERY_TASK_LABEL,
+)
 from memos.templates.mos_prompts import (
     FURTHER_SUGGESTION_PROMPT,
     get_memos_prompt,
@@ -244,7 +244,7 @@ def generate_chat_response() -> Generator[str, None, None]:
                         user_id=chat_req.user_id,
                         mem_cube_id=scheduler_cube_id,
                         query=chat_req.query,
-                        label=QUERY_LABEL,
+                        label=QUERY_TASK_LABEL,
                     )
                     # Extract memories from search results
                     memories_list = []
@@ -388,22 +388,6 @@ def generate_chat_response() -> Generator[str, None, None]:
                         [chat_req.mem_cube_id] if chat_req.mem_cube_id else [chat_req.user_id]
                     )
 
-                    search_req = APISearchRequest(
-                        query=chat_req.query,
-                        user_id=chat_req.user_id,
-                        readable_cube_ids=readable_cube_ids,
-                        mode=chat_req.mode,
-                        internet_search=chat_req.internet_search,
-                        top_k=chat_req.top_k,
-                        chat_history=chat_req.history,
-                        session_id=chat_req.session_id,
-                        include_preference=chat_req.include_preference,
-                        pref_top_k=chat_req.pref_top_k,
-                        filter=chat_req.filter,
-                        playground_search_goal_parser=True,
-                    )
-
-                    search_response = self.search_handler.handle_search_memories(search_req)
                     # for playground, add the query to memory without response
                     self._start_add_to_memory(
                         user_id=chat_req.user_id,
@@ -414,7 +398,6 @@ def generate_chat_response() -> Generator[str, None, None]:
                         async_mode="sync",
                     )
 
-                    yield f"data: {json.dumps({'type': 'status', 'data': '1'})}\n\n"
                     # Use first readable cube ID for scheduler (backward compatibility)
                     scheduler_cube_id = (
                         readable_cube_ids[0] if readable_cube_ids else chat_req.user_id
@@ -423,24 +406,42 @@ def generate_chat_response() -> Generator[str, None, None]:
                         user_id=chat_req.user_id,
                         mem_cube_id=scheduler_cube_id,
                         query=chat_req.query,
-                        label=QUERY_LABEL,
+                        label=QUERY_TASK_LABEL,
                     )
-                    # Extract memories from search results
+
+                    # ====== first search without parse goal ======
+                    search_req = APISearchRequest(
+                        query=chat_req.query,
+                        user_id=chat_req.user_id,
+                        readable_cube_ids=readable_cube_ids,
+                        mode=chat_req.mode,
+                        internet_search=False,
+                        top_k=chat_req.top_k,
+                        chat_history=chat_req.history,
+                        session_id=chat_req.session_id,
+                        include_preference=chat_req.include_preference,
+                        pref_top_k=chat_req.pref_top_k,
+                        filter=chat_req.filter,
+                    )
+                    search_response = self.search_handler.handle_search_memories(search_req)
+
+                    yield f"data: {json.dumps({'type': 'status', 'data': '1'})}\n\n"
+
+                    # Extract memories from search results (first search)
                     memories_list = []
                     if search_response.data and search_response.data.get("text_mem"):
                         text_mem_results = search_response.data["text_mem"]
                         if text_mem_results and text_mem_results[0].get("memories"):
                             memories_list = text_mem_results[0]["memories"]
 
                     # Filter memories by threshold
-                    filtered_memories = self._filter_memories_by_threshold(memories_list)
+                    first_filtered_memories = self._filter_memories_by_threshold(memories_list)
+
+                    # Prepare reference data (first search)
+                    reference = prepare_reference_data(first_filtered_memories)
+                    # get preference string
+                    pref_string = search_response.data.get("pref_string", "")
 
-                    # Prepare reference data
-                    reference = prepare_reference_data(filtered_memories)
-                    # get internet reference
-                    internet_reference = self._get_internet_reference(
-                        search_response.data.get("text_mem")[0]["memories"]
-                    )
                     yield f"data: {json.dumps({'type': 'reference', 'data': reference})}\n\n"
 
                     # Prepare preference markdown string
@@ -450,9 +451,52 @@ def generate_chat_response() -> Generator[str, None, None]:
                         pref_md_string = self._build_pref_md_string_for_playground(pref_memories)
                         yield f"data: {json.dumps({'type': 'pref_md_string', 'data': pref_md_string})}\n\n"
 
+                    # internet status
+                    yield f"data: {json.dumps({'type': 'status', 'data': 'start_internet_search'})}\n\n"
+
+                    # ====== second search with parse goal ======
+                    search_req = APISearchRequest(
+                        query=chat_req.query,
+                        user_id=chat_req.user_id,
+                        readable_cube_ids=readable_cube_ids,
+                        mode=chat_req.mode,
+                        internet_search=chat_req.internet_search,
+                        top_k=chat_req.top_k,
+                        chat_history=chat_req.history,
+                        session_id=chat_req.session_id,
+                        include_preference=False,
+                        filter=chat_req.filter,
+                        playground_search_goal_parser=True,
+                    )
+                    search_response = self.search_handler.handle_search_memories(search_req)
+
+                    # Extract memories from search results (second search)
+                    memories_list = []
+                    if search_response.data and search_response.data.get("text_mem"):
+                        text_mem_results = search_response.data["text_mem"]
+                        if text_mem_results and text_mem_results[0].get("memories"):
+                            memories_list = text_mem_results[0]["memories"]
+
+                    # Filter memories by threshold
+                    second_filtered_memories = self._filter_memories_by_threshold(memories_list)
+
+                    # dedup and supplement memories
+                    filtered_memories = self._dedup_and_supplement_memories(
+                        first_filtered_memories, second_filtered_memories
+                    )
+
+                    # Prepare remain reference data (second search)
+                    reference = prepare_reference_data(filtered_memories)
+                    # get internet reference
+                    internet_reference = self._get_internet_reference(
+                        search_response.data.get("text_mem")[0]["memories"]
+                    )
+
+                    yield f"data: {json.dumps({'type': 'reference', 'data': reference})}\n\n"
+
                     # Step 2: Build system prompt with memories
                     system_prompt = self._build_enhance_system_prompt(
-                        filtered_memories, search_response.data.get("pref_string", "")
+                        filtered_memories, pref_string
                     )
 
                     # Prepare messages
@@ -588,6 +632,19 @@ def generate_chat_response() -> Generator[str, None, None]:
             self.logger.error(f"Failed to start chat stream: {traceback.format_exc()}")
             raise HTTPException(status_code=500, detail=str(traceback.format_exc())) from err
 
+    def _dedup_and_supplement_memories(
+        self, first_filtered_memories: list, second_filtered_memories: list
+    ) -> list:
+        """Remove memory from second_filtered_memories that already exists in first_filtered_memories, return remaining memories"""
+        # Create a set of IDs from first_filtered_memories for efficient lookup
+        first_memory_ids = {memory["id"] for memory in first_filtered_memories}
+
+        remaining_memories = []
+        for memory in second_filtered_memories:
+            if memory["id"] not in first_memory_ids:
+                remaining_memories.append(memory)
+        return remaining_memories
+
     def _get_internet_reference(
         self, search_response: list[dict[str, any]]
     ) -> list[dict[str, any]]:
@@ -1034,7 +1091,7 @@ async def _post_chat_processing(
 
             # Send answer to scheduler
             self._send_message_to_scheduler(
-                user_id=user_id, mem_cube_id=cube_id, query=clean_response, label=ANSWER_LABEL
+                user_id=user_id, mem_cube_id=cube_id, query=clean_response, label=ANSWER_TASK_LABEL
             )
 
             self.logger.info(f"Post-chat processing completed for user {user_id}")

src/memos/api/handlers/memory_handler.py

@@ -209,12 +209,8 @@ def handle_delete_memories(delete_mem_req: DeleteMemoryRequest, naive_mem_cube:
             if naive_mem_cube.pref_mem is not None:
                 naive_mem_cube.pref_mem.delete(delete_mem_req.memory_ids)
         elif delete_mem_req.file_ids is not None:
-            # TODO: Implement deletion by file_ids
-            # Need to find memory_ids associated with file_ids and delete them
-            logger.warning("Deletion by file_ids not implemented yet")
-            return DeleteMemoryResponse(
-                message="Deletion by file_ids not implemented yet",
-                data={"status": "failure"},
+            naive_mem_cube.text_mem.delete_by_filter(
+                writable_cube_ids=delete_mem_req.writable_cube_ids, file_ids=delete_mem_req.file_ids
             )
         elif delete_mem_req.filter is not None:
             # TODO: Implement deletion by filter

src/memos/configs/mem_reader.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Any, ClassVar
 
-from pydantic import Field, field_validator, model_validator
+from pydantic import ConfigDict, Field, field_validator, model_validator
 
 from memos.configs.base import BaseConfig
 from memos.configs.chunker import ChunkerConfigFactory
@@ -44,6 +44,8 @@ def parse_datetime(cls, value):
 class SimpleStructMemReaderConfig(BaseMemReaderConfig):
     """SimpleStruct MemReader configuration class."""
 
+    model_config = ConfigDict(extra="allow", strict=True)
+
 
 class MultiModalStructMemReaderConfig(BaseMemReaderConfig):
     """MultiModalStruct MemReader configuration class."""
@@ -58,6 +60,8 @@ class MultiModalStructMemReaderConfig(BaseMemReaderConfig):
 class StrategyStructMemReaderConfig(BaseMemReaderConfig):
     """StrategyStruct MemReader configuration class."""
 
+    model_config = ConfigDict(extra="allow", strict=True)
+
 
 class MemReaderConfigFactory(BaseConfig):
     """Factory class for creating MemReader configurations."""

src/memos/graph_dbs/neo4j.py

@@ -1588,7 +1588,7 @@ def delete_node_by_prams(
                 file_id_and_conditions.append(f"${param_name} IN n.file_ids")
             if file_id_and_conditions:
                 # Use AND to require all file_ids to be present
-                where_clauses.append(f"({' AND '.join(file_id_and_conditions)})")
+                where_clauses.append(f"({' OR '.join(file_id_and_conditions)})")
 
         # Query nodes by filter if provided
         filter_ids = []