生产级RAG系统实战：从数据分块到混合检索的完整构建指南

1. 从零到一：我如何构建一个生产级的RAG系统

如果你和我一样，在过去一年里被各种大模型应用刷屏，并且尝试过用ChatGPT API直接构建一个问答机器人，那你大概率会遇到一个核心痛点：模型会一本正经地“胡说八道”。它可能会告诉你一个你公司内部文档里根本不存在的流程，或者编造一个看似合理但完全错误的答案。这就是所谓的“幻觉”问题。为了解决这个问题，检索增强生成技术，也就是RAG，迅速成为了构建可靠、可信大模型应用的事实标准。

我最近花了几个月时间，系统性地梳理和实践了RAG的整个技术栈，从最基础的数据加载到复杂的多模态检索优化，最终落地了一个能够处理公司内部数万份文档的智能知识库系统。这个过程踩了无数的坑，也积累了不少实战经验。今天，我想抛开那些高大上的概念，从一个一线开发者的角度，和你分享如何一步步搭建一个真正能用的、性能还不错的RAG系统。我会重点讲清楚每个环节“为什么”要这么做，以及我在实操中遇到的“坑”和解决方案。无论你是刚接触RAG的新手，还是已经有一定经验想深入优化，相信这篇长文都能给你带来一些实实在在的启发。

2. 核心思路拆解：RAG不是简单的“搜索+生成”

很多人对RAG的第一印象是“先用向量数据库搜一下，再把结果喂给大模型生成答案”。这个理解没错，但过于简化了。一个生产级的RAG系统，其复杂度和需要考虑的细节远超这个简单的两步流程。我的核心思路是将其视为一个由数据流、索引流和查询流三条主线构成的系统工程。

2.1 数据流：从原始文档到可检索的知识单元

数据是RAG的基石，垃圾进，垃圾出。数据流的核心目标是将各种格式、各种来源的非结构化文档（PDF、Word、网页、图片），转化为干净、结构化的“知识片段”。这里有几个关键决策点：

为什么需要文本分块？直接把整篇文档扔给模型不行吗？理论上可以，但实践中问题很大。首先，大模型有上下文长度限制，长文档会被截断。其次，检索精度会下降。想象一下，一篇50页的技术白皮书被编码成一个向量，当用户问一个非常具体的问题（比如“第三章第二节提到的API参数timeout默认值是多少？”），这个涵盖全篇的向量很难精准匹配到这个细节。因此，分块是将知识“颗粒化”，提高检索命中率的关键。

分块策略的权衡：固定大小 vs. 语义分割最简单的分块是按固定字符数（比如512个字符）滑动窗口切割。我刚开始就用这个，实现简单，但效果很糟糕。一个完整的表格被切成了两半，一个代码块被拦腰斩断，检索出来的片段前言不搭后语。 后来我转向了基于语义的分割，比如使用LangChain的RecursiveCharacterTextSplitter，它会优先按段落、换行符、句号等自然边界进行分割，尽量保证每个“块”在语义上是完整的。更进一步，对于技术文档，我引入了基于代码语法树的分割器，确保函数定义、类定义不被破坏。这里的经验是：没有一种分块策略适合所有场景。法律合同可能适合按条款分，技术文档适合按章节或函数分，聊天记录适合按对话轮次分。你需要根据你的数据特性进行定制。

一个容易被忽略的环节：元数据附加分块之后，每个“块”就变成了一个孤岛。为了在后续检索和生成中提供上下文，必须为每个块附加丰富的元数据。我通常会附加：

• source: 原始文档路径或URL。
• page_number: 在PDF或文档中的页码。
• section_title: 该块所属的章节标题。
• chunk_index: 在该文档中的顺序索引。 这些元数据在后续的“重排序”和“引用溯源”环节至关重要。大模型在生成答案时，可以明确告知用户“该信息来源于XX文档第Y页”，极大增强了可信度。

2.2 索引流：把知识“装进”向量数据库

有了干净的数据块，下一步就是为它们创建索引，以便快速检索。这里的核心是“向量化”和“向量数据库”。

嵌入模型的选择：通用 vs. 领域适配嵌入模型负责将文本转换为高维向量（比如768维或1024维）。text-embedding-ada-002曾是开源标杆，但现在有更多选择。我对比过BGE、M3E和OpenAI的嵌入模型。如果你的数据是通用领域（如维基百科），text-embedding-3-small效果很好且性价比高。但我的数据是垂直领域的金融科技文档，测试发现，在MTEB中文榜单上排名靠前的BGE-large-zh和M3E-large在我的任务上表现明显更好，因为它们用更多中文语料进行了训练。

