Merge branch 'dev' of github.com:MemTensor/MemOS into dev

Author: 黑布林

docker/.env.example

@@ -47,7 +47,7 @@ OLLAMA_API_BASE=http://localhost:11434     # required when backend=ollama
 MOS_RERANKER_BACKEND=http_bge             # http_bge | http_bge_strategy | cosine_local
 MOS_RERANKER_URL=http://localhost:8001     # required when backend=http_bge*
 MOS_RERANKER_MODEL=bge-reranker-v2-m3     # siliconflow → use BAAI/bge-reranker-v2-m3
-MOS_RERANKER_HEADERS_EXTRA=               # extra headers, JSON string
+MOS_RERANKER_HEADERS_EXTRA=               # extra headers, JSON string, e.g. {"Authorization":"Bearer your_token"}
 MOS_RERANKER_STRATEGY=single_turn
 MOS_RERANK_SOURCE=                        # optional rerank scope, e.g., history/stream/custom
 
@@ -93,6 +93,9 @@ NEO4J_DB_NAME=neo4j                       # required for shared-db mode
 MOS_NEO4J_SHARED_DB=false
 QDRANT_HOST=localhost
 QDRANT_PORT=6333
+# For Qdrant Cloud / remote endpoint (takes priority if set):
+QDRANT_URL=your_qdrant_url
+QDRANT_API_KEY=your_qdrant_key
 MILVUS_URI=http://localhost:19530         # required when ENABLE_PREFERENCE_MEMORY=true
 MILVUS_USER_NAME=root                     # same as above
 MILVUS_PASSWORD=12345678                  # same as above

docs/product-api-tests.md

@@ -0,0 +1,65 @@
+## Product API smoke tests (local 0.0.0.0:8001)
+
+Source: https://github.com/MemTensor/MemOS/issues/518
+
+### Prerequisites
+- Service is running: `python -m uvicorn memos.api.server_api:app --host 0.0.0.0 --port 8001`
+- `.env` is configured for Redis, embeddings, and the vector DB (current test setup: Redis reachable, Qdrant Cloud connected).
+
+### 1) /product/add
+- Purpose: Write a memory (sync/async).
+- Example request (sync):
+
+  ```bash
+  curl -s -X POST http://127.0.0.1:8001/product/add \
+    -H 'Content-Type: application/json' \
+    -d '{
+          "user_id": "tester",
+          "mem_cube_id": "default_cube",
+          "memory_content": "Apple is a fruit rich in fiber.",
+          "async_mode": "sync"
+        }'
+  ```
+
+- Observed result: `200`, message: "Memory added successfully", returns the written `memory_id` and related info.
+
+### 2) /product/get_all
+- Purpose: List all memories for the user/type to confirm writes.
+- Example request:
+
+  ```bash
+  curl -s -X POST http://127.0.0.1:8001/product/get_all \
+    -H 'Content-Type: application/json' \
+    -d '{
+          "user_id": "tester",
+          "memory_type": "text_mem",
+          "mem_cube_ids": ["default_cube"]
+        }'
+  ```
+
+- Observed result: `200`, shows the recently written apple memories (WorkingMemory/LongTermMemory/UserMemory present, `vector_sync=success`).
+
+### 3) /product/search
+- Purpose: Vector search memories.
+- Example request:
+
+  ```bash
+  curl -s -X POST http://127.0.0.1:8001/product/search \
+    -H 'Content-Type: application/json' \
+    -d '{
+          "query": "What fruit is rich in fiber?",
+          "user_id": "tester",
+          "mem_cube_id": "default_cube",
+          "top_k": 5,
+          "pref_top_k": 3,
+          "include_preference": false
+        }'
+  ```
+
+- Observed result: previously returned 400 because payload indexes (e.g., `vector_sync`) were missing in Qdrant. Index creation is now automatic during Qdrant initialization (memory_type/status/vector_sync/user_name).
+- If results are empty or errors persist, verify indexes exist (auto-created on restart) or recreate/clean the collection.
+
+### Notes / Next steps
+- `/product/add` and `/product/get_all` are healthy.
+- `/product/search` still returns empty results even with vectors present; likely related to search filters or vector retrieval.
+- Suggested follow-ups: inspect `SearchHandler` flow, filter conditions (user_id/session/cube_name), and vector DB search calls; capture logs or compare with direct `VecDBFactory.search` calls.

src/memos/api/config.py

@@ -500,6 +500,9 @@ def get_neo4j_community_config(user_id: str | None = None) -> dict[str, Any]:
                     "distance_metric": "cosine",
                     "host": os.getenv("QDRANT_HOST", "localhost"),
                     "port": int(os.getenv("QDRANT_PORT", "6333")),
+                    "path": os.getenv("QDRANT_PATH"),
+                    "url": os.getenv("QDRANT_URL"),
+                    "api_key": os.getenv("QDRANT_API_KEY"),
                 },
             },
         }

src/memos/configs/vec_db.py

@@ -27,10 +27,13 @@ class QdrantVecDBConfig(BaseVecDBConfig):
     host: str | None = Field(default=None, description="Host for Qdrant")
     port: int | None = Field(default=None, description="Port for Qdrant")
     path: str | None = Field(default=None, description="Path for Qdrant")
+    url: str | None = Field(default=None, description="Qdrant Cloud/remote endpoint URL")
+    api_key: str | None = Field(default=None, description="Qdrant Cloud API key")
 
     @model_validator(mode="after")
     def set_default_path(self):
-        if all(x is None for x in (self.host, self.port, self.path)):
+        # Only fall back to embedded/local path when no remote host/port/path/url is provided.
+        if all(x is None for x in (self.host, self.port, self.path, self.url)):
             logger.warning(
                 "No host, port, or path provided for Qdrant. Defaulting to local path: %s",
                 settings.MEMOS_DIR / "qdrant",

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,