RAG召回率从61%到89%：chunk重构与rerank二阶段调优实录

Python 3.10.13
langchain 0.2.11
chromadb 0.4.24
sentence-transformers 2.6.1
FlagEmbedding 1.2.8
torch 2.1.2+cu118

from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

# 按标题层级切片，保留标题作为上下文
headers_to_split_on = [
    ("#", "H1"),
    ("##", "H2"),
    ("###", "H3"),
]

markdown_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)

# 对每个标题块，再按段落和句子切分，但保证最小chunk和重叠
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=380,          # 调大了一点，适应表格和代码块
    chunk_overlap=64,        # 重叠从20提升到64，保证跨块语义连贯
    separators=["\n\n", "\n", "。", "；", " ", ""],  # 按中文句读优先
    length_function=len,
)

raw_documents = load_markdown_files("knowledge_base/")

all_chunks = []
for doc in raw_documents:
    # 先按标题拆分
    sections = markdown_splitter.split_text(doc.page_content)
    for section in sections:
        # 再按长度细分
        sub_chunks = text_splitter.split_text(section.page_content)
        for chunk in sub_chunks:
            # 将标题路径拼接到chunk内容前，作为语义锚点
            header_path = " > ".join([str(v) for k, v in section.metadata.items() if k.startswith("H")])
            full_text = f"[{header_path}] {chunk}"
            all_chunks.append(full_text)

print(f"Total chunks: {len(all_chunks)}")
# 输出：Total chunks: 18742 （相比原来硬切减少了23%）

from sentence_transformers import SentenceTransformer

# 旧模型：768维
# model = SentenceTransformer("BAAI/bge-large-zh-v1.5")

# 新模型：1024维，支持8192长度
model = SentenceTransformer("BAAI/bge-m3")

# 重要：bge系列需要加query指令前缀（对检索效果影响3-5%）
QUERY_PREFIX = "为这个句子生成表示以用于检索相关文章："

def embed_documents(texts: list[str]) -> list[list[float]]:
    # 批量编码，自动用GPU
    embeddings = model.encode(
        texts,
        batch_size=64,
        normalize_embeddings=True,  # 必须归一化，否则余弦相似度计算错误
        show_progress_bar=False
    )
    return embeddings.tolist()

def embed_query(query: str) -> list[float]:
    # 查询时加前缀
    return model.encode([QUERY_PREFIX + query], normalize_embeddings=True)[0].tolist()

from FlagEmbedding import FlagReranker

# 加载rerank模型，注意需要GPU
reranker = FlagReranker("BAAI/bge-reranker-large", use_fp16=True)

def rerank_search(query: str, top_k: int = 5, candidate_k: int = 20):
    # 第一步：向量检索粗排，取Top-20
    query_emb = embed_query(query)
    results = collection.query(
        query_embeddings=[query_emb],
        n_results=candidate_k,
        include=["documents", "metadatas", "distances"]
    )

    docs = results["documents"][0]
    distances = results["distances"][0]

    # 第二步：交叉编码器精排
    pairs = [[query, doc] for doc in docs]
    scores = reranker.compute_score(pairs, normalize=True)  # normalize=True输出0-1概率

    # 按分数排序
    sorted_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k]

    final_docs = [docs[i] for i in sorted_indices]
    final_scores = [scores[i] for i in sorted_indices]

    return final_docs, final_scores