personal-consumption-service/tests/TestEmbeddingStoreQueryApp.py

# @description: 
# @author: licanglong
# @date: 2025/12/19 17:34
from qdrant_client import QdrantClient
from sentence_transformers import SentenceTransformer

from app.App import App
from app.utils.pathutils import getpath


class TestEmbeddingStoreQueryApp(App):
    def run(self, *args, **kwargs):
        COLLECTION_NAME = "rule_embed_store"
        # COLLECTION_NAME = "case_embed_store"
        # COLLECTION_NAME = "merchants_embed_store"
        # COLLECTION_NAME = "edges_embed_store"
        client = QdrantClient(host="117.72.147.109", port=16333)
        model = SentenceTransformer(getpath(r"res\models\acge_text_embedding"))

        vector = model.encode("特定业务类型： 货物名称：*餐饮服务*餐费 规格型号：")
        results = client.query_points(
            collection_name=COLLECTION_NAME,
            query=vector.tolist(),
            limit=5,
            score_threshold=0.5
        )
        print(results)