feat: add Valkey as RAG vector store backend

config/config2.example.yaml

@@ -74,6 +74,19 @@ redis:
   password: "YOUR_PASSWORD"
   db: "0"
 
+# Valkey Vector Store (for RAG backend)
+# valkey:
+#   host: "localhost"
+#   port: 6379
+#   password: ""
+#   use_tls: false
+#   request_timeout: 5000
+#   index_name: "metagpt_rag"
+#   prefix: "metagpt:rag:"
+#   vector_dimensions: 1536
+#   distance_metric: "COSINE"  # COSINE, L2, or IP
+#   vector_algorithm: "HNSW"   # HNSW or FLAT
+
 s3:
   access_key: "YOUR_ACCESS_KEY"
   secret_key: "YOUR_SECRET_KEY"

metagpt/rag/factories/index.py

@@ -17,6 +17,7 @@
     ElasticsearchIndexConfig,
     ElasticsearchKeywordIndexConfig,
     FAISSIndexConfig,
+    ValkeyIndexConfig,
 )
 
 
@@ -28,6 +29,7 @@ def __init__(self):
             BM25IndexConfig: self._create_bm25,
             ElasticsearchIndexConfig: self._create_es,
             ElasticsearchKeywordIndexConfig: self._create_es,
+            ValkeyIndexConfig: self._create_valkey,
         }
         super().__init__(creators)
 
@@ -75,6 +77,12 @@ def _index_from_vector_store(
             embed_model=embed_model,
         )
 
+    def _create_valkey(self, config: ValkeyIndexConfig, **kwargs) -> VectorStoreIndex:
+        from metagpt.rag.vector_stores.valkey import ValkeyVectorStore
+
+        vector_store = ValkeyVectorStore(**config.store_config.model_dump())
+        return self._index_from_vector_store(vector_store=vector_store, config=config, **kwargs)
+
     def _extract_embed_model(self, config, **kwargs) -> BaseEmbedding:
         return self._val_from_config_or_kwargs("embed_model", config, **kwargs)
 

metagpt/rag/factories/retriever.py

@@ -28,6 +28,7 @@
     ElasticsearchKeywordRetrieverConfig,
     ElasticsearchRetrieverConfig,
     FAISSRetrieverConfig,
+    ValkeyRetrieverConfig,
 )
 
 
@@ -57,6 +58,7 @@ def __init__(self):
             ChromaRetrieverConfig: self._create_chroma_retriever,
             ElasticsearchRetrieverConfig: self._create_es_retriever,
             ElasticsearchKeywordRetrieverConfig: self._create_es_retriever,
+            ValkeyRetrieverConfig: self._create_valkey_retriever,
         }
         super().__init__(creators)
 
@@ -104,6 +106,18 @@ def _create_es_retriever(self, config: ElasticsearchRetrieverConfig, **kwargs) -
 
         return ElasticsearchRetriever(**config.model_dump())
 
+    def _create_valkey_retriever(self, config: ValkeyRetrieverConfig, **kwargs) -> "ValkeyRetriever":
+        from metagpt.rag.retrievers.valkey_retriever import ValkeyRetriever
+        from metagpt.rag.vector_stores.valkey import ValkeyVectorStore
+
+        vector_store = ValkeyVectorStore(**config.store_config.model_dump())
+        return ValkeyRetriever(
+            vector_store=vector_store,
+            similarity_top_k=config.similarity_top_k,
+            nodes=self._extract_nodes(config, **kwargs),
+            embed_model=self._extract_embed_model(config, **kwargs),
+        )
+
     def _extract_index(self, config: BaseRetrieverConfig = None, **kwargs) -> VectorStoreIndex:
         return self._val_from_config_or_kwargs("index", config, **kwargs)
 

metagpt/rag/retrievers/valkey_retriever.py

