suggestion_searcher.py

from qdrant_client import QdrantClient
from qdrant_client.models import Filter, FieldCondition, MatchValue
from fastembed import SparseTextEmbedding, LateInteractionTextEmbedding
from qdrant_client import QdrantClient, models
from sentence_transformers import SentenceTransformer
from config import DENSE_MODEL, SPARSE_MODEL, LATE_INTERACTION_MODEL, QDRANT_URL, QDRANT_API_KEY,HUGGING_FACE_API_KEY

class SuggestionSearcher:
    
    def __init__(self, collection_name):
        self.collection_name = collection_name
        self.dense_model = SentenceTransformer(DENSE_MODEL,device="cpu",token=HUGGING_FACE_API_KEY)
        self.sparse_model = SparseTextEmbedding(SPARSE_MODEL)
        self.late_interaction_model = LateInteractionTextEmbedding(LATE_INTERACTION_MODEL)
        self.qdrant_client = QdrantClient(QDRANT_URL,api_key=QDRANT_API_KEY,timeout=30)
    
    async def search(self, text: str,type:int):
        
        dense_query = self.dense_model.encode(text).tolist()
        sparse_query = next(self.sparse_model.query_embed(text))

        prefetch = [
                models.Prefetch(
                    query=dense_query,
                    params=models.SearchParams(
                        quantization=models.QuantizationSearchParams(
                            rescore=False,
                        ),
                    ),
                    using=DENSE_MODEL,
                    limit=10000
                ),
                models.Prefetch(
                    query=models.SparseVector(**sparse_query.as_object()),
                    params=models.SearchParams(
                        quantization=models.QuantizationSearchParams(
                            rescore=False,
                        ),
                    ),
                    using=SPARSE_MODEL,
                    limit=10000
                )
        ]

        self.qdrant_client.scroll
        search_result = self.qdrant_client.query_points(
            collection_name= self.collection_name,
            query_filter=Filter(
                must=[
                    FieldCondition(
                        key="tip",
                        match=MatchValue(value=type)
                    )
                ]
            ),
            search_params=models.SearchParams(
                hnsw_ef=64, 
                exact=False, 
                quantization=models.QuantizationSearchParams(
                        rescore=True,
                    ),
                ),
            prefetch=prefetch,
            query=models.FusionQuery(
                fusion=models.Fusion.RRF,
            ),
            with_payload=True,
            limit = 5
        ).points

        data = []

        for hit in search_result:
            data.append(hit.payload)
        
        return data