feat: Add conversation_turn tracking for session-based memory search

    - Add conversation_turn field to APIMemoryHistoryEntryItem schema with default value 0
    - Implement session counter in OptimizedScheduler to track turn count per session_id
    - Update sync_search_data method to accept and store conversation_turn parameter
    - Maintain session history with LRU eviction (max 5 sessions)
    - Rename conversation_id to session_id for consistency with request object
    - Enable direct access to session_id from search requests

    This feature allows tracking conversation turns within the same session,
    providing better context for memory retrieval and search history management.

Author: tangg555

src/memos/mem_scheduler/general_modules/api_misc.py

@@ -8,7 +8,6 @@
     APISearchHistoryManager,
     TaskRunningStatus,
 )
-from memos.mem_scheduler.utils.db_utils import get_utc_now
 from memos.memories.textual.item import TextualMemoryItem
 
 
@@ -45,7 +44,8 @@ def sync_search_data(
         query: str,
         memories: list[TextualMemoryItem],
         formatted_memories: Any,
-        conversation_id: str | None = None,
+        session_id: str | None = None,
+        conversation_turn: int = 0,
     ) -> Any:
         logger.info(
             f"Syncing search data for item_id: {item_id}, user_id: {user_id}, mem_cube_id: {mem_cube_id}"
@@ -66,7 +66,7 @@ def sync_search_data(
                 query=query,
                 formatted_memories=formatted_memories,
                 task_status=TaskRunningStatus.COMPLETED,  # Use the provided running_status
-                conversation_id=conversation_id,
+                session_id=session_id,
                 memories=memories,
             )
 
@@ -76,18 +76,18 @@ def sync_search_data(
                 logger.warning(f"Failed to update entry with item_id: {item_id}")
         else:
             # Add new entry based on running_status
-            search_entry = APIMemoryHistoryEntryItem(
+            entry_item = APIMemoryHistoryEntryItem(
                 item_id=item_id,
                 query=query,
                 formatted_memories=formatted_memories,
                 memories=memories,
                 task_status=TaskRunningStatus.COMPLETED,
-                conversation_id=conversation_id,
-                created_time=get_utc_now(),
+                session_id=session_id,
+                conversation_turn=conversation_turn,
             )
 
             # Add directly to completed list as APIMemoryHistoryEntryItem instance
-            search_history.completed_entries.append(search_entry)
+            search_history.completed_entries.append(entry_item)
 
             # Maintain window size
             if len(search_history.completed_entries) > search_history.window_size:

src/memos/mem_scheduler/optimized_scheduler.py

@@ -1,6 +1,7 @@
 import json
 import os
 
+from collections import OrderedDict
 from typing import TYPE_CHECKING
 
 from memos.api.product_models import APISearchRequest
@@ -39,6 +40,8 @@ def __init__(self, config: GeneralSchedulerConfig):
         super().__init__(config)
         self.window_size = int(os.getenv("API_SEARCH_WINDOW_SIZE", 5))
         self.history_memory_turns = int(os.getenv("API_SEARCH_HISTORY_TURNS", 5))
+        self.session_counter = OrderedDict()
+        self.max_session_history = 5
 
         self.api_module = SchedulerAPIModule(
             window_size=self.window_size,
@@ -54,13 +57,14 @@ def submit_memory_history_async_task(
         self,
         search_req: APISearchRequest,
         user_context: UserContext,
+        session_id: str | None = None,
     ):
         # Create message for async fine search
         message_content = {
             "search_req": {
                 "query": search_req.query,
                 "user_id": search_req.user_id,
-                "session_id": search_req.session_id,
+                "session_id": session_id,
                 "top_k": search_req.top_k,
                 "internet_search": search_req.internet_search,
                 "moscube": search_req.moscube,
@@ -163,6 +167,7 @@ def mix_search_memories(
         self.submit_memory_history_async_task(
             search_req=search_req,
             user_context=user_context,
+            session_id=search_req.session_id,
         )
 
         # Try to get pre-computed fine memories if available
@@ -171,6 +176,7 @@ def mix_search_memories(
             mem_cube_id=user_context.mem_cube_id,
             turns=self.history_memory_turns,
         )
+
         if not history_memories:
             fast_memories = searcher.post_retrieve(
                 retrieved_results=fast_retrieved_memories,
@@ -214,6 +220,23 @@ def update_search_memories_to_redis(
             search_req = content_dict["search_req"]
             user_context = content_dict["user_context"]
 
+            session_id = search_req.get("session_id")
+            if session_id:
+                if session_id not in self.session_counter:
+                    self.session_counter[session_id] = 0
+                else:
+                    self.session_counter[session_id] += 1
+                session_turn = self.session_counter[session_id]
+
+                # Move the current session to the end to mark it as recently used
+                self.session_counter.move_to_end(session_id)
+
+                # If the counter exceeds the max size, remove the oldest item
+                if len(self.session_counter) > self.max_session_history:
+                    self.session_counter.popitem(last=False)
+            else:
+                session_turn = 0
+
             memories: list[TextualMemoryItem] = self.search_memories(
                 search_req=APISearchRequest(**content_dict["search_req"]),
                 user_context=UserContext(**content_dict["user_context"]),
@@ -230,6 +253,8 @@ def update_search_memories_to_redis(
                 query=search_req["query"],
                 memories=memories,
                 formatted_memories=formatted_memories,
+                session_id=session_id,
+                conversation_turn=session_turn,
             )
 
     def _api_mix_search_message_consumer(self, messages: list[ScheduleMessageItem]) -> None:

src/memos/mem_scheduler/schemas/api_schemas.py

@@ -35,11 +35,10 @@ class APIMemoryHistoryEntryItem(BaseModel, DictConversionMixin):
     task_status: str = Field(
         default="running", description="Task status: running, completed, failed"
     )
-    conversation_id: str | None = Field(
-        default=None, description="Optional conversation identifier"
-    )
+    session_id: str | None = Field(default=None, description="Optional conversation identifier")
     created_time: datetime = Field(description="Entry creation time", default_factory=get_utc_now)
     timestamp: datetime | None = Field(default=None, description="Timestamp for the entry")
+    conversation_turn: int = Field(default=0, description="Turn count for the same session_id")
 
     model_config = ConfigDict(
         arbitrary_types_allowed=True,
@@ -107,11 +106,13 @@ def get_running_item_ids(self) -> list[str]:
         """Get all running task IDs"""
         return self.running_item_ids.copy()
 
-    def get_completed_entries(self) -> list[dict[str, Any]]:
+    def get_completed_entries(self) -> list[APIMemoryHistoryEntryItem]:
         """Get all completed entries"""
         return self.completed_entries.copy()
 
-    def get_history_memory_entries(self, turns: int | None = None) -> list[dict[str, Any]]:
+    def get_history_memory_entries(
+        self, turns: int | None = None
+    ) -> list[APIMemoryHistoryEntryItem]:
         """
         Get the most recent n completed search entries, sorted by created_time.
 
@@ -179,7 +180,7 @@ def update_entry_by_item_id(
         query: str,
         formatted_memories: Any,
         task_status: TaskRunningStatus,
-        conversation_id: str | None = None,
+        session_id: str | None = None,
         memories: list[TextualMemoryItem] | None = None,
     ) -> bool:
         """
@@ -191,7 +192,7 @@ def update_entry_by_item_id(
             query: New query string
             formatted_memories: New formatted memories
             task_status: New task status
-            conversation_id: New conversation ID
+            session_id: New conversation ID
             memories: List of TextualMemoryItem objects
 
         Returns:
@@ -204,8 +205,8 @@ def update_entry_by_item_id(
                 entry.query = query
                 entry.formatted_memories = formatted_memories
                 entry.task_status = task_status
-                if conversation_id is not None:
-                    entry.conversation_id = conversation_id
+                if session_id is not None:
+                    entry.session_id = session_id
                 if memories is not None:
                     entry.memories = memories