add try catch logic in server router, add dedup logic in explicit pref

Author: yuan.wang

src/memos/api/routers/server_router.py

@@ -388,36 +388,44 @@ def search_memories(search_req: APISearchRequest):
     search_mode = search_req.mode
 
     def _search_text():
-        if search_mode == SearchMode.FAST:
-            formatted_memories = fast_search_memories(
-                search_req=search_req, user_context=user_context
-            )
-        elif search_mode == SearchMode.FINE:
-            formatted_memories = fine_search_memories(
-                search_req=search_req, user_context=user_context
-            )
-        elif search_mode == SearchMode.MIXTURE:
-            formatted_memories = mix_search_memories(
-                search_req=search_req, user_context=user_context
-            )
-        else:
-            logger.error(f"Unsupported search mode: {search_mode}")
-            raise HTTPException(status_code=400, detail=f"Unsupported search mode: {search_mode}")
-        return formatted_memories
+        try:
+            if search_mode == SearchMode.FAST:
+                formatted_memories = fast_search_memories(
+                    search_req=search_req, user_context=user_context
+                )
+            elif search_mode == SearchMode.FINE:
+                formatted_memories = fine_search_memories(
+                    search_req=search_req, user_context=user_context
+                )
+            elif search_mode == SearchMode.MIXTURE:
+                formatted_memories = mix_search_memories(
+                    search_req=search_req, user_context=user_context
+                )
+            else:
+                logger.error(f"Unsupported search mode: {search_mode}")
+                raise HTTPException(status_code=400, detail=f"Unsupported search mode: {search_mode}")
+            return formatted_memories
+        except Exception as e:
+            logger.error("Error in search_text: %s; traceback: %s", e, traceback.format_exc())
+            return []
 
     def _search_pref():
         if os.getenv("ENABLE_PREFERENCE_MEMORY", "false").lower() != "true":
             return []
-        results = naive_mem_cube.pref_mem.search(
-            query=search_req.query,
-            top_k=search_req.pref_top_k,
-            info={
-                "user_id": search_req.user_id,
-                "session_id": search_req.session_id,
-                "chat_history": search_req.chat_history,
-            },
-        )
-        return [_format_memory_item(data) for data in results]
+        try:
+            results = naive_mem_cube.pref_mem.search(
+                query=search_req.query,
+                top_k=search_req.pref_top_k,
+                info={
+                    "user_id": search_req.user_id,
+                    "session_id": search_req.session_id,
+                    "chat_history": search_req.chat_history,
+                },
+            )
+            return [_format_memory_item(data) for data in results]
+        except Exception as e:
+            logger.error("Error in _search_pref: %s; traceback: %s", e, traceback.format_exc())
+            return []
 
     with ContextThreadPoolExecutor(max_workers=2) as executor:
         text_future = executor.submit(_search_text)

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

@@ -109,8 +109,10 @@ def _judge_dup_with_text_mem(self, new_pref: MilvusVecDBItem) -> bool:
         new_preference = {"id": new_pref.id, "memory": new_pref.payload["preference"]}
 
         prompt = NAIVE_JUDGE_DUP_WITH_TEXT_MEM_PROMPT.replace(
-            "{new_preference}", json.dumps(new_preference)
-        ).replace("{retrieved_memories}", json.dumps(text_mem_recalls))
+            "{new_preference}", json.dumps(new_preference, ensure_ascii=False)
+        ).replace(
+            "{retrieved_memories}", json.dumps(text_mem_recalls, ensure_ascii=False)
+        )
         try:
             response = self.llm_provider.generate([{"role": "user", "content": prompt}])
             response = response.strip().replace("```json", "").replace("```", "").strip()
@@ -142,7 +144,7 @@ def _dedup_explicit_pref_by_textual(
         self, new_prefs: list[MilvusVecDBItem]
     ) -> list[MilvusVecDBItem]:
         """Deduplicate explicit preferences by textual memory."""