注意：嵌入模型一旦选定，后续就不能轻易更换，除非你愿意重新为所有数据生成向量并重建索引。所以前期花时间做一个小规模的评测（比如100个问答对）是非常值得的。

向量数据库的选型：Milvus、Chroma还是Pgvector？这是一个工程决策。我最终选择了Milvus，原因如下：

1. 性能与规模：Milvus是专为向量检索设计的分布式系统，当你的数据量达到百万甚至千万级时，它的性能优势非常明显。支持GPU加速索引构建和查询，速度极快。
2. 丰富的索引类型：除了最常用的IVF_FLAT，还支持HNSW、SCANN等。HNSW图索引在追求极高召回率的场景下非常有用，虽然建索引慢、内存占用大，但查询速度极快。
3. 生产级特性：支持高可用、数据持久化、多租户、权限控制，这些都是线上系统必须考虑的。相比之下，Chroma更轻量，适合原型快速验证，但在数据持久化和并发性能上不如Milvus成熟。Pgvector适合那些已经重度使用PostgreSQL，且向量数据量不大的团队，可以省去维护另一个数据库的复杂度。

索引参数调优：IVF_FLAT的nlist怎么设？以Milvus最常用的IVF_FLAT索引为例，nlist这个参数至关重要。它决定了将向量空间聚成多少类。我的经验法则是：nlist = sqrt(总向量数)。例如，我有100万个向量，nlist可以设为1000。这个值太小，每个聚类里的向量太多，查询时要扫描的向量就多，速度慢；这个值太大，聚类本身的计算开销大，且可能因为数据分布不均导致空聚类。在实际部署前，一定要用你的真实数据集，在nlist为sqrt(N)的附近进行性能测试（查询速度vs召回率），找到最佳平衡点。

2.3 查询流：从用户问题到精准答案

这是用户直接感知的环节。流程是：用户提问 -> 查询理解/重构 -> 向量检索 -> 重排序 -> 上下文组装 -> 大模型生成。

查询重构：让问题变得更“好搜”用户的原始提问可能很模糊。例如，“怎么设置？” 这种问题对于检索系统是灾难。我们需要进行查询重构。简单的方法可以使用大模型进行查询扩展，比如将“怎么设置？” 扩展为“如何配置系统参数？请提供具体步骤和注意事项。”。更高级的可以用HyDE技术，让模型先根据问题生成一个假设的答案段落，然后用这个生成的段落去检索，有时能奇迹般地找到更相关的文档。

混合检索：向量搜索的“黄金搭档”单纯依赖向量检索（稠密检索）有时会漏掉一些关键词匹配的精确结果。例如，搜索“Python 3.12的新特性”，一些文档可能用“What‘s new in Python 3.12”作为标题，向量相似度可能不高，但关键词匹配度很高。因此，我引入了混合检索：同时进行向量检索和传统的关键词检索（如BM25），然后将两者的结果按分数融合。LangChain和LlamaIndex都提供了现成的融合器，如reciprocal_rank_fusion。实测下来，混合检索的召回率比单一方法有显著提升。

重排序：给检索结果“去粗取精”向量检索返回的Top K个结果（比如K=10），其相似度分数可能相差不大，但相关性却有高低。直接把这10个片段都塞给大模型，会引入噪声，浪费上下文窗口。这里需要引入一个重排序模型。它是一个轻量级的、专门做文本相关性判别的模型（如BGE-reranker、Cohere rerank）。它的作用是：对初步检索出的10个片段，根据原始问题重新打分和排序，只保留Top N个（比如N=3）最相关的片段送给大模型。这一步成本很低（相比大模型生成），但能极大提升最终答案的质量和准确性，是我强烈推荐的必做优化。

3. 实战搭建：一个模块化的RAG系统实现

光说不练假把式。下面我以一个企业内部知识库为例，拆解核心模块的代码实现和配置。我选择LangChain作为主要框架，因为它生态丰富，但也会指出其抽象可能带来的性能问题及解决方案。

3.1 数据准备模块：打造健壮的文档处理流水线

这个模块的目标是：无论来的是什么格式的文档，都能稳定地输出结构化的文本块。我将其设计为一个可插拔的管道。

