RAG召回率从61%到89%：chunk重构与Embedding选型全记录

langchain==0.1.16
langchain-community==0.0.36
sentence-transformers==2.6.1  
faiss-cpu==1.8.0.post1
FlagEmbedding==1.2.10
torch==2.2.1

import re
from langchain.text_splitter import TextSplitter

class SemanticParagraphSplitter(TextSplitter):
    """按标题层级+句子边界切分，保留元数据"""

    def __init__(self, max_chunk=512, min_chunk=128):
        super().__init__()
        self.max_chunk = max_chunk
        self.min_chunk = min_chunk
        self.heading_pattern = re.compile(r'^#{1,4}\s+', re.MULTILINE)
        self.sentence_delimiters = re.compile(r'(? list[str]:
        # 1. 按标题拆成一级结构
        sections = self._split_by_headings(text)
        chunks = []
        for heading, body in sections:
            # 2. 对每个body内部按句子聚合
            sentences = self.sentence_delimiters.split(body)
            current_chunk = f"{heading}\n"
            for sent in sentences:
                sent = sent.strip()
                if not sent:
                    continue
                if len(current_chunk) + len(sent) > self.max_chunk and len(current_chunk) > self.min_chunk:
                    chunks.append(current_chunk.strip())
                    current_chunk = f"{heading}\n"
                current_chunk += sent + "。"
            if current_chunk.strip():
                chunks.append(current_chunk.strip())
        return chunks

    def _split_by_headings(self, text):
        # 简化版：按行扫描，遇到标题行则开启新段落
        lines = text.split('\n')
        sections = []
        current_heading = ""
        current_body = []
        for line in lines:
            if self.heading_pattern.match(line):
                if current_body:
                    sections.append((current_heading, '\n'.join(current_body)))
                current_heading = line.strip()
                current_body = []
            else:
                current_body.append(line)
        if current_body:
            sections.append((current_heading, '\n'.join(current_body)))
        return sections

# 使用示例
splitter = SemanticParagraphSplitter(max_chunk=512, min_chunk=128)
docs = loader.load()  # 假设已加载文档
chunks = []
for doc in docs:
    chunks.extend(splitter.split_text(doc.page_content))
print(f"原始文档{len(docs)}份，切分为{len(chunks)}个chunk")
# 输出：原始文档3241份，切分为18762个chunk（旧方案是24530个，数量减少23%）

from FlagEmbedding import FlagReranker
from langchain.embeddings import HuggingFaceBgeEmbeddings
import faiss
import numpy as np

# 初始化bge-m3 embedding
embedder = HuggingFaceBgeEmbeddings(
    model_name="BAAI/bge-m3",
    model_kwargs={'device': 'cpu'},
    encode_kwargs={'normalize_embeddings': True}
)

# 构建Faiss索引（IndexFlatIP，内积）
dim = 1024
index = faiss.IndexFlatIP(dim)
# 假设已有chunk_texts和对应embeddings
# index.add(all_embeddings)

# 重排器
reranker = FlagReranker('BAAI/bge-reranker-v2-m3', use_fp16=False)

def hybrid_retrieve_with_rerank(query, top_k=20, final_k=5):
    # 1. 稠密检索
    q_emb = embedder.embed_query(query)
    scores, indices = index.search(np.array([q_emb]), top_k)

    # 2. 候选块
    candidates = [chunk_texts[i] for i in indices[0]]

    # 3. Rerank打分
    pairs = [[query, cand] for cand in candidates]
    rerank_scores = reranker.compute_score(pairs, normalize=True)  # 归一化到[0,1]

    # 4. 按重排分数取Top-5
    top_indices = np.argsort(rerank_scores)[::-1][:final_k]
    return [candidates[i] for i in top_indices], [rerank_scores[i] for i in top_indices]

# 测试一条query
query = "A类设备保修期内更换电源模块需要什么审批流程？"
result, scores = hybrid_retrieve_with_rerank(query)
for i, (chunk, score) in enumerate(zip(result, scores)):
    print(f"Rank {i+1} (score: {score:.4f}): {chunk[:80]}...")