diff --git a/benchmark/halumem/eval_reme.py b/benchmark/halumem/eval_reme.py index eb6b18ef..baebde2d 100644 --- a/benchmark/halumem/eval_reme.py +++ b/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)" ) diff --git a/reme/agent/memory/__init__.py b/reme/agent/memory/__init__.py index 024b68c4..f812a62f 100644 --- a/reme/agent/memory/__init__.py +++ b/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", diff --git a/reme/config/default.yaml b/reme/config/default.yaml index 836d866f..5908c50c 100644 --- a/reme/config/default.yaml +++ b/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() \ No newline at end of file diff --git a/reme/reme.py b/reme/reme.py index 5f587ea5..eabc4a63 100644 --- a/reme/reme.py +++ b/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 diff --git a/reme/tool/memory/__init__.py b/reme/tool/memory/__init__.py index ff58d44f..14a1c6f0 100644 --- a/reme/tool/memory/__init__.py +++ b/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", diff --git a/reme/tool/memory/profiles/profile_handler.py b/reme/tool/memory/profiles/profile_handler.py index 6f646ac3..3d0df400 100644 --- a/reme/tool/memory/profiles/profile_handler.py +++ b/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") diff --git a/reme/tool/memory/vector/add_draft_and_retrieve_similar_memory.py b/reme/tool/memory/vector/add_draft_and_retrieve_similar_memory.py index 040c475a..00aa0383 100644 --- a/reme/tool/memory/vector/add_draft_and_retrieve_similar_memory.py +++ b/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]