feat(agent): 添加 Halumem 版本的记忆检索器和摘要器

- 添加 PersonalHalumemRetriever 和 PersonalHalumemSummarizer 类
- 在 memory 模块中注册新的检索器和摘要器
- 添加 UpdateProfileFilterOlder、DeleteProfile 和 AddProfile 工具
- 将 AddDraftAndRetrieveSimilarMemory 重命名为 AddAndRetrieveSimilarMemory
- 修改配置文件中的默认模型名称为 qwen-flash
- 在 benchmark 中添加 Halumem 评估支持和实时更新功能
- 降低 ProfileHandler 的最大容量限制并添加重复节点过滤逻辑
- 在 ReMe 中添加 halumem 版本的记忆代理配置

benchmark/halumem/eval_reme.py

@@ -218,7 +218,7 @@ async def answer_question_with_memories(
 
     result = await reme.llm.simple_request_for_json(
         prompt=prompt,
-        model_name="qwen3-30b-a3b-instruct-2507"
+        model_name="qwen-flash"
     )
 
     return result
@@ -301,6 +301,8 @@ class MemoryProcessor:
                 user_name=user_id,
                 version=self.algo_version,
                 return_dict=True,
+                enable_time_filter=True,
+                enable_thinking_params=False
             )
 
             duration_ms = (time.time() - start) * 1000
@@ -333,6 +335,8 @@ class MemoryProcessor:
             user_name=user_id,
             version=self.algo_version,
             return_dict=True,
+            enable_time_filter=True,
+            enable_thinking_params=False
         )
 
         # Extract memories from response
@@ -404,6 +408,16 @@ class QuestionAnsweringEvaluator:
                 model_name=self.eval_model_name
             )
 
+            eval_result_original_answer = await evaluation_for_question(
+                reme=self.reme,
+                question=qa["question"],
+                reference_answer=qa["answer"],
+                key_memory_points=evidence_text,
+                response=retrieved_memories,
+                dialogue=formatted_dialogue,
+                model_name=self.eval_model_name
+            )
+
             # Build result record
             qa_result = {
                 **qa,
@@ -416,7 +430,9 @@ class QuestionAnsweringEvaluator:
                 "retrieve_messages": agent_messages,
                 "search_duration_ms": duration_ms,
                 "result_type": eval_result.get("evaluation_result"),
-                "question_answering_reasoning": eval_result.get("reasoning", "")
+                "question_answering_reasoning": eval_result.get("reasoning", ""),
+                "original_result_type": eval_result_original_answer.get("evaluation_result"),
+                "original_question_answering_reasoning": eval_result_original_answer.get("reasoning", ""),
             }
             results.append(qa_result)
 
@@ -427,8 +443,8 @@ class MetricsAggregator:
     """Aggregates evaluation metrics."""
 
     @staticmethod
-    def compute_qa_metrics(qa_records: list[dict]) -> dict[str, Any]:
-        """Compute question answering metrics."""
+    def _compute_single_metric(qa_records: list[dict], result_key: str) -> dict[str, Any]:
+        """Compute metrics for a single result type key."""
         total = len(qa_records)
         if total == 0:
             return {
@@ -448,7 +464,7 @@ class MetricsAggregator:
         valid = 0
 
         for qa in qa_records:
-            result_type = qa.get("result_type", "")
+            result_type = qa.get(result_key, "")
 
             if result_type in ["Correct", "Hallucination", "Omission"]:
                 valid += 1
@@ -482,6 +498,14 @@ class MetricsAggregator:
 
         return metrics
 
+    @staticmethod
+    def compute_qa_metrics(qa_records: list[dict]) -> dict[str, Any]:
+        """Compute question answering metrics for both result_type and original_result_type."""
+        return {
+            "with_llm_answer": MetricsAggregator._compute_single_metric(qa_records, "result_type"),
+            "with_original_memories": MetricsAggregator._compute_single_metric(qa_records, "original_result_type")
+        }
+
     @staticmethod
     def compute_time_metrics(eval_results_file: str) -> dict[str, float]:
         """Compute timing metrics from evaluation results."""
@@ -536,6 +560,10 @@ class HaluMemEvaluator:
         )
         self.data_loader = DataLoader()
 
