RAG检索增强生成:从文档切片到向量检索的完整链路

去年公司要做个内部知识库问答系统，要求能检索上千份技术文档然后准确回答。我选了RAG方案，从零搭了一套完整链路。这篇文章把关键环节和踩过的坑都写出来。

RAG到底是个什么流程

用ASCII画个简单架构图，一目了然：

核心就四个环节：切片→向量化→检索→生成。每个环节都有讲究，不是随便调调API就能用的。

文档切片：RAG的地基

切片质量直接决定检索精度。我试过好几种切片策略，最后发现LangChain的RecursiveCharacterTextSplitter最省心。

from langchain.text_splitter import RecursiveCharacterTextSplitter# 读取文档with open("技术文档_v2.md", "r", encoding="utf-8") as f:    document = f.read()# 递归字符分割器：优先按段落、句子、词分割splitter = RecursiveCharacterTextSplitter(    chunk_size=500,        # 每块500字符    chunk_overlap=100,     # 块之间重叠100字符    separators=["\n\n", "\n", "。", ".", " ", ""],  # 分割优先级    length_function=len,)chunks = splitter.create_documents([document])print(f"总共切了 {len(chunks)} 个chunk")# 输出：总共切了 247 个chunk

为什么用RecursiveCharacterTextSplitter？因为它会按优先级尝试分割符：先按段落(\n\n)分，分不开再按换行(\n)，再不行按句号，最后才按空格。这样能最大程度保持语义完整性。

不同切片策略对比

我拿同一份200页的技术文档做了测试：

踩坑：切片大小太关键了。 我一开始用chunk_size=1000，结果检索精度只有65%。因为chunk太大，里面混了太多无关信息，LLM生成答案时容易被干扰。后来改成500，精度提到85%。但也不能太小——300以下碎片化严重，连续的信息被拆散。

另一个坑：中文分段。 用英文标点作为分割符对中文文档效果很差。我后来在separators里加了中文标点。、！、？，检索精度提升了8个百分点。

Embedding生成：把文字变成向量

切好片后，要把每个chunk转成向量。我对比了几个模型：

from openai import OpenAIclient = OpenAI()defget_embedding(text: str, model="text-embedding-3-small") -> list:"""生成文本的向量表示"""    text = text.replace("\n", " ")    response = client.embeddings.create(        input=[text],        model=model    )return response.data[0].embedding  # 返回1536维向量

Embedding模型对比

我最后选了BGE-M3本地部署。虽说要自己搭服务，但中文检索精度确实比OpenAI高，而且免费。部署命令：

# 用TEI（Text Embeddings Inference）部署BGE-M3docker run -p 8080:80 --gpus all \  ghcr.io/huggingface/text-embeddings-inference:1.2 \  --model-id BAAI/bge-m3

向量数据库存储

我用Chroma搭了一个快速原型，几行代码就能跑：

import chromadbfrom chromadb.config import Settings# 初始化Chromaclient = chromadb.PersistentClient(path="./chroma_db")collection = client.get_or_create_collection(    name="tech_docs",    metadata={"hnsw:space": "cosine"}  # 用余弦相似度)# 批量写入向量documents = [chunk.page_content for chunk in chunks]ids = [f"doc_{i}"for i in range(len(documents))]embeddings = [get_embedding(doc) for doc in documents]collection.add(    documents=documents,    ids=ids,    embeddings=embeddings)print(f"写入了 {len(documents)} 条记录")

完整的RAG查询流程

把所有环节串起来，一个完整的RAG查询：

defrag_query(question: str, top_k: int = 5) -> str:"""完整的RAG查询流程"""# 1. 问题向量化    query_embedding = get_embedding(question)# 2. 向量检索    results = collection.query(        query_embeddings=[query_embedding],        n_results=top_k,        include=["documents", "distances"]    )# 3. 构建上下文    context = "\n\n---\n\n".join(results["documents"][0])    distances = results["distances"][0]    print(f"检索到 {len(results['documents'][0])} 条，最小距离: {min(distances):.4f}")# 4. LLM生成回答    prompt = f"""基于以下资料回答问题。如果资料中没有相关信息，请明确说明。资料：{context}问题：{question}回答："""    response = client.chat.completions.create(        model="gpt-4o",        messages=[{"role": "user", "content": prompt}],        temperature=0.3    )return response.choices[0].message.content# 测试answer = rag_query("Kubernetes的Service和Ingress有什么区别？")print(answer)

踩坑：检索结果排序。 向量相似度不等于语义相关性！有时候检索回来的top5里，距离最小的反而不是最相关的。我后来加了BM25做混合检索（用RRF融合排序），相关性提升了约15%。

又一个坑：中文问题的分词。 用户问"怎么配Nginx反向代理"，如果文档里写的是"Nginx反向代理配置方法"，向量相似度可能只有0.7。解决方案是在切片时保留一些关键词的同义词，或者在检索后加个rerank环节。

生产环境的几个优化点

