RAG检索增强生成系统优化实录：chunk策略、embedding切换与rerank效果对比

Python 3.10.13
torch 2.1.2 + cu121
transformers 4.36.2
sentence-transformers 2.2.2
langchain 0.1.0
langchain-community 0.0.10
faiss-cpu 1.7.4（离线评估用），线上用faiss-gpu 1.7.4
FlagEmbedding 1.2.5（reranker）

import re
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document

def structure_aware_split(md_text: str, doc_id: str, max_len: int = 800):
    """
    按 Markdown 标题层级切分，超长 section 再递归切。
    chunk 头部拼接标题路径，提升 embedding 的语义完整性。
    """
    # 按 ## / ### 标题切块，保留标题行
    blocks = re.split(r'\n(?=#{2,3}\s)', md_text)
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=max_len,
        chunk_overlap=100,
        separators=["\n\n", "\n", "。", "；", " ", ""],
        length_function=len,
    )
    docs = []
    current_h1 = ""
    for block in blocks:
        block = block.strip()
        if not block:
            continue
        # 记录一级标题作为路径前缀
        m1 = re.match(r'#\s+(.+)', block)
        if m1:
            current_h1 = m1.group(1).strip()
            continue
        m2 = re.match(r'(#{2,3})\s+(.+)', block)
        heading = m2.group(2).strip() if m2 else ""
        path = f"{current_h1} > {heading}" if heading else current_h1

        if len(block)  排查步骤”这样的路径进入embedding，解决孤立段落语义漂移的问题。

### 4.2 Embedding切换

中文场景下ada-002其实不差，但有两个硬伤：一是按token计费，12万文档全量重建索引成本不低；二是中文短query的区分度一般。我们对比了三个候选（都在评估集上跑）：

| 模型 | 维度 | Recall@4 | 单条推理(CPU) | 备注 |
|---|---|---|---|---|
| text-embedding-ada-002 | 1536 | 0.71 | API调用 | 基线，chunk优化后 |
| bge-large-zh-v1.5 | 1024 | 0.78 | ~18ms | 中文最优档 |
| bge-m3 | 1024 | 0.79 | ~45ms | 多语言强，但慢 |

最终选bge-large-zh-v1.5，Recall提升明显且能本地部署，批量编码12万chunk在单张A10上约25分钟。注意bge系列检索时要加query instruction，这是官方推荐的用法，不加会掉2-3个点：

```python
from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("BAAI/bge-large-zh-v1.5", device="cuda")
# 关键：query 侧加 instruction，doc 侧不加
QUERY_INSTR = "为这个句子生成表示以用于检索相关文章："

def encode_queries(texts):
    return model.encode(
        [QUERY_INSTR + t for t in texts],
        normalize_embeddings=True,
        batch_size=64,
    )

def encode_docs(texts):
    return model.encode(
        texts,
        normalize_embeddings=True,
        batch_size=128,
    )

from FlagEmbedding import FlagReranker

reranker = FlagReranker("BAAI/bge-reranker-base", use_fp16=True)

def rerank(query: str, candidates: list[dict], top_n: int = 4):
    pairs = [[query, c["content"]] for c in candidates]
    scores = reranker.compute_score(pairs, normalize=True)
    for c, s in zip(candidates, scores):
        c["rerank_score"] = float(s)
    return sorted(candidates, key=lambda x: x["rerank_score"],
                  reverse=True)[:top_n]