# 核心数据加载与处理管道 import os from langchain_community.document_loaders import ( PyPDFLoader, UnstructuredWordDocumentLoader, UnstructuredFileLoader, ) from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document from typing import List, Optional import hashlib class RobustDocumentProcessor: def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200): # 根据经验，1000字符的块大小，配合200字符的重叠，能在语义完整性和检索粒度间取得较好平衡 self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""], # 中文友好的分隔符 length_function=len, ) self._loaders = { '.pdf': PyPDFLoader, '.docx': UnstructuredWordDocumentLoader, '.doc': UnstructuredWordDocumentLoader, '.txt': UnstructuredFileLoader, '.md': UnstructuredFileLoader, } def load_single_document(self, file_path: str) -> List[Document]: """加载单个文档，自动识别格式""" ext = os.path.splitext(file_path)[1].lower() if ext not in self._loaders: # 兜底方案，用Unstructured尝试解析 loader = UnstructuredFileLoader(file_path, mode="elements") else: loader_class = self._loaders[ext] loader = loader_class(file_path) try: docs = loader.load() # 为每个文档添加唯一ID和源信息 for doc in docs: doc.metadata["source"] = file_path doc.metadata["file_hash"] = self._generate_file_hash(file_path) # 尝试提取标题，作为后续分块的参考 if "title" not in doc.metadata: # 简单启发式：第一行非空内容可能是标题 lines = doc.page_content.strip().split('\n') if lines: doc.metadata["title"] = lines[0][:100] # 截断避免过长 return docs except Exception as e: print(f"加载文档 {file_path} 失败: {e}") # 生产环境应记录日志并跳过，而非中断 return [] def _generate_file_hash(self, file_path: str) -> str: """生成文件哈希，用于内容变更检测，避免重复处理""" with open(file_path, 'rb') as f: return hashlib.md5(f.read()).hexdigest() def split_documents(self, documents: List[Document]) -> List[Document]: """分割文档，并增强元数据""" all_chunks = [] for doc in documents: chunks = self.text_splitter.split_documents([doc]) for idx, chunk in enumerate(chunks): # 为每个块附加更多上下文信息 chunk.metadata.update({ "chunk_id": f"{doc.metadata['file_hash']}_{idx}", "chunk_index": idx, "parent_title": doc.metadata.get("title", ""), }) # 如果原文档有页码，继承过来 if "page" in doc.metadata: chunk.metadata["page"] = doc.metadata["page"] all_chunks.extend(chunks) return all_chunks # 使用示例 processor = RobustDocumentProcessor(chunk_size=800, chunk_overlap=150) # 对于中文，800字符可能更合适 raw_docs = processor.load_single_document("产品手册.pdf") chunks = processor.split_documents(raw_docs) print(f"原始文档分割为 {len(chunks)} 个块。")

实操心得：

1. 错误处理是关键：线上环境会遇到各种奇葩文档（加密PDF、损坏的Word文件）。加载器必须要有健壮的try-catch，并记录失败日志，让流程能继续处理其他文件，而不是整体崩溃。
2. 元数据是黄金：尽可能多地从原始文档中提取和保留元数据（标题、作者、章节、页码）。这些信息在后续的检索结果展示和答案溯源时价值连城。
3. 分块大小不是固定的：我后来改进了这个类，允许根据文档类型动态调整分块策略。比如，对于API文档，我使用了一个基于Markdown标题的分割器，确保每个函数或类定义在一个独立的块里。

3.2 索引构建模块：与Milvus深度集成

这里展示如何将处理好的文本块，通过嵌入模型向量化后，存入Milvus。