这套RAG系统上线后，我还做了几个优化：

1. HyDE（假设文档嵌入）：先让LLM生成一个假设答案，用这个假设答案去检索，比直接用问题检索效果好
2. 检索后重排：用BGE-Reranker对初步检索结果重排，精度提升明显
3. 元数据过滤：在Chroma里存了文档类型、日期等元数据，检索时可以按条件过滤

整个RAG系统的核心就是：切片质量 > Embedding模型 > 检索策略 > LLM本身。地基打不好，上面的都白搭。

多种切片策略的完整实现

下面是一次性实现三种切片策略的完整代码，覆盖了实际项目中常用的所有方式：

from langchain.text_splitter import (    RecursiveCharacterTextSplitter,    MarkdownHeaderTextSplitter,    CharacterTextSplitter,)from langchain.schema import Documentimport redefchunk_document(filepath: str, strategy: str = "recursive"):"""支持三种切片策略的文档分块函数"""with open(filepath, "r", encoding="utf-8") as f:        text = f.read()if strategy == "recursive":# 策略1：递归分割（最通用）        splitter = RecursiveCharacterTextSplitter(            chunk_size=500, chunk_overlap=100,            separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?", " ", ""],            length_function=len,        )return splitter.create_documents([text])elif strategy == "markdown_header":# 策略2：按Markdown标题分割（保持文档结构）        headers_to_split_on = [            ("#", "h1"), ("##", "h2"), ("###", "h3"),        ]        splitter = MarkdownHeaderTextSplitter(headers_to_split_on)return splitter.split_text(text)elif strategy == "semantic":# 策略3：语义分割（按语义边界，基于句子）        sentences = re.split(r'[。！？\n](?![a-zA-Z])', text)        chunks = []        current_chunk = ""for sent in sentences:if len(current_chunk) + len(sent) > 500:                chunks.append(Document(page_content=current_chunk))                current_chunk = sentelse:                current_chunk += sent + "。"if current_chunk:            chunks.append(Document(page_content=current_chunk))return chunkselse:# 策略4：固定长度（最简单，但不推荐）        splitter = CharacterTextSplitter(            chunk_size=500, chunk_overlap=100, separator=""        )return splitter.create_documents([text])# 使用示例chunks_recursive = chunk_document("技术文档.md", "recursive")chunks_md = chunk_document("技术文档.md", "markdown_header")chunks_semantic = chunk_document("技术文档.md", "semantic")print(f"递归分割: {len(chunks_recursive)}块, 语义分割: {len(chunks_semantic)}块")

本地Embedding模型的完整实现

除了调用OpenAI的API，本地部署BGE模型能省成本且中文效果更好：

