feat: Multi-Model Memory Reader with Modular Parser Architecture

src/memos/mem_reader/multi_model_struct.py

@@ -39,8 +39,16 @@ def __init__(self, config: MultiModelStructMemReaderConfig):
             parser=None,
         )
 
+    def _concat_multi_model_memories(
+        self, all_memory_items: list[TextualMemoryItem]
+    ) -> list[TextualMemoryItem]:
+        # TODO: concat multi_model_memories
+        return all_memory_items
+
     @timed
-    def _process_multi_model_data(self, scene_data_info: MessagesType, info, **kwargs):
+    def _process_multi_model_data(
+        self, scene_data_info: MessagesType, info, **kwargs
+    ) -> list[TextualMemoryItem]:
         """
         Process multi-model data using MultiModelParser.
 
@@ -50,23 +58,81 @@ def _process_multi_model_data(self, scene_data_info: MessagesType, info, **kwarg
             **kwargs: Additional parameters (mode, etc.)
         """
         mode = kwargs.get("mode", "fine")
+        # Pop custom_tags from info (same as simple_struct.py)
+        # must pop here, avoid add to info, only used in sync fine mode
+        custom_tags = info.pop("custom_tags", None) if isinstance(info, dict) else None
 
         # Use MultiModelParser to parse the scene data
         # If it's a list, parse each item; otherwise parse as single message
         if isinstance(scene_data_info, list):
             # Parse each message in the list
             all_memory_items = []
             for msg in scene_data_info:
-                items = self.multi_model_parser.parse(msg, info, mode=mode, **kwargs)
+                items = self.multi_model_parser.parse(msg, info, mode="fast", **kwargs)
                 all_memory_items.extend(items)
-            return all_memory_items
+            fast_memory_items = self._concat_multi_model_memories(all_memory_items)
+
         else:
             # Parse as single message
-            return self.multi_model_parser.parse(scene_data_info, info, mode=mode, **kwargs)
+            fast_memory_items = self.multi_model_parser.parse(
+                scene_data_info, info, mode="fast", **kwargs
+            )
+
+        if mode == "fast":
+            return fast_memory_items
+        else:
+            # TODO: parallel call llm and get fine multi model items
+            # Part A: call llm
+            fine_memory_items = []
+            fine_memory_items_string_parser = []
+            fine_memory_items.extend(fine_memory_items_string_parser)
+            # Part B: get fine multi model items
+
+            for fast_item in fast_memory_items:
+                sources = fast_item.metadata.sources
+                for source in sources:
+                    items = self.multi_model_parser.process_transfer(
+                        source, context_items=[fast_item], custom_tags=custom_tags
+                    )
+                    fine_memory_items.extend(items)
+            logger.warning("Not Implemented Now!")
+            return fine_memory_items
 
     @timed
