Feat: sync test to dev (#375)

## Description

<!--
Please include a summary of the changes below;
Fill in the issue number that this PR addresses (if applicable);
Fill in the related MemOS-Docs repository issue or PR link (if
applicable);
Mention the person who will review this PR (if you know who it is);
Replace (summary), (issue), (docs-issue-or-pr-link), and (reviewer) with
the appropriate information.

请在下方填写更改的摘要；
填写此 PR 解决的问题编号（如果适用）；
填写相关的 MemOS-Docs 仓库 issue 或 PR 链接（如果适用）；
提及将审查此 PR 的人（如果您知道是谁）；
替换 (summary)、(issue)、(docs-issue-or-pr-link) 和 (reviewer) 为适当的信息。
-->

Summary: (summary)

Fix: #(issue)

Docs Issue/PR: (docs-issue-or-pr-link)

Reviewer: @(reviewer)

## Checklist:

- [ ] I have performed a self-review of my own code | 我已自行检查了自己的代码
- [ ] I have commented my code in hard-to-understand areas |
我已在难以理解的地方对代码进行了注释
- [ ] I have added tests that prove my fix is effective or that my
feature works | 我已添加测试以证明我的修复有效或功能正常
- [ ] I have created related documentation issue/PR in
[MemOS-Docs](https://github.com/MemTensor/MemOS-Docs) (if applicable) |
我已在 [MemOS-Docs](https://github.com/MemTensor/MemOS-Docs) 中创建了相关的文档
issue/PR（如果适用）
- [ ] I have linked the issue to this PR (if applicable) | 我已将 issue
链接到此 PR（如果适用）
- [ ] I have mentioned the person who will review this PR | 我已提及将审查此 PR
的人

Author: fridayL

evaluation/.env-example

@@ -9,3 +9,24 @@ ZEP_API_KEY="z_***REDACTED***"
 CHAT_MODEL="gpt-4o-mini"
 CHAT_MODEL_BASE_URL="http://***.***.***.***:3000/v1"
 CHAT_MODEL_API_KEY="sk-***REDACTED***"
+
+# Configuration Only For Scheduler
+# RabbitMQ Configuration
+MEMSCHEDULER_RABBITMQ_HOST_NAME=rabbitmq-cn-***.cn-***.amqp-32.net.mq.amqp.aliyuncs.com
+MEMSCHEDULER_RABBITMQ_USER_NAME=***
+MEMSCHEDULER_RABBITMQ_PASSWORD=***
+MEMSCHEDULER_RABBITMQ_VIRTUAL_HOST=memos
+MEMSCHEDULER_RABBITMQ_ERASE_ON_CONNECT=true
+MEMSCHEDULER_RABBITMQ_PORT=5672
+
+# OpenAI Configuration
+MEMSCHEDULER_OPENAI_API_KEY=sk-***
+MEMSCHEDULER_OPENAI_BASE_URL=http://***.***.***.***:3000/v1
+MEMSCHEDULER_OPENAI_DEFAULT_MODEL=gpt-4o-mini
+
+# Graph DB Configuration
+MEMSCHEDULER_GRAPHDBAUTH_URI=bolt://localhost:7687
+MEMSCHEDULER_GRAPHDBAUTH_USER=neo4j
+MEMSCHEDULER_GRAPHDBAUTH_PASSWORD=***
+MEMSCHEDULER_GRAPHDBAUTH_DB_NAME=neo4j
+MEMSCHEDULER_GRAPHDBAUTH_AUTO_CREATE=true

evaluation/__init__.py

@@ -9,7 +9,6 @@
 
 from bert_score import score as bert_score
 from dotenv import load_dotenv
-from modules.locomo_eval_module import LocomoEvalModelModules
 from nltk.translate.bleu_score import SmoothingFunction, sentence_bleu
 from nltk.translate.meteor_score import meteor_score
 from openai import AsyncOpenAI
@@ -19,6 +18,7 @@
 from sentence_transformers import SentenceTransformer
 from tqdm import tqdm
 
+from evaluation.scripts.temporal_locomo.modules.locomo_eval_module import LocomoEvalModelModules
 from memos.log import get_logger
 
 
@@ -281,33 +281,64 @@ def __init__(self, args):
             api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL")
         )
 
-    async def run(self):
-        print(
-            f"\n=== Starting LoCoMo evaluation for {self.frame} (version: {self.version}) with {self.num_runs} run(s) per question ==="
-        )
-        print(f"Using {self.max_workers} concurrent workers for processing groups")
+    def _load_response_data(self):
+        """
+        Load response data from the response path file.
 
+        Returns:
+            dict: The loaded response data
+        """
         with open(self.response_path) as file:
-            locomo_responses = json.load(file)
+            return json.load(file)
 
-        num_users = 10
+    def _load_existing_evaluation_results(self):
+        """
+        Attempt to load existing evaluation results from the judged path.
+        If the file doesn't exist or there's an error loading it, return an empty dict.
+
+        Returns:
+            dict: Existing evaluation results or empty dict if none available
+        """
         all_grades = {}
+        try:
+            if os.path.exists(self.judged_path):
+                with open(self.judged_path) as f:
+                    all_grades = json.load(f)
+                print(f"Loaded existing evaluation results from {self.judged_path}")
+        except Exception as e:
+            print(f"Error loading existing evaluation results: {e}")
 
-        total_responses_count = sum(
-            len(locomo_responses.get(f"locomo_exp_user_{i}", [])) for i in range(num_users)
-        )
-        print(f"Found {total_responses_count} total responses across {num_users} users to evaluate")
+        return all_grades
+
+    def _create_evaluation_tasks(self, locomo_responses, all_grades, num_users):
+        """
+        Create evaluation tasks for groups that haven't been evaluated yet.
+
+        Args:
+            locomo_responses (dict): The loaded response data
+            all_grades (dict): Existing evaluation results
+            num_users (int): Number of user groups to process
 
-        # Create tasks for processing each group
+        Returns:
+            tuple: (tasks list, active users count)
+        """
         tasks = []
         active_users = 0
+
         for group_idx in range(num_users):
             group_id = f"locomo_exp_user_{group_idx}"
             group_responses = locomo_responses.get(group_id, [])
+
             if not group_responses:
                 print(f"No responses found for group {group_id}")
                 continue
 
+            # Skip groups that already have evaluation results
+            if all_grades.get(group_id):
+                print(f"Skipping group {group_id} as it already has evaluation results")
+                active_users += 1
+                continue
+
             active_users += 1
             tasks.append(
                 process_single_group(
@@ -319,29 +350,50 @@ async def run(self):
                 )
             )
 
-        print(f"Starting evaluation of {active_users} user groups with responses")
+        return tasks, active_users
+
+    async def _process_tasks(self, tasks):
+        """
+        Process evaluation tasks with concurrency control.
+
+        Args:
+            tasks (list): List of tasks to process
+
+        Returns:
+            list: Results from processing all tasks
+        """
+        if not tasks:
+            return []
 
         semaphore = asyncio.Semaphore(self.max_workers)
 
         async def limited_task(task):
+            """Helper function to limit concurrent task execution"""
             async with semaphore:
                 return await task
 
         limited_tasks = [limited_task(task) for task in tasks]
-        group_results = await asyncio.gather(*limited_tasks)
+        return await asyncio.gather(*limited_tasks)
 
-        for group_id, graded_responses in group_results:
-            all_grades[group_id] = graded_responses
+    def _calculate_scores(self, all_grades):
+        """
+        Calculate evaluation scores based on all grades.
 
-        print("\n=== Evaluation Complete: Calculating final scores ===")
+        Args:
+            all_grades (dict): The complete evaluation results
 
+        Returns:
+            tuple: (run_scores, evaluated_count)
+        """
         run_scores = []
         evaluated_count = 0
+
         if self.num_runs > 0:
             for i in range(1, self.num_runs + 1):
                 judgment_key = f"judgment_{i}"
                 current_run_correct_count = 0
                 current_run_total_count = 0
+
                 for group in all_grades.values():
                     for response in group:
                         if judgment_key in response["llm_judgments"]:
@@ -355,6 +407,16 @@ async def limited_task(task):
 
             evaluated_count = current_run_total_count
 
+        return run_scores, evaluated_count
+
+    def _report_scores(self, run_scores, evaluated_count):
+        """
+        Report evaluation scores to the console.
+
+        Args:
+            run_scores (list): List of accuracy scores for each run
+            evaluated_count (int): Number of evaluated responses
+        """
         if evaluated_count > 0:
             mean_of_scores = np.mean(run_scores)
             std_of_scores = np.std(run_scores)
@@ -368,11 +430,63 @@ async def limited_task(task):
             print("No responses were evaluated")
             print("LLM-as-a-Judge score: N/A (0/0)")
 
+    def _save_results(self, all_grades):
+        """
+        Save evaluation results to the judged path file.
+
+        Args:
+            all_grades (dict): The complete evaluation results to save
+        """
         all_grades = convert_numpy_types(all_grades)
         with open(self.judged_path, "w") as f:
             json.dump(all_grades, f, indent=2)
             print(f"Saved detailed evaluation results to {self.judged_path}")
 
+    async def run(self):
+        """
+        Main execution method for the LoCoMo evaluation process.
+        This method orchestrates the entire evaluation workflow:
+        1. Loads existing evaluation results if available
+        2. Processes only groups that haven't been evaluated yet
+        3. Calculates and reports final evaluation scores
+        """
+        print(
+            f"\n=== Starting LoCoMo evaluation for {self.frame} (version: {self.version}) with {self.num_runs} run(s) per question ==="
+        )
+        print(f"Using {self.max_workers} concurrent workers for processing groups")
+
+        # Load response data and existing evaluation results
+        locomo_responses = self._load_response_data()
+        all_grades = self._load_existing_evaluation_results()
+
+        # Count total responses for reporting
+        num_users = 10
+        total_responses_count = sum(
+            len(locomo_responses.get(f"locomo_exp_user_{i}", [])) for i in range(num_users)
+        )
+        print(f"Found {total_responses_count} total responses across {num_users} users to evaluate")
+
+        # Create tasks only for groups that haven't been evaluated yet
+        tasks, active_users = self._create_evaluation_tasks(locomo_responses, all_grades, num_users)
+        print(
+            f"Starting evaluation of {len(tasks)} user groups with responses (out of {active_users} active users)"
+        )
+
+        # Process tasks and update all_grades with results
+        if tasks:
+            group_results = await self._process_tasks(tasks)
+            for group_id, graded_responses in group_results:
+                all_grades[group_id] = graded_responses
+
+        print("\n=== Evaluation Complete: Calculating final scores ===")
+
+        # Calculate and report scores
+        run_scores, evaluated_count = self._calculate_scores(all_grades)
+        self._report_scores(run_scores, evaluated_count)
+
+        # Save results
+        self._save_results(all_grades)
+
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser()

evaluation/scripts/__init__.py

@@ -6,16 +6,16 @@
 from datetime import datetime, timezone
 from pathlib import Path
 
-from modules.constants import (
+from tqdm import tqdm
+
+from evaluation.scripts.temporal_locomo.modules.constants import (
     MEM0_GRAPH_MODEL,
     MEM0_MODEL,
     MEMOS_MODEL,
     MEMOS_SCHEDULER_MODEL,
     ZEP_MODEL,
 )
-from modules.locomo_eval_module import LocomoEvalModelModules
-from tqdm import tqdm
-
+from evaluation.scripts.temporal_locomo.modules.locomo_eval_module import LocomoEvalModelModules
 from memos.log import get_logger
 
 

evaluation/scripts/temporal_locomo/models/__init__.py

@@ -4,7 +4,7 @@
 import numpy as np
 import pandas as pd
 
-from modules.locomo_eval_module import LocomoEvalModelModules
+from evaluation.scripts.temporal_locomo.modules.locomo_eval_module import LocomoEvalModelModules
 
 
 # Category mapping as per your request