personal-consumption-service/app/client/VectorStoreClient.py

# @description: 
# @author: licanglong
# @date: 2025/12/19 16:08
import logging

from qdrant_client import AsyncQdrantClient
from sentence_transformers import SentenceTransformer

from app.core import CTX
from app.utils.pathutils import getpath

log = logging.getLogger(__name__)


class VectorStoreClient:
    def __init__(
            self,
            host: str = None,
            port: int = None,
            model_path: str = None,
    ):
        if not host and not port:
            raise ValueError("host and port cannot be empty")
        if not model_path:
            raise ValueError("model_path cannot be empty")
        # 1. Qdrant 客户端
        self.client = AsyncQdrantClient(host=host, port=port)
        # 2. 向量模型
        self.embedding = SentenceTransformer(model_path)

    async def create_collection(self, collection_name: str, **kwargs) -> bool:
        if await self.client.collection_exists(collection_name):
            return False

        await self.client.create_collection(
            collection_name=collection_name,
            **kwargs,
        )
        return True


vector_store_client = VectorStoreClient(host=CTX.ENV.getprop("qdrant.host"),
                                        port=CTX.ENV.getprop("qdrant.port"),
                                        model_path=getpath(r"res\models\acge_text_embedding"))