深入理解RAG系统中的重排序技术：基于liu673/rag-all-techniques的实现

2025-07-05 08:19:15作者：幸俭卉

引言

在现代信息检索系统中，检索增强生成（Retrieval-Augmented Generation，RAG）已成为一种强大的技术范式。本文将重点探讨RAG系统中一个关键但常被忽视的组件——重排序（Reranking）技术，并基于liu673/rag-all-techniques项目中的实现进行详细解析。

重排序技术概述

什么是重排序？

重排序是RAG系统中继初始检索后的第二步过滤过程，其主要目的是对初步检索到的文档进行重新排序，确保最终用于生成响应的内容是最相关的。

为什么需要重排序？

初始检索的局限性：向量相似性搜索虽然快速，但可能无法准确捕捉语义相关性
精度提升：重排序可以显著提高最终结果的准确性
计算效率：相比直接使用大模型处理所有检索结果，重排序是一种更经济的方案

核心实现组件

1. 文档处理流程

def process_document(pdf_path, chunk_size=1000, chunk_overlap=200):
    # 提取PDF文本
    extracted_text = extract_text_from_pdf(pdf_path)
    
    # 分割文本块
    chunks = chunk_text(extracted_text, chunk_size, chunk_overlap)
    
    # 创建嵌入向量
    chunk_embeddings = create_embeddings(chunks)
    
    # 构建向量存储
    store = SimpleVectorStore()
    for i, (chunk, embedding) in enumerate(zip(chunks, chunk_embeddings)):
        store.add_item(text=chunk, embedding=embedding, metadata={"index": i})
    return store

2. 向量存储实现

项目实现了一个轻量级的SimpleVectorStore类，核心功能包括：

存储文本、嵌入向量和元数据
基于余弦相似度的相似性搜索
支持批量添加和检索

class SimpleVectorStore:
    def __init__(self):
        self.vectors = []  # 存储嵌入向量
        self.texts = []    # 存储原始文本
        self.metadata = [] # 存储元数据
        
    def similarity_search(self, query_embedding, k=5):
        # 计算余弦相似度
        similarities = []
        for i, vector in enumerate(self.vectors):
            similarity = np.dot(query_vector, vector) / \
                        (np.linalg.norm(query_vector) * np.linalg.norm(vector))
            similarities.append((i, similarity))
        
        # 排序并返回top-k结果
        similarities.sort(key=lambda x: x[1], reverse=True)
        return [{"text": self.texts[idx], ...} for idx, _ in similarities[:k]]

重排序技术实现

1. 基于LLM的重排序

这种方法利用大语言模型对文档相关性进行评分：

def rerank_with_llm(query, results, top_n=3):
    # 系统提示定义评分标准
    system_prompt = """
    您是文档相关性评估专家，擅长判断文档与搜索查询的匹配程度...
    评分标准：
    0-2分：完全无关
    3-5分：含部分相关信息但未直接回答问题
    6-8分：相关且能部分解答查询
    9-10分：高度相关且直接准确回答问题
    """
    
    # 对每个文档评分
    for result in results:
        response = client.chat.completions.create(
            model=llm_model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": f"查询: {query}\n文档: {result['text']}"}
            ]
        )
        # 提取评分并存储
        score = extract_score(response)
        result["relevance_score"] = score
    
    # 按评分排序
    return sorted(results, key=lambda x: x["relevance_score"], reverse=True)[:top_n]

优势：

能够理解复杂的语义关系
评分标准可灵活定义
适用于各种类型的查询

局限：

计算成本较高
延迟相对较大

2. 基于关键词的重排序

这是一种轻量级的替代方案：

def rerank_with_keywords(query, results, top_n=3):
    # 提取关键词
    keywords = [word.lower() for word in query.split() if len(word) > 3]
    
    # 计算每个文档的分数
    for result in results:
        score = result["similarity"] * 0.5  # 基础分
        
        # 关键词匹配加分
        for keyword in keywords:
            if keyword in document_text:
                score += 0.1  # 基础匹配分
                if first_position < len(text)/4:  # 位置加分
                    score += 0.1
                score += min(0.05 * frequency, 0.2)  # 频率加分
    
    # 排序并返回
    return sorted(results, key=lambda x: x["relevance_score"], reverse=True)[:top_n]

适用场景：

对延迟敏感的应用
查询包含明确关键词的情况
资源受限的环境

完整RAG管道集成

将各组件整合为端到端的RAG管道：

def rag_with_reranking(query, vector_store, reranking_method="llm", top_n=3):
    # 1. 创建查询嵌入
    query_embedding = create_embeddings(query)
    
    # 2. 初始检索（召回较多结果）
    initial_results = vector_store.similarity_search(query_embedding, k=10)
    
    # 3. 应用重排序
    if reranking_method == "llm":
        reranked_results = rerank_with_llm(query, initial_results, top_n)
    else:
        reranked_results = rerank_with_keywords(query, initial_results, top_n)
    
    # 4. 合并上下文并生成响应
    context = "\n\n===\n\n".join([r["text"] for r in reranked_results])
    response = generate_response(query, context)
    
    return {
        "query": query,
        "context": context,
        "response": response
    }