# 索引构建与向量入库 from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Milvus from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType, utility import time class MilvusIndexBuilder: def __init__(self, host: str, port: str, embedding_model_name: str = "BAAI/bge-large-zh"): # 连接Milvus connections.connect(host=host, port=port) self.collection_name = "rag_knowledge_base" self.embedding_model = HuggingFaceEmbeddings( model_name=embedding_model_name, model_kwargs={'device': 'cuda'}, # 使用GPU加速 encode_kwargs={'normalize_embeddings': True} # 归一化，对余弦相似度很重要 ) self.dim = 1024 # BGE-large-zh的向量维度 def create_collection_if_not_exists(self): """创建Milvus集合（表），如果不存在""" if utility.has_collection(self.collection_name): print(f"集合 {self.collection_name} 已存在。") return Collection(self.collection_name) # 1. 定义字段 fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="chunk_id", dtype=DataType.VARCHAR, max_length=256), FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=self.dim), # 元数据字段 FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=512), FieldSchema(name="page", dtype=DataType.INT64), FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=512), ] # 2. 定义Schema schema = CollectionSchema(fields, description="RAG知识库向量集合") # 3. 创建集合 collection = Collection(name=self.collection_name, schema=schema) print(f"集合 {self.collection_name} 创建成功。") # 4. 创建索引（在插入数据后创建性能更好，这里先定义） # 我们将在插入数据后统一创建索引 return collection def index_documents(self, chunks: List[Document], batch_size: int = 100): """将文档块向量化并索引到Milvus""" collection = self.create_collection_if_not_exists() # 准备批量插入的数据 texts = [chunk.page_content for chunk in chunks] metadatas = [chunk.metadata for chunk in chunks] print(f"开始生成 {len(texts)} 个文本的向量...") start = time.time() # 批量生成向量 embeddings = self.embedding_model.embed_documents(texts) print(f"向量生成完成，耗时 {time.time() - start:.2f} 秒。") # 准备插入Milvus的数据 data = [ [meta.get("chunk_id", "") for meta in metadatas], # chunk_id texts, # text embeddings, # embedding [meta.get("source", "") for meta in metadatas], # source [meta.get("page", 0) for meta in metadatas], # page [meta.get("parent_title", "") for meta in metadatas], # title ] # 分批插入，避免单次请求过大 total = len(texts) for i in range(0, total, batch_size): end_idx = min(i + batch_size, total) batch_data = [col[i:end_idx] for col in data] insert_result = collection.insert(batch_data) print(f"已插入批次 {i//batch_size + 1}, 条数: {end_idx - i}") print("数据插入完成。") # 数据插入后，创建索引 print("开始创建向量索引...") index_params = { "metric_type": "IP", # 内积，因为我们的向量是归一化的，内积等价于余弦相似度 "index_type": "IVF_FLAT", "params": {"nlist": 1024}, # 根据数据量调整，这里假设数据量在百万级 } collection.create_index(field_name="embedding", index_params=index_params) print("向量索引创建成功。") # 将集合加载到内存，加速查询 collection.load() print("集合已加载到内存。") return collection # 使用示例 index_builder = MilvusIndexBuilder(host="localhost", port="19530") # chunks 是上一节处理好的文档块列表 collection = index_builder.index_documents(chunks)

关键参数解析与避坑指南：

1. normalize_embeddings=True：这是很多新手会忽略但极其重要的一步。嵌入模型输出的向量如果不归一化（即模长不为1），那么使用内积计算相似度时，向量的模长会影响结果。归一化后，内积就等于余弦相似度，相似度计算更准确。Milvus的IP（内积）度量方式配合归一化向量是标准做法。
2. 索引创建时机：一定要在数据插入完毕后再创建索引。如果在空集合上先创建索引，再插入数据，Milvus不会自动更新索引，导致查询时走暴力扫描，速度极慢。
3. nlist参数：如前所述，需要根据数据量估算。对于这个示例，如果chunks数量在100万左右，nlist=1024是个合理的起点。你可以在插入少量数据后，用collection.search进行性能测试，调整nlist以达到查询速度和召回率的平衡。
4. collection.load()：创建索引后，必须调用load()方法将集合加载到内存，否则无法进行检索。在生产环境，如果内存紧张，可以考虑只加载索引文件，但性能会有所下降。

3.3 检索与生成模块：实现混合检索与重排序

这是RAG的查询端，集成了混合检索、重排序和大模型调用。

