最近在尝试将大模型应用到企业内部知识库问答时遇到了一个普遍难题模型经常“一本正经地胡说八道”要么回答得过于笼统要么就凭空捏造信息。为了解决这个痛点RAG检索增强生成技术成为了关键。然而网上资料要么只讲Embedding要么只讲向量数据库缺乏一个从数据准备到智能体编排的端到端实战指南。本文将为你拆解一套完整的RAG系统构建流程涵盖从文本Embedding、向量存储与检索到引入重排序优化最终利用LangGraph构建具备长期记忆和复杂决策能力的智能体。无论你是想快速搭建一个可用的知识库问答系统还是希望深入理解RAG的每个环节并进行优化这篇文章都能提供清晰的路径和可直接运行的代码。1. RAG系统核心概念与架构全景在深入代码之前我们有必要厘清RAG到底是什么以及一个完整的工业级RAG系统包含哪些核心组件。RAGRetrieval-Augmented Generation检索增强生成是一种将信息检索与大语言模型生成相结合的技术范式。其核心思想是当大模型需要回答一个问题或完成一项任务时先从外部知识库如文档、数据库中检索出最相关的信息片段然后将这些片段作为上下文连同用户问题一起提交给大模型从而生成更准确、更具事实依据的答案。一个典型的RAG系统工作流程可以概括为以下两个阶段索引Indexing将原始的非结构化文档如PDF、Word、网页进行预处理、分块然后通过Embedding模型转换为向量最后存入向量数据库。这个过程是离线的。检索与生成Retrieval Generation检索当用户提问时将问题同样转换为向量在向量数据库中进行相似度搜索找出最相关的文本块。增强将检索到的相关文本块作为“证据”或“上下文”。生成将原始问题和检索到的上下文一起构造提示词Prompt提交给大语言模型生成最终答案。然而一个健壮的RAG系统远不止“Embedding检索生成”这么简单。结合当前的技术趋势一个更先进的RAG系统架构应包含以下层次数据层原始文档的解析、清洗、分块策略。Embedding层选用合适的Embedding模型将文本转换为高维向量。这是检索质量的基石。向量存储层选择并部署向量数据库负责高效存储和检索向量。检索层核心的相似度搜索算法。进阶优化包括重排序Rerank即使用一个更精细的交叉编码器模型对初步检索结果进行二次排序显著提升精度。智能体层使用如LangGraph这样的框架将检索、生成、工具调用、记忆等环节编排成可控的工作流实现多轮对话、长期记忆和复杂任务分解。模型层大语言模型本身。对于特定领域可能还需要用到LoRALow-Rank Adaptation等微调技术让通用模型更好地适应专业术语和行文风格。本文将按照“基础搭建 - 核心优化 - 智能体进阶”的路线带你逐步实现一个功能不断增强的RAG系统。2. 环境准备与工具选型在开始动手之前我们需要搭建开发环境并选择合适的技术组件。以下配置是一个兼顾学习与开发的推荐方案。操作系统Linux (Ubuntu 20.04) / macOS / Windows (WSL2推荐)编程语言Python 3.9核心Python库# 基础数据处理与网络请求 pip install pandas numpy requests # 文档加载与处理LangChain 提供丰富工具 pip install langchain langchain-community # 文本分块与递归分割 pip install tiktoken # OpenAI分词器用于精确控制文本块长度 # 向量数据库客户端以Chroma为例 pip install chromadb # 嵌入模型以BGE为例本地运行 pip install sentence-transformers # 重排序模型以BGE Reranker为例 pip install FlagEmbedding # LangGraph 用于构建智能体 pip install langgraph # 大模型接口以OpenAI API为例也可替换为Ollama本地模型 pip install openai关键组件选型说明Embedding模型本地轻量级BAAI/bge-small-zh-v1.5中文效果好体积小。云端高性能BAAI/bge-m3支持多语言、多粒度检索能力更强可通过阿里云等平台获取服务。最新探索Qwen/Qwen2.5-7B-Instruct等大模型也具备优秀的Embedding能力但资源消耗较大。本文示例将使用BAAI/bge-small-zh-v1.5保证可在个人电脑上运行。向量数据库轻量易上手Chroma纯Python实现无需额外服务适合原型开发。高性能生产Milvus、Qdrant、Weaviate。其中Milvus开源社区活跃功能全面Qdrant以Rust编写性能优异Weaviate内置多模态和GraphQL。云服务各大云厂商均提供托管向量数据库服务。本文示例将使用Chroma进行演示便于快速验证流程。大语言模型LLM云端APIOpenAI GPT系列、Anthropic Claude、国内深度求索等。稳定但需网络和费用。本地部署通过Ollama运行Llama 3.1、Qwen2.5、Gemma2等开源模型。数据隐私性好但需要一定显卡资源。本文示例将使用OpenAI API (gpt-3.5-turbo) 进行生成步骤演示请注意替换为你自己的API KEY。智能体框架LangGraph。它是LangChain的一个扩展专注于通过有状态图StateGraph来构建复杂、可循环的智能体工作流比传统的LangChain Chain更灵活非常适合实现多轮对话和长期记忆。3. 基础RAG搭建从文档到答案让我们从零开始构建一个最基础的RAG流水线。这个例子将完成加载PDF文档、分块、生成向量、存储到Chroma、进行检索并回答。3.1 项目结构与文档准备创建一个新的项目目录结构如下rag_tutorial/ ├── data/ # 存放原始文档 │ └── sample.pdf # 示例PDF文件 ├── docs_processed/ # 处理后的文本块可选 ├── vector_db/ # Chroma数据库持久化路径 ├── config.py # 配置文件API密钥等 ├── 01_basic_rag.py # 基础RAG脚本 └── requirements.txt # 依赖列表在config.py中配置你的密钥# config.py OPENAI_API_KEY your-openai-api-key-here # 其他配置如Embedding模型路径等3.2 文档加载与文本分块文本分块是RAG中至关重要但常被忽视的一步。块太大检索会包含无关信息块太小会丢失上下文。我们使用LangChain的递归字符分割器。# 01_basic_rag.py 部分代码 import os from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document def load_and_split_documents(pdf_path): 加载PDF并分割成文本块 loader PyPDFLoader(pdf_path) raw_documents loader.load() # 使用递归字符分割器 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块的最大字符数 chunk_overlap50, # 块之间的重叠字符保持上下文连贯 length_functionlen, # 计算长度的函数 separators[\n\n, \n, 。, , , , , , ] # 中文优先分隔符 ) split_docs text_splitter.split_documents(raw_documents) print(f原始文档页数: {len(raw_documents)}) print(f分割后文本块数: {len(split_docs)}) # 查看前两个块的内容预览 for i, doc in enumerate(split_docs[:2]): print(f\n--- Chunk {i} ---\n{doc.page_content[:200]}...) return split_docs if __name__ __main__: pdf_path ./data/sample.pdf all_splits load_and_split_documents(pdf_path)3.3 生成Embedding并存入向量数据库这里我们使用sentence-transformers加载本地的BGE模型。# 01_basic_rag.py 续 from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings def create_vector_store(documents): 创建Chroma向量存储并插入文档 # 1. 初始化Embedding模型 print(正在加载Embedding模型...) embed_model SentenceTransformer(BAAI/bge-small-zh-v1.5) # 2. 为每个文本块生成向量 print(正在生成文本向量...) texts [doc.page_content for doc in documents] embeddings embed_model.encode(texts, normalize_embeddingsTrue).tolist() # 归一化便于余弦相似度计算 # 3. 创建或连接Chroma数据库 chroma_client chromadb.PersistentClient(path./vector_db) # 创建集合类似数据库的表 collection chroma_client.create_collection( nameknowledge_base, metadata{hnsw:space: cosine} # 使用余弦相似度 ) # 4. 向集合中添加文档、向量和元数据 ids [fdoc_{i} for i in range(len(texts))] metadatas [{source: doc.metadata.get(source, unknown), page: doc.metadata.get(page, 0)} for doc in documents] collection.add( documentstexts, embeddingsembeddings, metadatasmetadatas, idsids ) print(f成功将 {len(texts)} 个文本块存入向量数据库。) return embed_model, collection # 在主流程中调用 all_splits load_and_split_documents(pdf_path) embed_model, knowledge_collection create_vector_store(all_splits)3.4 检索与生成答案现在我们可以进行查询了。步骤是将问题向量化 - 在数据库中检索 - 构造Prompt - 调用LLM生成。# 01_basic_rag.py 续 from openai import OpenAI import config def retrieve_and_answer(question, collection, embed_model, top_k3): 检索并生成答案 # 1. 将问题转换为向量 query_embedding embed_model.encode([question], normalize_embeddingsTrue).tolist()[0] # 2. 在向量数据库中检索 results collection.query( query_embeddings[query_embedding], n_resultstop_k ) retrieved_docs results[documents][0] print(f\n检索到 {len(retrieved_docs)} 个相关片段) for i, doc in enumerate(retrieved_docs): print(f[{i1}] {doc[:150]}...) # 3. 构造Prompt上下文 context \n\n.join(retrieved_docs) prompt f基于以下上下文信息请回答用户的问题。如果上下文没有提供足够信息请直接说“根据已知信息无法回答该问题”。 上下文 {context} 问题{question} 答案 # 4. 调用大语言模型生成答案 client OpenAI(api_keyconfig.OPENAI_API_KEY) response client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.1 # 低温度使输出更确定更依赖上下文 ) answer response.choices[0].message.content return answer, retrieved_docs # 示例查询 if __name__ __main__: # ... 前面的加载和创建向量库代码 ... question 本文档中主要讨论了什么技术 answer, contexts retrieve_and_answer(question, knowledge_collection, embed_model) print(f\n 问题 \n{question}) print(f\n 答案 \n{answer})运行这个脚本你就得到了一个最基础的RAG系统。但这只是起点检索质量可能并不理想。4. 核心优化重排序Rerank与混合检索基础向量检索也称为“稠密检索”可能存在“语义相似但主题不相关”的问题。重排序Rerank技术使用一个计算量更大、更精确的交叉编码器模型对初步检索出的Top N个结果进行二次打分和排序从而将最相关的结果排到最前面。4.1 使用BGE Reranker模型我们使用FlagEmbedding库中的BGE Reranker模型。# 02_rag_with_rerank.py from FlagEmbedding import FlagReranker import numpy as np class RerankRetriever: def __init__(self, vector_collection, embed_model, rerank_model_nameBAAI/bge-reranker-v2-m3, top_k_vector10, top_k_final3): self.collection vector_collection self.embed_model embed_model self.top_k_vector top_k_vector # 第一轮向量检索返回的数量 self.top_k_final top_k_final # 重排序后最终返回的数量 print(f正在加载重排序模型: {rerank_model_name}) self.reranker FlagReranker(rerank_model_name, use_fp16False) # use_fp16可加速但需要GPU def retrieve(self, question): # 第一步向量检索召回 query_embedding self.embed_model.encode([question], normalize_embeddingsTrue).tolist()[0] vector_results self.collection.query( query_embeddings[query_embedding], n_resultsself.top_k_vector ) candidate_docs vector_results[documents][0] candidate_ids vector_results[ids][0] print(f向量检索召回 {len(candidate_docs)} 个候选文档。) if not candidate_docs: return [], [] # 第二步重排序精排 # 构建 (query, document) 对 pairs [[question, doc] for doc in candidate_docs] # 计算相关性分数 scores self.reranker.compute_score(pairs) # 处理分数输出compute_score可能返回单个分数或列表 if isinstance(scores, float): scores [scores] elif isinstance(scores, np.ndarray): scores scores.tolist() # 根据分数排序 scored_docs list(zip(candidate_docs, candidate_ids, scores)) scored_docs.sort(keylambda x: x[2], reverseTrue) # 按分数降序排列 # 取最终Top K final_docs [doc for doc, _, _ in scored_docs[:self.top_k_final]] final_ids [doc_id for _, doc_id, _ in scored_docs[:self.top_k_final]] final_scores [score for _, _, score in scored_docs[:self.top_k_final]] print(f重排序后Top {self.top_k_final} 文档分数: {final_scores}) return final_docs, final_ids # 使用示例 if __name__ __main__: # 假设已有 embed_model 和 knowledge_collection rerank_retriever RerankRetriever(knowledge_collection, embed_model, top_k_vector10, top_k_final3) question 详细说明RAG系统中重排序的作用是什么 final_docs, final_ids rerank_retriever.retrieve(question) print(\n最终检索结果) for i, (doc, doc_id) in enumerate(zip(final_docs, final_ids)): print(f[{i1}] ID:{doc_id}\n{doc[:200]}...\n)重排序模型计算的是查询和每个文档之间的相关性分数比单纯的向量余弦相似度更能理解细微的语义关联尤其在处理否定、比较和复杂逻辑时优势明显。4.2 混合检索策略更进一步我们可以结合稀疏检索如BM25和稠密检索向量检索形成混合检索。稀疏检索基于关键词匹配擅长处理实体、术语的精确匹配稠密检索擅长语义匹配。将两者的结果融合可以提升召回率。# 03_hybrid_retrieval.py (简化示例需安装rank_bm25) from rank_bm25 import BM25Okapi import jieba # 用于中文分词 class HybridRetriever: def __init__(self, vector_collection, embed_model, documents, document_ids): self.vector_retriever vector_collection # Chroma集合 self.embed_model embed_model # 构建BM25稀疏检索 self.documents documents # 原始文本列表 self.document_ids document_ids # 对应的ID列表 # 对文档进行分词 tokenized_docs [list(jieba.cut(doc)) for doc in self.documents] self.bm25 BM25Okapi(tokenized_docs) def retrieve(self, question, vector_top_k5, bm25_top_k5, final_top_k3): # 1. 稠密检索 query_embedding self.embed_model.encode([question], normalize_embeddingsTrue).tolist()[0] vector_results self.vector_retriever.query( query_embeddings[query_embedding], n_resultsvector_top_k ) vector_doc_ids set(vector_results[ids][0]) # 2. 稀疏检索 (BM25) tokenized_query list(jieba.cut(question)) bm25_scores self.bm25.get_scores(tokenized_query) # 获取BM25的top_k索引 bm25_top_indices np.argsort(bm25_scores)[::-1][:bm25_top_k] bm25_doc_ids set([self.document_ids[i] for i in bm25_top_indices]) # 3. 结果融合 (简单取并集) combined_ids list(vector_doc_ids.union(bm25_doc_ids)) # 根据ID获取文档内容 combined_docs [] for doc_id in combined_ids: idx self.document_ids.index(doc_id) # 简化处理实际应建立映射 combined_docs.append(self.documents[idx]) # 4. (可选) 对融合后的结果进行重排序 # ... 可以调用前面的RerankRetriever ... return combined_docs[:final_top_k], combined_ids[:final_top_k]混合检索能有效应对查询多样化是生产系统中提升召回能力的常用手段。5. 进阶实战用LangGraph构建智能体工作流基础RAG是线性的“检索-生成”。而LangGraph允许我们构建有状态、可循环、可分支的图Graph从而创建更强大的智能体。例如一个智能体可以根据初次生成结果判断是否需要进一步检索或者调用计算工具。5.1 LangGraph核心概念状态State与节点Node在LangGraph中你定义一个有向图。图的节点是函数或可运行对象边定义了节点间的流转条件。整个图共享一个状态State它是一个字典随着图的执行而更新。让我们构建一个具备“自我验证”能力的RAG智能体如果模型对生成的答案置信度不高它会自动进行一次更广泛的检索来验证或补充信息。# 04_langgraph_self_check_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage import config # 1. 定义状态结构 class AgentState(TypedDict): question: str retrieved_docs: List[str] initial_answer: str confidence: float # 模型自我评估的置信度 final_answer: str verification_triggered: bool verification_docs: List[str] # 2. 初始化组件 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.1, api_keyconfig.OPENAI_API_KEY) # 假设我们已经有了一个检索器实例 # retriever ... (可以是基础检索器或带重排序的检索器) # 3. 定义各个节点函数 def retrieve_node(state: AgentState): 检索节点根据问题检索相关文档 print([节点] 执行检索...) docs, _ retriever.retrieve(state[question]) # 使用之前定义的检索器 return {retrieved_docs: docs} def generate_initial_answer_node(state: AgentState): 生成初始答案节点 print([节点] 生成初始答案...) context \n\n.join(state[retrieved_docs]) prompt f请基于以下上下文回答问题。请先给出答案然后在单独一行以“置信度评估”开头给出一个0到1之间的分数表示你对这个答案的确定程度。 上下文 {context} 问题{state[question]} 答案 response llm.invoke([HumanMessage(contentprompt)]) content response.content # 简单解析答案和置信度 if 置信度评估 in content: answer_part, confidence_part content.split(置信度评估) answer answer_part.strip() try: confidence float(confidence_part.strip()) except: confidence 0.5 else: answer content confidence 0.5 return {initial_answer: answer, confidence: confidence} def check_confidence_node(state: AgentState): 检查置信度节点决定下一步流程 print(f[节点] 检查置信度: {state[confidence]}) if state[confidence] 0.7: # 置信度阈值 print( - 置信度低触发验证流程) return {verification_triggered: True} else: print( - 置信度高直接结束) return {verification_triggered: False} def verification_retrieve_node(state: AgentState): 验证检索节点进行更广泛的检索例如增加检索数量 print([节点] 执行验证检索扩大范围...) # 这里可以换用不同的检索策略比如top_k更大或者用混合检索 docs, _ retriever.retrieve(state[question]) # 假设retriever可以配置top_k return {verification_docs: docs} def generate_final_answer_node(state: AgentState): 生成最终答案节点 print([节点] 生成最终答案...) if state[verification_triggered]: # 结合初始答案和验证文档生成最终答案 context \n\n.join(state[verification_docs]) prompt f你之前给出了一个初步答案但信心不足。现在提供了更多参考资料。请结合新旧信息给出一个更准确、完整的最终答案。 初步答案{state[initial_answer]} 补充参考资料 {context} 原始问题{state[question]} 请输出最终答案 response llm.invoke([HumanMessage(contentprompt)]) final_answer response.content else: final_answer state[initial_answer] return {final_answer: final_answer} # 4. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(retrieve, retrieve_node) workflow.add_node(generate_initial, generate_initial_answer_node) workflow.add_node(check_confidence, check_confidence_node) workflow.add_node(verification_retrieve, verification_retrieve_node) workflow.add_node(generate_final, generate_final_answer_node) # 设置边定义流程 workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, generate_initial) workflow.add_edge(generate_initial, check_confidence) # 条件边根据置信度决定流向 workflow.add_conditional_edges( check_confidence, lambda x: verification_retrieve if x[verification_triggered] else generate_final, { True: verification_retrieve, False: generate_final } ) workflow.add_edge(verification_retrieve, generate_final) workflow.add_edge(generate_final, END) # 编译图 app workflow.compile() # 5. 运行智能体 if __name__ __main__: # 需要先初始化 retriever # retriever RerankRetriever(...) initial_state AgentState( questionLangGraph在RAG系统中主要解决什么问题, retrieved_docs[], initial_answer, confidence0.0, final_answer, verification_triggeredFalse, verification_docs[] ) print(开始运行LangGraph智能体...) final_state app.invoke(initial_state) print(\n 智能体运行结束 ) print(f问题: {final_state[question]}) print(f最终答案:\n{final_state[final_answer]}) print(f是否触发验证: {final_state[verification_triggered]})这个例子展示了LangGraph如何将简单的线性流程升级为一个具备决策能力的智能体。你可以在此基础上扩展更多节点例如调用搜索引擎、查询数据库、或者让人类介入审核Human-in-the-loop。5.2 实现长期记忆Conversational RAG多轮对话是RAG的常见场景需要让系统记住之前的对话历史。这可以通过在State中维护一个消息列表来实现。# 05_langgraph_conversational_rag.py (部分代码) from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage class ConversationState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 关键这是一个累加的消息列表 question: str retrieved_docs: List[str] def retrieve_with_history_node(state: ConversationState): 结合历史消息重新生成查询进行检索 # 从消息历史中提取最近的用户问题或综合历史生成一个搜索查询 # 这里简化处理只使用最新的一条用户消息 recent_user_msg None for msg in reversed(state[messages]): if isinstance(msg, HumanMessage): recent_user_msg msg.content break if not recent_user_msg: recent_user_msg state.get(question, ) # 使用历史信息增强查询简单示例将最后一条AI回复也作为上下文 search_query recent_user_msg # 更复杂的策略可以用一个小型LLM根据对话历史重写查询 docs, _ retriever.retrieve(search_query) return {retrieved_docs: docs} def generate_response_node(state: ConversationState): 生成回复并将对话存入历史 context \n\n.join(state[retrieved_docs]) # 构建包含对话历史和上下文的Prompt prompt_messages [ SystemMessage(content你是一个有帮助的助手请根据提供的上下文和对话历史回答问题。) ] # 添加历史消息除了最新的用户消息因为它即将被单独添加 for msg in state[messages][:-1] if len(state[messages]) 1 else []: prompt_messages.append(msg) # 添加当前上下文和问题 latest_human_msg state[messages][-1] if state[messages] else HumanMessage(contentstate[question]) enhanced_content f上下文信息 {context} 请基于以上上下文和我们的对话历史回答以下问题 {latest_human_msg.content} prompt_messages.append(HumanMessage(contentenhanced_content)) response llm.invoke(prompt_messages) # 将AI的回复也添加到状态中以便下一轮使用 return {messages: [AIMessage(contentresponse.content)]} # 构建图... # workflow.add_node(retrieve_with_history, retrieve_with_history_node) # workflow.add_node(generate_with_history, generate_response_node) # ... 设置边通过维护messages列表智能体就具备了对话记忆能力。LangGraph的状态管理让这种功能的实现变得非常清晰。6. 生产环境考量与最佳实践将一个实验性的RAG管道部署到生产环境需要关注以下方面1. 数据预处理与分块策略不要一刀切不同文档类型技术手册、法律合同、会议纪要需要不同的分块策略。可以尝试按段落、按标题、按固定长度或使用语义分割模型。保留元数据在分块时务必保留文件名、页码、章节标题等元数据并在回答中引用来源增强可信度。处理表格和图片对于非文本内容需要使用OCR或专用解析器提取文字或使用多模态模型。2. Embedding模型选择与优化领域适配如果领域专业性强如医学、法律考虑在领域数据上继续训练微调Embedding模型或使用领域专用的模型。维度与速度权衡向量维度和检索速度。维度越高通常表征能力越强但存储和计算成本也越高。标准化对生成的向量进行L2标准化以便使用余弦相似度等度量方式。3. 向量数据库部署与调优索引选择生产环境通常使用HNSW近似最近邻或IVF倒排文件索引。HNSW查询快但建索引慢、内存占用高IVF建索引快查询精度需调参。持久化与备份确保向量数据库有可靠的持久化机制和备份策略。监控监控查询延迟、召回率、内存和CPU使用情况。4. 检索质量评估与迭代构建测试集准备一批代表性的问题Q和对应的标准答案A及来源文档D。定义评估指标命中率Hit Rate检索到的Top K文档中包含标准答案来源的比例。平均精度均值Mean Average Precision, MAP衡量检索结果排序的好坏。答案相关性使用LLM或人工评估生成答案与标准答案的匹配程度。A/B测试对比不同分块大小、不同Embedding模型、是否使用重排序等策略的效果。5. 提示词工程与生成优化清晰的指令在Prompt中明确要求模型“基于上下文”、“引用来源”、“不知道就说不知道”。上下文管理当检索到的上下文过长时需要智能地截断或摘要以适配LLM的上下文窗口限制。后处理对模型生成的答案进行后处理如格式化、引用来源标注、敏感信息过滤等。6. 安全与权限数据隔离确保不同用户或租户的数据在向量数据库中物理或逻辑隔离。访问控制在检索前加入权限过滤层只检索用户有权访问的文档。输入输出过滤对用户输入和模型输出进行内容安全过滤防止注入攻击和不当内容生成。7. 常见问题与排查清单在开发和运维RAG系统中你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案检索结果完全不相关1. Embedding模型与领域不匹配。2. 文本分块不合理破坏了语义。3. 查询问题表述不清。1. 尝试更换或微调Embedding模型。2. 调整分块大小和分隔符尝试按语义分割。3. 实现查询重写Query Rewriting让LLM将用户问题改写成更利于检索的形式。答案未包含在检索到的上下文中1. 检索的top_k太小。2. 答案所需信息分散在多个块中。1. 增大检索数量top_k。2. 尝试在检索后对相关块进行合并或摘要。3. 使用句子窗口检索检索到相关块后将其前后相邻的块也一并纳入上下文。答案包含幻觉胡编乱造1. 模型未严格遵守上下文。2. 上下文本身信息不足或矛盾。1. 强化Prompt指令如“严格仅根据上下文回答”。2. 在Prompt中提供更明确的格式例如“答案... [来源文档X第Y页]”。3. 引入Self-Check或Verification机制让模型对自己答案的置信度进行评估低置信度时触发重新检索或提示用户。多轮对话中遗忘历史未将对话历史有效地纳入检索和生成环节。1. 在State中维护完整的对话历史消息列表。2. 检索时将最近几轮对话的历史或其摘要与当前问题拼接作为新的查询向量。3. 生成时将历史消息作为Prompt的一部分输入给LLM。系统响应速度慢1. Embedding模型推理慢。2. 向量数据库索引未优化。3. LLM生成速度慢。1. 使用更轻量的Embedding模型或启用GPU推理、模型量化。2. 优化向量数据库索引参数如HNSW的ef_construction和ef_search。3. 对LLM回答进行缓存对常见问题预生成答案。4. 考虑使用流式输出改善用户体验。无法处理最新信息知识库更新后向量索引未重建。建立增量更新机制定期或触发式地处理新文档更新向量数据库索引。对于删除和修改也需要有相应的处理逻辑。构建一个高效的RAG系统是一个持续迭代和优化的过程。从最简单的管道开始逐步引入重排序、混合检索、查询扩展、智能体工作流等高级技术。同时建立一套可靠的评估体系来量化每一步改进的效果至关重要。希望这份从基础到进阶的实战指南能帮助你搭建起属于自己的、可用的RAG系统并为进一步的优化和创新打下坚实基础。
最近在尝试将大模型应用到企业内部知识库问答时遇到了一个普遍难题模型经常“一本正经地胡说八道”要么回答得过于笼统要么就凭空捏造信息。为了解决这个痛点RAG检索增强生成技术成为了关键。然而网上资料要么只讲Embedding要么只讲向量数据库缺乏一个从数据准备到智能体编排的端到端实战指南。本文将为你拆解一套完整的RAG系统构建流程涵盖从文本Embedding、向量存储与检索到引入重排序优化最终利用LangGraph构建具备长期记忆和复杂决策能力的智能体。无论你是想快速搭建一个可用的知识库问答系统还是希望深入理解RAG的每个环节并进行优化这篇文章都能提供清晰的路径和可直接运行的代码。1. RAG系统核心概念与架构全景在深入代码之前我们有必要厘清RAG到底是什么以及一个完整的工业级RAG系统包含哪些核心组件。RAGRetrieval-Augmented Generation检索增强生成是一种将信息检索与大语言模型生成相结合的技术范式。其核心思想是当大模型需要回答一个问题或完成一项任务时先从外部知识库如文档、数据库中检索出最相关的信息片段然后将这些片段作为上下文连同用户问题一起提交给大模型从而生成更准确、更具事实依据的答案。一个典型的RAG系统工作流程可以概括为以下两个阶段索引Indexing将原始的非结构化文档如PDF、Word、网页进行预处理、分块然后通过Embedding模型转换为向量最后存入向量数据库。这个过程是离线的。检索与生成Retrieval Generation检索当用户提问时将问题同样转换为向量在向量数据库中进行相似度搜索找出最相关的文本块。增强将检索到的相关文本块作为“证据”或“上下文”。生成将原始问题和检索到的上下文一起构造提示词Prompt提交给大语言模型生成最终答案。然而一个健壮的RAG系统远不止“Embedding检索生成”这么简单。结合当前的技术趋势一个更先进的RAG系统架构应包含以下层次数据层原始文档的解析、清洗、分块策略。Embedding层选用合适的Embedding模型将文本转换为高维向量。这是检索质量的基石。向量存储层选择并部署向量数据库负责高效存储和检索向量。检索层核心的相似度搜索算法。进阶优化包括重排序Rerank即使用一个更精细的交叉编码器模型对初步检索结果进行二次排序显著提升精度。智能体层使用如LangGraph这样的框架将检索、生成、工具调用、记忆等环节编排成可控的工作流实现多轮对话、长期记忆和复杂任务分解。模型层大语言模型本身。对于特定领域可能还需要用到LoRALow-Rank Adaptation等微调技术让通用模型更好地适应专业术语和行文风格。本文将按照“基础搭建 - 核心优化 - 智能体进阶”的路线带你逐步实现一个功能不断增强的RAG系统。2. 环境准备与工具选型在开始动手之前我们需要搭建开发环境并选择合适的技术组件。以下配置是一个兼顾学习与开发的推荐方案。操作系统Linux (Ubuntu 20.04) / macOS / Windows (WSL2推荐)编程语言Python 3.9核心Python库# 基础数据处理与网络请求 pip install pandas numpy requests # 文档加载与处理LangChain 提供丰富工具 pip install langchain langchain-community # 文本分块与递归分割 pip install tiktoken # OpenAI分词器用于精确控制文本块长度 # 向量数据库客户端以Chroma为例 pip install chromadb # 嵌入模型以BGE为例本地运行 pip install sentence-transformers # 重排序模型以BGE Reranker为例 pip install FlagEmbedding # LangGraph 用于构建智能体 pip install langgraph # 大模型接口以OpenAI API为例也可替换为Ollama本地模型 pip install openai关键组件选型说明Embedding模型本地轻量级BAAI/bge-small-zh-v1.5中文效果好体积小。云端高性能BAAI/bge-m3支持多语言、多粒度检索能力更强可通过阿里云等平台获取服务。最新探索Qwen/Qwen2.5-7B-Instruct等大模型也具备优秀的Embedding能力但资源消耗较大。本文示例将使用BAAI/bge-small-zh-v1.5保证可在个人电脑上运行。向量数据库轻量易上手Chroma纯Python实现无需额外服务适合原型开发。高性能生产Milvus、Qdrant、Weaviate。其中Milvus开源社区活跃功能全面Qdrant以Rust编写性能优异Weaviate内置多模态和GraphQL。云服务各大云厂商均提供托管向量数据库服务。本文示例将使用Chroma进行演示便于快速验证流程。大语言模型LLM云端APIOpenAI GPT系列、Anthropic Claude、国内深度求索等。稳定但需网络和费用。本地部署通过Ollama运行Llama 3.1、Qwen2.5、Gemma2等开源模型。数据隐私性好但需要一定显卡资源。本文示例将使用OpenAI API (gpt-3.5-turbo) 进行生成步骤演示请注意替换为你自己的API KEY。智能体框架LangGraph。它是LangChain的一个扩展专注于通过有状态图StateGraph来构建复杂、可循环的智能体工作流比传统的LangChain Chain更灵活非常适合实现多轮对话和长期记忆。3. 基础RAG搭建从文档到答案让我们从零开始构建一个最基础的RAG流水线。这个例子将完成加载PDF文档、分块、生成向量、存储到Chroma、进行检索并回答。3.1 项目结构与文档准备创建一个新的项目目录结构如下rag_tutorial/ ├── data/ # 存放原始文档 │ └── sample.pdf # 示例PDF文件 ├── docs_processed/ # 处理后的文本块可选 ├── vector_db/ # Chroma数据库持久化路径 ├── config.py # 配置文件API密钥等 ├── 01_basic_rag.py # 基础RAG脚本 └── requirements.txt # 依赖列表在config.py中配置你的密钥# config.py OPENAI_API_KEY your-openai-api-key-here # 其他配置如Embedding模型路径等3.2 文档加载与文本分块文本分块是RAG中至关重要但常被忽视的一步。块太大检索会包含无关信息块太小会丢失上下文。我们使用LangChain的递归字符分割器。# 01_basic_rag.py 部分代码 import os from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document def load_and_split_documents(pdf_path): 加载PDF并分割成文本块 loader PyPDFLoader(pdf_path) raw_documents loader.load() # 使用递归字符分割器 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个块的最大字符数 chunk_overlap50, # 块之间的重叠字符保持上下文连贯 length_functionlen, # 计算长度的函数 separators[\n\n, \n, 。, , , , , , ] # 中文优先分隔符 ) split_docs text_splitter.split_documents(raw_documents) print(f原始文档页数: {len(raw_documents)}) print(f分割后文本块数: {len(split_docs)}) # 查看前两个块的内容预览 for i, doc in enumerate(split_docs[:2]): print(f\n--- Chunk {i} ---\n{doc.page_content[:200]}...) return split_docs if __name__ __main__: pdf_path ./data/sample.pdf all_splits load_and_split_documents(pdf_path)3.3 生成Embedding并存入向量数据库这里我们使用sentence-transformers加载本地的BGE模型。# 01_basic_rag.py 续 from sentence_transformers import SentenceTransformer import chromadb from chromadb.config import Settings def create_vector_store(documents): 创建Chroma向量存储并插入文档 # 1. 初始化Embedding模型 print(正在加载Embedding模型...) embed_model SentenceTransformer(BAAI/bge-small-zh-v1.5) # 2. 为每个文本块生成向量 print(正在生成文本向量...) texts [doc.page_content for doc in documents] embeddings embed_model.encode(texts, normalize_embeddingsTrue).tolist() # 归一化便于余弦相似度计算 # 3. 创建或连接Chroma数据库 chroma_client chromadb.PersistentClient(path./vector_db) # 创建集合类似数据库的表 collection chroma_client.create_collection( nameknowledge_base, metadata{hnsw:space: cosine} # 使用余弦相似度 ) # 4. 向集合中添加文档、向量和元数据 ids [fdoc_{i} for i in range(len(texts))] metadatas [{source: doc.metadata.get(source, unknown), page: doc.metadata.get(page, 0)} for doc in documents] collection.add( documentstexts, embeddingsembeddings, metadatasmetadatas, idsids ) print(f成功将 {len(texts)} 个文本块存入向量数据库。) return embed_model, collection # 在主流程中调用 all_splits load_and_split_documents(pdf_path) embed_model, knowledge_collection create_vector_store(all_splits)3.4 检索与生成答案现在我们可以进行查询了。步骤是将问题向量化 - 在数据库中检索 - 构造Prompt - 调用LLM生成。# 01_basic_rag.py 续 from openai import OpenAI import config def retrieve_and_answer(question, collection, embed_model, top_k3): 检索并生成答案 # 1. 将问题转换为向量 query_embedding embed_model.encode([question], normalize_embeddingsTrue).tolist()[0] # 2. 在向量数据库中检索 results collection.query( query_embeddings[query_embedding], n_resultstop_k ) retrieved_docs results[documents][0] print(f\n检索到 {len(retrieved_docs)} 个相关片段) for i, doc in enumerate(retrieved_docs): print(f[{i1}] {doc[:150]}...) # 3. 构造Prompt上下文 context \n\n.join(retrieved_docs) prompt f基于以下上下文信息请回答用户的问题。如果上下文没有提供足够信息请直接说“根据已知信息无法回答该问题”。 上下文 {context} 问题{question} 答案 # 4. 调用大语言模型生成答案 client OpenAI(api_keyconfig.OPENAI_API_KEY) response client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.1 # 低温度使输出更确定更依赖上下文 ) answer response.choices[0].message.content return answer, retrieved_docs # 示例查询 if __name__ __main__: # ... 前面的加载和创建向量库代码 ... question 本文档中主要讨论了什么技术 answer, contexts retrieve_and_answer(question, knowledge_collection, embed_model) print(f\n 问题 \n{question}) print(f\n 答案 \n{answer})运行这个脚本你就得到了一个最基础的RAG系统。但这只是起点检索质量可能并不理想。4. 核心优化重排序Rerank与混合检索基础向量检索也称为“稠密检索”可能存在“语义相似但主题不相关”的问题。重排序Rerank技术使用一个计算量更大、更精确的交叉编码器模型对初步检索出的Top N个结果进行二次打分和排序从而将最相关的结果排到最前面。4.1 使用BGE Reranker模型我们使用FlagEmbedding库中的BGE Reranker模型。# 02_rag_with_rerank.py from FlagEmbedding import FlagReranker import numpy as np class RerankRetriever: def __init__(self, vector_collection, embed_model, rerank_model_nameBAAI/bge-reranker-v2-m3, top_k_vector10, top_k_final3): self.collection vector_collection self.embed_model embed_model self.top_k_vector top_k_vector # 第一轮向量检索返回的数量 self.top_k_final top_k_final # 重排序后最终返回的数量 print(f正在加载重排序模型: {rerank_model_name}) self.reranker FlagReranker(rerank_model_name, use_fp16False) # use_fp16可加速但需要GPU def retrieve(self, question): # 第一步向量检索召回 query_embedding self.embed_model.encode([question], normalize_embeddingsTrue).tolist()[0] vector_results self.collection.query( query_embeddings[query_embedding], n_resultsself.top_k_vector ) candidate_docs vector_results[documents][0] candidate_ids vector_results[ids][0] print(f向量检索召回 {len(candidate_docs)} 个候选文档。) if not candidate_docs: return [], [] # 第二步重排序精排 # 构建 (query, document) 对 pairs [[question, doc] for doc in candidate_docs] # 计算相关性分数 scores self.reranker.compute_score(pairs) # 处理分数输出compute_score可能返回单个分数或列表 if isinstance(scores, float): scores [scores] elif isinstance(scores, np.ndarray): scores scores.tolist() # 根据分数排序 scored_docs list(zip(candidate_docs, candidate_ids, scores)) scored_docs.sort(keylambda x: x[2], reverseTrue) # 按分数降序排列 # 取最终Top K final_docs [doc for doc, _, _ in scored_docs[:self.top_k_final]] final_ids [doc_id for _, doc_id, _ in scored_docs[:self.top_k_final]] final_scores [score for _, _, score in scored_docs[:self.top_k_final]] print(f重排序后Top {self.top_k_final} 文档分数: {final_scores}) return final_docs, final_ids # 使用示例 if __name__ __main__: # 假设已有 embed_model 和 knowledge_collection rerank_retriever RerankRetriever(knowledge_collection, embed_model, top_k_vector10, top_k_final3) question 详细说明RAG系统中重排序的作用是什么 final_docs, final_ids rerank_retriever.retrieve(question) print(\n最终检索结果) for i, (doc, doc_id) in enumerate(zip(final_docs, final_ids)): print(f[{i1}] ID:{doc_id}\n{doc[:200]}...\n)重排序模型计算的是查询和每个文档之间的相关性分数比单纯的向量余弦相似度更能理解细微的语义关联尤其在处理否定、比较和复杂逻辑时优势明显。4.2 混合检索策略更进一步我们可以结合稀疏检索如BM25和稠密检索向量检索形成混合检索。稀疏检索基于关键词匹配擅长处理实体、术语的精确匹配稠密检索擅长语义匹配。将两者的结果融合可以提升召回率。# 03_hybrid_retrieval.py (简化示例需安装rank_bm25) from rank_bm25 import BM25Okapi import jieba # 用于中文分词 class HybridRetriever: def __init__(self, vector_collection, embed_model, documents, document_ids): self.vector_retriever vector_collection # Chroma集合 self.embed_model embed_model # 构建BM25稀疏检索 self.documents documents # 原始文本列表 self.document_ids document_ids # 对应的ID列表 # 对文档进行分词 tokenized_docs [list(jieba.cut(doc)) for doc in self.documents] self.bm25 BM25Okapi(tokenized_docs) def retrieve(self, question, vector_top_k5, bm25_top_k5, final_top_k3): # 1. 稠密检索 query_embedding self.embed_model.encode([question], normalize_embeddingsTrue).tolist()[0] vector_results self.vector_retriever.query( query_embeddings[query_embedding], n_resultsvector_top_k ) vector_doc_ids set(vector_results[ids][0]) # 2. 稀疏检索 (BM25) tokenized_query list(jieba.cut(question)) bm25_scores self.bm25.get_scores(tokenized_query) # 获取BM25的top_k索引 bm25_top_indices np.argsort(bm25_scores)[::-1][:bm25_top_k] bm25_doc_ids set([self.document_ids[i] for i in bm25_top_indices]) # 3. 结果融合 (简单取并集) combined_ids list(vector_doc_ids.union(bm25_doc_ids)) # 根据ID获取文档内容 combined_docs [] for doc_id in combined_ids: idx self.document_ids.index(doc_id) # 简化处理实际应建立映射 combined_docs.append(self.documents[idx]) # 4. (可选) 对融合后的结果进行重排序 # ... 可以调用前面的RerankRetriever ... return combined_docs[:final_top_k], combined_ids[:final_top_k]混合检索能有效应对查询多样化是生产系统中提升召回能力的常用手段。5. 进阶实战用LangGraph构建智能体工作流基础RAG是线性的“检索-生成”。而LangGraph允许我们构建有状态、可循环、可分支的图Graph从而创建更强大的智能体。例如一个智能体可以根据初次生成结果判断是否需要进一步检索或者调用计算工具。5.1 LangGraph核心概念状态State与节点Node在LangGraph中你定义一个有向图。图的节点是函数或可运行对象边定义了节点间的流转条件。整个图共享一个状态State它是一个字典随着图的执行而更新。让我们构建一个具备“自我验证”能力的RAG智能体如果模型对生成的答案置信度不高它会自动进行一次更广泛的检索来验证或补充信息。# 04_langgraph_self_check_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage import config # 1. 定义状态结构 class AgentState(TypedDict): question: str retrieved_docs: List[str] initial_answer: str confidence: float # 模型自我评估的置信度 final_answer: str verification_triggered: bool verification_docs: List[str] # 2. 初始化组件 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.1, api_keyconfig.OPENAI_API_KEY) # 假设我们已经有了一个检索器实例 # retriever ... (可以是基础检索器或带重排序的检索器) # 3. 定义各个节点函数 def retrieve_node(state: AgentState): 检索节点根据问题检索相关文档 print([节点] 执行检索...) docs, _ retriever.retrieve(state[question]) # 使用之前定义的检索器 return {retrieved_docs: docs} def generate_initial_answer_node(state: AgentState): 生成初始答案节点 print([节点] 生成初始答案...) context \n\n.join(state[retrieved_docs]) prompt f请基于以下上下文回答问题。请先给出答案然后在单独一行以“置信度评估”开头给出一个0到1之间的分数表示你对这个答案的确定程度。 上下文 {context} 问题{state[question]} 答案 response llm.invoke([HumanMessage(contentprompt)]) content response.content # 简单解析答案和置信度 if 置信度评估 in content: answer_part, confidence_part content.split(置信度评估) answer answer_part.strip() try: confidence float(confidence_part.strip()) except: confidence 0.5 else: answer content confidence 0.5 return {initial_answer: answer, confidence: confidence} def check_confidence_node(state: AgentState): 检查置信度节点决定下一步流程 print(f[节点] 检查置信度: {state[confidence]}) if state[confidence] 0.7: # 置信度阈值 print( - 置信度低触发验证流程) return {verification_triggered: True} else: print( - 置信度高直接结束) return {verification_triggered: False} def verification_retrieve_node(state: AgentState): 验证检索节点进行更广泛的检索例如增加检索数量 print([节点] 执行验证检索扩大范围...) # 这里可以换用不同的检索策略比如top_k更大或者用混合检索 docs, _ retriever.retrieve(state[question]) # 假设retriever可以配置top_k return {verification_docs: docs} def generate_final_answer_node(state: AgentState): 生成最终答案节点 print([节点] 生成最终答案...) if state[verification_triggered]: # 结合初始答案和验证文档生成最终答案 context \n\n.join(state[verification_docs]) prompt f你之前给出了一个初步答案但信心不足。现在提供了更多参考资料。请结合新旧信息给出一个更准确、完整的最终答案。 初步答案{state[initial_answer]} 补充参考资料 {context} 原始问题{state[question]} 请输出最终答案 response llm.invoke([HumanMessage(contentprompt)]) final_answer response.content else: final_answer state[initial_answer] return {final_answer: final_answer} # 4. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(retrieve, retrieve_node) workflow.add_node(generate_initial, generate_initial_answer_node) workflow.add_node(check_confidence, check_confidence_node) workflow.add_node(verification_retrieve, verification_retrieve_node) workflow.add_node(generate_final, generate_final_answer_node) # 设置边定义流程 workflow.set_entry_point(retrieve) workflow.add_edge(retrieve, generate_initial) workflow.add_edge(generate_initial, check_confidence) # 条件边根据置信度决定流向 workflow.add_conditional_edges( check_confidence, lambda x: verification_retrieve if x[verification_triggered] else generate_final, { True: verification_retrieve, False: generate_final } ) workflow.add_edge(verification_retrieve, generate_final) workflow.add_edge(generate_final, END) # 编译图 app workflow.compile() # 5. 运行智能体 if __name__ __main__: # 需要先初始化 retriever # retriever RerankRetriever(...) initial_state AgentState( questionLangGraph在RAG系统中主要解决什么问题, retrieved_docs[], initial_answer, confidence0.0, final_answer, verification_triggeredFalse, verification_docs[] ) print(开始运行LangGraph智能体...) final_state app.invoke(initial_state) print(\n 智能体运行结束 ) print(f问题: {final_state[question]}) print(f最终答案:\n{final_state[final_answer]}) print(f是否触发验证: {final_state[verification_triggered]})这个例子展示了LangGraph如何将简单的线性流程升级为一个具备决策能力的智能体。你可以在此基础上扩展更多节点例如调用搜索引擎、查询数据库、或者让人类介入审核Human-in-the-loop。5.2 实现长期记忆Conversational RAG多轮对话是RAG的常见场景需要让系统记住之前的对话历史。这可以通过在State中维护一个消息列表来实现。# 05_langgraph_conversational_rag.py (部分代码) from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage class ConversationState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 关键这是一个累加的消息列表 question: str retrieved_docs: List[str] def retrieve_with_history_node(state: ConversationState): 结合历史消息重新生成查询进行检索 # 从消息历史中提取最近的用户问题或综合历史生成一个搜索查询 # 这里简化处理只使用最新的一条用户消息 recent_user_msg None for msg in reversed(state[messages]): if isinstance(msg, HumanMessage): recent_user_msg msg.content break if not recent_user_msg: recent_user_msg state.get(question, ) # 使用历史信息增强查询简单示例将最后一条AI回复也作为上下文 search_query recent_user_msg # 更复杂的策略可以用一个小型LLM根据对话历史重写查询 docs, _ retriever.retrieve(search_query) return {retrieved_docs: docs} def generate_response_node(state: ConversationState): 生成回复并将对话存入历史 context \n\n.join(state[retrieved_docs]) # 构建包含对话历史和上下文的Prompt prompt_messages [ SystemMessage(content你是一个有帮助的助手请根据提供的上下文和对话历史回答问题。) ] # 添加历史消息除了最新的用户消息因为它即将被单独添加 for msg in state[messages][:-1] if len(state[messages]) 1 else []: prompt_messages.append(msg) # 添加当前上下文和问题 latest_human_msg state[messages][-1] if state[messages] else HumanMessage(contentstate[question]) enhanced_content f上下文信息 {context} 请基于以上上下文和我们的对话历史回答以下问题 {latest_human_msg.content} prompt_messages.append(HumanMessage(contentenhanced_content)) response llm.invoke(prompt_messages) # 将AI的回复也添加到状态中以便下一轮使用 return {messages: [AIMessage(contentresponse.content)]} # 构建图... # workflow.add_node(retrieve_with_history, retrieve_with_history_node) # workflow.add_node(generate_with_history, generate_response_node) # ... 设置边通过维护messages列表智能体就具备了对话记忆能力。LangGraph的状态管理让这种功能的实现变得非常清晰。6. 生产环境考量与最佳实践将一个实验性的RAG管道部署到生产环境需要关注以下方面1. 数据预处理与分块策略不要一刀切不同文档类型技术手册、法律合同、会议纪要需要不同的分块策略。可以尝试按段落、按标题、按固定长度或使用语义分割模型。保留元数据在分块时务必保留文件名、页码、章节标题等元数据并在回答中引用来源增强可信度。处理表格和图片对于非文本内容需要使用OCR或专用解析器提取文字或使用多模态模型。2. Embedding模型选择与优化领域适配如果领域专业性强如医学、法律考虑在领域数据上继续训练微调Embedding模型或使用领域专用的模型。维度与速度权衡向量维度和检索速度。维度越高通常表征能力越强但存储和计算成本也越高。标准化对生成的向量进行L2标准化以便使用余弦相似度等度量方式。3. 向量数据库部署与调优索引选择生产环境通常使用HNSW近似最近邻或IVF倒排文件索引。HNSW查询快但建索引慢、内存占用高IVF建索引快查询精度需调参。持久化与备份确保向量数据库有可靠的持久化机制和备份策略。监控监控查询延迟、召回率、内存和CPU使用情况。4. 检索质量评估与迭代构建测试集准备一批代表性的问题Q和对应的标准答案A及来源文档D。定义评估指标命中率Hit Rate检索到的Top K文档中包含标准答案来源的比例。平均精度均值Mean Average Precision, MAP衡量检索结果排序的好坏。答案相关性使用LLM或人工评估生成答案与标准答案的匹配程度。A/B测试对比不同分块大小、不同Embedding模型、是否使用重排序等策略的效果。5. 提示词工程与生成优化清晰的指令在Prompt中明确要求模型“基于上下文”、“引用来源”、“不知道就说不知道”。上下文管理当检索到的上下文过长时需要智能地截断或摘要以适配LLM的上下文窗口限制。后处理对模型生成的答案进行后处理如格式化、引用来源标注、敏感信息过滤等。6. 安全与权限数据隔离确保不同用户或租户的数据在向量数据库中物理或逻辑隔离。访问控制在检索前加入权限过滤层只检索用户有权访问的文档。输入输出过滤对用户输入和模型输出进行内容安全过滤防止注入攻击和不当内容生成。7. 常见问题与排查清单在开发和运维RAG系统中你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案检索结果完全不相关1. Embedding模型与领域不匹配。2. 文本分块不合理破坏了语义。3. 查询问题表述不清。1. 尝试更换或微调Embedding模型。2. 调整分块大小和分隔符尝试按语义分割。3. 实现查询重写Query Rewriting让LLM将用户问题改写成更利于检索的形式。答案未包含在检索到的上下文中1. 检索的top_k太小。2. 答案所需信息分散在多个块中。1. 增大检索数量top_k。2. 尝试在检索后对相关块进行合并或摘要。3. 使用句子窗口检索检索到相关块后将其前后相邻的块也一并纳入上下文。答案包含幻觉胡编乱造1. 模型未严格遵守上下文。2. 上下文本身信息不足或矛盾。1. 强化Prompt指令如“严格仅根据上下文回答”。2. 在Prompt中提供更明确的格式例如“答案... [来源文档X第Y页]”。3. 引入Self-Check或Verification机制让模型对自己答案的置信度进行评估低置信度时触发重新检索或提示用户。多轮对话中遗忘历史未将对话历史有效地纳入检索和生成环节。1. 在State中维护完整的对话历史消息列表。2. 检索时将最近几轮对话的历史或其摘要与当前问题拼接作为新的查询向量。3. 生成时将历史消息作为Prompt的一部分输入给LLM。系统响应速度慢1. Embedding模型推理慢。2. 向量数据库索引未优化。3. LLM生成速度慢。1. 使用更轻量的Embedding模型或启用GPU推理、模型量化。2. 优化向量数据库索引参数如HNSW的ef_construction和ef_search。3. 对LLM回答进行缓存对常见问题预生成答案。4. 考虑使用流式输出改善用户体验。无法处理最新信息知识库更新后向量索引未重建。建立增量更新机制定期或触发式地处理新文档更新向量数据库索引。对于删除和修改也需要有相应的处理逻辑。构建一个高效的RAG系统是一个持续迭代和优化的过程。从最简单的管道开始逐步引入重排序、混合检索、查询扩展、智能体工作流等高级技术。同时建立一套可靠的评估体系来量化每一步改进的效果至关重要。希望这份从基础到进阶的实战指南能帮助你搭建起属于自己的、可用的RAG系统并为进一步的优化和创新打下坚实基础。