-    def _process_transfer_multi_model_data(self, raw_node: TextualMemoryItem):
-        raise NotImplementedError
+    def _process_transfer_multi_model_data(
+        self,
+        raw_node: TextualMemoryItem,
+        custom_tags: list[str] | None = None,
+    ) -> list[TextualMemoryItem]:
+        """
+        Process transfer for multi-model data.
+
+        Each source is processed independently by its corresponding parser,
+        which knows how to rebuild the original message and parse it in fine mode.
+        """
+        sources = raw_node.metadata.sources or []
+        if not sources:
+            logger.warning("[MultiModelStruct] No sources found in raw_node")
+            return []
+
+        # Extract info from raw_node (same as simple_struct.py)
+        info = {
+            "user_id": raw_node.metadata.user_id,
+            "session_id": raw_node.metadata.session_id,
+            **(raw_node.metadata.info or {}),
+        }
+
+        fine_memory_items = []
+        # Part A: call llm
+        fine_memory_items_string_parser = []
+        fine_memory_items.extend(fine_memory_items_string_parser)
+        # Part B: get fine multi model items
+        for source in sources:
+            items = self.multi_model_parser.process_transfer(
+                source, context_items=[raw_node], info=info, custom_tags=custom_tags
+            )
+            fine_memory_items.extend(items)
+        return fine_memory_items
 
     def get_scene_data_info(self, scene_data: list, type: str) -> list[list[Any]]:
         """
@@ -85,7 +151,7 @@ def get_scene_data_info(self, scene_data: list, type: str) -> list[list[Any]]:
 
     def _read_memory(
         self, messages: list[MessagesType], type: str, info: dict[str, Any], mode: str = "fine"
-    ):
+    ) -> list[list[TextualMemoryItem]]:
         list_scene_data_info = self.get_scene_data_info(messages, type)
 
         memory_list = []
@@ -106,7 +172,10 @@ def _read_memory(
         return memory_list
 
     def fine_transfer_simple_mem(
-        self, input_memories: list[TextualMemoryItem], type: str
+        self,
+        input_memories: list[TextualMemoryItem],
+        type: str,
+        custom_tags: list[str] | None = None,
     ) -> list[list[TextualMemoryItem]]:
         if not input_memories:
             return []
@@ -116,7 +185,9 @@ def fine_transfer_simple_mem(
         # Process Q&A pairs concurrently with context propagation
         with ContextThreadPoolExecutor() as executor:
             futures = [
-                executor.submit(self._process_transfer_multi_model_data, scene_data_info)
+                executor.submit(
+                    self._process_transfer_multi_model_data, scene_data_info, custom_tags
+                )
                 for scene_data_info in input_memories
             ]
             for future in concurrent.futures.as_completed(futures):

src/memos/mem_reader/read_multi_model/assistant_parser.py

@@ -5,10 +5,10 @@
 from memos.embedders.base import BaseEmbedder
 from memos.llms.base import BaseLLM
 from memos.log import get_logger
-from memos.memories.textual.item import TextualMemoryItem
+from memos.memories.textual.item import SourceMessage, TextualMemoryItem
 from memos.types.openai_chat_completion_types import ChatCompletionAssistantMessageParam
 
-from .base import BaseMessageParser
+from .base import BaseMessageParser, _extract_text_from_content
 
 
 logger = get_logger(__name__)
@@ -25,16 +25,45 @@ def __init__(self, embedder: BaseEmbedder, llm: BaseLLM | None = None):
             embedder: Embedder for generating embeddings
             llm: Optional LLM for fine mode processing
         """
-        self.embedder = embedder
-        self.llm = llm
+        super().__init__(embedder, llm)
+
+    def create_source(
+        self,
+        message: ChatCompletionAssistantMessageParam,
+        info: dict[str, Any],
+    ) -> SourceMessage:
+        """Create SourceMessage from assistant message."""
+        if not isinstance(message, dict):
+            return SourceMessage(type="chat", role="assistant")
+
+        content = _extract_text_from_content(message.get("content", ""))
+        return SourceMessage(
+            type="chat",
+            role="assistant",
+            chat_time=message.get("chat_time"),
+            message_id=message.get("message_id"),
+            content=content,
+        )
+
+    def rebuild_from_source(
+        self,
+        source: SourceMessage,
+    ) -> ChatCompletionAssistantMessageParam:
+        """Rebuild assistant message from SourceMessage."""
+        return {
+            "role": "assistant",
+            "content": source.content or "",
+            "chat_time": source.chat_time,
+            "message_id": source.message_id,
+        }
 
     def parse_fast(
         self,
         message: ChatCompletionAssistantMessageParam,
         info: dict[str, Any],
         **kwargs,
     ) -> list[TextualMemoryItem]:
