chore(ltm): optimize memory structure

Author: yaozheng-fang

docs/content/6.memory/3.long-term-memory.md

@@ -7,13 +7,13 @@ navigation:
 
 ## 使用方法
 
-VeADK 的长期记忆通常存储在数据库中，通过如下方式定义一个长期记忆：
+VeADK 的长期记忆通常存储在数据库中，你需要在初始化长期记忆时定义 `index` 来指定后端索引名称。通过如下方式定义一个长期记忆：
 
 ```python
 from veadk.memory.long_term_memory import LongTermMemory
 
-# 由于长期记忆需要构建索引，因此你必须在初始化长期记忆时定义 `app_name` 以及 `user_id`
-long_term_memory = LongTermMemory(app_name="my_app_name", user_id="user_id")
+# 
+long_term_memory = LongTermMemory(index="my_index")
 ```
 
 通过如下例子说明长期记忆：
@@ -32,7 +32,7 @@ user_id = "temp_user"
 teaching_session_id = "teaching_session"
 student_session_id = "student_session"
 
-long_term_memory = LongTermMemory(backend="local", app_name=app_name, user_id=user_id)
+long_term_memory = LongTermMemory(backend="local", index=app_name)
 
 agent = Agent(long_term_memory=long_term_memory)
 
@@ -90,8 +90,4 @@ print(response)
   ::field{name="app_name" type="string"}
   Agent 应用名称，用于多应用区分。默认空字符串。
   ::
-
-  ::field{name="user_id" type="string"}
-  Agent 用户 ID，用于区分不同用户的长期记忆。默认空字符串。
-  ::
 ::

docs/content/7.knowledgebase/1.knowledgebase.md

@@ -9,7 +9,7 @@ navigation:
 
 VeADK 基于 Llama-index 作为知识库的主要处理入口。开发者可上传文本、文件、目录，我们会为您进行自动切片。
 
