| 1234567891011121314151617181920212223242526272829303132333435363738394041424344 |
- # @description:
- # @author: licanglong
- # @date: 2025/12/19 16:08
- import logging
- from qdrant_client import QdrantClient
- 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 = QdrantClient(host=host, port=port)
- # 2. 向量模型
- self.embedding = SentenceTransformer(model_path)
- def create_collection(self, collection_name: str, **kwargs) -> bool:
- if self.client.collection_exists(collection_name):
- return False
- 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"))
|