-        if os.getenv("PREF_DEDUP_EXP_BY_TEXTUAL", "false").lower() != "true":
+        if os.getenv("DEDUP_PREF_EXP_BY_TEXTUAL", "false").lower() != "true" or not self.text_mem:
             return new_prefs
         dedup_prefs = []
         with ContextThreadPoolExecutor(max_workers=max(1, min(len(new_prefs), 5))) as executor:
@@ -205,8 +207,11 @@ def _update_memory_op_trace(
         ]
 
         rsp = self._judge_update_or_add_trace_op(
-            new_mems=json.dumps(new_mem_inputs),
-            retrieved_mems=json.dumps(retrieved_mem_inputs) if retrieved_mem_inputs else "",
+            new_mems=json.dumps(new_mem_inputs, ensure_ascii=False),
+            retrieved_mems=
+                json.dumps(retrieved_mem_inputs, ensure_ascii=False)
+                if retrieved_mem_inputs
+                else "",
         )
         if not rsp:
             dedup_rsp = self._dedup_explicit_pref_by_textual(new_vec_db_items)
@@ -293,8 +298,11 @@ def _update_memory_fine(
             if mem.payload.get("preference", None)
         ]
         rsp = self._judge_update_or_add_fine(
-            new_mem=json.dumps(new_mem_input),
-            retrieved_mems=json.dumps(retrieved_mem_inputs) if retrieved_mem_inputs else "",
+            new_mem=json.dumps(new_mem_input, ensure_ascii=False),
+            retrieved_mems=
+                json.dumps(retrieved_mem_inputs, ensure_ascii=False)
+                if retrieved_mem_inputs
+                else "",
         )
         need_update = rsp.get("need_update", False) if rsp else False
         need_update = (

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

@@ -119,6 +119,9 @@ def retrieve(
             if pref.payload.get("preference", None)
         ]
 
+        # store explicit id and score, use it after reranker
+        explicit_id_scores = {item.id: item.score for item in explicit_prefs}
+
         reranker_map = {
             "naive": self._naive_reranker,
             "original_text": self._original_text_reranker,
@@ -131,4 +134,9 @@ def retrieve(
             query=query, prefs_mem=implicit_prefs_mem, prefs=implicit_prefs, top_k=top_k
         )
 
+        # filter explicit mem by score bigger than threshold 
+        explicit_prefs_mem = [
+            item for item in explicit_prefs_mem if explicit_id_scores.get(item.id, 0) >= 0.2
+        ]
+
         return explicit_prefs_mem + implicit_prefs_mem

src/memos/vec_dbs/milvus.py

@@ -236,29 +236,32 @@ def search(
             "sparse": self._sparse_search,
             "hybrid": self._hybrid_search,
         }
+        try:
+            results = search_func_map[search_type](
+                collection_name=collection_name,
+                query_vector=query_vector,
+                query=query,
+                top_k=top_k,
+                filter=expr,
+            )
 
-        results = search_func_map[search_type](
-            collection_name=collection_name,
-            query_vector=query_vector,
-            query=query,
-            top_k=top_k,
-            filter=expr,
-        )
-
-        items = []
-        for hit in results[0]:
-            entity = hit.get("entity", {})
-
-            items.append(
-                MilvusVecDBItem(
-                    id=str(entity.get("id")),
-                    memory=entity.get("memory"),
-                    original_text=entity.get("original_text"),
-                    vector=entity.get("vector"),
-                    payload=entity.get("payload", {}),
-                    score=1 - float(hit["distance"]),
+            items = []
+            for hit in results[0]:
+                entity = hit.get("entity", {})
+
+                items.append(
+                    MilvusVecDBItem(
+                        id=str(entity.get("id")),
+                        memory=entity.get("memory"),
+                        original_text=entity.get("original_text"),
+                        vector=entity.get("vector"),
+                        payload=entity.get("payload", {}),
+                        score=1 - float(hit["distance"]),
+                    )
                 )
-            )
+        except Exception as e:
+            logger.error("Error in _%s_search: %s", search_type, e)
+            return []
 
         logger.info(f"Milvus search completed with {len(items)} results.")
         return items