From 52a4b66d598059f74ef96a479bdcc3272299143d Mon Sep 17 00:00:00 2001
From: "jinli.yl" <jinli.yl@alibaba-inc.com>
Date: Wed, 14 Jan 2026 00:13:55 +0800
Subject: [PATCH] refactor(bench): update halumem evaluation with concurrent
 session processing

- Increased max_rps from 9 to 20 in default config
- Removed tqdm progress bar for session processing
- Implemented asyncio semaphore for session concurrency control
- Added parallel processing of sessions within each user
- Updated logging to show session count per user
- Changed user processing from concurrent to sequential
- Added completion tracking with progress indicators
- Modified output formatting for better readability
---
 bench/halumem/eval_baseline_simple.py | 66 +++++++++++++--------------
 reme_ai/core/config/default.yaml      |  3 +-
 2 files changed, 34 insertions(+), 35 deletions(-)

diff --git a/bench/halumem/eval_baseline_simple.py b/bench/halumem/eval_baseline_simple.py
index f06d69f5..7e0e4aef 100644
--- a/bench/halumem/eval_baseline_simple.py
+++ b/bench/halumem/eval_baseline_simple.py
@@ -24,7 +24,6 @@ from pathlib import Path
 from typing import Any
 
 from loguru import logger
-from tqdm import tqdm
 
 from eval_tools import evaluation_for_question2
 from llms import llm_request_for_json
@@ -424,16 +423,15 @@ class HaluMemBaselineEvaluator:
         user_name = self.data_loader.extract_user_name(user_data["persona_info"])
         uuid = user_data["uuid"]
         
-        logger.info(f"Processing user: {user_name}")
+        total_sessions = len(user_data["sessions"])
+        logger.info(f"Processing user: {user_name} ({total_sessions} sessions)")
         
-        # Create progress bar for sessions
-        with tqdm(
-            total=len(user_data["sessions"]),
-            desc=f"Sessions [{user_name}]",
-            leave=False,
-            ncols=100
-        ) as pbar:
-            for idx, session in enumerate(user_data["sessions"]):
+        # Semaphore for concurrency control within user sessions
+        semaphore = asyncio.Semaphore(self.config.max_concurrency)
+        completed_count = [0]  # Use list to allow modification in nested async function
+        
+        async def process_session_with_log(idx: int, session: dict):
+            async with semaphore:
                 session_data = await self.process_session(
                     session=session,
                     session_id=idx,
@@ -442,7 +440,17 @@ class HaluMemBaselineEvaluator:
                 )
                 
                 self.file_manager.save_session(user_name, idx, session_data)
-                pbar.update(1)
+                
+                # Update and log completion
+                completed_count[0] += 1
+                print(f"✅ {user_name} complete {completed_count[0]}/{total_sessions}")
+        
+        # Process all sessions in parallel
+        tasks = [
+            process_session_with_log(idx, session)
+            for idx, session in enumerate(user_data["sessions"])
+        ]
+        await asyncio.gather(*tasks)
         
         return {"uuid": uuid, "user_name": user_name, "status": "ok"}
     
@@ -456,31 +464,21 @@ class HaluMemBaselineEvaluator:
         
         print("\n" + "=" * 80)
         print("HALUMEM BASELINE EVALUATION - DIRECT QA WITHOUT MEMORY SYSTEM")
-        print(f"Users: {len(users_to_process)} | Concurrency: {self.config.max_concurrency}")
+        print(f"Users: {len(users_to_process)} | Session Concurrency: {self.config.max_concurrency}")
         print("=" * 80 + "\n")
         
-        # Process users with concurrency control
-        semaphore = asyncio.Semaphore(self.config.max_concurrency)
-        
-        async def process_with_cache_check(idx: int, user_data: dict):
-            async with semaphore:
-                user_name = self.data_loader.extract_user_name(user_data["persona_info"])
-                
-                # 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"}
-                
-                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 = [
-            process_with_cache_check(idx, user) 
-            for idx, user in enumerate(users_to_process, 1)
-        ]
-        await asyncio.gather(*tasks)
+        # Process users sequentially (for loop)
+        for idx, user_data in enumerate(users_to_process, 1):
+            user_name = self.data_loader.extract_user_name(user_data["persona_info"])
+            
+            # Check cache
+            if self.file_manager.user_has_cache(user_name):
+                print(f"⚡ [{idx}/{len(users_to_process)}] Skipping {user_name} (cached)")
+                continue
+            
+            print(f"🔄 [{idx}/{len(users_to_process)}] Processing {user_name}...")
+            await self.process_user(user_data)
+            print(f"✅ [{idx}/{len(users_to_process)}] User {user_name} completed\n")
         
         # Combine results
         output_file = os.path.join(self.config.output_dir, "eval_results.jsonl")
diff --git a/reme_ai/core/config/default.yaml b/reme_ai/core/config/default.yaml
index 16b03f60..f8280b17 100644
--- a/reme_ai/core/config/default.yaml
+++ b/reme_ai/core/config/default.yaml
@@ -23,7 +23,8 @@ llm:
     backend: openai
     model_name: qwen3-max
 #    temperature: 0.6
-    max_rps: 9
+    max_rps: 20
+#    max_rps: 9
     rps_window: 10
 
 embedding_model: