feat(vector-store): add base vector store interface and multiple implementations

docs/deprecated.txt

@@ -1,6 +1,6 @@
 from loguru import logger
 
-用英文注释，完善module/class/function docstring，要一句话简洁，不要变更代码
+用英文注释，完善module/class/function docstring，要一句话简洁，不要变更代码逻辑，符合pep和pylint规范，使用list而不是typing.List/Dict，不使用typing.Union
 
 看看代码有什么问题
 用英文注释，完善module/class/function docstring，要一句话简洁，代码要简洁，符合pep和pylint规范，使用list而不是typing.List，不使用typing.Union

reme_ai/core/vector_store/__init__.py

@@ -0,0 +1,17 @@
+"""vector store"""
+
+from .base_vector_store import BaseVectorStore
+from .chroma_vector_store import ChromaVectorStore
+from .es_vector_store import ESVectorStore
+from .local_vector_store import LocalVectorStore
+from .pgvector_store import PGVectorStore
+from .qdrant_vector_store import QdrantVectorStore
+
+__all__ = [
+    "BaseVectorStore",
+    "ChromaVectorStore",
+    "ESVectorStore",
+    "LocalVectorStore",
+    "PGVectorStore",
+    "QdrantVectorStore",
+]

reme_ai/core/vector_store/base_vector_store.py

@@ -0,0 +1,92 @@
+"""Base vector store interface for managing vector embeddings and similarity search."""
+
+import asyncio
+from abc import ABC, abstractmethod
+from collections.abc import Callable
+from functools import partial
+
+from reme_ai.core.context import C
+from reme_ai.core.embedding import BaseEmbeddingModel
+from reme_ai.core.schema import VectorNode
+
+
+class BaseVectorStore(ABC):
+    """Abstract base class defining the interface for vector storage and retrieval."""
+
+    def __init__(
+        self,
+        collection_name: str,
+        embedding_model: BaseEmbeddingModel,
+        **kwargs,
+    ):
+        """Initialize the vector store with a collection name and an embedding model."""
+        if embedding_model is None:
+            raise ValueError("embedding_model is required")
+        self.collection_name: str = collection_name
+        self.embedding_model: BaseEmbeddingModel = embedding_model
+        self.kwargs: dict = kwargs
+
+    @staticmethod
+    async def _run_sync_in_executor(sync_func: Callable, *args, **kwargs):
+        """Run a synchronous function in the context-defined thread pool executor."""
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(C.thread_pool, partial(sync_func, *args, **kwargs))
+
+    async def get_node_embedding(self, node: VectorNode) -> VectorNode:
+        """Generate and assign embedding for a single vector node."""
+        return await self.embedding_model.get_node_embedding(node)
+
+    async def get_node_embeddings(self, nodes: list[VectorNode]) -> list[VectorNode]:
+        """Generate and assign embeddings for multiple vector nodes."""
+        return await self.embedding_model.get_node_embeddings(nodes)
+
+    async def get_embedding(self, query: str) -> list[float]:
+        """Convert a single text query into vector embedding using the configured model."""
+        return await self.embedding_model.get_embedding(query)
+
+    async def get_embeddings(self, queries: list[str]) -> list[list[float]]:
+        """Convert multiple text queries into vector embeddings using the configured model."""
+        return await self.embedding_model.get_embeddings(queries)
+
+    @abstractmethod
+    async def list_collections(self) -> list[str]:
+        """Retrieve a list of all existing collection names in the store."""
+
+    @abstractmethod
+    async def create_collection(self, collection_name: str, **kwargs) -> None:
+        """Create a new vector collection with the specified name and configuration."""
+
+    @abstractmethod
+    async def delete_collection(self, collection_name: str, **kwargs) -> None:
+        """Permanently remove a collection from the vector store."""
+
+    @abstractmethod
+    async def copy_collection(self, collection_name: str, **kwargs) -> None:
+        """Duplicate the current collection to a new one with the given name."""
+
+    @abstractmethod
+    async def insert(self, nodes: VectorNode | list[VectorNode], **kwargs) -> None:
+        """Add one or more vector nodes into the current collection."""
+
+    @abstractmethod
+    async def search(self, query: str, limit: int = 5, filters: dict | None = None, **kwargs) -> list[VectorNode]:
+        """Find the most similar vector nodes based on a text query."""
+
+    @abstractmethod
+    async def delete(self, vector_ids: str | list[str], **kwargs) -> None:
+        """Remove specific vectors from the collection using their identifiers."""
+
+    @abstractmethod
+    async def update(self, nodes: VectorNode | list[VectorNode], **kwargs) -> None:
+        """Update the data or metadata of existing vectors in the collection."""
+
+    @abstractmethod
+    async def get(self, vector_ids: str | list[str]) -> VectorNode | list[VectorNode]:
+        """Fetch specific vector nodes from the collection by their IDs."""
+
+    @abstractmethod
+    async def list(self, filters: dict | None = None, limit: int | None = None) -> list[VectorNode]:
+        """Retrieve vectors from the collection that match the given filters."""
+
+    async def close(self) -> None:
+        """Release resources and close active connections to the vector store."""

reme_ai/core/vector_store/chroma_vector_store.py

@@ -0,0 +1,397 @@
+"""ChromaDB vector store implementation for the ReMe framework."""
+
+from typing import Any
+
+from loguru import logger
+
+from .base_vector_store import BaseVectorStore
+from ..context import C
+from ..embedding import BaseEmbeddingModel
+from ..schema import VectorNode
+
+_CHROMADB_IMPORT_ERROR = None
+
+try:
+    import chromadb
+    from chromadb.config import Settings
+except ImportError as e:
+    _CHROMADB_IMPORT_ERROR = e
+    chromadb = None
+    Settings = None
+
+
+@C.register_vector_store("chroma")
+class ChromaVectorStore(BaseVectorStore):
+    """ChromaDB-based vector store implementation for local or remote storage."""
+
+    def __init__(
+        self,
+        collection_name: str,
+        embedding_model: BaseEmbeddingModel,
+        client: chromadb.ClientAPI | None = None,
+        host: str | None = None,
+        port: int | None = None,
+        path: str | None = None,
+        api_key: str | None = None,
+        tenant: str | None = None,
+        database: str | None = None,
+        **kwargs,
+    ):
+        """Initialize the ChromaDB vector store with the provided configuration."""
+        if _CHROMADB_IMPORT_ERROR is not None:
+            raise ImportError(
+                "ChromaDB requires extra dependencies. Install with `pip install chromadb`",
+            ) from _CHROMADB_IMPORT_ERROR
+
+        super().__init__(
+            collection_name=collection_name,
+            embedding_model=embedding_model,
+            **kwargs,
+        )
+
+        self.client: chromadb.ClientAPI
+        self.collection: chromadb.Collection
+
+        if client:
+            self.client = client
+        elif api_key and tenant:
+            logger.info("Initializing ChromaDB Cloud client")
+            self.client = chromadb.CloudClient(
+                api_key=api_key,
+                tenant=tenant,
+                database=database or "default",
+            )
+        elif host and port:
+            logger.info(f"Initializing ChromaDB HTTP client at {host}:{port}")
+            self.client = chromadb.HttpClient(host=host, port=port)
+        else:
+            if path is None:
+                path = "./chroma_db"
+            logger.info(f"Initializing local ChromaDB at {path}")
+            self.client = chromadb.PersistentClient(
+                path=path,
+                settings=Settings(anonymized_telemetry=False),
+            )
+
+        self.collection = self.client.get_or_create_collection(
+            name=collection_name,
+            metadata={"hnsw:space": "cosine"},
+        )
+
+    @staticmethod
+    def _parse_results(
+        results: dict,
+        include_score: bool = False,
+    ) -> list[VectorNode]:
+        """Convert ChromaDB query results into a list of VectorNode objects."""
+        nodes = []
+
+        ids = results.get("ids", [])
+        documents = results.get("documents", [])
+        metadatas = results.get("metadatas", [])
+        embeddings = results.get("embeddings") if results.get("embeddings") is not None else []
+        distances = results.get("distances") if results.get("distances") is not None else []
+
+        if ids and isinstance(ids[0], list):
+            ids = ids[0] if ids else []
+            documents = documents[0] if documents else []
+            metadatas = metadatas[0] if metadatas else []
+            embeddings = embeddings[0] if embeddings and len(embeddings) > 0 else []
+            distances = distances[0] if distances and len(distances) > 0 else []
+
+        for i, vector_id in enumerate(ids):
+            metadata = metadatas[i] if i < len(metadatas) and metadatas[i] else {}
+
+            if include_score and distances and i < len(distances):
+                metadata["_score"] = 1.0 - distances[i]
+
+            node = VectorNode(
+                vector_id=vector_id,
+                content=documents[i] if i < len(documents) and documents[i] else "",
+                vector=embeddings[i] if len(embeddings) > i else None,
+                metadata=metadata,
+            )
+            nodes.append(node)
+
+        return nodes
+
+    @staticmethod
+    def _generate_where_clause(filters: dict | None) -> dict | None:
+        """Convert the universal filter format to a ChromaDB-compatible where clause."""
+        if not filters:
+            return None
+
+        def convert_condition(k: str, v: Any) -> dict | None:
+            """Convert a single filter condition to ChromaDB operator format."""
+            if v == "*":
+                return None
+            if isinstance(v, dict):
+                chroma_condition = {}
+                for op, val in v.items():
+                    mapping = {
+                        "eq": "$eq",
+                        "ne": "$ne",
+                        "gt": "$gt",
+                        "gte": "$gte",
+                        "lt": "$lt",
+                        "lte": "$lte",
+                        "in": "$in",
+                        "nin": "$nin",
+                    }
+                    chroma_op = mapping.get(op, "$eq")
+                    chroma_condition[k] = {chroma_op: val}
+                return chroma_condition
+            if isinstance(v, list):
+                return {k: {"$in": v}}
+            return {k: {"$eq": v}}
+
+        processed_filters = []
+
+        for key, value in filters.items():
+            if key == "$or":
+                or_conditions = []
+                for condition in value:
+                    or_condition = {}
+                    for sub_key, sub_value in condition.items():
+                        converted = convert_condition(sub_key, sub_value)
+                        if converted:
+                            or_condition.update(converted)
+                    if or_condition:
+                        or_conditions.append(or_condition)
+                if len(or_conditions) > 1:
+                    processed_filters.append({"$or": or_conditions})
+                elif len(or_conditions) == 1:
+                    processed_filters.append(or_conditions[0])
+
+            elif key == "$and":
+                for condition in value:
+                    for sub_key, sub_value in condition.items():
+                        converted = convert_condition(sub_key, sub_value)
+                        if converted:
+                            processed_filters.append(converted)
+            elif key == "$not":
+                continue
+            else:
+                converted = convert_condition(key, value)
+                if converted:
+                    processed_filters.append(converted)
+
+        if not processed_filters:
+            return None
+        return processed_filters[0] if len(processed_filters) == 1 else {"$and": processed_filters}
+
+    async def list_collections(self) -> list[str]:
+        """Retrieve a list of all existing collection names."""
+
+        def _list():
+            return [col.name for col in self.client.list_collections()]
+
+        return await self._run_sync_in_executor(_list)
+
+    async def create_collection(self, collection_name: str, **kwargs):
+        """Create a new collection with specified distance metrics and metadata."""
+
+        def _create():
+            distance_metric = kwargs.get("distance_metric", "cosine")
+            metadata = kwargs.get("metadata", {})
+            metadata["hnsw:space"] = distance_metric
+            return self.client.get_or_create_collection(name=collection_name, metadata=metadata)
+
+        new_collection = await self._run_sync_in_executor(_create)
+        if collection_name == self.collection_name:
+            self.collection = new_collection
+        logger.info(f"Created collection {collection_name}")
+
+    async def delete_collection(self, collection_name: str, **kwargs):
+        """Delete a specified collection from the database."""
+
+        def _delete():
+            try:
+                self.client.delete_collection(name=collection_name)
+                return True
+            except Exception as e:
+                logger.warning(f"Failed to delete collection {collection_name}: {e}")
+                return False
+
+        deleted = await self._run_sync_in_executor(_delete)
+        if deleted and collection_name == self.collection_name:
+            self.collection = None
+        logger.info(f"Deleted collection {collection_name}")
+
+    async def copy_collection(self, collection_name: str, **kwargs):
+        """Copy all data from the current collection to a new collection."""
+
+        def _copy():
+            source_data = self.collection.get(include=["documents", "metadatas", "embeddings"])
+            if not source_data["ids"]:
+                logger.warning(f"Source collection {self.collection_name} is empty")
+                return
+
+            target_collection = self.client.get_or_create_collection(
+                name=collection_name,
+                metadata={"hnsw:space": "cosine"},
+            )
+            target_collection.add(
+                ids=source_data["ids"],
+                documents=source_data["documents"],
+                metadatas=source_data["metadatas"],
+                embeddings=source_data["embeddings"],
+            )
+
+        await self._run_sync_in_executor(_copy)
+        logger.info(f"Copied collection {self.collection_name} to {collection_name}")
+
+    async def insert(self, nodes: VectorNode | list[VectorNode], **kwargs):
+        """Insert vector nodes into the current collection in batches."""
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+        if not nodes:
+            return
+
+        # Batch generate embeddings for nodes that need them
+        nodes_without_vectors = [node for node in nodes if node.vector is None]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            # Create a mapping for quick lookup
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_insert = [vector_map.get(n.vector_id, n) if n.vector is None else n for n in nodes]
+        else:
+            nodes_to_insert = nodes
+
+        batch_size = kwargs.get("batch_size", 100)
+
+        def _insert_batch(batch_nodes: list[VectorNode]):
+            self.collection.add(
+                ids=[n.vector_id for n in batch_nodes],
+                documents=[n.content for n in batch_nodes],
+                embeddings=[n.vector for n in batch_nodes],
+                metadatas=[n.metadata for n in batch_nodes],
+            )
+
+        for i in range(0, len(nodes_to_insert), batch_size):
+            await self._run_sync_in_executor(_insert_batch, nodes_to_insert[i : i + batch_size])
+        logger.info(f"Inserted {len(nodes_to_insert)} nodes into {self.collection_name}")
+
+    async def search(
+        self,
+        query: str,
+        limit: int = 5,
+        filters: dict | None = None,
+        **kwargs,
+    ) -> list[VectorNode]:
+        """Search for the most similar vector nodes based on a text query."""
+        query_vector = await self.get_embedding(query)
+        where_clause = self._generate_where_clause(filters)
+        include_embeddings = kwargs.get("include_embeddings", False)
+
+        def _search():
+            include: list = ["documents", "metadatas", "distances"]
+            if include_embeddings:
+                include.append("embeddings")
+            return self.collection.query(
+                query_embeddings=[query_vector],
+                n_results=limit,
+                where=where_clause,
+                include=include,
+            )
+
+        results = await self._run_sync_in_executor(_search)
+        nodes = self._parse_results(results, include_score=True)
+
+        score_threshold = kwargs.get("score_threshold")
+        if score_threshold is not None:
+            nodes = [n for n in nodes if n.metadata.get("_score", 0) >= score_threshold]
+        return nodes
+
+    async def delete(self, vector_ids: str | list[str], **kwargs):
+        """Delete specific vector nodes by their IDs."""
+        if isinstance(vector_ids, str):
+            vector_ids = [vector_ids]
+        if not vector_ids:
+            return
+
+        def _delete():
+            self.collection.delete(ids=vector_ids)
+
+        await self._run_sync_in_executor(_delete)
+        logger.info(f"Deleted {len(vector_ids)} nodes from {self.collection_name}")
+
+    async def update(self, nodes: VectorNode | list[VectorNode], **kwargs):
+        """Update existing vector nodes with new content or metadata."""
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+        if not nodes:
+            return
+
+        # Batch generate embeddings for nodes that need them
+        nodes_without_vectors = [node for node in nodes if node.vector is None and node.content]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            # Create a mapping for quick lookup
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_update = [vector_map.get(n.vector_id, n) if n.vector is None and n.content else n for n in nodes]
+        else:
+            nodes_to_update = nodes
+
+        def _update():
+            self.collection.upsert(
+                ids=[n.vector_id for n in nodes_to_update],
+                documents=[n.content for n in nodes_to_update],
+                embeddings=[n.vector for n in nodes_to_update if n.vector] or None,
+                metadatas=[n.metadata for n in nodes_to_update],
+            )
+
+        await self._run_sync_in_executor(_update)
+        logger.info(f"Updated {len(nodes_to_update)} nodes in {self.collection_name}")
+
+    async def get(self, vector_ids: str | list[str]) -> VectorNode | list[VectorNode] | None:
+        """Fetch vector nodes by their IDs from the collection."""
+        is_single = isinstance(vector_ids, str)
+        ids = [vector_ids] if is_single else vector_ids
+
+        def _get():
+            return self.collection.get(ids=ids, include=["documents", "metadatas", "embeddings"])
+
+        results = await self._run_sync_in_executor(_get)
+        nodes = self._parse_results(results)
+        return nodes[0] if is_single and nodes else (nodes if not is_single else None)
+
+    async def list(
+        self,
+        filters: dict | None = None,
+        limit: int | None = None,
+    ) -> list[VectorNode]:
+        """List vector nodes matching optional metadata filters."""
+        where_clause = self._generate_where_clause(filters)
+
+        def _list():
+            return self.collection.get(
+                where=where_clause,
+                limit=limit,
+                include=["documents", "metadatas", "embeddings"],
+            )
+
+        results = await self._run_sync_in_executor(_list)
+        return self._parse_results(results)
+
+    async def count(self) -> int:
+        """Return the total number of vectors in the current collection."""
+        return await self._run_sync_in_executor(self.collection.count)
+
+    async def reset(self):
+        """Reset the current collection by clearing all its data."""
+        logger.warning(f"Resetting collection {self.collection_name}...")
+        await self.delete_collection(self.collection_name)
+
+        def _recreate():
+            self.collection = self.client.get_or_create_collection(
+                name=self.collection_name,
+                metadata={"hnsw:space": "cosine"},
+            )
+
+        await self._run_sync_in_executor(_recreate)
+        logger.info(f"Collection {self.collection_name} has been reset")
+
+    async def close(self):
+        """Close the vector store and log the shutdown process."""
+        logger.info(f"ChromaDB vector store for collection {self.collection_name} closed")

reme_ai/core/vector_store/es_vector_store.py

@@ -0,0 +1,458 @@
+"""Elasticsearch vector store implementation for ReMe.
+
+This module provides an Elasticsearch-based vector store that implements the BaseVectorStore
+interface for high-performance dense vector storage and retrieval.
+"""
+
+from typing import Any
+
+from loguru import logger
+
+from .base_vector_store import BaseVectorStore
+from ..context import C
+from ..embedding import BaseEmbeddingModel
+from ..schema import VectorNode
+
+_ELASTICSEARCH_IMPORT_ERROR = None
+
+try:
+    from elasticsearch import AsyncElasticsearch
+    from elasticsearch.helpers import async_bulk
+except ImportError as e:
+    _ELASTICSEARCH_IMPORT_ERROR = e
+    AsyncElasticsearch = None
+    async_bulk = None
+
+
+@C.register_vector_store("es")
+class ESVectorStore(BaseVectorStore):
+    """Elasticsearch-based vector store for dense vector storage and kNN search."""
+
+    def __init__(
+        self,
+        collection_name: str,
+        embedding_model: BaseEmbeddingModel,
+        hosts: str | list[str] | None = None,
+        basic_auth: tuple[str, str] | None = None,
+        cloud_id: str | None = None,
+        api_key: str | None = None,
+        verify_certs: bool = True,
+        headers: dict[str, str] | None = None,
+        **kwargs,
+    ):
+        """Initialize the Elasticsearch client and vector store configuration.
+
+        Args:
+            collection_name: Name of the Elasticsearch index (converted to lowercase).
+            embedding_model: Model instance used to generate vector embeddings.
+            hosts: Connection host(s) for the Elasticsearch cluster.
+            basic_auth: Credentials for basic authentication.
+            cloud_id: Deployment ID for Elastic Cloud.
+            api_key: API key for authentication.
+            verify_certs: Enable or disable SSL certificate verification.
+            headers: Custom HTTP headers for requests.
+            **kwargs: Additional configuration passed to the base class.
+        """
+        if _ELASTICSEARCH_IMPORT_ERROR is not None:
+            raise ImportError(
+                "Elasticsearch requires extra dependencies. Install with `pip install elasticsearch`",
+            ) from _ELASTICSEARCH_IMPORT_ERROR
+
+        # Elasticsearch requires lowercase index names
+        collection_name = collection_name.lower()
+
+        super().__init__(collection_name=collection_name, embedding_model=embedding_model, **kwargs)
+
+        # Initialize AsyncElasticsearch client
+        self.client = AsyncElasticsearch(
+            hosts=hosts,
+            cloud_id=cloud_id,
+            api_key=api_key,
+            basic_auth=basic_auth,
+            verify_certs=verify_certs,
+            headers=headers or {},
+        )
+
+    async def list_collections(self) -> list[str]:
+        """List all available index names in the Elasticsearch cluster."""
+        aliases = await self.client.indices.get_alias()
+        return list(aliases.keys())
+
+    async def create_collection(self, collection_name: str, **kwargs):
+        """Create a new index with dense vector mappings for kNN search.
+
+        Args:
+            collection_name: Name of the index to create.
+            **kwargs: Settings like dimensions, similarity, shards, and replicas.
+        """
+        collection_name = collection_name.lower()
+
+        if await self.client.indices.exists(index=collection_name):
+            return
+
+        dimensions = kwargs.get("dimensions", self.embedding_model.dimensions)
+        similarity = kwargs.get("similarity", "cosine")
+        number_of_shards = kwargs.get("number_of_shards", 5)
+        number_of_replicas = kwargs.get("number_of_replicas", 1)
+        refresh_interval = kwargs.get("refresh_interval", "1s")
+
+        index_settings = {
+            "settings": {
+                "index": {
+                    "number_of_replicas": number_of_replicas,
+                    "number_of_shards": number_of_shards,
+                    "refresh_interval": refresh_interval,
+                },
+            },
+            "mappings": {
+                "properties": {
+                    "vector_id": {"type": "keyword"},
+                    "content": {"type": "text"},
+                    "vector": {
+                        "type": "dense_vector",
+                        "dims": dimensions,
+                        "index": True,
+                        "similarity": similarity,
+                    },
+                    "metadata": {"type": "object", "enabled": True},
+                },
+            },
+        }
+
+        if not await self.client.indices.exists(index=collection_name):
+            await self.client.indices.create(index=collection_name, body=index_settings)
+            logger.info(f"Created index {collection_name} with dimensions={dimensions}")
+        else:
+            logger.info(f"Index {collection_name} already exists")
+
+    async def delete_collection(self, collection_name: str, **kwargs):
+        """Permanently delete an Elasticsearch index.
+
+        Args:
+            collection_name: Name of the index to delete.
+            **kwargs: Additional parameters for the deletion request.
+        """
+        collection_name = collection_name.lower()
+
+        if await self.client.indices.exists(index=collection_name):
+            await self.client.indices.delete(index=collection_name)
+            logger.info(f"Deleted index {collection_name}")
+        else:
+            logger.warning(f"Index {collection_name} does not exist")
+
+    async def copy_collection(self, collection_name: str, **kwargs):
+        """Reindex the current collection into a new index with identical mappings.
+
+        Args:
+            collection_name: Name of the destination index.
+            **kwargs: Additional parameters for the reindexing process.
+        """
+        collection_name = collection_name.lower()
+
+        current_index = await self.client.indices.get(index=self.collection_name)
+        current_settings = current_index[self.collection_name]
+
+        settings_to_copy = current_settings.get("settings", {}).copy()
+        if "index" in settings_to_copy:
+            index_settings = settings_to_copy["index"].copy()
+            internal_keys = [
+                "uuid",
+                "creation_date",
+                "provided_name",
+                "version",
+                "store",
+                "routing",
+                "replication",
+            ]
+            for key in internal_keys:
+                index_settings.pop(key, None)
+            settings_to_copy["index"] = index_settings
+
+        await self.client.indices.create(
+            index=collection_name,
+            body={
+                "settings": settings_to_copy,
+                "mappings": current_settings.get("mappings", {}),
+            },
+        )
+
+        await self.client.reindex(
+            body={
+                "source": {"index": self.collection_name},
+                "dest": {"index": collection_name},
+            },
+        )
+
+        logger.info(f"Copied collection {self.collection_name} to {collection_name}")
+
+    async def insert(self, nodes: VectorNode | list[VectorNode], refresh: bool = True, **kwargs):
+        """Insert nodes into the index, generating embeddings if missing.
+
+        Args:
+            nodes: Single or multiple VectorNode objects to index.
+            refresh: If True, makes the operation visible to search immediately.
+            **kwargs: Additional insertion options.
+        """
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_insert = [vector_map.get(n.vector_id, n) if n.vector is None else n for n in nodes]
+        else:
+            nodes_to_insert = nodes
+
+        actions = []
+        for node in nodes_to_insert:
+            action = {
+                "_index": self.collection_name,
+                "_id": node.vector_id,
+                "_source": {
+                    "vector_id": node.vector_id,
+                    "content": node.content,
+                    "vector": node.vector,
+                    "metadata": node.metadata,
+                },
+            }
+            actions.append(action)
+
+        success, failed = await async_bulk(self.client, actions, raise_on_error=False)
+
+        if failed:
+            logger.warning(f"Failed to insert {len(failed)} documents")
+
+        logger.info(f"Inserted {success} documents into {self.collection_name}")
+
+        if refresh:
+            await self.client.indices.refresh(index=self.collection_name)
+
+    async def search(
+        self,
+        query: str,
+        limit: int = 5,
+        filters: dict | None = None,
+        **kwargs,
+    ) -> list[VectorNode]:
+        """Perform a kNN similarity search based on a text query.
+
+        Args:
+            query: The text to search for.
+            limit: Maximum number of nearest neighbors to return.
+            filters: Metadata filters for exact match or 'IN' operations.
+            **kwargs: Search parameters like num_candidates or score_threshold.
+
+        Returns:
+            List of VectorNode objects ordered by similarity.
+        """
+        query_vector = await self.get_embedding(query)
+        num_candidates = kwargs.get("num_candidates", limit * 2)
+
+        search_query: dict = {
+            "knn": {
+                "field": "vector",
+                "query_vector": query_vector,
+                "k": limit,
+                "num_candidates": num_candidates,
+            },
+            "size": limit,
+        }
+
+        if filters:
+            filter_conditions = []
+            for key, value in filters.items():
+                if isinstance(value, list):
+                    filter_conditions.append({"terms": {f"metadata.{key}": value}})
+                else:
+                    filter_conditions.append({"term": {f"metadata.{key}": value}})
+            search_query["knn"]["filter"] = {"bool": {"must": filter_conditions}}
+
+        response = await self.client.search(index=self.collection_name, body=search_query)
+
+        results = []
+        for hit in response["hits"]["hits"]:
+            source = hit["_source"]
+            node = VectorNode(
+                vector_id=source.get("vector_id", hit["_id"]),
+                content=source.get("content", ""),
+                vector=source.get("vector"),
+                metadata=source.get("metadata", {}),
+            )
+            node.metadata["_score"] = hit["_score"]
+            results.append(node)
+
+        return results
+
+    async def delete(self, vector_ids: str | list[str], refresh: bool = True, **kwargs):
+        """Delete specific vectors from the index by their IDs.
+
+        Args:
+            vector_ids: Single ID or list of IDs to remove.
+            refresh: If True, refreshes the index after deletion.
+            **kwargs: Additional deletion parameters.
+        """
+        if isinstance(vector_ids, str):
+            vector_ids = [vector_ids]
+
+        actions = []
+        for vector_id in vector_ids:
+            actions.append(
+                {
+                    "_op_type": "delete",
+                    "_index": self.collection_name,
+                    "_id": vector_id,
+                },
+            )
+
+        success, failed = await async_bulk(
+            self.client,
+            actions,
+            raise_on_error=False,
+            raise_on_exception=False,
+        )
+
+        if failed:
+            logger.warning(f"Failed to delete {len(failed)} documents")
+
+        logger.info(f"Deleted {success} documents from {self.collection_name}")
+
+        if refresh:
+            await self.client.indices.refresh(index=self.collection_name)
+
+    async def update(self, nodes: VectorNode | list[VectorNode], refresh: bool = True, **kwargs):
+        """Update existing documents with new content or metadata.
+
+        Args:
+            nodes: Single or multiple VectorNode objects with updated data.
+            refresh: If True, refreshes the index after update.
+            **kwargs: Additional update parameters.
+        """
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None and node.content]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_update = [vector_map.get(n.vector_id, n) if n.vector is None and n.content else n for n in nodes]
+        else:
+            nodes_to_update = nodes
+
+        actions = []
+        for node in nodes_to_update:
+            doc = {
+                "vector_id": node.vector_id,
+                "content": node.content,
+                "metadata": node.metadata,
+            }
+            if node.vector is not None:
+                doc["vector"] = node.vector
+
+            actions.append(
+                {
+                    "_op_type": "update",
+                    "_index": self.collection_name,
+                    "_id": node.vector_id,
+                    "doc": doc,
+                },
+            )
+
+        success, failed = await async_bulk(
+            self.client,
+            actions,
+            raise_on_error=False,
+            raise_on_exception=False,
+        )
+
+        if failed:
+            logger.warning(f"Failed to update {len(failed)} documents")
+
+        logger.info(f"Updated {success} documents in {self.collection_name}")
+
+        if refresh:
+            await self.client.indices.refresh(index=self.collection_name)
+
+    async def get(self, vector_ids: str | list[str]) -> VectorNode | list[VectorNode]:
+        """Fetch documents by their IDs from the current index.
+
+        Args:
+            vector_ids: Single ID or list of IDs to retrieve.
+
+        Returns:
+            A single VectorNode or a list of VectorNodes.
+        """
+        single_result = isinstance(vector_ids, str)
+        if single_result:
+            vector_ids = [vector_ids]
+
+        response = await self.client.mget(
+            index=self.collection_name,
+            body={"ids": vector_ids},
+        )
+
+        results = []
+        for doc in response["docs"]:
+            if doc.get("found"):
+                source = doc["_source"]
+                node = VectorNode(
+                    vector_id=source.get("vector_id", doc["_id"]),
+                    content=source.get("content", ""),
+                    vector=source.get("vector"),
+                    metadata=source.get("metadata", {}),
+                )
+                results.append(node)
+            else:
+                logger.warning(f"Document with ID {doc['_id']} not found")
+
+        return results[0] if single_result and results else results
+
+    async def list(
+        self,
+        filters: dict | None = None,
+        limit: int | None = None,
+    ) -> list[VectorNode]:
+        """Retrieve a list of nodes filtered by metadata or limit.
+
+        Args:
+            filters: Optional metadata filtering criteria.
+            limit: Maximum number of nodes to return.
+
+        Returns:
+            A list of matching VectorNode objects.
+        """
+        query: dict[str, Any] = {"query": {"match_all": {}}}
+
+        if filters:
+            filter_conditions = []
+            for key, value in filters.items():
+                if isinstance(value, list):
+                    filter_conditions.append({"terms": {f"metadata.{key}": value}})
+                else:
+                    filter_conditions.append({"term": {f"metadata.{key}": value}})
+            query["query"] = {"bool": {"must": filter_conditions}}
+
+        if limit:
+            query["size"] = limit
+        else:
+            query["size"] = 10000
+
+        response = await self.client.search(index=self.collection_name, body=query)
+
+        results = []
+        for hit in response["hits"]["hits"]:
+            source = hit["_source"]
+            node = VectorNode(
+                vector_id=source.get("vector_id", hit["_id"]),
+                content=source.get("content", ""),
+                vector=source.get("vector"),
+                metadata=source.get("metadata", {}),
+            )
+            results.append(node)
+
+        return results
+
+    async def close(self):
+        """Terminate the Elasticsearch client session and release resources."""
+        await self.client.close()
+        logger.info("Elasticsearch client connection closed")

