从62%到89%：一次RAG检索增强生成系统的三轮迭代优化记录

Python 3.10.13
milvus 2.3.1 (standalone, HNSW, M=16, efConstruction=200)
FlagEmbedding 1.2.10
sentence-transformers 2.2.2
torch 2.1.0 + cu118
langchain 0.1.0 (只用了text splitter部分)
openai 1.6.1 (仅用于生成)

from langchain.text_splitter import RecursiveCharacterTextSplitter, MarkdownHeaderTextSplitter

def semantic_chunking(md_text: str, max_chunk_size: int = 400, overlap: int = 80):
    # 一级：按Markdown标题切
    headers_to_split_on = [
        ("#", "h1"),
        ("##", "h2"),
        ("###", "h3"),
    ]
    md_splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=headers_to_split_on,
        strip_headers=False
    )
    sections = md_splitter.split_text(md_text)

    # 二级：章节内递归切分
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=max_chunk_size,
        chunk_overlap=overlap,
        separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""],
        length_function=lambda x: len(x)  # 中文按字符数近似token
    )

    final_chunks = []
    for sec in sections:
        # 把标题拼回正文，保证chunk自带上下文
        header_prefix = " > ".join([f"{k}:{v}" for k, v in sec.metadata.items()])
        content = f"[{header_prefix}]\n{sec.page_content}"
        if len(content)  错误码」这种前缀能显著提升相关性；二是中文的 `length_function` 直接用字符数，别傻乎乎用tiktoken，中文token和字符差不多1:1，没必要多一次编码开销。

Round 1结果：Hit@5从62% → 71%，MRR@5从0.41 → 0.52。chunk数量从原来的平均每篇8个涨到13个，库总量从96万涨到156万条向量。存储成本涨了60%，但召回质量实打实提升。

### Round 2：切换embedding模型

ada-002是通用多语言模型，但在中文技术文档上表现一般。我换成了智源的 `bge-large-zh-v1.5`，768维，中文语义对齐明显更好。

这里有个坑：**query和passage要用不同的前缀**。bge系列训练时query加了指令前缀，passage没有。如果两边都不加，效果会掉一截。

```python
from FlagEmbedding import FlagModel
import numpy as np

class BGEEmbedder:
    def __init__(self, model_path: str = "BAAI/bge-large-zh-v1.5", device: str = "cuda:0"):
        self.model = FlagModel(
            model_path,
            query_instruction_for_retrieval="为这个句子生成表示以用于检索相关文章：",
            use_fp16=True,
            device=device
        )

    def encode_queries(self, queries):
        # 自动加query instruction
        return self.model.encode_queries(queries, batch_size=64, max_length=512)

    def encode_passages(self, passages):
        # passage不加instruction
        return self.model.encode(passages, batch_size=64, max_length=512)

# 批量重建索引
embedder = BGEEmbedder()
# 假设chunks是Round 1产出的所有chunk
vectors = embedder.encode_passages(chunks)
# 写入Milvus，collection维度从1536改成768

from FlagEmbedding import FlagReranker

class RerankerPipeline:
    def __init__(self, reranker_path: str = "BAAI/bge-reranker-large"):
        self.reranker = FlagReranker(reranker_path, use_fp16=True)

    def retrieve_and_rerank(self, query: str, milvus_client, collection, top_k_recall=50, top_k_final=5):
        # 1. 向量召回
        query_vec = embedder.encode_queries([query])[0]
        search_params = {"metric_type": "IP", "params": {"ef": 128}}
        results = milvus_client.search(
            collection_name=collection,
            data=[query_vec],
            anns_field="embedding",
            param=search_params,
            limit=top_k_recall,
            output_fields=["chunk_id", "content"]
        )
        candidates = [(hit.entity.get("content"), hit.entity.get("chunk_id")) for hit in results[0]]

        # 2. 构造query-doc对
        pairs = [[query, doc] for doc, _ in candidates]
        scores = self.reranker.compute_score(pairs, normalize=True)

        # 3. 排序取Top-K
        ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
        return [(doc, cid, score) for (doc, cid), score in ranked[:top_k_final]]