# 高级检索与生成链 from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain_community.vectorstores import Milvus from langchain_core.runnables import RunnablePassthrough from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from typing import List, Tuple import numpy as np class AdvancedRAGQueryEngine: def __init__(self, collection_name: str, embedding_model, llm_api_key: str): self.vector_store = Milvus( embedding_function=embedding_model, collection_name=collection_name, connection_args={"host": "localhost", "port": "19530"} ) # 初始化一个内存中的BM25检索器（需要先有文本列表） # 注意：这里仅为示例，生产环境BM25索引应持久化（如用Elasticsearch） self._all_texts_for_bm25 = [] # 需要在初始化后从向量库或别处加载 self.bm25_retriever = None # 初始化大模型（这里以OpenAI GPT-4为例，可替换为国内模型） self.llm = ChatOpenAI( model="gpt-4-turbo-preview", api_key=llm_api_key, temperature=0.1, # 低温度，保证答案稳定性 max_tokens=2000 ) # 定义提示词模板 self.prompt_template = ChatPromptTemplate.from_messages([ ("system", "你是一个专业的助手，请严格根据以下提供的上下文信息来回答问题。如果上下文信息不足以回答问题，请直接说“根据提供的信息，我无法回答这个问题”，不要编造信息。"), ("human", "上下文：\n{context}\n\n问题：{question}") ]) def _initialize_bm25(self, texts: List[str]): """初始化BM25检索器。在实际应用中，文本列表应从数据库或文件加载。""" from langchain.retrievers import BM25Retriever from langchain.schema import Document docs = [Document(page_content=text) for text in texts] self.bm25_retriever = BM25Retriever.from_documents(docs, k=10) # 初步检索10个 self._all_texts_for_bm25 = texts def hybrid_retrieve(self, query: str, top_k_vector: int = 10, top_k_bm25: int = 10) -> List[Tuple[str, float, dict]]: """执行混合检索，返回（文本，分数，元数据）列表""" # 1. 向量检索 vector_results = self.vector_store.similarity_search_with_score(query, k=top_k_vector) # 结果格式: [(Document, score), ...] # 2. BM25检索 (如果已初始化) bm25_results = [] if self.bm25_retriever: bm25_docs = self.bm25_retriever.get_relevant_documents(query) # 为BM25结果赋予一个分数（这里简化处理，可按排名给分） for i, doc in enumerate(bm25_docs): # 分数可以简单化为 1/(rank+1)，排名越靠前分数越高 score = 1.0 / (i + 1) bm25_results.append((doc.page_content, score, doc.metadata)) # 3. 结果融合 (简化版加权平均) all_results = {} # 处理向量结果 for doc, score in vector_results: # Milvus IP分数范围可能很大，这里做归一化到[0,1]的简化处理 norm_score = (score + 1) / 2 if score < 1 else 1.0 # 假设IP分数在[-1,1]附近 key = doc.page_content[:100] # 用文本前100字符作为去重key（不严谨，仅示例） if key not in all_results: all_results[key] = {"text": doc.page_content, "score": 0.0, "metadata": doc.metadata} all_results[key]["score"] += norm_score * 0.7 # 向量检索权重0.7 # 处理BM25结果 for text, score, metadata in bm25_results: key = text[:100] if key not in all_results: all_results[key] = {"text": text, "score": 0.0, "metadata": metadata} all_results[key]["score"] += score * 0.3 # BM25检索权重0.3 # 按融合分数排序 sorted_items = sorted(all_results.values(), key=lambda x: x["score"], reverse=True) return [(item["text"], item["score"], item["metadata"]) for item in sorted_items] def rerank(self, query: str, candidates: List[Tuple[str, float, dict]], top_n: int = 3): """使用重排序模型对候选结果进行精排""" # 这里使用一个轻量级重排序模型，例如BGE Reranker # 假设我们有一个本地部署的BGE Reranker API # 实际使用时，可调用 `from FlagEmbedding import FlagReranker` reranked = [] # 模拟重排序过程：这里简化实现，实际应调用模型API # 真实代码示例（需安装FlagEmbedding）: # from FlagEmbedding import FlagReranker # reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True) # pairs = [[query, cand[0]] for cand in candidates] # scores = reranker.compute_score(pairs) # reranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True) # 为演示，我们假设前top_n个就是重排序后的结果 reranked = candidates[:top_n] print(f"重排序完成，保留Top-{top_n}个最相关片段。") return reranked def generate_answer(self, query: str, context_docs: List[Tuple[str, float, dict]]) -> str: """根据检索到的上下文生成答案""" # 组装上下文 context_text = "\n\n---\n\n".join([ f"来源：{meta.get('source', '未知')} (页码：{meta.get('page', 'N/A')})\n内容：{text}" for text, score, meta in context_docs ]) # 构造提示词 prompt = self.prompt_template.invoke({"context": context_text, "question": query}) # 调用大模型 response = self.llm.invoke(prompt) answer = response.content # 附上引用来源 sources = list(set([meta.get('source', '未知') for _, _, meta in context_docs])) answer += f"\n\n**参考来源**：{', '.join(sources)}" return answer def query(self, question: str) -> str: """完整的查询流程""" print(f"用户问题：{question}") # 1. 混合检索 print("执行混合检索...") candidates = self.hybrid_retrieve(question, top_k_vector=10, top_k_bm25=10) # 2. 重排序 print("执行重排序...") reranked = self.rerank(question, candidates, top_n=3) # 3. 生成答案 print("生成答案...") answer = self.generate_answer(question, reranked) return answer # 使用示例 # 注意：需要先初始化BM25检索器所需的数据 # query_engine = AdvancedRAGQueryEngine("rag_knowledge_base", embedding_model, "your_openai_key") # query_engine._initialize_bm25([chunk.page_content for chunk in chunks]) # 传入所有文本 # result = query_engine.query("我们产品的退货政策是什么？") # print(result)