reme_ai/core/vector_store/local_vector_store.py

@@ -0,0 +1,281 @@
+"""Local file system vector store implementation for ReMe."""
+
+import json
+from pathlib import Path
+
+from loguru import logger
+
+from .base_vector_store import BaseVectorStore
+from ..context import C
+from ..embedding import BaseEmbeddingModel
+from ..schema import VectorNode
+
+
+@C.register_vector_store("local")
+class LocalVectorStore(BaseVectorStore):
+    """Local file system-based vector store using JSON files and manual cosine similarity."""
+
+    def __init__(
+        self,
+        collection_name: str,
+        embedding_model: BaseEmbeddingModel,
+        root_path: str = "./local_vector_store",
+        **kwargs,
+    ):
+        """Initialize the local vector store with a root path and collection name."""
+        super().__init__(collection_name=collection_name, embedding_model=embedding_model, **kwargs)
+        self.root_path = Path(root_path)
+        self.collection_path = self.root_path / collection_name
+        self.root_path.mkdir(parents=True, exist_ok=True)
+
+    def _get_collection_path(self, collection_name: str) -> Path:
+        """Get the file system path for a specific collection."""
+        return self.root_path / collection_name
+
+    def _get_node_file_path(self, vector_id: str, collection_name: str | None = None) -> Path:
+        """Get the JSON file path for a specific vector node."""
+        col_path = self._get_collection_path(collection_name or self.collection_name)
+        return col_path / f"{vector_id}.json"
+
+    def _save_node(self, node: VectorNode, collection_name: str | None = None):
+        """Save a vector node to a JSON file on disk."""
+        file_path = self._get_node_file_path(node.vector_id, collection_name)
+        file_path.parent.mkdir(parents=True, exist_ok=True)
+
+        with open(file_path, "w", encoding="utf-8") as f:
+            json.dump(node.model_dump(), f, ensure_ascii=False, indent=2)
+
+    def _load_node(self, vector_id: str, collection_name: str | None = None) -> VectorNode | None:
+        """Load a vector node from a JSON file."""
+        file_path = self._get_node_file_path(vector_id, collection_name)
+
+        if not file_path.exists():
+            return None
+
+        with open(file_path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+            return VectorNode(**data)
+
+    def _load_all_nodes(self, collection_name: str | None = None) -> list[VectorNode]:
+        """Load all vector nodes existing in a collection."""
+        col_path = self._get_collection_path(collection_name or self.collection_name)
+
+        if not col_path.exists():
+            return []
+
+        nodes = []
+        for file_path in col_path.glob("*.json"):
+            try:
+                with open(file_path, "r", encoding="utf-8") as f:
+                    data = json.load(f)
+                    nodes.append(VectorNode(**data))
+            except Exception as e:
+                logger.warning(f"Failed to load node from {file_path}: {e}")
+
+        return nodes
+
+    @staticmethod
+    def _cosine_similarity(vec1: list[float], vec2: list[float]) -> float:
+        """Calculate the cosine similarity between two numeric vectors."""
+        if len(vec1) != len(vec2):
+            raise ValueError(f"Vectors must have same length: {len(vec1)} != {len(vec2)}")
+
+        dot_product = sum(a * b for a, b in zip(vec1, vec2))
+        magnitude1 = sum(a * a for a in vec1) ** 0.5
+        magnitude2 = sum(b * b for b in vec2) ** 0.5
+
+        if magnitude1 == 0 or magnitude2 == 0:
+            return 0.0
+
+        return dot_product / (magnitude1 * magnitude2)
+
+    @staticmethod
+    def _match_filters(node: VectorNode, filters: dict | None) -> bool:
+        """Check if a vector node matches the provided metadata filters."""
+        if not filters:
+            return True
+
+        for key, value in filters.items():
+            node_value = node.metadata.get(key)
+
+            if isinstance(value, list):
+                if node_value not in value:
+                    return False
+            else:
+                if node_value != value:
+                    return False
+
+        return True
+
+    async def list_collections(self) -> list[str]:
+        """List all collection directories in the root path."""
+        if not self.root_path.exists():
+            return []
+
+        return [d.name for d in self.root_path.iterdir() if d.is_dir() and not d.name.startswith(".")]
+
+    async def create_collection(self, collection_name: str, **kwargs):
+        """Create a new collection directory."""
+        col_path = self._get_collection_path(collection_name)
+        col_path.mkdir(parents=True, exist_ok=True)
+        logger.info(f"Created collection {collection_name} at {col_path}")
+
+    async def delete_collection(self, collection_name: str, **kwargs):
+        """Delete a collection directory and all its JSON files."""
+        col_path = self._get_collection_path(collection_name)
+
+        if not col_path.exists():
+            logger.warning(f"Collection {collection_name} does not exist")
+            return
+
+        for file_path in col_path.glob("*.json"):
+            file_path.unlink()
+
+        col_path.rmdir()
+        logger.info(f"Deleted collection {collection_name}")
+
+    async def copy_collection(self, collection_name: str, **kwargs):
+        """Copy all nodes from the current collection to a new one."""
+        source_path = self._get_collection_path(self.collection_name)
+        target_path = self._get_collection_path(collection_name)
+
+        if not source_path.exists():
+            logger.warning(f"Source collection {self.collection_name} does not exist")
+            return
+
+        target_path.mkdir(parents=True, exist_ok=True)
+
+        for file_path in source_path.glob("*.json"):
+            target_file = target_path / file_path.name
+            target_file.write_text(file_path.read_text(encoding="utf-8"), encoding="utf-8")
+
+        logger.info(f"Copied collection {self.collection_name} to {collection_name}")
+
+    async def insert(self, nodes: VectorNode | list[VectorNode], **kwargs):
+        """Insert vector nodes into the local store, generating embeddings if necessary."""
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_insert = [vector_map.get(n.vector_id, n) if n.vector is None else n for n in nodes]
+        else:
+            nodes_to_insert = nodes
+
+        for node in nodes_to_insert:
+            self._save_node(node)
+
+        logger.info(f"Inserted {len(nodes_to_insert)} nodes into {self.collection_name}")
+
+    async def search(
+        self,
+        query: str,
+        limit: int = 5,
+        filters: dict | None = None,
+        **kwargs,
+    ) -> list[VectorNode]:
+        """Search for nodes similar to the query using brute-force cosine similarity."""
+        query_vector = await self.get_embedding(query)
+        all_nodes = self._load_all_nodes()
+        filtered_nodes = [node for node in all_nodes if self._match_filters(node, filters)]
+
+        scored_nodes = []
+        for node in filtered_nodes:
+            if node.vector is None:
+                logger.warning(f"Node {node.vector_id} has no vector, skipping")
+                continue
+
+            try:
+                score = self._cosine_similarity(query_vector, node.vector)
+                scored_nodes.append((node, score))
+            except ValueError as e:
+                logger.warning(f"Failed to calculate similarity for node {node.vector_id}: {e}")
+
+        scored_nodes.sort(key=lambda x: x[1], reverse=True)
+
+        score_threshold = kwargs.get("score_threshold")
+        if score_threshold is not None:
+            scored_nodes = [(node, score) for node, score in scored_nodes if score >= score_threshold]
+
+        scored_nodes = scored_nodes[:limit]
+        results = []
+        for node, score in scored_nodes:
+            node.metadata["_score"] = score
+            results.append(node)
+
+        return results
+
+    async def delete(self, vector_ids: str | list[str], **kwargs):
+        """Delete specific vector nodes by their IDs."""
+        if isinstance(vector_ids, str):
+            vector_ids = [vector_ids]
+
+        deleted_count = 0
+        for vector_id in vector_ids:
+            file_path = self._get_node_file_path(vector_id)
+            if file_path.exists():
+                file_path.unlink()
+                deleted_count += 1
+            else:
+                logger.warning(f"Node {vector_id} does not exist")
+
+        logger.info(f"Deleted {deleted_count} nodes from {self.collection_name}")
+
+    async def update(self, nodes: VectorNode | list[VectorNode], **kwargs):
+        """Update existing vector nodes with new data or embeddings."""
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None and node.content]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_update = [vector_map.get(n.vector_id, n) if n.vector is None and n.content else n for n in nodes]
+        else:
+            nodes_to_update = nodes
+
+        updated_count = 0
+        for node in nodes_to_update:
+            file_path = self._get_node_file_path(node.vector_id)
+            if file_path.exists():
+                self._save_node(node)
+                updated_count += 1
+            else:
+                logger.warning(f"Node {node.vector_id} does not exist, skipping update")
+
+        logger.info(f"Updated {updated_count} nodes in {self.collection_name}")
+
+    async def get(self, vector_ids: str | list[str]) -> VectorNode | list[VectorNode]:
+        """Retrieve one or more vector nodes by their unique IDs."""
+        is_single = isinstance(vector_ids, str)
+        ids = [vector_ids] if is_single else vector_ids
+
+        results = []
+        for vector_id in ids:
+            node = self._load_node(vector_id)
+            if node:
+                results.append(node)
+            else:
+                logger.warning(f"Node {vector_id} not found")
+
+        return results[0] if is_single and results else results
+
+    async def list(
+        self,
+        filters: dict | None = None,
+        limit: int | None = None,
+    ) -> list[VectorNode]:
+        """List vector nodes in the collection with optional filtering and limits."""
+        all_nodes = self._load_all_nodes()
+        filtered_nodes = [node for node in all_nodes if self._match_filters(node, filters)]
+
+        if limit is not None:
+            filtered_nodes = filtered_nodes[:limit]
+
+        return filtered_nodes
+
+    async def close(self):
+        """Close the vector store (no-op for local file system)."""
+        logger.info("Local vector store closed")