@@ -0,0 +1,96 @@
+"""Valkey retriever."""
+
+from typing import Any, List, Optional
+
+from llama_index.core.schema import BaseNode, NodeWithScore, QueryBundle, QueryType
+from llama_index.core.vector_stores.types import VectorStoreQuery
+
+from metagpt.rag.retrievers.base import RAGRetriever
+from metagpt.rag.vector_stores.valkey import ValkeyVectorStore
+
+
+class ValkeyRetriever(RAGRetriever):
+    """Valkey-based retriever using ValkeyVectorStore for KNN search."""
+
+    def __init__(
+        self,
+        vector_store: ValkeyVectorStore,
+        similarity_top_k: int = 5,
+        nodes: Optional[List[BaseNode]] = None,
+        embed_model: Any = None,
+        **kwargs,
+    ):
+        super().__init__()
+        self._vector_store = vector_store
+        self._similarity_top_k = similarity_top_k
+        self._embed_model = embed_model
+
+        # If nodes are provided, add them to the store
+        if nodes:
+            self.add_nodes(nodes)
+
+    @property
+    def vector_store(self) -> ValkeyVectorStore:
+        """Access the underlying vector store."""
+        return self._vector_store
+
+    def _embed_query(self, query: QueryBundle) -> List[float]:
+        """Resolve the query embedding, preferring the async embed call when available."""
+        if query.embedding is not None:
+            return query.embedding
+        if self._embed_model is None:
+            raise ValueError("Query embedding is required. Provide an embed_model or set query.embedding.")
+        return self._embed_model.get_query_embedding(query.query_str)
+
+    async def _aembed_query(self, query: QueryBundle) -> List[float]:
+        """Async embedding resolution that does not block the event loop when the model supports it."""
+        if query.embedding is not None:
+            return query.embedding
+        if self._embed_model is None:
+            raise ValueError("Query embedding is required. Provide an embed_model or set query.embedding.")
+        if hasattr(self._embed_model, "aget_query_embedding"):
+            return await self._embed_model.aget_query_embedding(query.query_str)
+        return self._embed_model.get_query_embedding(query.query_str)
+
+    def _build_result(self, result) -> List[NodeWithScore]:
+        return [
+            NodeWithScore(node=node, score=similarity) for node, similarity in zip(result.nodes, result.similarities)
+        ]
+
+    async def _aretrieve(self, query: QueryType) -> List[NodeWithScore]:
+        """Async retrieve nodes matching the query."""
+        if isinstance(query, str):
+            query = QueryBundle(query_str=query)
+
+        query_embedding = await self._aembed_query(query)
+        store_query = VectorStoreQuery(query_embedding=query_embedding, similarity_top_k=self._similarity_top_k)
+        result = self._vector_store.query(store_query)
+        return self._build_result(result)
+
+    def _retrieve(self, query: QueryType) -> List[NodeWithScore]:
+        """Sync retrieve nodes matching the query."""
+        if isinstance(query, str):
+            query = QueryBundle(query_str=query)
+
+        query_embedding = self._embed_query(query)
+        store_query = VectorStoreQuery(query_embedding=query_embedding, similarity_top_k=self._similarity_top_k)
+        result = self._vector_store.query(store_query)
+        return self._build_result(result)
+
+    def add_nodes(self, nodes: List[BaseNode], **kwargs) -> None:
+        """Add nodes to the underlying vector store."""
+        self._vector_store.add(nodes, **kwargs)
+
+    def persist(self, persist_dir: str = "", **kwargs) -> None:
+        """Persist is a no-op since Valkey auto-persists.
+
+        Valkey automatically saves data, so there is no need to implement."""
+
+    def query_total_count(self) -> int:
+        """Query total count of documents in the store."""
+        return len(self._vector_store.scan_all_docs())
+
+    def clear(self, **kwargs) -> None:
+        """Clear all documents from the store and recreate the index."""
+        self._vector_store.drop_index()
+        self._vector_store.ensure_index()

metagpt/rag/schema.py

@@ -108,6 +108,32 @@ class ElasticsearchKeywordRetrieverConfig(ElasticsearchRetrieverConfig):
     )
 
 
+class ValkeyStoreConfig(BaseModel):
+    """Configuration for Valkey vector store connection."""
+
+    host: str = Field(default="localhost", description="Valkey server host.")
+    port: int = Field(default=6379, description="Valkey server port.")
+    password: Optional[str] = Field(default=None, description="Valkey server password.", repr=False)
+    use_tls: bool = Field(default=False, description="Whether to use TLS for connection.")
+    request_timeout: int = Field(default=5000, description="Request timeout in milliseconds.")
+    index_name: str = Field(default="metagpt_rag", description="Name of the Valkey Search index.")
+    prefix: str = Field(default="metagpt:rag:", description="Key prefix for stored documents.")
+    vector_dimensions: int = Field(default=1536, description="Dimensionality of embedding vectors.")
+    distance_metric: Literal["COSINE", "L2", "IP"] = Field(
+        default="COSINE", description="Distance metric: COSINE, L2, or IP."
+    )
+    vector_algorithm: Literal["HNSW", "FLAT"] = Field(
+        default="HNSW", description="Vector index algorithm: HNSW or FLAT."
+    )
+    client_name: str = Field(default="metagpt_rag_client", description="Client name for Valkey connection.")
+
+
+class ValkeyRetrieverConfig(IndexRetrieverConfig):
+    """Config for Valkey-based retrievers."""
+
+    store_config: ValkeyStoreConfig = Field(..., description="ValkeyStore config.")
+
+
 class BaseRankerConfig(BaseModel):
     """Common config for rankers.
 
@@ -193,6 +219,13 @@ class ElasticsearchIndexConfig(VectorIndexConfig):
     persist_path: Union[str, Path] = ""
 
 
+class ValkeyIndexConfig(VectorIndexConfig):
+    """Config for Valkey-based index."""
+
+    store_config: ValkeyStoreConfig = Field(..., description="ValkeyStore config.")
+    persist_path: Union[str, Path] = ""
+
+
 class ElasticsearchKeywordIndexConfig(ElasticsearchIndexConfig):
     """Config for es-based index. no embedding."""
 

metagpt/rag/vector_stores/__init__.py

@@ -0,0 +1 @@
+"""RAG Vector Stores."""