核心要点与优化方向：

1. 混合检索权重：向量检索和BM25的权重（0.7和0.3）需要根据你的数据和查询类型进行A/B测试调整。对于事实性、关键词明确的问题，BM25权重可以调高；对于语义复杂、需要理解意图的问题，向量检索权重应更高。
2. 重排序是性价比最高的优化：即使你暂时无法实现复杂的混合检索，也强烈建议加上重排序这一步。一个轻量级的重排序模型（如BGE-reranker，仅几百MB）可以过滤掉大量不相关片段，显著提升最终答案质量，而增加的延迟通常只有几十毫秒。
3. 提示词工程：系统提示词（system）至关重要。我强调“严格根据上下文”和“不要编造”，这能有效抑制幻觉。你还可以在提示词中要求模型以特定格式（如Markdown列表、表格）输出，或者先判断问题是否可回答。
4. 引用溯源：在答案末尾附上参考来源，不仅能增加可信度，也方便用户追溯和验证。这是生产级RAG应用的必备特性。

4. 性能调优与问题排查实录

系统搭起来只是第一步，让它跑得快、答得准才是真正的挑战。下面是我在真实项目中遇到的一些典型问题及解决方案。

4.1 检索速度慢，响应延迟高

现象：用户查询需要等待5-10秒才返回结果，体验很差。

排查与解决：

1. 检查向量索引：首先确认Milvus集合是否已创建索引并加载到内存。通过Milvus的collection.index()和collection.load()状态确认。
2. 调整检索参数：Milvus的search接口有一个search_params参数。对于IVF_FLAT索引，关键参数是nprobe，它代表搜索时探查的聚类数量。默认值可能较小。适当增加nprobe可以提升召回率，但会降低速度。我的经验是，在保证召回率的前提下，找到最小的nprobe。可以通过在测试集上绘制nprobe-召回率曲线来确定。

   # 在查询时指定search_params search_params = {"metric_type": "IP", "params": {"nprobe": 20}} # 调整nprobe results = collection.search(vectors=[query_vector], anns_field="embedding", param=search_params, limit=10)

3. 审视查询流程：是否在每次查询时都重新计算查询问题的向量？可以将嵌入模型缓存起来，或者对常见问题做向量预计算。另外，检查重排序模型调用是否成为瓶颈，考虑将其服务化并用批量推理。
4. 硬件与部署：Milvus的索引构建和查询可以受益于GPU。确保你的Milvus是GPU版本，并且embedding模型也跑在GPU上。对于超大规模数据（亿级），需要考虑Milvus的分布式部署。

4.2 答案不准确或包含幻觉

现象：模型给出的答案与提供的上下文不符，或者凭空捏造细节。

排查与解决：

1. 增强检索相关性：这是根本。首先检查检索到的Top K个片段是否真的与问题相关。可以人工审核一批查询的检索结果。如果不相关，回溯检查：
   • 嵌入模型是否匹配领域？用你的领域数据做一个相似度匹配的小测试。
   • 分块是否合理？是不是块太大导致噪声多，或者块太小导致信息不完整？尝试调整分块策略。
   • 是否使用了混合检索和重排序？这是提升相关性的最有效手段。
2. 优化提示词：在系统提示词中更严厉地约束模型。例如：“你必须且只能使用以下上下文信息。如果答案不在上下文中，请说‘我不知道’。严禁编造任何信息。” 同时，在上下文中明确标注每个片段的来源，让模型“知道”它应该引用。
3. 实施“答案可验证性”检查：在生成答案后，增加一个验证步骤。用答案中的关键事实作为查询，再次进行检索，检查这些事实是否存在于检索到的上下文中。如果不存在，则判定为幻觉，可以要求模型重新生成或直接返回“无法确认”。
4. 降低模型“温度”：将LLM的temperature参数设为0或接近0（如0.1），这会使模型的输出更确定、更少“创造性”，从而减少幻觉。