reme_ai/core/vector_store/pgvector_store.py

@@ -0,0 +1,533 @@
+"""PostgreSQL pgvector implementation for vector storage and retrieval."""
+
+import json
+from typing import Any
+
+from loguru import logger
+
+from .base_vector_store import BaseVectorStore
+from ..context import C
+from ..embedding import BaseEmbeddingModel
+from ..schema import VectorNode
+
+_ASYNCPG_IMPORT_ERROR = None
+
+try:
+    import asyncpg
+    from asyncpg import Pool
+except ImportError as e:
+    _ASYNCPG_IMPORT_ERROR = e
+    asyncpg = None
+    Pool = None
+
+
+@C.register_vector_store("pgvector")
+class PGVectorStore(BaseVectorStore):
+    """Vector store implementation using PostgreSQL and pgvector for efficient similarity search."""
+
+    def __init__(
+        self,
+        collection_name: str,
+        embedding_model: BaseEmbeddingModel,
+        host: str = "localhost",
+        port: int = 5432,
+        database: str = "postgres",
+        user: str = "postgres",
+        password: str = "",
+        min_size: int = 1,
+        max_size: int = 10,
+        dsn: str | None = None,
+        use_hnsw: bool = True,
+        use_diskann: bool = False,
+        **kwargs,
+    ):
+        """Initialize the PGVector store with connection parameters and index settings."""
+        if _ASYNCPG_IMPORT_ERROR is not None:
+            raise ImportError(
+                "PGVector requires extra dependencies. Install with `pip install asyncpg pgvector`",
+            ) from _ASYNCPG_IMPORT_ERROR
+
+        super().__init__(collection_name=collection_name, embedding_model=embedding_model, **kwargs)
+
+        self.dsn = dsn
+        self.host = host
+        self.port = port
+        self.database = database
+        self.user = user
+        self.password = password
+        self.min_size = min_size
+        self.max_size = max_size
+        self.use_hnsw = use_hnsw
+        self.use_diskann = use_diskann
+        self._pool: Pool | None = None
+        self.embedding_model_dims = embedding_model.dimensions
+
+    async def _get_pool(self) -> Pool:
+        """Create or return the existing asyncpg connection pool."""
+        if self._pool is None:
+            if self.dsn:
+                self._pool = await asyncpg.create_pool(
+                    dsn=self.dsn,
+                    min_size=self.min_size,
+                    max_size=self.max_size,
+                )
+            else:
+                self._pool = await asyncpg.create_pool(
+                    host=self.host,
+                    port=self.port,
+                    database=self.database,
+                    user=self.user,
+                    password=self.password,
+                    min_size=self.min_size,
+                    max_size=self.max_size,
+                )
+
+            async with self._pool.acquire() as conn:
+                await conn.execute("CREATE EXTENSION IF NOT EXISTS vector")
+
+            logger.info(f"PGVector connection pool created for database {self.database}")
+
+        return self._pool
+
+    async def _ensure_collection_exists(self):
+        """Check if the collection table exists and create it if missing."""
+        collections = await self.list_collections()
+        if self.collection_name not in collections:
+            await self.create_collection(self.collection_name)
+
+    async def list_collections(self) -> list[str]:
+        """List all available table names in the current database."""
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            rows = await conn.fetch(
+                "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'",
+            )
+            return [row["table_name"] for row in rows]
+
+    async def create_collection(self, collection_name: str, **kwargs):
+        """Create a new PostgreSQL table with vector support and appropriate indexing."""
+        pool = await self._get_pool()
+        dimensions = kwargs.get("dimensions", self.embedding_model_dims)
+
+        async with pool.acquire() as conn:
+            await conn.execute(
+                f"""
+                CREATE TABLE IF NOT EXISTS {collection_name} (
+                    id TEXT PRIMARY KEY,
+                    content TEXT,
+                    vector vector({dimensions}),
+                    metadata JSONB
+                )
+            """,
+            )
+
+            if self.use_diskann and dimensions < 2000:
+                result = await conn.fetchval(
+                    "SELECT 1 FROM pg_extension WHERE extname = 'vectorscale'",
+                )
+                if result:
+                    await conn.execute(
+                        f"""
+                        CREATE INDEX IF NOT EXISTS {collection_name}_diskann_idx
+                        ON {collection_name}
+                        USING diskann (vector)
+                    """,
+                    )
+                    logger.info(f"Created DiskANN index for collection {collection_name}")
+                else:
+                    logger.warning("vectorscale extension not available, skipping DiskANN index")
+            elif self.use_hnsw:
+                await conn.execute(
+                    f"""
+                    CREATE INDEX IF NOT EXISTS {collection_name}_hnsw_idx
+                    ON {collection_name}
+                    USING hnsw (vector vector_cosine_ops)
+                """,
+                )
+                logger.info(f"Created HNSW index for collection {collection_name}")
+
+        logger.info(f"Created collection {collection_name} with dimensions={dimensions}")
+
+    async def delete_collection(self, collection_name: str, **kwargs):
+        """Remove the specified collection table from the database."""
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            await conn.execute(f"DROP TABLE IF EXISTS {collection_name}")
+        logger.info(f"Deleted collection {collection_name}")
+
+    async def copy_collection(self, collection_name: str, **kwargs):
+        """Duplicate the structure and content of the current collection to a new table."""
+        pool = await self._get_pool()
+
+        async with pool.acquire() as conn:
+            columns = await conn.fetch(
+                """
+                SELECT column_name, data_type, udt_name
+                FROM information_schema.columns
+                WHERE table_name = $1 AND table_schema = 'public'
+            """,
+                self.collection_name,
+            )
+
+            if not columns:
+                raise ValueError(f"Source collection {self.collection_name} does not exist")
+
+            await conn.execute(f"CREATE TABLE {collection_name} AS TABLE {self.collection_name}")
+            await conn.execute(f"ALTER TABLE {collection_name} ADD PRIMARY KEY (id)")
+
+            if self.use_hnsw:
+                await conn.execute(
+                    f"""
+                    CREATE INDEX IF NOT EXISTS {collection_name}_hnsw_idx
+                    ON {collection_name}
+                    USING hnsw (vector vector_cosine_ops)
+                """,
+                )
+
+        logger.info(f"Copied collection {self.collection_name} to {collection_name}")
+
+    async def insert(self, nodes: VectorNode | list[VectorNode], **kwargs):
+        """Insert or upsert vector nodes into the PostgreSQL collection."""
+        await self._ensure_collection_exists()
+
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        if not nodes:
+            return
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_insert = [vector_map.get(n.vector_id, n) if n.vector is None else n for n in nodes]
+        else:
+            nodes_to_insert = nodes
+
+        pool = await self._get_pool()
+        data = [
+            (
+                node.vector_id,
+                node.content,
+                f"[{','.join(map(str, node.vector))}]",
+                json.dumps(node.metadata),
+            )
+            for node in nodes_to_insert
+        ]
+
+        async with pool.acquire() as conn:
+            on_conflict = kwargs.get("on_conflict", "update")
+
+            if on_conflict == "update":
+                await conn.executemany(
+                    f"""
+                    INSERT INTO {self.collection_name} (id, content, vector, metadata)
+                    VALUES ($1, $2, $3::vector, $4::jsonb)
+                    ON CONFLICT (id) DO UPDATE SET
+                        content = EXCLUDED.content,
+                        vector = EXCLUDED.vector,
+                        metadata = EXCLUDED.metadata
+                """,
+                    data,
+                )
+            elif on_conflict == "ignore":
+                await conn.executemany(
+                    f"""
+                    INSERT INTO {self.collection_name} (id, content, vector, metadata)
+                    VALUES ($1, $2, $3::vector, $4::jsonb)
+                    ON CONFLICT (id) DO NOTHING
+                """,
+                    data,
+                )
+            else:
+                await conn.executemany(
+                    f"""
+                    INSERT INTO {self.collection_name} (id, content, vector, metadata)
+                    VALUES ($1, $2, $3::vector, $4::jsonb)
+                """,
+                    data,
+                )
+
+        logger.info(f"Inserted {len(nodes_to_insert)} documents into {self.collection_name}")
+
+    @staticmethod
+    def _build_filter_clause(filters: dict | None) -> tuple[str, list]:
+        """Generate an SQL WHERE clause and parameter list from a filter dictionary."""
+        if not filters:
+            return "", []
+
+        conditions = []
+        params = []
+        param_idx = 1
+
+        for key, value in filters.items():
+            if isinstance(value, list):
+                placeholders = ", ".join([f"${param_idx + i}" for i in range(len(value))])
+                conditions.append(f"metadata->>'{key}' IN ({placeholders})")
+                params.extend([str(v) for v in value])
+                param_idx += len(value)
+            else:
+                conditions.append(f"metadata->>'{key}' = ${param_idx}")
+                params.append(str(value))
+                param_idx += 1
+
+        filter_clause = "WHERE " + " AND ".join(conditions) if conditions else ""
+        return filter_clause, params
+
+    async def search(
+        self,
+        query: str,
+        limit: int = 5,
+        filters: dict | None = None,
+        **kwargs,
+    ) -> list[VectorNode]:
+        """Perform vector similarity search with optional metadata filtering."""
+        await self._ensure_collection_exists()
+
+        query_vector = await self.get_embedding(query)
+        vector_str = f"[{','.join(map(str, query_vector))}]"
+        pool = await self._get_pool()
+
+        filter_clause, filter_params = self._build_filter_clause(filters)
+
+        if filter_clause:
+            for i in range(len(filter_params)):
+                old_idx = i + 1
+                new_idx = i + 2
+                filter_clause = filter_clause.replace(f"${old_idx}", f"${new_idx}", 1)
+
+        async with pool.acquire() as conn:
+            sql = f"""
+                SELECT id, content, vector, metadata, vector <=> $1::vector AS distance
+                FROM {self.collection_name}
+                {filter_clause}
+                ORDER BY distance
+                LIMIT ${len(filter_params) + 2}
+            """
+            rows = await conn.fetch(sql, vector_str, *filter_params, limit)
+
+        results = []
+        score_threshold = kwargs.get("score_threshold")
+
+        for row in rows:
+            distance = row["distance"]
+            if score_threshold is not None and distance > score_threshold:
+                continue
+
+            vector_data = None
+            if row["vector"]:
+                vector_str_raw = str(row["vector"])
+                if vector_str_raw.startswith("[") and vector_str_raw.endswith("]"):
+                    vector_data = [float(x) for x in vector_str_raw[1:-1].split(",")]
+
+            metadata = row["metadata"] if row["metadata"] else {}
+            if isinstance(metadata, str):
+                metadata = json.loads(metadata)
+
+            metadata["_score"] = 1 - distance
+            metadata["_distance"] = distance
+
+            node = VectorNode(
+                vector_id=row["id"],
+                content=row["content"] or "",
+                vector=vector_data,
+                metadata=metadata,
+            )
+            results.append(node)
+
+        return results
+
+    async def delete(self, vector_ids: str | list[str], **kwargs):
+        """Remove specific vector records from the collection by their IDs."""
+        await self._ensure_collection_exists()
+
+        if isinstance(vector_ids, str):
+            vector_ids = [vector_ids]
+
+        if not vector_ids:
+            return
+
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            placeholders = ", ".join([f"${i + 1}" for i in range(len(vector_ids))])
+            await conn.execute(
+                f"DELETE FROM {self.collection_name} WHERE id IN ({placeholders})",
+                *vector_ids,
+            )
+
+        logger.info(f"Deleted {len(vector_ids)} documents from {self.collection_name}")
+
+    async def update(self, nodes: VectorNode | list[VectorNode], **kwargs):
+        """Update existing vector nodes with new content, embeddings, or metadata."""
+        await self._ensure_collection_exists()
+
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        if not nodes:
+            return
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None and node.content]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_update = [vector_map.get(n.vector_id, n) if n.vector is None and n.content else n for n in nodes]
+        else:
+            nodes_to_update = nodes
+
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            for node in nodes_to_update:
+                update_fields = []
+                params = []
+                idx = 1
+
+                if node.content:
+                    update_fields.append(f"content = ${idx}")
+                    params.append(node.content)
+                    idx += 1
+
+                if node.vector:
+                    vector_str = f"[{','.join(map(str, node.vector))}]"
+                    update_fields.append(f"vector = ${idx}::vector")
+                    params.append(vector_str)
+                    idx += 1
+
+                if node.metadata:
+                    update_fields.append(f"metadata = ${idx}::jsonb")
+                    params.append(json.dumps(node.metadata))
+                    idx += 1
+
+                if update_fields:
+                    params.append(node.vector_id)
+                    await conn.execute(
+                        f"UPDATE {self.collection_name} SET {', '.join(update_fields)} WHERE id = ${idx}",
+                        *params,
+                    )
+
+        logger.info(f"Updated {len(nodes_to_update)} documents in {self.collection_name}")
+
+    async def get(self, vector_ids: str | list[str]) -> VectorNode | list[VectorNode] | None:
+        """Retrieve vector nodes by their unique identifiers."""
+        await self._ensure_collection_exists()
+
+        single_result = isinstance(vector_ids, str)
+        if single_result:
+            vector_ids = [vector_ids]
+
+        if not vector_ids:
+            return [] if not single_result else None
+
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            placeholders = ", ".join([f"${i + 1}" for i in range(len(vector_ids))])
+            rows = await conn.fetch(
+                f"SELECT id, content, vector, metadata FROM {self.collection_name} WHERE id IN ({placeholders})",
+                *vector_ids,
+            )
+
+        results = []
+        for row in rows:
+            vector_data = None
+            if row["vector"]:
+                vector_str_raw = str(row["vector"])
+                if vector_str_raw.startswith("[") and vector_str_raw.endswith("]"):
+                    vector_data = [float(x) for x in vector_str_raw[1:-1].split(",")]
+
+            metadata = row["metadata"] if row["metadata"] else {}
+            if isinstance(metadata, str):
+                metadata = json.loads(metadata)
+
+            results.append(
+                VectorNode(
+                    vector_id=row["id"],
+                    content=row["content"] or "",
+                    vector=vector_data,
+                    metadata=metadata,
+                ),
+            )
+
+        if single_result:
+            return results[0] if results else None
+        return results
+
+    async def list(
+        self,
+        filters: dict | None = None,
+        limit: int | None = None,
+    ) -> list[VectorNode]:
+        """Return a list of vector nodes matching the provided filters and limit."""
+        await self._ensure_collection_exists()
+
+        pool = await self._get_pool()
+        filter_clause, filter_params = self._build_filter_clause(filters)
+
+        limit_clause = ""
+        if limit:
+            limit_clause = f"LIMIT ${len(filter_params) + 1}"
+            filter_params.append(limit)
+
+        async with pool.acquire() as conn:
+            sql = f"""
+                SELECT id, content, vector, metadata
+                FROM {self.collection_name}
+                {filter_clause}
+                {limit_clause}
+            """
+            rows = await conn.fetch(sql, *filter_params)
+
+        results = []
+        for row in rows:
+            vector_data = None
+            if row["vector"]:
+                vector_str_raw = str(row["vector"])
+                if vector_str_raw.startswith("[") and vector_str_raw.endswith("]"):
+                    vector_data = [float(x) for x in vector_str_raw[1:-1].split(",")]
+
+            metadata = row["metadata"] if row["metadata"] else {}
+            if isinstance(metadata, str):
+                metadata = json.loads(metadata)
+
+            results.append(
+                VectorNode(
+                    vector_id=row["id"],
+                    content=row["content"] or "",
+                    vector=vector_data,
+                    metadata=metadata,
+                ),
+            )
+
+        return results
+
+    async def collection_info(self) -> dict[str, Any]:
+        """Fetch metadata including record count and disk usage for the collection."""
+        pool = await self._get_pool()
+
+        async with pool.acquire() as conn:
+            row = await conn.fetchrow(
+                f"""
+                SELECT
+                    '{self.collection_name}' as name,
+                    (SELECT COUNT(*) FROM {self.collection_name}) as row_count,
+                    pg_size_pretty(pg_total_relation_size('{self.collection_name}')) as total_size
+            """,
+            )
+
+        return {
+            "name": row["name"],
+            "count": row["row_count"],
+            "size": row["total_size"],
+        }
+
+    async def reset(self):
+        """Purge all data by dropping and recreating the collection table."""
+        logger.warning(f"Resetting collection {self.collection_name}...")
+        await self.delete_collection(self.collection_name)
+        await self.create_collection(self.collection_name)
+
+    async def close(self):
+        """Terminate the database connection pool and release associated resources."""
+        if self._pool is not None:
+            await self._pool.close()
+            self._pool = None
+            logger.info("PGVector connection pool closed")

