如何提升RAG的召回率：从原理到实战 #

def sliding_window_chunk(text: str, chunk_size: int = 300, overlap: int = 50) -> list[str]:
    """
    滑动窗口分块：保证边界信息不丢失
    chunk_size: 每个块的token数
    overlap: 相邻块的重叠token数
    """
    words = text.split()
    chunks = []
    start = 0
    while start < len(words):
        end = start + chunk_size
        chunk = ' '.join(words[start:end])
        chunks.append(chunk)
        start += chunk_size - overlap
    return chunks

# LlamaIndex 父子索引示例
from llama_index.core.node_parser import SentenceSplitter, HierarchicalNodeParser
from llama_index.core import VectorStoreIndex

# 创建父子节点
hierarchical = HierarchicalNodeParser.from_defaults(
    chunk_sizes=[2048, 512, 128],  # 父->子
    chunk_overlap=50
)
nodes = hierarchical.get_nodes_from_documents(documents)

# 子节点用于检索，父节点用于上下文
child_nodes = nodes[1:]  # 排除根节点
child_index = VectorStoreIndex(child_nodes)

import spacy
from typing import List

nlp = spacy.load("zh_core_web_sm")  # 或 en_core_web_sm

def semantic_chunk(text: str, max_tokens: int = 300) -> List[str]:
    """
    基于句子边界的语义分块
    保证不切断句子，同时控制chunk大小
    """
    doc = nlp(text)
    chunks = []
    current_chunk = []
    current_length = 0
    
    for sent in doc.sents:
        sent_tokens = len(sent.text.split())
        if current_length + sent_tokens > max_tokens and current_chunk:
            chunks.append(" ".join(current_chunk))
            current_chunk = [sent.text]
            current_length = sent_tokens
        else:
            current_chunk.append(sent.text)
            current_length += sent_tokens
    
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    return chunks

from sentence_transformers import SentenceTransformer, losses, InputExample
from torch.utils.data import DataLoader

# 1. 加载预训练模型
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 2. 准备训练数据：(query, positive_doc, negative_doc)
train_examples = [
    InputExample(texts=["如何重置密码", "密码重置指南...", "账户登录页面..."]),
    InputExample(texts=["API超时设置", "timeout参数配置...", "API接口列表..."]),
    # ... 更多样本
]

# 3. 构建数据加载器
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)

# 4. 定义损失函数：MultipleNegativesRankingLoss
train_loss = losses.MultipleNegativesRankingLoss(model)

# 5. 训练
model.fit(
    train_objectives=[(train_dataloader, train_loss)],
    epochs=3,
    warmup_steps=100,
    output_path='./bge-domain-finetuned'
)

def reciprocal_rank_fusion(results: list[list], k: int = 60) -> dict:
    """
    RRF融合多个检索结果
    results: 每个元素是一个检索器返回的有序文档列表
    """
    fused_scores = {}
    for ranked_list in results:
        for rank, doc in enumerate(ranked_list):
            if doc not in fused_scores:
                fused_scores[doc] = 0
            fused_scores[doc] += 1 / (k + rank)
    
    # 按RRF分数降序排序
    return dict(sorted(fused_scores.items(), key=lambda x: x[1], reverse=True))

from elasticsearch import Elasticsearch

es = Elasticsearch("http://localhost:9200")

def hybrid_search(query: str, index: str, top_k: int = 5):
    """
    Elasticsearch 混合检索：BM25 + 向量检索
    """
    response = es.search(
        index=index,
        query={
            "bool": {
                "should": [
                    {
                        "multi_match": {  # BM25关键词匹配
                            "query": query,
                            "fields": ["content", "title^2"],
                            "boost": 0.4
                        }
                    },
                    {
                        "knn": {  # 向量检索
                            "field": "embedding",
                            "query_vector": get_embedding(query),
                            "k": top_k * 2,
                            "boost": 0.6
                        }
                    }
                ]
            }
        },
        size=top_k
    )
    return [hit["_source"] for hit in response["hits"]["hits"]]

from llama_index.core.query_engine import MultiQueryRetriever

# 使用LLM自动生成多角度查询
retriever = MultiQueryRetriever.from_defaults(
    vector_index,
    llm=llm,
    num_queries=4
)
nodes = retriever.retrieve("如何优化数据库查询性能？")
# 实际执行的查询可能包括：
# - 数据库查询优化方法
# - SQL性能调优技巧
# - 索引优化策略
# - 查询执行计划分析

from FlagEmbedding import FlagReranker

reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True)

def rerank_results(query: str, candidates: list[str], top_k: int = 5) -> list:
    """
    使用Cross-Encoder对候选结果重排序
    """
    # 构造(查询, 文档)对
    pairs = [[query, doc] for doc in candidates]
    # 批量计算相关性分数
    scores = reranker.compute_score(pairs, normalize=True)
    
    # 按分数排序
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, score in ranked[:top_k]]

from ragas import evaluate
from ragas.metrics import context_recall
from datasets import Dataset

# 准备评估数据
data_samples = {
    'question': ['如何重置密码？', 'API超时怎么设置？'],
    'contexts': [['密码重置步骤...', '账户安全指南...']],
    'ground_truth': ['访问设置页面，点击安全选项...']
}
dataset = Dataset.from_dict(data_samples)

# 评估
score = evaluate(dataset, metrics=[context_recall])
print(f"Context Recall: {score['context_recall']}")

import os
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

class HighRecallRAG:
    def __init__(self, docs, embedding_model_name="BAAI/bge-large-zh-v1.5"):
        # 1. 文档分块（带overlap）
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=300,
            chunk_overlap=50,
            separators=["\n\n", "\n", "。", "，", " ", ""]
        )
        self.chunks = splitter.split_documents(docs)
        
        # 2. Embedding模型
        self.embeddings = HuggingFaceEmbeddings(
            model_name=embedding_model_name,
            model_kwargs={'device': 'cuda'},
            encode_kwargs={'normalize_embeddings': True}
        )
        
        # 3. 构建向量索引
        self.vectorstore = FAISS.from_documents(self.chunks, self.embeddings)
        
        # 4. 构建BM25索引
        self.bm25_retriever = BM25Retriever.from_documents(self.chunks)
        self.bm25_retriever.k = 10
        
        # 5. 混合检索器
        self.vector_retriever = self.vectorstore.as_retriever(search_kwargs={"k": 10})
        self.ensemble_retriever = EnsembleRetriever(
            retrievers=[self.bm25_retriever, self.vector_retriever],
            weights=[0.4, 0.6]
        )
    
    def retrieve(self, query: str, top_k: int = 5) -> list:
        # 第一阶段：混合检索
        candidates = self.ensemble_retriever.invoke(query)
        
        # 第二阶段：Re-ranker（伪代码，需安装FlagEmbedding）
        # reranked = self._rerank(query, candidates, top_k)
        
        return candidates[:top_k]
    
    def _rerank(self, query, candidates, top_k):
        from FlagEmbedding import FlagReranker
        reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True)
        pairs = [[query, doc.page_content] for doc in candidates]
        scores = reranker.compute_score(pairs, normalize=True)
        ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
        return [doc for doc, _ in ranked[:top_k]]