from sentence_transformers import SentenceTransformerimport numpy as npclassLocalEmbedder:"""本地Embedding模型封装，支持BGE-M3和批量处理"""def__init__(self, model_name: str = "BAAI/bge-m3"):        self.model = SentenceTransformer(model_name)        self.dim = self.model.get_sentence_embedding_dimension()defencode(self, texts: list, batch_size: int = 32) -> np.ndarray:"""批量编码文本，返回numpy数组"""        embeddings = self.model.encode(            texts, batch_size=batch_size,            normalize_embeddings=True,  # 归一化，方便余弦相似度计算            show_progress_bar=True,        )return embeddingsdefencode_query(self, query: str) -> np.ndarray:"""对查询文本编码（BGE模型查询需要加前缀）"""return self.model.encode(            [f"为这个句子生成表示以用于检索相关文章：{query}"],            normalize_embeddings=True,        )[0]# 使用embedder = LocalEmbedder("BAAI/bge-m3")texts = ["Kubernetes是一个容器编排平台", "Docker是一种容器化技术"]embeddings = embedder.encode(texts)print(f"向量维度: {embedder.dim}, 形状: {embeddings.shape}")# 输出: 向量维度: 1024, 形状: (2, 1024)

注意BGE-M3的查询编码要加前缀"为这个句子生成表示以用于检索相关文章："，不加前缀的话检索精度会掉约5个百分点。这是BGE模型的特殊要求，其他模型不一定需要。

Chroma向量存储的完整实现（含元数据过滤）

import chromadbfrom chromadb.config import Settingsfrom datetime import datetimeclassChromaStore:"""Chroma向量数据库的完整封装"""def__init__(self, persist_path: str = "./chroma_db"):        self.client = chromadb.PersistentClient(            path=persist_path,            settings=Settings(anonymized_telemetry=False)        )defcreate_collection(self, name: str, dim: int = 1024):"""创建集合，使用余弦相似度"""try:            self.client.delete_collection(name)except:pass        self.collection = self.client.create_collection(            name=name,            metadata={"hnsw:space": "cosine"},        )return self.collectiondefinsert_batch(self, chunks: list, embeddings: np.ndarray,                     metadatas: list = None, batch_size: int = 100):"""批量插入，支持元数据"""        total = len(chunks)for i in range(0, total, batch_size):            batch = chunks[i:i+batch_size]            self.collection.add(                documents=[c.page_content for c in batch],                embeddings=embeddings[i:i+batch_size].tolist(),                ids=[f"chunk_{j}"for j in range(i, min(i+batch_size, total))],                metadatas=metadatas[i:i+batch_size] if metadatas elseNone,            )        print(f"写入 {total} 条记录")defsearch(self, query_embedding, top_k=5, filter_meta=None):"""向量检索，支持元数据过滤"""        kwargs = {"query_embeddings": [query_embedding.tolist()],"n_results": top_k,"include": ["documents", "distances", "metadatas"],        }if filter_meta:            kwargs["where"] = filter_meta  # 如 {"category": "技术文档"}return self.collection.query(**kwargs)

完整的RAG查询流程（含HyDE和Rerank）

下面是一个生产级的RAG查询实现，包含HyDE假设文档生成和检索后重排：

from openai import OpenAIfrom FlagEmbedding import FlagRerankerclassProductionRAG:"""生产级RAG查询，含HyDE和Rerank"""def__init__(self, embedder, store, llm_client, reranker_model="BAAI/bge-reranker-v2-m3"):        self.embedder = embedder        self.store = store        self.llm = llm_client        self.reranker = FlagReranker(reranker_model, use_fp16=True)defquery(self, question: str, top_k: int = 10, rerank_k: int = 5, use_hyde: bool = True):# 步骤1：HyDE——生成假设答案用于检索        search_query = questionif use_hyde:            hyde_prompt = f"请用一段话回答以下问题（不需要准确，只需要提供相关上下文）：\n{question}"            hyde_response = self.llm.chat.completions.create(                model="gpt-4o", messages=[{"role": "user", "content": hyde_prompt}],                max_tokens=200, temperature=0.5,            )            search_query = hyde_response.choices[0].message.content# 步骤2：向量检索        query_emb = self.embedder.encode_query(search_query)        results = self.store.search(query_emb, top_k=top_k)# 步骤3：Rerank——对检索结果重排序        pairs = [[question, doc] for doc in results["documents"][0]]        rerank_scores = self.reranker.compute_score(pairs)# 取重排后的top rerank_k        sorted_indices = np.argsort(rerank_scores)[::-1][:rerank_k]        final_docs = [results["documents"][0][i] for i in sorted_indices]        final_scores = [rerank_scores[i] for i in sorted_indices]# 步骤4：构建上下文并生成回答        context = "\n\n---\n\n".join(final_docs)        prompt = f"""基于以下资料回答问题。如果资料中没有相关信息，请明确说明"未找到相关信息"。资料：{context}问题：{question}回答："""        response = self.llm.chat.completions.create(            model="gpt-4o",            messages=[{"role": "user", "content": prompt}],            temperature=0.3,        )return {"answer": response.choices[0].message.content,"sources": final_docs,"scores": final_scores,"retrieved_count": len(results["documents"][0]),"reranked_count": len(final_docs),        }

这个实现的关键优化点：HyDE把检索从"问题-文档"匹配变成了"答案-文档"匹配，语义空间更接近；Rerank用Cross-encoder重新打分，比Bi-encoder的向量相似度准确得多。代价是多了两次模型调用（一次LLM生成假设答案，一次Reranker打分），总延迟增加约1-2秒，但检索精度提升约20%。

不同切片策略效果对比（含具体指标）

用同一份150页的中文技术文档（含Kubernetes、Docker、Spring Boot内容），在BGE-M3 + GPT-4o上测试：

结论：语义+递归混合策略在所有指标上都最优，但实现复杂度也是最高的。Markdown标题分割在语义完整性上满分，因为天然按文档结构切分。固定长度1000字符在所有指标上垫底，千万不要用。

不同Embedding模型效果对比（中文场景）

用同一份中文技术文档，在相同检索策略（递归分割500/100 + Chroma）下测试：

中文场景下，BGE-M3以Recall@5=0.91和MRR=0.88全面领先。OpenAI的embedding模型在中文上表现一般，text-embedding-3-small的中文Recall只有0.78，比BGE-M3低了13个百分点。如果你做的是中文RAG，强烈建议用BGE系列本地部署——精度高、免费、延迟低，一举三得。

RAG常见问题与解决方案

其中"父子文档"模式解决切片信息断裂的效果最好——检索时用小chunk（200-300字符），保证精度；生成答案时取回大chunk（1000-1500字符），保证上下文完整。实现上就是在Chroma里存两份：一份小chunk的embedding用于检索，一份大chunk的原文用于生成。

另外，跨语言检索问题在中英文混合文档中特别常见。比如用户问"Kubernetes的Service怎么用"，文档里可能写的是"Kubernetes Service usage guide"。BGE-M3天然支持多语言，可以缓解这个问题，但如果文档语种混杂严重，建议在Embedding之前统一做语言检测和翻译。