+        # For real-time updates
+        self._update_lock: asyncio.Lock | None = None
+        self._output_file: str | None = None
+
     async def __aenter__(self):
         """Async context manager entry."""
         return self
@@ -626,8 +654,20 @@ class HaluMemEvaluator:
 
             self.file_manager.save_session(user_name, idx, session_data)
 
+            # Update results file after each session completes
+            await self._trigger_update()
+
         return {"uuid": uuid, "user_name": user_name, "status": "ok"}
 
+    async def _trigger_update(self):
+        """Trigger real-time update of results and statistics."""
+        if self._update_lock is None or self._output_file is None:
+            return
+
+        async with self._update_lock:
+            self.file_manager.combine_results(self._output_file)
+            self._update_statistics(self._output_file)
+
     async def run_evaluation(self):
         """Run the complete evaluation pipeline using ReMe."""
         start_time = time.time()
@@ -661,6 +701,12 @@ class HaluMemEvaluator:
         print(f"Users: {len(users_to_process)} | Concurrency: {self.config.max_concurrency}")
         print("=" * 80 + "\n")
 
+        # Output file path for real-time updates
+        self._output_file = os.path.join(self.config.output_dir, "eval_results.jsonl")
+
+        # Lock for thread-safe file updates
+        self._update_lock = asyncio.Lock()
+
         # Process users with concurrency control
         semaphore = asyncio.Semaphore(self.config.max_concurrency)
 
@@ -671,11 +717,14 @@ class HaluMemEvaluator:
                 # Check cache
                 if self.file_manager.user_has_cache(user_name):
                     print(f"⚡ [{idx}/{len(users_to_process)}] Skipping {user_name} (cached)")
-                    return {"user_name": user_name, "status": "cached"}
+                    result = {"user_name": user_name, "status": "cached"}
+                    # Also trigger update for cached users
+                    await self._trigger_update()
+                else:
+                    print(f"🔄 [{idx}/{len(users_to_process)}] Processing {user_name}...")
+                    result = await self.process_user(user_data)
+                    print(f"✅ [{idx}/{len(users_to_process)}] Completed {user_name}")
 
-                print(f"🔄 [{idx}/{len(users_to_process)}] Processing {user_name}...")
-                result = await self.process_user(user_data)
-                print(f"✅ [{idx}/{len(users_to_process)}] Completed {user_name}")
                 return result
 
         tasks = [
@@ -684,16 +733,57 @@ class HaluMemEvaluator:
         ]
         await asyncio.gather(*tasks)
 
-        # Combine results
-        output_file = os.path.join(self.config.output_dir, "eval_results.jsonl")
-        self.file_manager.combine_results(output_file)
-
         elapsed = time.time() - start_time
         print(f"\n✅ Processing completed in {elapsed:.2f}s")
-        print(f"📁 Results: {output_file}\n")
+        print(f"📁 Results: {self._output_file}\n")
 
-        # Aggregate metrics
-        await self.aggregate_and_report(output_file)
+        # Final aggregation and report
+        await self.aggregate_and_report(self._output_file)
+
+    def _update_statistics(self, results_file: str):
+        """Update statistics file based on current results (for real-time monitoring)."""
+        if not os.path.exists(results_file):
+            return
+
+        # Collect all QA records
+        qa_records = []
+        try:
+            with open(results_file, "r", encoding="utf-8") as f:
+                for line in f:
+                    if not line.strip():
+                        continue
+                    user_data = json.loads(line)
+
+                    for session in user_data["sessions"]:
+                        if session.get("is_generated_qa_session"):
+                            continue
+
+                        eval_results = session.get("evaluation_results", {})
+                        qa_records.extend(
+                            eval_results.get("question_answering_records", [])
+                        )
+        except (json.JSONDecodeError, KeyError):
+            return
+
+        if not qa_records:
+            return
+
+        # Compute metrics
+        qa_metrics = MetricsAggregator.compute_qa_metrics(qa_records)
+        time_metrics = MetricsAggregator.compute_time_metrics(results_file)
+
+        final_results = {
+            "overall_score": {
+                "question_answering": qa_metrics,
+                "time_consuming": time_metrics
+            },
+            "question_answering_records": qa_records
+        }
+
+        # Save statistics
+        report_file = os.path.join(self.config.output_dir, "eval_statistics.json")
+        with open(report_file, "w", encoding="utf-8") as f:
+            json.dump(final_results, f, ensure_ascii=False, indent=4)
 
     async def aggregate_and_report(self, results_file: str):
         """Aggregate results and generate final report."""
