告别“搜不到、搜不准”：用这套查询优化，让你的RAG检索召回率飙升

from langchain.retrievers import MultiQueryRetriever# 1. 从LLM和向量数据库创建一个MultiQueryRetriever#    它会自动处理“生成查询 -> 检索 -> 合并去重”的整个流程multiquery_retriever = MultiQueryRetriever.from_llm(    retriever=vectorstore.as_retriever(), # 使用我们创建的向量数据库作为基础检索器    llm=llm # 使用我们初始化的LLM来生成子查询)# 2. 使用原始查询进行调用user_query = "如何通过修改问题来改进检索效果？"retrieved_docs = multiquery_retriever.invoke(user_query)# 3. 打印结果print_docs(retrieved_docs, f"查询扩展 (MultiQuery) 对 '{user_query}' 的检索结果")# 深入了解它生成了哪些子查询import logginglogging.basicConfig()logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)retrieved_docs = multiquery_retriever.invoke(user_query)

from langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.documents import Documentimport operator# 1. 定义一个用于生成子查询的链 (Chain)query_gen_prompt = ChatPromptTemplate.from_messages([    ("user", "你是一位AI研究员。请根据以下问题，生成3个不同角度的、语义相似的查询。\n"             "每个查询占一行，不要有其他前缀或编号。\n\n"             "原始问题: {original_question}")])generate_queries_chain = query_gen_prompt | llm | StrOutputParser() | (lambda x: x.split("\n"))# 2. 定义RRF算法函数def reciprocal_rank_fusion(retrieval_results: list[list[Document]], k=60):    fused_scores = {}       for doc_list in retrieval_results:        for rank, doc in enumerate(doc_list):            doc_id = doc.page_content            if doc_id not in fused_scores:                fused_scores[doc_id] = 0            fused_scores[doc_id] += 1 / (k + rank)
    reranked_results = [        next((doc for doc_list in retrieval_results for doc in doc_list if doc.page_content == doc_id), None)        for doc_id, score in sorted(fused_scores.items(), key=operator.itemgetter(1), reverse=True)    ]    return [doc for doc in reranked_results if doc is not None]def rag_fusion_pipeline(original_question: str):    # 生成多个查询    generated_queries = generate_queries_chain.invoke({"original_question": original_question})    all_queries = [original_question] + generated_queries    print(f"生成的查询: {all_queries}")    # 独立检索每个查询    retriever = vectorstore.as_retriever()    retrieval_results = [retriever.invoke(q) for q in all_queries]    # 应用RRF算法对结果进行融合和重排    final_docs = reciprocal_rank_fusion(retrieval_results)    return final_docs# 4. 调用user_query = "如何处理用户提问太具体的情况？"fusion_docs = rag_fusion_pipeline(user_query)print(fusion_docs, f"RAG-Fusion 对 '{user_query}' 的检索结果")

from langchain_core.runnables import RunnableParallel, RunnablePassthrough# 1. 定义生成“后退一步”问题的Prompt和链step_back_prompt_template = ChatPromptTemplate.from_messages([    ("user", "你是一位善于提炼核心问题的专家。请将以下可能很具体的问题，"             "抽象成一个更通用、更高层次的“后退一步”的问题。\n\n"             "例如：'LangChain的LCEL和Python的asyncio库是如何交互的？' -> 'LangChain LCEL的异步执行机制是怎样的？'\n\n"             "原始问题: {original_question}")])step_back_chain = step_back_prompt_template | llm | StrOutputParser()retriever = vectorstore.as_retriever()chain = (    {        # 第一个分支：对原始问题进行检索        "original_docs": RunnablePassthrough() | retriever,        # 第二个分支：先生成后退问题，再用它进行检索        "step_back_docs": step_back_chain | retriever,    }    # 将两个分支的结果合并、去重    | (lambda x: remove_duplicates_by_id(x["original_docs"] + x["step_back_docs"])))# 辅助函数去重def remove_duplicates_by_id(documents):    seen_ids = set()    unique_docs = []    for doc in documents:        # 假设 page_content 是唯一标识        if doc.page_content not in seen_ids:            unique_docs.append(doc)            seen_ids.add(doc.page_content)    return unique_docsuser_query = "RAG-Fusion里那个RRF算法的平滑参数k有什么用？"step_back_docs = chain.invoke({"original_question": user_query})

# 示例: 使用 EnsembleRetriever 实现混合搜索from langchain.retrievers import EnsembleRetrieverfrom langchain_community.retrievers import BM25Retrieverfrom langchain.retrievers import EnsembleRetriever# 假设 all_splits 和 vectorstore 已准备好# 1. 初始化关键词检索器 (Sparse Retriever)bm25_retriever = BM25Retriever.from_documents(all_splits)bm25_retriever.k = 3 # 检索3个结果# 2. 初始化向量检索器 (Dense Retriever)vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})# 3. 初始化 EnsembleRetriever，并设置权重# weights 参数决定了最终排序时，两种检索器结果的权重ensemble_retriever = EnsembleRetriever(    retrievers=[bm25_retriever, vector_retriever],    weights=[0.4, 0.6] # 稍微偏重向量搜索的语义理解能力)# 4. 使用query = "LangChain中的LCEL是什么?"retrieved_docs = ensemble_retriever.invoke(query)print(f"混合搜索召回了 {len(retrieved_docs)} 个文档。")