从论文到代码：手把手实现RAG Fusion中的RRF算法，让你的LangChain应用更智能

当我们在构建基于大语言模型的问答系统时，检索增强生成(RAG)已经成为提升回答质量的关键技术。然而，传统的RAG系统往往面临一个核心挑战：如何从海量文档中检索出最相关的信息？这正是RAG Fusion技术大显身手的舞台，而其中的RRF(Reciprocal Rank Fusion)算法则是让整个系统变得更智能的秘密武器。

1. RAG Fusion技术全景解析

在深入RRF算法之前，我们需要先理解RAG Fusion的整体架构。与基础RAG系统不同，RAG Fusion通过多查询生成和结果重排序两个关键步骤，显著提升了检索质量。

核心工作流程 ：

1. 多查询生成 ：基于用户原始问题，LLM生成多个相关但视角不同的查询
2. 并行检索 ：对每个生成的查询分别执行文档检索
3. 结果融合 ：使用RRF算法对所有检索结果进行智能重排序
4. 答案生成 ：将重排序后的最相关文档输入LLM生成最终答案

这种架构的优势在于能够克服单一查询可能存在的表述局限，通过多角度检索和智能融合，确保返回的文档集合既全面又精准。

2. RRF算法深度剖析

RRF算法源自信息检索领域，其核心思想是通过考虑文档在多个排序列表中的位置，计算出一个融合分数。让我们拆解其数学本质：

算法公式 ：

RRF分数(d) = Σ(1/(k + rank(d,i)))

其中：

• d 表示特定文档
• rank(d,i) 表示文档d在第i个排序列表中的位置(从0开始)
• k 是一个调节参数，通常设为60

这个设计的精妙之处在于：

• 位置倒数的非线性衰减 ：排名越靠前，贡献的分数越大，但衰减速度逐渐变缓
• 参数k的平滑作用 ：防止排名靠后的文档对结果产生过大影响
• 多列表协同验证 ：只有在多个列表中均排名靠前的文档才能获得高分

3. Python实现RRF算法

理解了算法原理后，我们来看如何在Python中实现RRF。以下是完整的代码实现：

from langchain.load import dumps, loads

def reciprocal_rank_fusion(results: list[list], k=60):
    fused_scores = {}
    
    # 遍历每个查询的检索结果列表
    for docs in results:
        # 遍历当前列表中的每个文档及其排名
        for rank, doc in enumerate(docs):
            doc_str = dumps(doc)
            
            # 初始化文档分数
            if doc_str not in fused_scores:
                fused_scores[doc_str] = 0
            
            # 累加RRF分数
            fused_scores[doc_str] += 1 / (rank + k)
    
    # 按分数降序排序
    reranked_results = [
        (loads(doc), score) 
        for doc, score in sorted(fused_scores.items(), 
                               key=lambda x: x[1], 
                               reverse=True)
    ]
    
    return reranked_results

关键实现细节 ：

1. 文档序列化 ：使用 dumps/loads 处理文档对象，确保能作为字典键使用
2. 分数累加 ：对每个文档在所有列表中的表现进行综合评估
3. 结果排序 ：最终按融合分数降序排列，得到重排序结果

4. 集成到LangChain工作流

现在我们将RRF算法整合到完整的LangChain流程中。以下是构建端到端RAG Fusion系统的步骤：

4.1 设置基础组件

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
from langchain import hub
from langchain_core.runnables import RunnablePassthrough

# 初始化LLM和prompt
llm = ChatOpenAI(temperature=0)
prompt = hub.pull("langchain-ai/rag-fusion-query-generation")

# 构建多查询生成链
generate_queries = (
    prompt 
    | llm
    | StrOutputParser() 
    | (lambda x: x.split("\n"))
)

4.2 构建完整检索链

# 假设已经初始化了retriever
retrieval_chain = (
    generate_queries 
    | retriever.map()  # 对每个查询并行检索
    | reciprocal_rank_fusion  # 应用RRF算法
)

4.3 添加答案生成环节

from operator import itemgetter

answer_prompt = ChatPromptTemplate.from_template(
    "根据以下上下文回答提问：\n\n{context}\n\n问题：{question}"
)

final_chain = (
    {"context": retrieval_chain, "question": itemgetter("question")}
    | answer_prompt
    | llm
    | StrOutputParser()
)

# 使用完整链回答问题
question = "气候变化对经济有哪些影响？"
result = final_chain.invoke({"question": question})
print(result)

5. 参数调优与性能分析

RRF算法中的k值对结果有重要影响。通过实验，我们发现：

k值影响分析 ：

k值	特点	适用场景
较小(10-30)	更强调排名靠前的文档	需要高精度的场景
中等(40-80)	平衡排名前后文档	一般用途
较大(>100)	更考虑长尾文档	需要高召回率的场景

实际测试数据 ：

# 测试不同k值的效果
for k in [30, 60, 100]:
    print(f"\nk={k}时的前3个文档：")
    results = reciprocal_rank_fusion(retrieval_results, k=k)
    for doc, score in results[:3]:
        print(f"分数：{score:.4f} | 内容：{doc.page_content[:50]}...")

在真实项目中，建议通过A/B测试确定最优k值。通常从60开始，根据具体需求上下调整。

6. 进阶优化技巧

要让RRF发挥最大效用，可以考虑以下优化策略：

1. 查询质量提升 ：

   • 为多查询生成设计更精细的prompt
   • 控制生成查询的多样性，避免偏离原问题

2. 混合检索策略 ：

   def hybrid_retrieval(question):
       # 同时使用关键词检索和向量检索
       keyword_results = keyword_retriever(question)
       vector_results = vector_retriever(question)
       return reciprocal_rank_fusion([keyword_results, vector_results])
   

3. 结果后处理 ：

   • 对重排序结果进行去重
   • 添加基于文档质量的过滤
   • 考虑文档的新鲜度因素

7. 真实案例：构建智能问答系统

让我们看一个完整的实现案例，使用百度百科关于"恐龙"的数据：

from langchain_community.document_loaders import WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma

# 加载并处理数据
loader = WebBaseLoader("https://baike.baidu.com/item/恐龙/139019")
docs = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
splits = text_splitter.split_documents(docs)

# 创建向量库
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()

# 定义完整流程
def rag_fusion_qa(question):
    # 生成多个查询
    queries = generate_queries.invoke({"original_query": question})
    
    # 并行检索
    retrieved_docs = [retriever.invoke(q) for q in queries]
    
    # RRF重排序
    reranked = reciprocal_rank_fusion(retrieved_docs)
    
    # 生成最终答案
    context = "\n\n".join([d.page_content for d, _ in reranked[:3]])
    response = llm.invoke(f"根据以下信息回答问题：\n{context}\n\n问题：{question}")
    
    return response

# 使用示例
answer = rag_fusion_qa("恐龙是如何灭绝的？")
print(answer)

这个实现展示了如何将RRF算法应用于真实数据，构建出能够给出高质量答案的智能系统。