feat(qdrant):support qdrant cloud and add index (#522)

* docs: update .env.example with comprehensive variables and comments

* hotfix:hotfix

* feat(qdrant):support qdrant cloud and add index

* chore: format qdrant test

---------

Co-authored-by: HarveyXiang <[email protected]>
Co-authored-by: fridayL <[email protected]>
Co-authored-by: chunyu li <[email protected]>
Co-authored-by: CaralHsi <[email protected]>

Author: 4 people

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/reranker/factory.py

@@ -1,6 +1,7 @@
 # memos/reranker/factory.py
 from __future__ import annotations
 
+import json
 from typing import TYPE_CHECKING, Any
 
 # Import singleton decorator
@@ -28,12 +29,19 @@ def from_config(cfg: RerankerConfigFactory | None) -> BaseReranker | None:
         backend = (cfg.backend or "").lower()
         c: dict[str, Any] = cfg.config or {}
 
+        headers_extra = c.get("headers_extra")
+        if isinstance(headers_extra, str):
+            try:
+                headers_extra = json.loads(headers_extra)
+            except Exception:
+                headers_extra = None
+
         if backend in {"http_bge", "bge"}:
             return HTTPBGEReranker(
                 reranker_url=c.get("url") or c.get("endpoint") or c.get("reranker_url"),
                 model=c.get("model", "bge-reranker-v2-m3"),
                 timeout=int(c.get("timeout", 10)),
-                headers_extra=c.get("headers_extra"),
+                headers_extra=headers_extra,
                 rerank_source=c.get("rerank_source"),
             )
 
@@ -51,7 +59,7 @@ def from_config(cfg: RerankerConfigFactory | None) -> BaseReranker | None:
                 reranker_url=c.get("url") or c.get("endpoint") or c.get("reranker_url"),
                 model=c.get("model", "bge-reranker-v2-m3"),
                 timeout=int(c.get("timeout", 10)),
-                headers_extra=c.get("headers_extra"),
+                headers_extra=headers_extra,
                 rerank_source=c.get("rerank_source"),
                 reranker_strategy=c.get("reranker_strategy"),
             )

src/memos/vec_dbs/qdrant.py

@@ -23,24 +23,49 @@ def __init__(self, config: QdrantVecDBConfig):
         from qdrant_client import QdrantClient
 
         self.config = config
+        # Default payload fields we always index because query filters rely on them
+        self._default_payload_index_fields = [
+            "memory_type",
+            "status",
+            "vector_sync",
+            "user_name",
+        ]
 
-        # If both host and port are None, we are running in local mode
-        if self.config.host is None and self.config.port is None:
-            logger.warning(
-                "Qdrant is running in local mode (host and port are both None). "
-                "In local mode, there may be race conditions during concurrent reads/writes. "
-                "It is strongly recommended to deploy a standalone Qdrant server "
-                "(e.g., via Docker: https://qdrant.tech/documentation/quickstart/)."
+        client_kwargs: dict[str, Any] = {}
+        if self.config.url:
+            client_kwargs["url"] = self.config.url
+            if self.config.api_key:
+                client_kwargs["api_key"] = self.config.api_key
+        else:
+            client_kwargs.update(
+                {
+                    "host": self.config.host,
+                    "port": self.config.port,
+                    "path": self.config.path,
+                }
             )
 
-        self.client = QdrantClient(
-            host=self.config.host, port=self.config.port, path=self.config.path
-        )
+            # If both host and port are None, we are running in local/embedded mode
+            if self.config.host is None and self.config.port is None:
+                logger.warning(
+                    "Qdrant is running in local mode (host and port are both None). "
+                    "In local mode, there may be race conditions during concurrent reads/writes. "
+                    "It is strongly recommended to deploy a standalone Qdrant server "
+                    "(e.g., via Docker: https://qdrant.tech/documentation/quickstart/)."
+                )
+
+        self.client = QdrantClient(**client_kwargs)
         self.create_collection()
+        # Ensure common payload indexes exist (idempotent)
+        try:
+            self.ensure_payload_indexes(self._default_payload_index_fields)
+        except Exception as e:
+            logger.warning(f"Failed to ensure default payload indexes: {e}")
 
     def create_collection(self) -> None:
         """Create a new collection with specified parameters."""
         from qdrant_client.http import models
+        from qdrant_client.http.exceptions import UnexpectedResponse
 
         if self.collection_exists(self.config.collection_name):
             collection_info = self.client.get_collection(self.config.collection_name)
@@ -57,13 +82,25 @@ def create_collection(self) -> None:
             "dot": models.Distance.DOT,
         }
 
-        self.client.create_collection(
-            collection_name=self.config.collection_name,
-            vectors_config=models.VectorParams(
-                size=self.config.vector_dimension,
-                distance=distance_map[self.config.distance_metric],
-            ),
-        )
+        try:
+            self.client.create_collection(
+                collection_name=self.config.collection_name,
+                vectors_config=models.VectorParams(
+                    size=self.config.vector_dimension,
+                    distance=distance_map[self.config.distance_metric],
+                ),
+            )
+        except UnexpectedResponse as err:
+            # Cloud Qdrant returns 409 when the collection already exists; tolerate and continue.
+            if getattr(err, "status_code", None) == 409 or "already exists" in str(err).lower():
+                logger.warning(
+                    f"Collection '{self.config.collection_name}' already exists. Skipping creation."
+                )
+                return
+            raise
+        except Exception:
+            # Bubble up other exceptions so callers can observe failures
+            raise
 
         logger.info(
             f"Collection '{self.config.collection_name}' created with {self.config.vector_dimension} dimensions."

tests/configs/test_vec_db.py

@@ -40,7 +40,15 @@ def test_qdrant_vec_db_config():
         required_fields=[
             "collection_name",
         ],
-        optional_fields=["vector_dimension", "distance_metric", "host", "port", "path"],
+        optional_fields=[
+            "vector_dimension",
+            "distance_metric",
+            "host",
+            "port",
+            "path",
+            "url",
+            "api_key",
+        ],
     )
 
     check_config_instantiation_valid(
@@ -53,6 +61,17 @@ def test_qdrant_vec_db_config():
         },
     )
 
+    check_config_instantiation_valid(
+        QdrantVecDBConfig,
+        {
+            "collection_name": "test_collection",
+            "vector_dimension": 768,
+            "distance_metric": "cosine",
+            "url": "https://cloud.qdrant.example",
+            "api_key": "dummy",
+        },
+    )
+
     check_config_instantiation_invalid(QdrantVecDBConfig)
 
 

tests/vec_dbs/test_qdrant.py

@@ -113,3 +113,26 @@ def test_get_all(vec_db):
     results = vec_db.get_all()
     assert len(results) == 1
     assert isinstance(results[0], VecDBItem)
+
+
+def test_qdrant_client_cloud_init():
+    config = VectorDBConfigFactory.model_validate(
+        {
+            "backend": "qdrant",
+            "config": {
+                "collection_name": "cloud_collection",
+                "vector_dimension": 3,
+                "distance_metric": "cosine",
+                "url": "https://cloud.qdrant.example",
+                "api_key": "secret-key",
+            },
+        }
+    )
+
+    with patch("qdrant_client.QdrantClient") as mockclient:
+        mock_instance = mockclient.return_value
+        mock_instance.get_collection.side_effect = Exception("Not found")
+
+        VecDBFactory.from_config(config)
+
+        mockclient.assert_called_once_with(url="https://cloud.qdrant.example", api_key="secret-key")