Merge branch 'dev_new' into feat/deep-search

Author: fridayL

examples/mem_reader/compare_simple_vs_multimodal.py

@@ -469,7 +469,7 @@ class APIADDRequest(BaseRequest):
         ),
     )
 
-    info: dict[str, str] | None = Field(
+    info: dict[str, Any] | None = Field(
         None,
         description=(
             "Additional metadata for the add request. "

src/memos/api/product_models.py

@@ -171,6 +171,57 @@ def _build_window_from_items(
 
         return aggregated_item
 
+    def _process_string_fine(
+        self,
+        fast_memory_items: list[TextualMemoryItem],
+        info: dict[str, Any],
+        custom_tags: list[str] | None = None,
+    ) -> list[TextualMemoryItem]:
+        """
+        Process fast mode memory items through LLM to generate fine mode memories.
+        """
+        if not fast_memory_items:
+            return []
+
+        fine_memory_items = []
+
+        for fast_item in fast_memory_items:
+            # Extract memory text (string content)
+            mem_str = fast_item.memory or ""
+            if not mem_str.strip():
+                continue
+            sources = fast_item.metadata.sources or []
+            if not isinstance(sources, list):
+                sources = [sources]
+            try:
+                resp = self._get_llm_response(mem_str, custom_tags)
+            except Exception as e:
+                logger.error(f"[MultiModalFine] Error calling LLM: {e}")
+                continue
+            for m in resp.get("memory list", []):
+                try:
+                    # Normalize memory_type (same as simple_struct)
+                    memory_type = (
+                        m.get("memory_type", "LongTermMemory")
+                        .replace("长期记忆", "LongTermMemory")
+                        .replace("用户记忆", "UserMemory")
+                    )
+                    # Create fine mode memory item (same as simple_struct)
+                    node = self._make_memory_item(
+                        value=m.get("value", ""),
+                        info=info,
+                        memory_type=memory_type,
+                        tags=m.get("tags", []),
+                        key=m.get("key", ""),
+                        sources=sources,  # Preserve sources from fast item
+                        background=resp.get("summary", ""),
+                    )
+                    fine_memory_items.append(node)
+                except Exception as e:
+                    logger.error(f"[MultiModalFine] parse error: {e}")
+
+        return fine_memory_items
+
     @timed
     def _process_multi_modal_data(
         self, scene_data_info: MessagesType, info, mode: str = "fine", **kwargs
@@ -208,21 +259,21 @@ def _process_multi_modal_data(
         if mode == "fast":
             return fast_memory_items
         else:
-            # TODO: parallel call llm and get fine multimodal items
             # Part A: call llm
             fine_memory_items = []
-            fine_memory_items_string_parser = fast_memory_items
+            fine_memory_items_string_parser = self._process_string_fine(
+                fast_memory_items, info, custom_tags
+            )
             fine_memory_items.extend(fine_memory_items_string_parser)
-            # Part B: get fine multimodal items
 
+            # Part B: get fine multimodal items
             for fast_item in fast_memory_items:
                 sources = fast_item.metadata.sources
                 for source in sources:
                     items = self.multi_modal_parser.process_transfer(
                         source, context_items=[fast_item], custom_tags=custom_tags
                     )
                     fine_memory_items.extend(items)
-            logger.warning("Not Implemented Now!")
             return fine_memory_items
 
     @timed
@@ -251,7 +302,7 @@ def _process_transfer_multi_modal_data(
 
         fine_memory_items = []
         # Part A: call llm
-        fine_memory_items_string_parser = []
+        fine_memory_items_string_parser = self._process_string_fine([raw_node], info, custom_tags)
         fine_memory_items.extend(fine_memory_items_string_parser)
         # Part B: get fine multimodal items
         for source in sources:

src/memos/mem_reader/multi_modal_struct.py

@@ -138,7 +138,8 @@ def mix_search_memories(
         target_session_id = search_req.session_id
         if not target_session_id:
             target_session_id = "default_session"
-        search_filter = {"session_id": search_req.session_id} if search_req.session_id else None
+        search_priority = {"session_id": search_req.session_id} if search_req.session_id else None
+        search_filter = search_req.filter
 
         # Rerank Memories - reranker expects TextualMemoryItem objects
 
@@ -155,6 +156,7 @@ def mix_search_memories(
             mode=SearchMode.FAST,
             manual_close_internet=not search_req.internet_search,
             search_filter=search_filter,
+            search_priority=search_priority,
             info=info,
         )
 
@@ -178,7 +180,7 @@ def mix_search_memories(
                 query=search_req.query,  # Use search_req.query instead of undefined query
                 graph_results=history_memories,  # Pass TextualMemoryItem objects directly
                 top_k=search_req.top_k,  # Use search_req.top_k instead of undefined top_k
-                search_filter=search_filter,
+                search_priority=search_priority,
             )
             logger.info(f"Reranked {len(sorted_history_memories)} history memories.")
             processed_hist_mem = self.searcher.post_retrieve(
@@ -234,6 +236,7 @@ def mix_search_memories(
                             mode=SearchMode.FAST,
                             memory_type="All",
                             search_filter=search_filter,
+                            search_priority=search_priority,
                             info=info,
                         )
                     else:

src/memos/mem_scheduler/optimized_scheduler.py

@@ -17,7 +17,11 @@ def __init__(self, llm_provider=None, embedder=None, reranker=None, vector_db=No
 
     @abstractmethod
     def retrieve(
-        self, query: str, top_k: int, info: dict[str, Any] | None = None
+        self,
+        query: str,
+        top_k: int,
+        info: dict[str, Any] | None = None,
+        search_filter: dict[str, Any] | None = None,
     ) -> list[TextualMemoryItem]:
         """Retrieve memories from the retriever."""
 
@@ -76,14 +80,19 @@ def _original_text_reranker(
         return prefs_mem
 
     def retrieve(
-        self, query: str, top_k: int, info: dict[str, Any] | None = None
+        self,
+        query: str,
+        top_k: int,
+        info: dict[str, Any] | None = None,
+        search_filter: dict[str, Any] | None = None,
     ) -> list[TextualMemoryItem]:
         """Retrieve memories from the naive retriever."""
         # TODO: un-support rewrite query and session filter now
         if info:
             info = info.copy()  # Create a copy to avoid modifying the original
             info.pop("chat_history", None)
             info.pop("session_id", None)
+        search_filter = {"and": [info, search_filter]}
         query_embeddings = self.embedder.embed([query])  # Pass as list to get list of embeddings
         query_embedding = query_embeddings[0]  # Get the first (and only) embedding
 
@@ -96,15 +105,15 @@ def retrieve(
                 query,
                 "explicit_preference",
                 top_k * 2,
-                info,
+                search_filter,
             )
             future_implicit = executor.submit(
                 self.vector_db.search,
                 query_embedding,
                 query,
                 "implicit_preference",
                 top_k * 2,
-                info,
+                search_filter,
             )
 
             # Wait for all results

src/memos/memories/textual/prefer_text_memory/retrievers.py

@@ -76,7 +76,9 @@ def get_memory(
         """
         return self.extractor.extract(messages, type, info)
 
-    def search(self, query: str, top_k: int, info=None, **kwargs) -> list[TextualMemoryItem]:
+    def search(
+        self, query: str, top_k: int, info=None, search_filter=None, **kwargs
+    ) -> list[TextualMemoryItem]:
         """Search for memories based on a query.
         Args:
             query (str): The query to search for.
@@ -85,7 +87,8 @@ def search(self, query: str, top_k: int, info=None, **kwargs) -> list[TextualMem
         Returns:
             list[TextualMemoryItem]: List of matching memories.
         """
-        return self.retriever.retrieve(query, top_k, info)
+        logger.info(f"search_filter for preference memory: {search_filter}")
+        return self.retriever.retrieve(query, top_k, info, search_filter)
 
     def load(self, dir: str) -> None:
         """Load memories from the specified directory.

src/memos/memories/textual/preference.py

@@ -50,7 +50,9 @@ def get_memory(
         """
         return self.extractor.extract(messages, type, info)
 
-    def search(self, query: str, top_k: int, info=None, **kwargs) -> list[TextualMemoryItem]:
+    def search(
+        self, query: str, top_k: int, info=None, search_filter=None, **kwargs
+    ) -> list[TextualMemoryItem]:
         """Search for memories based on a query.
         Args:
             query (str): The query to search for.
@@ -59,7 +61,7 @@ def search(self, query: str, top_k: int, info=None, **kwargs) -> list[TextualMem
         Returns:
             list[TextualMemoryItem]: List of matching memories.
         """
-        return self.retriever.retrieve(query, top_k, info)
+        return self.retriever.retrieve(query, top_k, info, search_filter)
 
     def add(self, memories: list[TextualMemoryItem | dict[str, Any]]) -> list[str]:
         """Add memories.

src/memos/memories/textual/simple_preference.py

@@ -162,6 +162,7 @@ def search(
         mode: str = "fast",
         memory_type: str = "All",
         manual_close_internet: bool = True,
+        search_priority: dict | None = None,
         search_filter: dict | None = None,
         user_name: str | None = None,
     ) -> list[TextualMemoryItem]:
@@ -209,7 +210,14 @@ def search(
                 manual_close_internet=manual_close_internet,
             )
         return searcher.search(
-            query, top_k, info, mode, memory_type, search_filter, user_name=user_name
+            query,
+            top_k,
+            info,
+            mode,
+            memory_type,
+            search_filter,
+            search_priority,
+            user_name=user_name,
         )
 
     def get_relevant_subgraph(

src/memos/memories/textual/tree.py

@@ -38,6 +38,7 @@ def retrieve(
         memory_scope: str,
         query_embedding: list[list[float]] | None = None,
         search_filter: dict | None = None,
+        search_priority: dict | None = None,
         user_name: str | None = None,
         id_filter: dict | None = None,
         use_fast_graph: bool = False,
@@ -62,9 +63,12 @@ def retrieve(
             raise ValueError(f"Unsupported memory scope: {memory_scope}")
 
         if memory_scope == "WorkingMemory":
-            # For working memory, retrieve all entries (no filtering)
+            # For working memory, retrieve all entries (no session-oriented filtering)
             working_memories = self.graph_store.get_all_memory_items(
-                scope="WorkingMemory", include_embedding=False, user_name=user_name
+                scope="WorkingMemory",
+                include_embedding=False,
+                user_name=user_name,
+                filter=search_filter,
             )
             return [TextualMemoryItem.from_dict(record) for record in working_memories[:top_k]]
 
@@ -84,6 +88,7 @@ def retrieve(
                 memory_scope,
                 top_k,
                 search_filter=search_filter,
+                search_priority=search_priority,
                 user_name=user_name,
             )
             if self.use_bm25:
@@ -274,6 +279,7 @@ def _vector_recall(
         status: str = "activated",
         cube_name: str | None = None,
         search_filter: dict | None = None,
+        search_priority: dict | None = None,
         user_name: str | None = None,
     ) -> list[TextualMemoryItem]:
         """
@@ -283,39 +289,41 @@ def _vector_recall(
         if not query_embedding:
             return []
 
-        def search_single(vec, filt=None):
+        def search_single(vec, search_priority=None, search_filter=None):
             return (
                 self.graph_store.search_by_embedding(
                     vector=vec,
                     top_k=top_k,
                     status=status,
                     scope=memory_scope,
                     cube_name=cube_name,
-                    search_filter=filt,
+                    search_filter=search_priority,
+                    filter=search_filter,
                     user_name=user_name,
                 )
                 or []
             )
 
         def search_path_a():
-            """Path A: search without filter"""
+            """Path A: search without priority"""
             path_a_hits = []
             with ContextThreadPoolExecutor() as executor:
                 futures = [
-                    executor.submit(search_single, vec, None) for vec in query_embedding[:max_num]
+                    executor.submit(search_single, vec, None, search_filter)
+                    for vec in query_embedding[:max_num]
                 ]
                 for f in concurrent.futures.as_completed(futures):
                     path_a_hits.extend(f.result() or [])
             return path_a_hits
 
         def search_path_b():
-            """Path B: search with filter"""
-            if not search_filter:
+            """Path B: search with priority"""
+            if not search_priority:
                 return []
             path_b_hits = []
             with ContextThreadPoolExecutor() as executor:
                 futures = [
-                    executor.submit(search_single, vec, search_filter)
+                    executor.submit(search_single, vec, search_priority, search_filter)
                     for vec in query_embedding[:max_num]
                 ]
                 for f in concurrent.futures.as_completed(futures):