reme_ai/core/vector_store/qdrant_vector_store.py

@@ -0,0 +1,444 @@
+"""Qdrant vector store implementation for the ReMe project."""
+
+from typing import Any
+
+from loguru import logger
+
+from .base_vector_store import BaseVectorStore
+from ..context import C
+from ..embedding import BaseEmbeddingModel
+from ..schema import VectorNode
+
+_QDRANT_IMPORT_ERROR = None
+
+try:
+    from qdrant_client import AsyncQdrantClient
+    from qdrant_client.models import (
+        Distance,
+        FieldCondition,
+        Filter,
+        MatchValue,
+        PointIdsList,
+        PointStruct,
+        Range,
+        VectorParams,
+    )
+except ImportError as e:
+    _QDRANT_IMPORT_ERROR = e
+    AsyncQdrantClient = None
+    Distance = None
+    FieldCondition = None
+    Filter = None
+    MatchValue = None
+    PointIdsList = None
+    PointStruct = None
+    Range = None
+    VectorParams = None
+
+
+@C.register_vector_store("qdrant")
+class QdrantVectorStore(BaseVectorStore):
+    """Vector store implementation using Qdrant for dense vector search."""
+
+    def __init__(
+        self,
+        collection_name: str,
+        embedding_model: BaseEmbeddingModel,
+        host: str | None = None,
+        port: int = 6333,
+        path: str | None = None,
+        url: str | None = None,
+        api_key: str | None = None,
+        https: bool | None = None,
+        grpc_port: int = 6334,
+        prefer_grpc: bool = False,
+        distance: str = "cosine",
+        on_disk: bool = False,
+        **kwargs: Any,
+    ):
+        """Initialize the Qdrant client and collection configuration.
+
+        Args:
+            collection_name: Name of the collection.
+            embedding_model: Model used for generating vector embeddings.
+            host: Server host address.
+            port: HTTP port for the server.
+            path: Local storage path for on-disk/in-memory mode.
+            url: Full connection URL.
+            api_key: Authentication key for Qdrant Cloud.
+            https: Use secure connection if True.
+            grpc_port: gRPC interface port.
+            prefer_grpc: Use gRPC instead of HTTP if True.
+            distance: Metric for similarity (cosine, euclid, dot).
+            on_disk: Enable persistent storage for vectors.
+            **kwargs: Additional client configuration.
+        """
+        if _QDRANT_IMPORT_ERROR is not None:
+            raise ImportError(
+                "Qdrant requires extra dependencies. Install with `pip install qdrant-client`",
+            ) from _QDRANT_IMPORT_ERROR
+
+        super().__init__(collection_name=collection_name, embedding_model=embedding_model, **kwargs)
+
+        self.client = AsyncQdrantClient(
+            host=host,
+            port=port,
+            path=path,
+            url=url,
+            api_key=api_key,
+            https=https,
+            grpc_port=grpc_port,
+            prefer_grpc=prefer_grpc,
+            **kwargs,
+        )
+
+        self.is_local = path is not None
+        distance_map = {
+            "cosine": Distance.COSINE,
+            "euclid": Distance.EUCLID,
+            "dot": Distance.DOT,
+        }
+        self.distance = distance_map.get(distance.lower(), Distance.COSINE)
+        self.on_disk = on_disk
+
+    async def list_collections(self) -> list[str]:
+        """Retrieve names of all existing collections in the Qdrant instance."""
+        collections = await self.client.get_collections()
+        return [collection.name for collection in collections.collections]
+
+    async def create_collection(self, collection_name: str, **kwargs: Any):
+        """Create a new collection with the specified vector configuration.
+
+        Args:
+            collection_name: Name of the collection to create.
+            **kwargs: Overrides for dimensions, distance, or on_disk settings.
+        """
+        collections = await self.list_collections()
+        if collection_name in collections:
+            logger.info(f"Collection {collection_name} already exists")
+            return
+
+        dimensions = kwargs.get("dimensions", self.embedding_model.dimensions)
+        distance = kwargs.get("distance", self.distance)
+        on_disk = kwargs.get("on_disk", self.on_disk)
+
+        await self.client.create_collection(
+            collection_name=collection_name,
+            vectors_config=VectorParams(
+                size=dimensions,
+                distance=distance,
+                on_disk=on_disk,
+            ),
+        )
+
+        logger.info(f"Created collection {collection_name} with dimensions={dimensions}")
+
+        if not self.is_local:
+            await self._create_payload_indexes(collection_name)
+
+    async def _create_payload_indexes(self, collection_name: str):
+        """Create keyword indexes for common metadata fields to optimize filtering."""
+        common_fields = ["user_id", "agent_id", "run_id", "actor_id", "source"]
+
+        for field in common_fields:
+            try:
+                await self.client.create_payload_index(
+                    collection_name=collection_name,
+                    field_name=field,
+                    field_schema="keyword",
+                )
+                logger.debug(f"Created index for {field} in collection {collection_name}")
+            except Exception as e:
+                logger.debug(f"Index for {field} might already exist: {e}")
+
+    async def delete_collection(self, collection_name: str, **kwargs: Any):
+        """Permanently remove a collection from the Qdrant instance."""
+        collections = await self.list_collections()
+        if collection_name in collections:
+            await self.client.delete_collection(collection_name=collection_name)
+            logger.info(f"Deleted collection {collection_name}")
+        else:
+            logger.warning(f"Collection {collection_name} does not exist")
+
+    async def copy_collection(self, collection_name: str, **kwargs: Any):
+        """Duplicate an existing collection to a new one including all data."""
+        collection_info = await self.client.get_collection(collection_name=self.collection_name)
+
+        await self.client.create_collection(
+            collection_name=collection_name,
+            vectors_config=collection_info.config.params.vectors,
+        )
+
+        offset = None
+        batch_size = 100
+
+        while True:
+            records, next_offset = await self.client.scroll(
+                collection_name=self.collection_name,
+                limit=batch_size,
+                offset=offset,
+                with_payload=True,
+                with_vectors=True,
+            )
+
+            if not records:
+                break
+
+            points = [
+                PointStruct(
+                    id=record.id,
+                    vector=record.vector,
+                    payload=record.payload,
+                )
+                for record in records
+            ]
+
+            await self.client.upsert(
+                collection_name=collection_name,
+                points=points,
+            )
+
+            offset = next_offset
+            if offset is None:
+                break
+
+        logger.info(f"Copied collection {self.collection_name} to {collection_name}")
+
+    async def insert(self, nodes: VectorNode | list[VectorNode], **kwargs: Any):
+        """Insert vector nodes into the collection, generating embeddings as needed."""
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_insert = [vector_map.get(n.vector_id, n) if n.vector is None else n for n in nodes]
+        else:
+            nodes_to_insert = nodes
+
+        points = []
+        for node in nodes_to_insert:
+            try:
+                point_id = int(node.vector_id)
+            except ValueError:
+                point_id = abs(hash(node.vector_id)) % (10**18)
+
+            point = PointStruct(
+                id=point_id,
+                vector=node.vector,
+                payload={
+                    "vector_id": node.vector_id,
+                    "content": node.content,
+                    "metadata": node.metadata,
+                },
+            )
+            points.append(point)
+
+        wait = kwargs.get("wait", True)
+        await self.client.upsert(
+            collection_name=self.collection_name,
+            points=points,
+            wait=wait,
+        )
+
+        logger.info(f"Inserted {len(points)} documents into {self.collection_name}")
+
+    @staticmethod
+    def _create_filter(filters: dict) -> Filter | None:
+        """Convert a dictionary of filter conditions into a Qdrant Filter object."""
+        if not filters:
+            return None
+
+        conditions = []
+        for key, value in filters.items():
+            if isinstance(value, dict) and ("gte" in value or "lte" in value):
+                range_params = {}
+                if "gte" in value:
+                    range_params["gte"] = value["gte"]
+                if "lte" in value:
+                    range_params["lte"] = value["lte"]
+                conditions.append(
+                    FieldCondition(
+                        key=f"metadata.{key}",
+                        range=Range(**range_params),
+                    ),
+                )
+            elif isinstance(value, list):
+                conditions.append(
+                    FieldCondition(key=f"metadata.{key}", match=MatchValue(value=value[0])),
+                )
+            else:
+                conditions.append(
+                    FieldCondition(key=f"metadata.{key}", match=MatchValue(value=value)),
+                )
+
+        return Filter(must=conditions) if conditions else None
+
+    async def search(
+        self,
+        query: str,
+        limit: int = 5,
+        filters: dict | None = None,
+        **kwargs: Any,
+    ) -> list[VectorNode]:
+        """Search for the most similar vectors based on a text query."""
+        query_vector = await self.get_embedding(query)
+        query_filter = self._create_filter(filters) if filters else None
+        score_threshold = kwargs.get("score_threshold", None)
+
+        results = await self.client.query_points(
+            collection_name=self.collection_name,
+            query=query_vector,
+            query_filter=query_filter,
+            limit=limit,
+            score_threshold=score_threshold,
+        )
+
+        nodes = []
+        for point in results.points:
+            payload = point.payload or {}
+            node = VectorNode(
+                vector_id=payload.get("vector_id", str(point.id)),
+                content=payload.get("content", ""),
+                vector=point.vector if hasattr(point, "vector") else None,
+                metadata=payload.get("metadata", {}),
+            )
+            node.metadata["_score"] = point.score
+            nodes.append(node)
+
+        return nodes
+
+    async def delete(self, vector_ids: str | list[str], **kwargs: Any):
+        """Delete specific vectors from the collection using their IDs."""
+        if isinstance(vector_ids, str):
+            vector_ids = [vector_ids]
+
+        point_ids = []
+        for vector_id in vector_ids:
+            try:
+                point_id = int(vector_id)
+            except ValueError:
+                point_id = abs(hash(vector_id)) % (10**18)
+            point_ids.append(point_id)
+
+        wait = kwargs.get("wait", True)
+        await self.client.delete(
+            collection_name=self.collection_name,
+            points_selector=PointIdsList(points=point_ids),
+            wait=wait,
+        )
+
+        logger.info(f"Deleted {len(point_ids)} documents from {self.collection_name}")
+
+    async def update(self, nodes: VectorNode | list[VectorNode], **kwargs: Any):
+        """Update existing vector nodes with new content or metadata."""
+        if isinstance(nodes, VectorNode):
+            nodes = [nodes]
+
+        nodes_without_vectors = [node for node in nodes if node.vector is None and node.content]
+        if nodes_without_vectors:
+            nodes_with_vectors = await self.get_node_embeddings(nodes_without_vectors)
+            vector_map = {n.vector_id: n for n in nodes_with_vectors}
+            nodes_to_update = [vector_map.get(n.vector_id, n) if n.vector is None and n.content else n for n in nodes]
+        else:
+            nodes_to_update = nodes
+
+        points = []
+        for node in nodes_to_update:
+            try:
+                point_id = int(node.vector_id)
+            except ValueError:
+                point_id = abs(hash(node.vector_id)) % (10**18)
+
+            point = PointStruct(
+                id=point_id,
+                vector=node.vector,
+                payload={
+                    "vector_id": node.vector_id,
+                    "content": node.content,
+                    "metadata": node.metadata,
+                },
+            )
+            points.append(point)
+
+        wait = kwargs.get("wait", True)
+        await self.client.upsert(
+            collection_name=self.collection_name,
+            points=points,
+            wait=wait,
+        )
+
+        logger.info(f"Updated {len(points)} documents in {self.collection_name}")
+
+    async def get(self, vector_ids: str | list[str]) -> VectorNode | list[VectorNode]:
+        """Retrieve vector nodes by their IDs from the collection."""
+        single_result = isinstance(vector_ids, str)
+        if single_result:
+            vector_ids = [vector_ids]
+
+        point_ids = []
+        for vector_id in vector_ids:
+            try:
+                point_id = int(vector_id)
+            except ValueError:
+                point_id = abs(hash(vector_id)) % (10**18)
+            point_ids.append(point_id)
+
+        points = await self.client.retrieve(
+            collection_name=self.collection_name,
+            ids=point_ids,
+            with_payload=True,
+            with_vectors=True,
+        )
+
+        results = []
+        for point in points:
+            if point:
+                payload = point.payload or {}
+                node = VectorNode(
+                    vector_id=payload.get("vector_id", str(point.id)),
+                    content=payload.get("content", ""),
+                    vector=point.vector,
+                    metadata=payload.get("metadata", {}),
+                )
+                results.append(node)
+            else:
+                logger.warning("Point not found")
+
+        return results[0] if single_result and results else results
+
+    async def list(
+        self,
+        filters: dict | None = None,
+        limit: int | None = None,
+    ) -> list[VectorNode]:
+        """List all vector nodes in the collection matching the filter criteria."""
+        scroll_filter = self._create_filter(filters) if filters else None
+
+        limit = limit or 10000
+        records, _ = await self.client.scroll(
+            collection_name=self.collection_name,
+            scroll_filter=scroll_filter,
+            limit=limit,
+            with_payload=True,
+            with_vectors=True,
+        )
+
+        results = []
+        for record in records:
+            payload = record.payload or {}
+            node = VectorNode(
+                vector_id=payload.get("vector_id", str(record.id)),
+                content=payload.get("content", ""),
+                vector=record.vector,
+                metadata=payload.get("metadata", {}),
+            )
+            results.append(node)
+
+        return results
+
+    async def close(self):
+        """Close the AsyncQdrantClient connection and release resources."""
+        await self.client.close()
+        logger.info("Qdrant client connection closed")