-        return []
+        return super().parse_fast(message, info, **kwargs)
 
     def parse_fine(
         self,

src/memos/mem_reader/read_multi_model/base.py

@@ -4,24 +4,125 @@
 in both fast and fine modes.
 """
 
+import re
+
 from abc import ABC, abstractmethod
 from typing import Any
 
-from memos.memories.textual.item import TextualMemoryItem
+from memos import log
+from memos.memories.textual.item import (
+    SourceMessage,
+    TextualMemoryItem,
+    TreeNodeTextualMemoryMetadata,
+)
+
+
+logger = log.get_logger(__name__)
+
+
+def _derive_key(text: str, max_len: int = 80) -> str:
+    """Default key when without LLM: first max_len words."""
+    if not text:
+        return ""
+    sent = re.split(r"[。！？!?]\s*|\n", text.strip())[0]
+    return (sent[:max_len]).strip()
+
+
+def _extract_text_from_content(content: Any) -> str:
+    """
+    Extract text from message content.
+    Handles str, list of parts, or None.
+    """
+    if content is None:
+        return ""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        texts = []
+        for part in content:
+            if isinstance(part, dict):
+                part_type = part.get("type", "")
+                if part_type == "text":
+                    texts.append(part.get("text", ""))
+                elif part_type == "file":
+                    file_info = part.get("file", {})
+                    texts.append(file_info.get("file_data") or file_info.get("filename", "[file]"))
+                else:
+                    texts.append(f"[{part_type}]")
+            else:
+                texts.append(str(part))
+        return " ".join(texts)
+    return str(content)
 
 
 class BaseMessageParser(ABC):
     """Base interface for message type parsers."""
 
+    def __init__(self, embedder, llm=None):
+        """
+        Initialize BaseMessageParser.
+
+        Args:
+            embedder: Embedder for generating embeddings
+            llm: Optional LLM for fine mode processing
+        """
+        self.embedder = embedder
+        self.llm = llm
+
+    @abstractmethod
+    def create_source(
+        self,
+        message: Any,
+        info: dict[str, Any],
+    ) -> SourceMessage | list[SourceMessage]:
+        """
+        Create SourceMessage(s) from the message.
+
+        Each parser decides how to create sources:
+        - Simple messages: return single SourceMessage
+        - Multimodal messages: return list of SourceMessage (one per part)
+
+        Args:
+            message: The message to create source from
+            info: Dictionary containing user_id and session_id
+
+        Returns:
+            SourceMessage or list of SourceMessage
+        """
+
     @abstractmethod
+    def rebuild_from_source(
+        self,
+        source: SourceMessage,
+    ) -> Any:
+        """
+        Rebuild original message from SourceMessage.
+
+        Each parser knows how to reconstruct its own message type.
+
+        Args:
+            source: SourceMessage to rebuild from
+
+        Returns:
+            Rebuilt message in original format
+        """
+
     def parse_fast(
         self,
         message: Any,
         info: dict[str, Any],
         **kwargs,
     ) -> list[TextualMemoryItem]:
         """
-        Parse message in fast mode (no LLM calls, quick processing).
+        Default parse_fast implementation (equivalent to simple_struct fast mode).
+
+        Fast mode logic:
+        - Extract text content from message
+        - Determine memory_type based on role (UserMemory for user, LongTermMemory otherwise)
+        - Create TextualMemoryItem with tags=["mode:fast"]
+        - No LLM calls, quick processing
+
+        Subclasses can override this method for custom behavior.
 
         Args:
             message: The message to parse
@@ -31,6 +132,52 @@ def parse_fast(
         Returns:
             List of TextualMemoryItem objects
         """
+        if not isinstance(message, dict):
+            logger.warning(f"[BaseParser] Expected dict, got {type(message)}")
+            return []
+
+        # Extract text content
+        content = _extract_text_from_content(message.get("content"))
+        if not content:
+            return []
+
+        # Determine memory_type based on role (equivalent to simple_struct logic)
+        role = message.get("role", "").strip().lower()
+        memory_type = "UserMemory" if role == "user" else "LongTermMemory"
+
+        # Create source(s) using parser's create_source method
+        sources = self.create_source(message, info)
+        if isinstance(sources, SourceMessage):
+            sources = [sources]
+        elif not sources:
+            return []
+
+        # Extract info fields
+        info_ = info.copy()
+        user_id = info_.pop("user_id", "")
+        session_id = info_.pop("session_id", "")
+
+        # Create memory item (equivalent to _make_memory_item)
+        memory_item = TextualMemoryItem(
+            memory=content,
+            metadata=TreeNodeTextualMemoryMetadata(
+                user_id=user_id,
+                session_id=session_id,
+                memory_type=memory_type,
+                status="activated",
+                tags=["mode:fast"],
+                key=_derive_key(content),
+                embedding=self.embedder.embed([content])[0],
+                usage=[],
+                sources=sources,
+                background="",
+                confidence=0.99,
+                type="fact",
+                info=info_,
+            ),
+        )
+
+        return [memory_item]
 
     @abstractmethod
     def parse_fine(