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原因如下性能与规模Milvus是专为向量检索设计的分布式系统当你的数据量达到百万甚至千万级时它的性能优势非常明显。支持GPU加速索引构建和查询速度极快。丰富的索引类型除了最常用的IVF_FLAT还支持HNSW、SCANN等。HNSW图索引在追求极高召回率的场景下非常有用虽然建索引慢、内存占用大但查询速度极快。生产级特性支持高可用、数据持久化、多租户、权限控制这些都是线上系统必须考虑的。相比之下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个结果比如K10其相似度分数可能相差不大但相关性却有高低。直接把这10个片段都塞给大模型会引入噪声浪费上下文窗口。这里需要引入一个重排序模型。它是一个轻量级的、专门做文本相关性判别的模型如BGE-reranker、Cohere rerank。它的作用是对初步检索出的10个片段根据原始问题重新打分和排序只保留Top N个比如N3最相关的片段送给大模型。这一步成本很低相比大模型生成但能极大提升最终答案的质量和准确性是我强烈推荐的必做优化。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_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , , ], # 中文友好的分隔符 length_functionlen, ) 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, modeelements) 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_size800, chunk_overlap150) # 对于中文800字符可能更合适 raw_docs processor.load_single_document(产品手册.pdf) chunks processor.split_documents(raw_docs) print(f原始文档分割为 {len(chunks)} 个块。)实操心得错误处理是关键线上环境会遇到各种奇葩文档加密PDF、损坏的Word文件。加载器必须要有健壮的try-catch并记录失败日志让流程能继续处理其他文件而不是整体崩溃。元数据是黄金尽可能多地从原始文档中提取和保留元数据标题、作者、章节、页码。这些信息在后续的检索结果展示和答案溯源时价值连城。分块大小不是固定的我后来改进了这个类允许根据文档类型动态调整分块策略。比如对于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(hosthost, portport) self.collection_name rag_knowledge_base self.embedding_model HuggingFaceEmbeddings( model_nameembedding_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(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namechunk_id, dtypeDataType.VARCHAR, max_length256), FieldSchema(nametext, dtypeDataType.VARCHAR, max_length65535), FieldSchema(nameembedding, dtypeDataType.FLOAT_VECTOR, dimself.dim), # 元数据字段 FieldSchema(namesource, dtypeDataType.VARCHAR, max_length512), FieldSchema(namepage, dtypeDataType.INT64), FieldSchema(nametitle, dtypeDataType.VARCHAR, max_length512), ] # 2. 定义Schema schema CollectionSchema(fields, descriptionRAG知识库向量集合) # 3. 创建集合 collection Collection(nameself.collection_name, schemaschema) 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_nameembedding, index_paramsindex_params) print(向量索引创建成功。) # 将集合加载到内存加速查询 collection.load() print(集合已加载到内存。) return collection # 使用示例 index_builder MilvusIndexBuilder(hostlocalhost, port19530) # chunks 是上一节处理好的文档块列表 collection index_builder.index_documents(chunks)关键参数解析与避坑指南normalize_embeddingsTrue这是很多新手会忽略但极其重要的一步。嵌入模型输出的向量如果不归一化即模长不为1那么使用内积计算相似度时向量的模长会影响结果。归一化后内积就等于余弦相似度相似度计算更准确。Milvus的IP内积度量方式配合归一化向量是标准做法。索引创建时机一定要在数据插入完毕后再创建索引。如果在空集合上先创建索引再插入数据Milvus不会自动更新索引导致查询时走暴力扫描速度极慢。nlist参数如前所述需要根据数据量估算。对于这个示例如果chunks数量在100万左右nlist1024是个合理的起点。你可以在插入少量数据后用collection.search进行性能测试调整nlist以达到查询速度和召回率的平衡。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_functionembedding_model, collection_namecollection_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( modelgpt-4-turbo-preview, api_keyllm_api_key, temperature0.1, # 低温度保证答案稳定性 max_tokens2000 ) # 定义提示词模板 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_contenttext) for text in texts] self.bm25_retriever BM25Retriever.from_documents(docs, k10) # 初步检索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, ktop_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/(rank1)排名越靠前分数越高 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(), keylambda x: x[score], reverseTrue) 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_fp16True) # pairs [[query, cand[0]] for cand in candidates] # scores reranker.compute_score(pairs) # reranked sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue) # 为演示我们假设前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_vector10, top_k_bm2510) # 2. 重排序 print(执行重排序...) reranked self.rerank(question, candidates, top_n3) # 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)核心要点与优化方向混合检索权重向量检索和BM25的权重0.7和0.3需要根据你的数据和查询类型进行A/B测试调整。对于事实性、关键词明确的问题BM25权重可以调高对于语义复杂、需要理解意图的问题向量检索权重应更高。重排序是性价比最高的优化即使你暂时无法实现复杂的混合检索也强烈建议加上重排序这一步。一个轻量级的重排序模型如BGE-reranker仅几百MB可以过滤掉大量不相关片段显著提升最终答案质量而增加的延迟通常只有几十毫秒。提示词工程系统提示词system至关重要。我强调“严格根据上下文”和“不要编造”这能有效抑制幻觉。你还可以在提示词中要求模型以特定格式如Markdown列表、表格输出或者先判断问题是否可回答。引用溯源在答案末尾附上参考来源不仅能增加可信度也方便用户追溯和验证。这是生产级RAG应用的必备特性。4. 性能调优与问题排查实录系统搭起来只是第一步让它跑得快、答得准才是真正的挑战。下面是我在真实项目中遇到的一些典型问题及解决方案。4.1 检索速度慢响应延迟高现象用户查询需要等待5-10秒才返回结果体验很差。排查与解决检查向量索引首先确认Milvus集合是否已创建索引并加载到内存。通过Milvus的collection.index()和collection.load()状态确认。调整检索参数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_fieldembedding, paramsearch_params, limit10)审视查询流程是否在每次查询时都重新计算查询问题的向量可以将嵌入模型缓存起来或者对常见问题做向量预计算。另外检查重排序模型调用是否成为瓶颈考虑将其服务化并用批量推理。硬件与部署Milvus的索引构建和查询可以受益于GPU。确保你的Milvus是GPU版本并且embedding模型也跑在GPU上。对于超大规模数据亿级需要考虑Milvus的分布式部署。4.2 答案不准确或包含幻觉现象模型给出的答案与提供的上下文不符或者凭空捏造细节。排查与解决增强检索相关性这是根本。首先检查检索到的Top K个片段是否真的与问题相关。可以人工审核一批查询的检索结果。如果不相关回溯检查嵌入模型是否匹配领域用你的领域数据做一个相似度匹配的小测试。分块是否合理是不是块太大导致噪声多或者块太小导致信息不完整尝试调整分块策略。是否使用了混合检索和重排序这是提升相关性的最有效手段。优化提示词在系统提示词中更严厉地约束模型。例如“你必须且只能使用以下上下文信息。如果答案不在上下文中请说‘我不知道’。严禁编造任何信息。” 同时在上下文中明确标注每个片段的来源让模型“知道”它应该引用。实施“答案可验证性”检查在生成答案后增加一个验证步骤。用答案中的关键事实作为查询再次进行检索检查这些事实是否存在于检索到的上下文中。如果不存在则判定为幻觉可以要求模型重新生成或直接返回“无法确认”。降低模型“温度”将LLM的temperature参数设为0或接近0如0.1这会使模型的输出更确定、更少“创造性”从而减少幻觉。4.3 如何处理长文档和复杂多轮对话现象文档很长超过了模型上下文窗口或者用户的问题需要结合多轮对话历史来理解。解决方案针对长文档采用“分层检索”或“映射-归约”策略。首先为整个文档生成一个摘要或提取关键主题并将其向量化存储。当用户查询时先匹配到相关文档或章节然后再深入该章节内部进行更细粒度的检索。LlamaIndex在这方面提供了很好的抽象如SummaryIndex和TreeIndex。针对多轮对话需要维护对话历史。将历史对话包括问题和之前的答案也作为上下文的一部分。但要注意不能无限堆积会导致上下文爆炸。常见的策略是只保留最近N轮对话。对历史对话进行总结将总结后的文本作为上下文而不是完整的对话记录。在查询时将当前问题与最近的历史对话进行拼接或重写形成一个更完整的查询语句再去检索。例如用户问“它有什么优点”系统需要自动将上文提到的“产品X”补充进去形成“产品X有什么优点”的查询。4.4 系统评估如何量化RAG的好坏不能凭感觉说系统“好”或“不好”需要建立评估体系。我主要从三个维度评估检索质量命中率对于一组标准问题检索到的Top K个片段中包含正确答案的比例。平均排序倒数正确答案在检索结果中的平均排名的倒数。这个值越高说明正确答案排得越靠前。可以使用ragas、TruLens等专门评估RAG的库来计算这些指标。生成质量事实一致性生成的答案与提供的上下文事实是否一致。可以用NLI模型自动判断。答案相关性生成的答案是否直接回答了问题。信息完整性答案是否涵盖了上下文中所有关键信息点。这部分通常需要人工标注一部分测试集或者利用GPT-4作为裁判进行自动评估。系统性能端到端延迟从用户提问到收到答案的总时间。区分检索时间、重排序时间、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系统的过程就像在组装一个精密的仪器。每个环节——数据清洗、分块、嵌入、索引、检索、重排序、提示工程——都需要仔细调试。没有银弹最好的系统永远是针对你的特定数据和业务场景深度定制出来的。希望我分享的这些实战经验和踩过的坑能帮你少走一些弯路更快地构建出属于你自己的、高效可靠的智能知识系统。