-创建知识库时，您必须要提供您的 `app_name`（将会用来自动构建索引名称），或指定一个知识库的索引：
+创建知识库时，您必须要提供您的知识库后端索引名称 `index`，或指定 `app_name` 来作为索引名称：
 
 ```python
 from veadk.knowledgebase import KnowledgeBase

veadk/knowledgebase/knowledgebase.py

@@ -73,9 +73,7 @@ class KnowledgeBase(BaseModel):
     """Configuration for the backend"""
 
     top_k: int = 10
-    """Number of top similar documents to retrieve during search.
-
-    Default is 10."""
+    """Number of top similar documents to retrieve during search"""
 
     app_name: str = ""
 
@@ -90,33 +88,18 @@ def model_post_init(self, __context: Any) -> None:
             )
             return
 
-        # must provide at least one of them
-        if not self.app_name and not self.index:
-            raise ValueError(
-                "Either `app_name` or `index` must be provided one of them."
-            )
-
-        # priority use index
-        if self.app_name and self.index:
-            logger.warning(
-                "`app_name` and `index` are both provided, using `index` as the knowledgebase index name."
-            )
-
-        # generate index name if `index` not provided but `app_name` is provided
-        if self.app_name and not self.index:
-            self.index = build_knowledgebase_index(self.app_name)
-            logger.info(
-                f"Knowledgebase index is set to {self.index} (generated by the app_name: {self.app_name})."
-            )
+        self.index = self.index or self.app_name
+        if not self.index:
+            raise ValueError("Either `index` or `app_name` must be provided.")
 
         logger.info(
-            f"Initializing knowledgebase: backend={self.backend} top_k={self.top_k}"
+            f"Initializing knowledgebase: backend={self.backend} index={self.index} top_k={self.top_k}"
         )
         self._backend = _get_backend_cls(self.backend)(
             index=self.index, **self.backend_config if self.backend_config else {}
         )
         logger.info(
-            f"Initialized knowledgebase with backend {self._backend.__class__.__name__}"
+            f"Initialized knowledgebase with backend {self.backend.__class__.__name__}"
         )
 
     def add_from_directory(self, directory: str, **kwargs) -> bool:
@@ -133,8 +116,7 @@ def add_from_text(self, text: str | list[str], **kwargs) -> bool:
 
     def search(self, query: str, top_k: int = 0, **kwargs) -> list[KnowledgebaseEntry]:
         """Search knowledge from knowledgebase"""
-        if top_k == 0:
-            top_k = self.top_k
+        top_k = top_k if top_k != 0 else self.top_k
 
         _entries = self._backend.search(query=query, top_k=top_k, **kwargs)
 

veadk/memory/long_term_memory.py

@@ -72,10 +72,6 @@ def _get_backend_cls(backend: str) -> type[BaseLongTermMemoryBackend]:
     raise ValueError(f"Unsupported long term memory backend: {backend}")
 
 
-def build_long_term_memory_index(app_name: str, user_id: str):
-    return f"{app_name}_{user_id}"
-
-
 class LongTermMemory(BaseMemoryService, BaseModel):
     backend: Union[
         Literal["local", "opensearch", "redis", "viking", "viking_mem", "mem0"],
@@ -89,19 +85,14 @@ class LongTermMemory(BaseMemoryService, BaseModel):
     top_k: int = 5
     """Number of top similar documents to retrieve during search."""
 
+    index: str = ""
+
     app_name: str = ""
 
     user_id: str = ""
+    """Deprecated attribute"""
 
     def model_post_init(self, __context: Any) -> None:
-        if self.backend == "viking_mem":
-            logger.warning(
-                "The `viking_mem` backend is deprecated, please use `viking` instead."
-            )
-            self.backend = "viking"
-
-        self._backend = None
-
         # Once user define a backend instance, use it directly
         if isinstance(self.backend, BaseLongTermMemoryBackend):
             self._backend = self.backend
@@ -110,33 +101,23 @@ def model_post_init(self, __context: Any) -> None:
             )
             return
 
-        if self.backend_config:
-            logger.warning(
-                f"Initialized long term memory backend {self.backend} with config. We will ignore `app_name` and `user_id` if provided."
+        # Check index
+        self.index = self.index or self.app_name
+        if not self.index:
+            raise ValueError(
+                "Attribute `index` or `app_name` must be provided one of both."
             )
-            self._backend = _get_backend_cls(self.backend)(**self.backend_config)
-            _index = self.backend_config.get("index", None)
-            if _index:
-                self._index = _index
-                logger.info(f"Long term memory index set to {self._index}.")
-            else:
-                logger.warning(
-                    "Cannot find index via backend_config, please set `index` parameter."
-                )
-            return
 
-        if self.app_name and self.user_id:
-            self._index = build_long_term_memory_index(
-                app_name=self.app_name, user_id=self.user_id
-            )
-            logger.info(f"Long term memory index set to {self._index}.")
-            self._backend = _get_backend_cls(self.backend)(
-                index=self._index, **self.backend_config if self.backend_config else {}
-            )
-        else:
+        # Forward compliance
+        if self.backend == "viking_mem":
             logger.warning(
-                "Neither `backend_instance`, `backend_config`, nor (`app_name`/`user_id`) is provided, the long term memory storage will initialize when adding a session."
+                "The `viking_mem` backend is deprecated, change to `viking` instead."
             )
+            self.backend = "viking"
+
+        self._backend = _get_backend_cls(self.backend)(
+            index=self.index, **self.backend_config if self.backend_config else {}
+        )
 
     def _filter_and_convert_events(self, events: list[Event]) -> list[str]:
         final_events = []
@@ -164,75 +145,32 @@ async def add_session_to_memory(
         self,
         session: Session,
     ):
-        app_name = session.app_name
         user_id = session.user_id
-
-        if not self._backend and isinstance(self.backend, str):
-            self._index = build_long_term_memory_index(app_name, user_id)
-            self._backend = _get_backend_cls(self.backend)(
-                index=self._index, **self.backend_config if self.backend_config else {}
-            )
-            logger.info(
-                f"Initialize long term memory backend now, index is {self._index}"
-            )
-
-        if not self._index and self._index != build_long_term_memory_index(
-            app_name, user_id
-        ):
-            logger.warning(
-                f"The `app_name` or `user_id` is different from the initialized one, skip add session to memory. Initialized index: {self._index}, current built index: {build_long_term_memory_index(app_name, user_id)}"
-            )
-            return
         event_strings = self._filter_and_convert_events(session.events)
 
         logger.info(
-            f"Adding {len(event_strings)} events to long term memory: index={self._index}"
+            f"Adding {len(event_strings)} events to long term memory: index={self.index}"
+        )
+        self._backend.save_memory(user_id=user_id, event_strings=event_strings)
+        logger.info(
+            f"Added {len(event_strings)} events to long term memory: index={self.index}, user_id={user_id}"
         )
-
-        if self._backend:
-            self._backend.save_memory(event_strings=event_strings, user_id=user_id)
-
-            logger.info(
-                f"Added {len(event_strings)} events to long term memory: index={self._index}"
-            )
-        else:
-            logger.error(
-                "Long term memory backend initialize failed, cannot add session to memory."
-            )
 
     @override
-    async def search_memory(self, *, app_name: str, user_id: str, query: str):
-        # prevent model invoke `load_memory` before add session to this memory
-        if not self._backend and isinstance(self.backend, str):
-            self._index = build_long_term_memory_index(app_name, user_id)
-            self._backend = _get_backend_cls(self.backend)(
-                index=self._index, **self.backend_config if self.backend_config else {}
-            )
-            logger.info(
-                f"Initialize long term memory backend now, index is {self._index}"
-            )
+    async def search_memory(
+        self, *, app_name: str, user_id: str, query: str
+    ) -> SearchMemoryResponse:
+        logger.info(f"Search memory with query={query}")
 
-        if not self._index and self._index != build_long_term_memory_index(
-            app_name, user_id
-        ):
-            logger.warning(
-                f"The `app_name` or `user_id` is different from the initialized one. Initialized index: {self._index}, current built index: {build_long_term_memory_index(app_name, user_id)}. Search memory return empty list."
+        memory_chunks = []
+        try:
+            memory_chunks = self._backend.search_memory(
+                query=query, top_k=self.top_k, user_id=user_id
             )
-            return SearchMemoryResponse(memories=[])
-
-        if not self._backend:
+        except Exception as e:
             logger.error(
-                "Long term memory backend is not initialized, cannot search memory."
+                f"Exception orrcus during memory search: {e}. Return empty memory chunks"
             )
-            return SearchMemoryResponse(memories=[])
-
-        logger.info(
-            f"Searching long term memory: query={query} index={self._index} top_k={self.top_k}"
-        )
-
-        memory_chunks = self._backend.search_memory(
-            query=query, top_k=self.top_k, user_id=user_id
-        )
 
         memory_events = []
         for memory in memory_chunks:
@@ -260,6 +198,6 @@ async def search_memory(self, *, app_name: str, user_id: str, query: str):
             )
 
         logger.info(
-            f"Return {len(memory_events)} memory events for query: {query} index={self._index}"
+            f"Return {len(memory_events)} memory events for query: {query} index={self.index} user_id={user_id}"
         )
         return SearchMemoryResponse(memories=memory_events)

veadk/memory/long_term_memory_backends/base_backend.py

@@ -25,9 +25,11 @@ def precheck_index_naming(self):
         """Check the index name is valid or not"""
 
     @abstractmethod
-    def save_memory(self, event_strings: list[str], **kwargs) -> bool:
+    def save_memory(self, user_id: str, event_strings: list[str], **kwargs) -> bool:
         """Save memory to long term memory backend"""
 
     @abstractmethod
-    def search_memory(self, query: str, top_k: int, **kwargs) -> list[str]:
+    def search_memory(
+        self, user_id: str, query: str, top_k: int, **kwargs
+    ) -> list[str]:
         """Retrieve memory from long term memory backend"""

veadk/memory/long_term_memory_backends/in_memory_backend.py

@@ -29,10 +29,6 @@ class InMemoryLTMBackend(BaseLongTermMemoryBackend):
     embedding_config: EmbeddingModelConfig = Field(default_factory=EmbeddingModelConfig)
     """Embedding model configs"""
 
-    def precheck_index_naming(self):
-        # no checking
-        pass
-
     def model_post_init(self, __context: Any) -> None:
         self._embed_model = OpenAILikeEmbedding(
             model_name=self.embedding_config.name,
@@ -41,16 +37,22 @@ def model_post_init(self, __context: Any) -> None:
         )
         self._vector_index = VectorStoreIndex([], embed_model=self._embed_model)
 
+    def precheck_index_naming(self):
+        # no checking
+        pass
+
     @override
-    def save_memory(self, event_strings: list[str], **kwargs) -> bool:
+    def save_memory(self, user_id: str, event_strings: list[str], **kwargs) -> bool:
         for event_string in event_strings:
             document = Document(text=event_string)
             nodes = self._split_documents([document])
             self._vector_index.insert_nodes(nodes)
         return True
 
     @override
-    def search_memory(self, query: str, top_k: int, **kwargs) -> list[str]:
+    def search_memory(
+        self, user_id: str, query: str, top_k: int, **kwargs
+    ) -> list[str]:
         _retriever = self._vector_index.as_retriever(similarity_top_k=top_k)
         retrieved_nodes = _retriever.retrieve(query)
         return [node.text for node in retrieved_nodes]

veadk/memory/long_term_memory_backends/mem0_backend.py

@@ -13,12 +13,11 @@
 # limitations under the License.
 
 from typing import Any
-from typing_extensions import override
+
 from pydantic import Field
+from typing_extensions import override
 
 from veadk.configs.database_configs import Mem0Config
-
-
 from veadk.memory.long_term_memory_backends.base_backend import (
     BaseLongTermMemoryBackend,
 )
@@ -66,7 +65,9 @@ def precheck_index_naming(self):
         pass
 
     @override
-    def save_memory(self, event_strings: list[str], **kwargs) -> bool:
+    def save_memory(
+        self, event_strings: list[str], user_id: str = "default_user", **kwargs
+    ) -> bool:
         """Save memory to Mem0
 
         Args:
@@ -76,8 +77,6 @@ def save_memory(self, event_strings: list[str], **kwargs) -> bool:
         Returns:
             bool: True if saved successfully, False otherwise
         """
-        user_id = kwargs.get("user_id", "default_user")
-
         try:
             logger.info(
                 f"Saving {len(event_strings)} events to Mem0 for user: {user_id}"
@@ -100,7 +99,9 @@ def save_memory(self, event_strings: list[str], **kwargs) -> bool:
             return False
 
     @override
-    def search_memory(self, query: str, top_k: int, **kwargs) -> list[str]:
+    def search_memory(
+        self, query: str, top_k: int, user_id: str = "default_user", **kwargs
+    ) -> list[str]:
         """Search memory from Mem0
 
         Args:
@@ -111,7 +112,6 @@ def search_memory(self, query: str, top_k: int, **kwargs) -> list[str]:
         Returns:
             list[str]: List of memory strings
         """
-        user_id = kwargs.get("user_id", "default_user")
 
         try:
             logger.info(