address bugs

Author: tangg555

src/memos/mem_reader/simple_struct.py

@@ -557,105 +557,6 @@ def filter_hallucination_in_memories(
 
         return memory_list
 
-    def add_before_search(
-        self,
-        messages: list[dict],
-        memory_list: list[TextualMemoryItem],
-        user_name: str,
-        info: dict[str, Any],
-    ) -> list[TextualMemoryItem]:
-        # Build input objects with memory text and metadata (timestamps, sources, etc.)
-        template = PROMPT_MAPPING["add_before_search"]
-
-        if not self.searcher:
-            try:
-                from memos.mem_reader.utils import init_searcher
-
-                self.searcher = init_searcher(self.llm, self.embedder)
-            except Exception as e:
-                logger.error(f"[add_before_search] Failed to init searcher: {e}")
-                return memory_list
-
-        # 1. Gather candidates and search for related memories
-        candidates_data = []
-        for idx, mem in enumerate(memory_list):
-            try:
-                related_memories = self.searcher.search(
-                    query=mem.memory, top_k=3, mode="fast", user_nam=user_name, info=info
-                )
-                related_text = "None"
-                if related_memories:
-                    related_text = "\n".join([f"- {r.memory}" for r in related_memories])
-
-                candidates_data.append(
-                    {"idx": idx, "new_memory": mem.memory, "related_memories": related_text}
-                )
-            except Exception as e:
-                logger.error(f"[add_before_search] Search error for memory '{mem.memory}': {e}")
-                # If search fails, we can either skip this check or treat related as empty
-                candidates_data.append(
-                    {
-                        "idx": idx,
-                        "new_memory": mem.memory,
-                        "related_memories": "None (Search Failed)",
-                    }
-                )
-
-        if not candidates_data:
-            return memory_list
-
-        # 2. Build Prompt
-        messages_inline = "\n".join(
-            [
-                f"- [{message.get('role', 'unknown')}]: {message.get('content', '')}"
-                for message in messages
-            ]
-        )
-
-        candidates_inline_dict = {
-            str(item["idx"]): {
-                "new_memory": item["new_memory"],
-                "related_memories": item["related_memories"],
-            }
-            for item in candidates_data
-        }
-
-        candidates_inline = json.dumps(candidates_inline_dict, ensure_ascii=False, indent=2)
-
-        prompt = template.format(
-            messages_inline=messages_inline, candidates_inline=candidates_inline
-        )
-
-        # 3. Call LLM
-        try:
-            raw = self.llm.generate([{"role": "user", "content": prompt}])
-            success, parsed_result = parse_keep_filter_response(raw)
-
-            if not success:
-                logger.warning("[add_before_search] Failed to parse LLM response, keeping all.")
-                return memory_list
-
-            # 4. Filter
-            filtered_list = []
-            for idx, mem in enumerate(memory_list):
-                res = parsed_result.get(idx)
-                if not res:
-                    filtered_list.append(mem)
-                    continue
-
-                if res.get("keep", True):
-                    filtered_list.append(mem)
-                else:
-                    logger.info(
-                        f"[add_before_search] Dropping memory: '{mem.memory}', reason: '{res.get('reason')}'"
-                    )
-
-            return filtered_list
-
-        except Exception as e:
-            logger.error(f"[add_before_search] LLM execution error: {e}")
-            return memory_list
-
     def _read_memory(
         self, messages: list[MessagesType], type: str, info: dict[str, Any], mode: str = "fine"
     ) -> list[list[TextualMemoryItem]]:

tests/mem_reader/test_simple_structure.py

@@ -116,98 +116,6 @@ def test_parse_json_result_failure(self):
 
         self.assertEqual(result, {})
 
-    def test_add_before_search(self):
-        """Test add_before_search method."""
-        import json
-
-        from memos.memories.textual.item import TextualMemoryItem, TreeNodeTextualMemoryMetadata
-
-        # Mock searcher
-        self.reader.searcher = MagicMock()
-        self.reader.searcher.search.return_value = [
-            TextualMemoryItem(
-                memory="Related memory 1",
-                metadata=TreeNodeTextualMemoryMetadata(
-                    user_id="user1",
-                    session_id="session1",
-                    memory_type="LongTermMemory",
-                    status="activated",
-                    tags=[],
-                    key="key1",
-                    embedding=[0.1],
-                    usage=[],
-                    sources=[],
-                    background="",
-                    confidence=0.99,
-                    type="fact",
-                    info={},
-                ),
-            )
-        ]
-
-        # Mock LLM response for filter
-        # The method expects a JSON response with keep/drop decisions
-        mock_response = json.dumps(
-            {
-                "0": {"keep": True, "reason": "Relevant"},
-                "1": {"keep": False, "reason": "Duplicate"},
-            }
-        )
-        self.reader.llm.generate.return_value = mock_response
-
-        messages = [{"role": "user", "content": "test message"}]
-        memory_list = [
-            TextualMemoryItem(
-                memory="Mem 1",
-                metadata=TreeNodeTextualMemoryMetadata(
-                    user_id="user1",
-                    session_id="session1",
-                    memory_type="LongTermMemory",
-                    status="activated",
-                    tags=[],
-                    key="key1",
-                    embedding=[0.1],
-                    usage=[],
-                    sources=[],
-                    background="",
-                    confidence=0.99,
-                    type="fact",
-                    info={},
-                ),
-            ),
-            TextualMemoryItem(
-                memory="Mem 2",
-                metadata=TreeNodeTextualMemoryMetadata(
-                    user_id="user1",
-                    session_id="session1",
-                    memory_type="LongTermMemory",
-                    status="activated",
-                    tags=[],
-                    key="key2",
-                    embedding=[0.1],
-                    usage=[],
-                    sources=[],
-                    background="",
-                    confidence=0.99,
-                    type="fact",
-                    info={},
-                ),
-            ),
-        ]
-        info = {"user_id": "user1", "session_id": "session1"}
-
-        # Call the method
-        result = self.reader.add_before_search(messages, memory_list, info)
-
-        # Assertions
-        # Check if searcher.search was called with correct info
-        self.reader.searcher.search.assert_called_with(
-            query="Mem 2", top_k=3, mode="fast", info=info
-        )
-        # Check result
-        self.assertEqual(len(result), 1)
-        self.assertEqual(result[0].memory, "Mem 1")
-
 
 if __name__ == "__main__":
     unittest.main()