夜雨聆风去年公司要做个内部知识库问答系统，要求能检索上千份技术文档然后准确回答。我选了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系统上线后，我还做了几个优化：
HyDE（假设文档嵌入）：先让LLM生成一个假设答案，用这个假设答案去检索，比直接用问题检索效果好 检索后重排：用BGE-Reranker对初步检索结果重排，精度提升明显 元数据过滤：在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之前统一做语言检测和翻译。