@@ -746,14 +836,27 @@ class HaluMemEvaluator:
         print("EVALUATION SUMMARY - REME")
         print("=" * 80 + "\n")
 
-        print("📊 Question Answering:")
-        print(f"  Correct (all):       {qa_metrics['correct_qa_ratio(all)']:.4f}")
-        print(f"  Hallucination (all): {qa_metrics['hallucination_qa_ratio(all)']:.4f}")
-        print(f"  Omission (all):      {qa_metrics['omission_qa_ratio(all)']:.4f}")
-        print(f"  Correct (valid):     {qa_metrics['correct_qa_ratio(valid)']:.4f}")
-        print(f"  Hallucination (valid): {qa_metrics['hallucination_qa_ratio(valid)']:.4f}")
-        print(f"  Omission (valid):    {qa_metrics['omission_qa_ratio(valid)']:.4f}")
-        print(f"  Valid/Total:         {qa_metrics['qa_valid_num']}/{qa_metrics['qa_num']}")
+        # Print metrics for LLM-generated answer (result_type)
+        llm_metrics = qa_metrics["with_llm_answer"]
+        print("📊 Question Answering (with LLM answer):")
+        print(f"  Correct (all):       {llm_metrics['correct_qa_ratio(all)']:.4f}")
+        print(f"  Hallucination (all): {llm_metrics['hallucination_qa_ratio(all)']:.4f}")
+        print(f"  Omission (all):      {llm_metrics['omission_qa_ratio(all)']:.4f}")
+        print(f"  Correct (valid):     {llm_metrics['correct_qa_ratio(valid)']:.4f}")
+        print(f"  Hallucination (valid): {llm_metrics['hallucination_qa_ratio(valid)']:.4f}")
+        print(f"  Omission (valid):    {llm_metrics['omission_qa_ratio(valid)']:.4f}")
+        print(f"  Valid/Total:         {llm_metrics['qa_valid_num']}/{llm_metrics['qa_num']}")
+
+        # Print metrics for original retrieved memories (original_result_type)
+        orig_metrics = qa_metrics["with_original_memories"]
+        print("\n📊 Question Answering (with original memories):")
+        print(f"  Correct (all):       {orig_metrics['correct_qa_ratio(all)']:.4f}")
+        print(f"  Hallucination (all): {orig_metrics['hallucination_qa_ratio(all)']:.4f}")
+        print(f"  Omission (all):      {orig_metrics['omission_qa_ratio(all)']:.4f}")
+        print(f"  Correct (valid):     {orig_metrics['correct_qa_ratio(valid)']:.4f}")
+        print(f"  Hallucination (valid): {orig_metrics['hallucination_qa_ratio(valid)']:.4f}")
+        print(f"  Omission (valid):    {orig_metrics['omission_qa_ratio(valid)']:.4f}")
+        print(f"  Valid/Total:         {orig_metrics['qa_valid_num']}/{orig_metrics['qa_num']}")
 
         print(f"\n⏱️  Time Metrics:")
         print(f"  Memory Addition:  {time_metrics['add_dialogue_duration_time']:.2f} min")
