Aryan2301/YouTube_RAG_Intelligence

1from typing import List2from langchain_core.documents import Document3from langchain_classic.retrievers import EnsembleRetriever4import asyncio5 6from utils.constants import SEMANTIC_TOP_K, BM25_TOP_K, FINAL_TOP_K, RRF_K7 8async def hybrid_retrieve(9    query: str,10    vector_store,11    bm25_retriever,12    semantic_k: int = SEMANTIC_TOP_K,13    bm25_k: int = BM25_TOP_K,14    final_k: int = FINAL_TOP_K,15    rrf_k: int = RRF_K16) -> List[Document]:17    """18    Performs hybrid retrieval asynchronously using Langchain's EnsembleRetriever.19    """20    # 1. Prepare retrievers21    semantic_retriever = vector_store.as_retriever(search_kwargs={"k": semantic_k})22    bm25_retriever.k = bm25_k23    24    # 2. Build Ensemble Retriever25    ensemble_retriever = EnsembleRetriever(26        retrievers=[bm25_retriever, semantic_retriever],27        weights=[0.5, 0.5],28        c=rrf_k29    )30    31    # 3. Retrieve and fuse32    fused_documents = await ensemble_retriever.ainvoke(query)33    34    return fused_documents[:final_k]35