4.3 如何处理长文档和复杂多轮对话？

现象：文档很长，超过了模型上下文窗口；或者用户的问题需要结合多轮对话历史来理解。

解决方案：

1. 针对长文档：采用“分层检索”或“映射-归约”策略。首先，为整个文档生成一个摘要或提取关键主题，并将其向量化存储。当用户查询时，先匹配到相关文档或章节，然后再深入该章节内部进行更细粒度的检索。LlamaIndex在这方面提供了很好的抽象，如SummaryIndex和TreeIndex。
2. 针对多轮对话：需要维护对话历史。将历史对话（包括问题和之前的答案）也作为上下文的一部分。但要注意不能无限堆积，会导致上下文爆炸。常见的策略是：
   • 只保留最近N轮对话。
   • 对历史对话进行总结，将总结后的文本作为上下文，而不是完整的对话记录。
   • 在查询时，将当前问题与最近的历史对话进行拼接或重写，形成一个更完整的查询语句，再去检索。例如，用户问“它有什么优点？”，系统需要自动将上文提到的“产品X”补充进去，形成“产品X有什么优点？”的查询。

4.4 系统评估：如何量化RAG的好坏？

不能凭感觉说系统“好”或“不好”，需要建立评估体系。我主要从三个维度评估：

1. 检索质量：

   • 命中率：对于一组标准问题，检索到的Top K个片段中包含正确答案的比例。
   • 平均排序倒数：正确答案在检索结果中的平均排名的倒数。这个值越高，说明正确答案排得越靠前。
   • 可以使用ragas、TruLens等专门评估RAG的库来计算这些指标。

2. 生成质量：

   • 事实一致性：生成的答案与提供的上下文事实是否一致。可以用NLI模型自动判断。
   • 答案相关性：生成的答案是否直接回答了问题。
   • 信息完整性：答案是否涵盖了上下文中所有关键信息点。
   • 这部分通常需要人工标注一部分测试集，或者利用GPT-4作为裁判进行自动评估。

3. 系统性能：

   • 端到端延迟：从用户提问到收到答案的总时间。区分检索时间、重排序时间、LLM生成时间。
   • 吞吐量：系统每秒能处理多少查询。
   • 成本：每次查询消耗的Token数（特别是LLM调用）和向量数据库的计算资源。

建立一个持续运行的评估流水线，每次对系统做重大改动（如更换嵌入模型、调整分块大小）后，都跑一遍评估集，用数据说话，是保证系统持续优化的唯一途径。

5. 进阶思考：从RAG到智能体

当你把基础的RAG系统跑通并优化稳定后，可以开始思考更高级的架构。RAG本质上是一个“检索-生成”的固定管道。而智能体赋予了系统决策和调用工具的能力。

一个简单的演进思路是：让RAG系统成为智能体手中的一个“工具”。智能体根据用户的问题，决定是否需要调用RAG工具（查询知识库），还是调用其他工具（如计算器、API查询、数据库查询）。例如，用户问“去年我们部门在A项目上的开支是多少？”，智能体可以分解任务：1）调用RAG工具，检索“A项目的财务报告”；2）从报告中提取数字；3）调用计算工具进行汇总。

LangChain和LlamaIndex都提供了构建智能体的高级框架。你可以将现有的RAG查询引擎封装成一个Tool，然后让一个主控LLM（如GPT-4）来协调多个工具的调用。这会让你的应用从“问答机”升级为“智能助手”，能够处理更复杂、多步骤的任务。

这条路挑战更大，涉及到智能体的规划、工具调用纠错、长期记忆管理等。但这也是当前大模型应用最前沿和最有价值的方向之一。我的建议是，先夯实RAG基础，再逐步向智能体架构探索。

整个搭建和优化RAG系统的过程，就像在组装一个精密的仪器。每个环节——数据清洗、分块、嵌入、索引、检索、重排序、提示工程——都需要仔细调试。没有银弹，最好的系统永远是针对你的特定数据和业务场景深度定制出来的。希望我分享的这些实战经验和踩过的坑，能帮你少走一些弯路，更快地构建出属于你自己的、高效可靠的智能知识系统。