@@ -815,7 +918,8 @@ if __name__ == "__main__":
     parser.add_argument(
         "--data_path",
         type=str,
-        required=True,
+        # required=True,
+        default="/Users/zhouwk/PycharmProjects/MemAgent/dataset/halumem/HaluMem-Medium.jsonl",
         help="Path to HaluMem JSONL file"
     )
     parser.add_argument(
@@ -833,20 +937,20 @@ if __name__ == "__main__":
     parser.add_argument(
         "--max_concurrency",
         type=int,
-        default=100,
+        default=1,
         help="Maximum concurrent user processing (default: 100)"
     )
     parser.add_argument(
         "--eval_model_name",
         type=str,
-        default="qwen3-max",
+        default="qwen-flash",
         # default="qwen3-235b-a22b-instruct-2507",
         help="Model name for evaluation (default: qwen3-max)"
     )
     parser.add_argument(
         "--algo_version",
         type=str,
-        default="v1",
+        default="halumem",
         help="Algorithm version for summary and retrieval (default: v1)"
     )
 

reme/agent/memory/__init__.py

@@ -5,6 +5,8 @@ from .personal.personal_retriever import PersonalRetriever
 from .personal.personal_summarizer import PersonalSummarizer
 from .personal.personal_v1_retriever import PersonalV1Retriever
 from .personal.personal_v1_summarizer import PersonalV1Summarizer
+from .personal.personal_halumem_retriever import PersonalHalumemRetriever
+from .personal.personal_halumem_summarizer import PersonalHalumemSummarizer
 from .procedural.procedural_retriever import ProceduralRetriever
 from .procedural.procedural_summarizer import ProceduralSummarizer
 from .reme_retriever import ReMeRetriever
@@ -19,6 +21,8 @@ __all__ = [
     "PersonalSummarizer",
     "PersonalV1Retriever",
     "PersonalV1Summarizer",
+    "PersonalHalumemRetriever",
+    "PersonalHalumemSummarizer",
     "ProceduralRetriever",
     "ProceduralSummarizer",
     "ReMeRetriever",

reme/config/default.yaml

@@ -15,8 +15,8 @@ http:
 llm:
   default:
     backend: openai
-    model_name: qwen3-30b-a3b-instruct-2507
-#    model_name: qwen-flash
+#    model_name: qwen3-30b-a3b-instruct-2507
+    model_name: qwen-flash
     request_interval: 1
     temperature: 0.0001
 
@@ -46,3 +46,9 @@ token_counter:
     backend: hf
     model_name: Qwen/Qwen3-Coder-30B-A3B-Instruct
     use_mirror: true
+
+
+
+flow:
+  AddMemory:
+    flow_content: AddMemory()

reme/reme.py

@@ -9,6 +9,8 @@ from .agent.memory import (
     ReMeRetriever,
     PersonalV1Summarizer,
     PersonalV1Retriever,
+    PersonalHalumemSummarizer,
+    PersonalHalumemRetriever,
     PersonalSummarizer,
     PersonalRetriever,
     ProceduralSummarizer,
@@ -26,10 +28,13 @@ from .tool.memory import (
     ReadHistory,
     ProfileHandler,
     MemoryHandler,
-    AddDraftAndRetrieveSimilarMemory,
+    AddAndRetrieveSimilarMemory,
     UpdateMemoryV2,
     AddDraftAndReadAllProfiles,
     UpdateProfile,
+    UpdateProfileFilterOlder,
+    DeleteProfile,
+    AddProfile,
     AddHistory,
     ReadAllProfiles,
 )
@@ -104,6 +109,7 @@ class ReMe(Application):
         task_name: str | list[str] = "",
         tool_name: str | list[str] = "",
         enable_thinking_params: bool = True,
+        enable_time_filter: bool = True,
         version: str = "default",
         retrieve_top_k: int = 20,
         return_dict: bool = False,
@@ -121,7 +127,7 @@ class ReMe(Application):
         if version == "default":
             personal_summarizer = PersonalSummarizer(
                 tools=[
-                    AddDraftAndRetrieveSimilarMemory(
+                    AddAndRetrieveSimilarMemory(
                         enable_thinking_params=enable_thinking_params,
                         top_k=retrieve_top_k,
                     ),
@@ -140,7 +146,7 @@ class ReMe(Application):
         elif version == "v1":
             personal_summarizer = PersonalV1Summarizer(
                 tools=[
-                    AddDraftAndRetrieveSimilarMemory(
+                    AddAndRetrieveSimilarMemory(
                         enable_thinking_params=enable_thinking_params,
                         top_k=retrieve_top_k,
                     ),
@@ -151,18 +157,57 @@ class ReMe(Application):
                     ),
                 ],
             )
+        elif version == "halumem":
+            personal_summarizer = PersonalHalumemSummarizer(
+                tools=[
+                    AddAndRetrieveSimilarMemory(
+                        enable_thinking_params=enable_thinking_params,
+                        top_k=retrieve_top_k,
+                    ),
+                    UpdateMemoryV2(
+                        enable_thinking_params=enable_thinking_params
+                    ),
+                    # RetrieveMemory(
+                    #     enable_thinking_params=enable_thinking_params,
+                    #     top_k=retrieve_top_k,
+                    #     enable_time_filter=enable_time_filter,
+                    # ),
 
+                    # 处理userprofile
+                    ReadAllProfiles(
+                        enable_thinking_params=enable_thinking_params,
+                        profile_dir=self.profile_dir,
+                    ),
+                    # UpdateProfileFilterOlder(
+                    #     enable_thinking_params=enable_thinking_params,
+                    #     max_profile_count=50,
+                    #     profile_dir=self.profile_dir,
+                    # ),
+                    UpdateProfile(
+                        enable_thinking_params=enable_thinking_params,
+                        profile_dir=self.profile_dir,
+                    ),
+                    # AddProfile(
+                    #     enable_thinking_params=enable_thinking_params,
+                    #     profile_dir=self.profile_dir,
+                    # ),
+                    # DeleteProfile(
+                    #     enable_thinking_params=enable_thinking_params,
+                    #     profile_dir=self.profile_dir,
+                    # ),
+                ],
+            )
         else:
             raise NotImplementedError
 
         procedural_summarizer: BaseMemoryAgent
-        if version in ["default", "v1"]:
+        if version in ["default", "v1", "halumem"]:
             procedural_summarizer = ProceduralSummarizer(tools=[])
         else:
             raise NotImplementedError
 
         tool_summarizer: BaseMemoryAgent
-        if version in ["default", "v1"]:
+        if version in ["default", "v1", "halumem"]:
             tool_summarizer = ToolSummarizer(tools=[])
         else:
             raise NotImplementedError
@@ -204,7 +249,7 @@ class ReMe(Application):
             memory_agents = [personal_summarizer, procedural_summarizer, tool_summarizer]
 
         reme_summarizer: BaseMemoryAgent
-        if version in ["default", "v1"]:
+        if version in ["default", "v1", "halumem"]:
             reme_summarizer = ReMeSummarizer(tools=[AddHistory(), DelegateTask(memory_agents=memory_agents)])
         else:
             raise NotImplementedError
@@ -270,18 +315,32 @@ class ReMe(Application):
                     ReadHistory(enable_thinking_params=enable_thinking_params),
                 ],
             )
-
+        elif version == "halumem":
+            personal_retriever = PersonalHalumemRetriever(
+                tools=[
+                    ReadAllProfiles(
+                        enable_thinking_params=enable_thinking_params,
+                        profile_dir=self.profile_dir,
+                    ),
+                    RetrieveMemory(
+                        enable_thinking_params=enable_thinking_params,
+                        top_k=retrieve_top_k,
+                        enable_time_filter=enable_time_filter,
+                    ),
+                    ReadHistory(enable_thinking_params=enable_thinking_params),
+                ]
+            )
         else:
             raise NotImplementedError
 
         procedural_retriever: BaseMemoryAgent
-        if version in ["default", "v1"]:
+        if version in ["default", "v1", "halumem"]:
             procedural_retriever = ProceduralRetriever(tools=[])
         else:
             raise NotImplementedError
 
         tool_retriever: BaseMemoryAgent
-        if version in ["default", "v1"]:
+        if version in ["default", "v1", "halumem"]:
             tool_retriever = ToolRetriever(tools=[])
         else:
             raise NotImplementedError
@@ -321,7 +380,7 @@ class ReMe(Application):
             memory_agents = [personal_retriever, procedural_retriever, tool_retriever]
 
         reme_retriever: BaseMemoryAgent
-        if version in ["default", "v1"]:
+        if version in ["default", "v1", "halumem"]:
             reme_retriever = ReMeRetriever(tools=[DelegateTask(memory_agents=memory_agents)])
         else:
             raise NotImplementedError

reme/tool/memory/__init__.py

@@ -7,8 +7,11 @@ from .history.read_history import ReadHistory
 from .profiles.add_draft_and_read_all_profiles import AddDraftAndReadAllProfiles
 from .profiles.profile_handler import ProfileHandler
 from .profiles.read_all_profiles import ReadAllProfiles
+from .profiles.add_profile import AddProfile
 from .profiles.update_profile import UpdateProfile
-from .vector.add_draft_and_retrieve_similar_memory import AddDraftAndRetrieveSimilarMemory
+from .profiles.update_profile_filter_older import UpdateProfileFilterOlder
+from .profiles.delete_profile import DeleteProfile
+from .vector.add_draft_and_retrieve_similar_memory import AddAndRetrieveSimilarMemory
 from .vector.add_memory import AddMemory
 from .vector.delete_memory import DeleteMemory
 from .vector.memory_handler import MemoryHandler
@@ -30,8 +33,10 @@ __all__ = [
     "ProfileHandler",
     "ReadAllProfiles",
     "UpdateProfile",
+    "UpdateProfileFilterOlder",
+    "DeleteProfile",
     # Vector
-    "AddDraftAndRetrieveSimilarMemory",
+    "AddAndRetrieveSimilarMemory",
     "AddMemory",
     "DeleteMemory",
     "MemoryHandler",

reme/tool/memory/profiles/profile_handler.py

@@ -12,7 +12,7 @@ from ....core.utils import CacheHandler, deduplicate_memories
 class ProfileHandler:
     """User profile CRUD handler"""
 
-    def __init__(self, profile_path: str | Path, memory_target: str, max_capacity: int = 100):
+    def __init__(self, profile_path: str | Path, memory_target: str, max_capacity: int = 50):
         """init"""
         self.memory_target: str = memory_target
         self.cache_key: str = self.memory_target.replace(" ", "_").lower()
@@ -96,6 +96,12 @@ class ProfileHandler:
             ref_memory_id=ref_memory_id,
         )
 
+        # Remove existing nodes with the same when_to_use (profile_key)
+        original_count = len(nodes)
+        nodes = [n for n in nodes if n.when_to_use != profile_key]
+        if len(nodes) < original_count:
+            logger.info(f"Removed {original_count - len(nodes)} duplicate profile(s) with key: {profile_key}")
+
         nodes.append(new_node)
         self._save_nodes(nodes)
         logger.info(f"Added profile: {profile_key}={profile_value}")
@@ -120,6 +126,13 @@ class ProfileHandler:
             for p in profiles
         ]
 
+        # Remove existing nodes with the same when_to_use (profile_key)
+        new_keys = {n.when_to_use for n in new_nodes}
+        original_count = len(nodes)
+        nodes = [n for n in nodes if n.when_to_use not in new_keys]
+        if len(nodes) < original_count:
+            logger.info(f"Removed {original_count - len(nodes)} duplicate profile(s) with matching keys")
+
         nodes.extend(new_nodes)
         self._save_nodes(nodes)
         logger.info(f"Batch added {len(new_nodes)} profiles")

reme/tool/memory/vector/add_draft_and_retrieve_similar_memory.py

@@ -8,7 +8,7 @@ from ....core.schema import ToolCall, MemoryNode
 from ....core.utils import deduplicate_memories
 
 
-class AddDraftAndRetrieveSimilarMemory(BaseMemoryTool):
+class AddAndRetrieveSimilarMemory(BaseMemoryTool):
     """Tool to add draft memory and retrieve similar memories"""
 
     def __init__(
@@ -60,7 +60,7 @@ class AddDraftAndRetrieveSimilarMemory(BaseMemoryTool):
     def _build_tool_call(self) -> ToolCall:
         return ToolCall(
             **{
-                "description": "Add draft memory and retrieve similar memories from the vector store.",
+                "description": "Add memory and retrieve similar memories from the vector store.",
                 "parameters": self._build_query_parameters(),
             },
         )
@@ -68,24 +68,24 @@ class AddDraftAndRetrieveSimilarMemory(BaseMemoryTool):
     def _build_multiple_tool_call(self) -> ToolCall:
         return ToolCall(
             **{
-                "description": "Add draft memory and retrieve similar memories from the vector store.",
+                "description": "Add memory and retrieve similar memories from the vector store.",
                 "parameters": {
                     "type": "object",
                     "properties": {
-                        "draft_items": {
+                        "items": {
                             "type": "array",
-                            "description": "draft_items",
+                            "description": "items",
                             "items": self._build_query_parameters(),
                         },
                     },
-                    "required": ["draft_items"],
+                    "required": ["items"],
                 },
             },
         )
 
     async def execute(self):
         if self.enable_multiple:
-            draft_items = self.context.get("draft_items", [])
+            draft_items = self.context.get("items", [])
         else:
             draft_items = [self.context]