1. RAG 到底解决了什么问题，为什么现在非学不可最近两年，大模型相关的技术方向非常多，微调、Agent、多模态、RAG……如果不深入去看，很容易被各种名词绕晕。尤其是 RAG（Retrieval-Augmented Generation，检索增强生成），被很多文章说成“解决大模型幻觉的银弹”，导致一批人带着过高预期来学，结果做完一个 Demo 后又发现效果并没有想象中那么理想。先给一个清晰判断：RAG 不是用来解决“模型能力不足”的，它解决的是“模型不知道、记不住、容易编造”的问题。如果你希望模型回答你私有文档里的内容，或者回答最近才发生、不在模型训练数据里的事情，RAG 是目前工程落地性价比最高的方案。它不需要重新训练模型，不需要昂贵的 GPU 集群，也不需要你掌握从零预训练的语言模型知识。一个普通后端工程师，只要能写接口、会用数据库，就能在一到两周内做出一个可用版本的知识库问答系统。那为什么很多人做出来的 RAG 效果很差？因为 RAG 的链路比表面上看到的要长。大多数新手把 RAG 简单理解成“先把文档切碎，再往向量数据库里存，然后检索出来拼进 Prompt 交给大模型回答”。这个流程没有错，但只做对了一半。真正的工程难点集中在四个地方：文档解析质量、切片策略、召回精度、以及答案生成时的 Prompt 组织。任何一个环节粗糙处理，最后的效果都会大打折扣。这篇文章就是围绕这些关键环节来写的。我会从 RAG 的工作原理讲起，然后逐步覆盖向量数据库选型、环境搭建、完整代码实现、召回优化、效果验证、常见问题排查和工程落地建议。内容主线基于一个典型的“知识库问答”项目，不涉及过于底层的机器学习推导，尽量用“能直接照着做”的方式来组织。如果你是以下三类读者，这篇文章的收益最大：刚接触大模型应用开发，想用最快路径做出一个知识库问答系统的新手；已经做过简单 RAG Demo，但觉得回答质量不稳定、不知道从哪里优化的人；团队准备把 RAG 接入生产环境，需要一份相对完整的方案参考的技术负责人。不需要太多前置知识。有一点 Python 基础、会使用命令行，理解基本 HTTP 接口的概念，就足够跟完本文的实操部分。2. RAG 的核心概念与适用场景2.1 什么是 RAGRAG 的全称是 Retrieval-Augmented Generation，中文通常叫“检索增强生成”。它的基本思路是：在大模型回答问题之前，先从外部知识库中检索出与问题相关的资料片段，把这些片段作为上下文拼接到 Prompt 中，再让大模型根据这些资料生成回答。用一句话概括：先给模型开卷考试的参考资料，再让它答题。对比一下两种传统方案，能更清楚地看出 RAG 的定位：直接让大模型回答（无检索）：模型靠内部参数里的记忆来回答，对于私有知识、新知识无能为力，而且容易一本正经地编答案；微调模型：用特定数据进一步训练模型，让模型“记住”知识，但成本高、周期长，知识更新时需要重新训练；RAG：不改模型参数，只改“输入给模型的外部资料”，所以知识可以实时更新，维护成本低。这三者的选择并不互斥，但在大多数“知识库问答”场景里，RAG 是投入产出比最高的。2.2 RAG 的完整工作链路一个标准的 RAG 系统由两条链路组成：离线索引链路和在线问答链路。离线索引链路负责把原始文档处理成可检索的向量和文本片段，主要步骤如下：文档加载：读取 PDF、Word、Markdown、HTML 等格式的源文件；文本清洗：去掉页眉页脚、多余空行、特殊符号等噪声；切片处理：将长文本切分成多个 chunk（文本块），每个 chunk 是一个独立检索单元；向量化：用嵌入模型（Embedding Model）把每个 chunk 转换成向量；存入向量数据库：向量与原始文本一起存储，并建立索引。在线问答链路则面向用户查询，主要步骤如下：接收用户问题；用同一个嵌入模型把问题转换成向量；在向量数据库中执行相似度检索，召回 TopK 个最相关的文本片段；将用户问题和检索到的片段拼接到 Prompt 模板中；调用大模型生成答案；返回答案给用户，并可选附上引用来源。这个过程看起来不复杂，但每一步都有很多细节会直接影响最终效果。下面几节会逐一展开。2.3 适用场景与不适合场景RAG 最适合的场景有这些：企业内部知识库问答：比如员工手册、规章制度、产品文档、客服知识库；私有化资料问答：合同、研究报告、技术文档、历史档案等不便公开训练的内容；动态内容问答：模型训练截止日期之后的新文档、新公告、新政策；跨文档信息聚合：问题答案分散在多个文档中，需要先检索再综合回答。同时也要清醒地看到 RAG 不适合什么场景：对答案的数学推理、逻辑推理要求极高的场景，RAG 帮助有限；问题非常简短且依赖多轮上下文时，单纯 RAG 效果不稳定，需要做会话历史和意图改写；实时性要求极高、毫秒级响应的场景，RAG 链路本身有检索耗时，需要额外优化；需要模型“创造性地续写代码”等任务时，RAG 只是辅助，不能替代训练阶段的代码能力。换句话说，RAG 适合“从资料里找答案”的问题，不适合“凭空推理”的问题。理解这个边界，比多跑几个 Demo 更重要。3. RAG 技术选型：嵌入模型、向量数据库与大模型在动手写代码之前，先把关键组件选好。很多新手在环境搭建阶段就被各种库和框架搞混，核心原因是没搞清楚每一层承担什么职责。3.1 嵌入模型（Embedding Model）嵌入模型负责把文本转换成向量。这是 RAG 召回效果的基石，因为“检索得准不准”很大程度上取决于嵌入模型对语义的理解能力。选嵌入模型时，注意几个维度：中文支持能力：如果语料以中文为主，优先选择中文效果好的模型；向量维度：维度越高通常表示能力越强，但存储和计算成本也越高；最大输入长度：切片时不能让单个 chunk 超出模型的长度上限；是否可本地部署：数据敏感场景下，不能调用云 API 的模型，需要本地部署；与检索算法的兼容性：不同向量数据库对向量距离算法支持不同，一般默认余弦相似度即可。常见选择包括 OpenAI 的 text-embedding-ada-002、text-embedding-3-small，以及开源的中文嵌入模型如 BGE（BAAI General Embedding）系列、M3E 等。本地部署一般推荐 BGE 或 M3E，它们在中文语义检索上表现不错。3.2 向量数据库向量数据库负责存储向量，并提供基于向量相似度的检索能力。它和传统关系型数据库的区别在于：传统数据库擅长精确匹配和范围查询，向量数据库擅长“找语义上最接近的内容”。当前常用的向量数据库和方案主要有：方案类型特点Milvus分布式向量数据库适合大规模数据，支持多种索引，生产级能力强Qdrant向量数据库轻量易用，Rust 编写，性能好Chroma嵌入式向量数据库适合学习和原型验证，安装简单FAISS向量检索库不是完整数据库，只负责检索计算Elasticsearch + 向量插件搜索引擎扩展适合已有 ES 体系的团队关系型数据库 + pgvectorPostgreSQL 扩展复用现有 PG，适合中小数据量对于新手，我建议分两个阶段：学习阶段用 Chroma 或 Qdrant，快速跑通流程；进入生产阶段再根据数据量和并发需求迁移到 Milvus 或 Qdrant 集群。本文示例采用 Qdrant，因为它安装简单、有完整的 Python SDK，性能和功能也能覆盖真实项目。3.3 大模型RAG 的生成环节需要一个可调用的大模型。常见选择是 OpenAI 系列，或者通过国内云服务商提供的模型 API。如果要求数据不出内网，可以使用 Ollama、vLLM 等方式本地部署开源模型。这里要特别提醒：在 RAG 系统里，大模型的选择更多影响“表达质量和逻辑能力”，不影响“知识覆盖范围”。也就是说，模型再强，如果检索到的内容不对，答案也不会对。因此不要一上来就追求最强模型，先用一个性价比适中的模型跑通链路，把精力放在召回环节更划算。3.4 框架选择：LangChain、LlamaIndex 还是自己写很多教程一上来就引导你用 LangChain，但 LangChain 抽象层级较多，新手在报错时很难判断问题出在哪一层。相比之下，我更推荐新手先“裸写”一遍 RAG 流程：直接用 Embedding API、向量数据库 SDK 和 LLM API 分别调用，手工把链路串起来。这样你才能真正理解每一步。如果输入材料较少、想快速构建应用，也可以用 LlamaIndex，它对文档处理做了更好的封装。但无论如何，理解手写流程的价值不会因为使用框架而消失。本文示例将采用接近原生调用的方式，方便你看到每一步到底做了什么。4. 项目环境准备与前置条件开始写代码之前，先把环境准备好。本文的示例代码基于 Python 3，使用 Qdrant 作为向量数据库，嵌入模型和大模型接口可以按你的实际情况替换。4.1 安装 Python 与虚拟环境建议使用 Python 3.9 以上版本。创建一个新的虚拟环境，避免和本机其他项目依赖冲突。python3 -m venv rag_env source rag_env/bin/activateWindows 环境下激活命令略有区别：rag_env\Scripts\activate4.2 安装依赖包本文示例需要以下依赖：pip install qdrant-client pip install openai pip install pypdf pip install python-docx pip install beautifulsoup4 pip install tiktoken如果使用本地嵌入模型，可以根据所选模型额外安装 sentence-transformers：pip install sentence-transformers没有把握的版本号不要写死，直接用 pip 安装最新稳定版即可。如果网络环境无法直接安装，可以配置国内 pip 镜像源。4.3 启动向量数据库使用 Docker 启动 Qdrant 是最快的方式：docker run -d --name qdrant -p 6333:6333 -v $(pwd)/qdrant_storage:/qdrant/storage qdrant/qdrant启动后，通过如下方式检查是否正常运行：curl http://localhost:6333/collections正常情况下会返回一个 JSON 对象，其中 collections 字段是一个列表。此时本地向量数据库已经就绪。4.4 准备 OpenAI 密钥或本地模型如果使用 OpenAI API，需要设置环境变量：export OPENAI_API_KEY="your-api-key"如果使用本地模型，需要先启动一个兼容 OpenAI 格式的服务，例如通过 Ollama 或 vLLM。这样在代码里只需要修改 base_url，就能复用同一套调用逻辑。5. 知识库问答完整实现：从文档加载到答案生成现在开始实现一个最小的知识库问答系统。为了让你看清完整链路，我将流程拆成六个步骤，每个步骤对应一个可独立运行的模块。5.1 文档加载与文本清洗先把一个 PDF 文档加载进来。这里使用 pypdf 库读取文本内容。# 文件路径：rag_pipeline/document_loader.py from pypdf import PdfReader def load_pdf_text(pdf_path: str) -> str: reader = PdfReader(pdf_path) pages_text = [] for page in reader.pages: page_text = page.extract_text() if page_text: pages_text.append(page_text) return "\n".join(pages_text) if __name__ == "__main__": text = load_pdf_text("sample_docs/product_manual.pdf") print(f"读取字符数: {len(text)}") # 简单输出前 200 个字符，确认文本解析正常 print(text[:200])这一步看起来简单，但实际项目中 PDF 解析最容易出问题。很多 PDF 的文本层是图片或扫描件，pypdf 提取出来全是乱码或空白。遇到这种情况，需要改用 OCR 方案，例如 PaddleOCR 或 Tesseract。如果文档是 Word 格式，用 python-docx；如果是网页，用 BeautifulSoup 解析 HTML。文本清洗的主要任务是去掉噪声。页眉页脚、页码、版权信息、连续换行等都会干扰后续的切片和向量化，建议做一个统一的清洗函数。# 文件路径：rag_pipeline/text_cleaner.py import re def clean_text(text: str) -> str: # 去掉多余空行 text = re.sub(r"\n\s*\n+", "\n", text) # 去掉页眉页脚常见模式，例如 "第 1 页" text = re.sub(r"第\s*\d+\s*页", "", text) # 去掉多余空白字符 text = re.sub(r"[ \t]+", " ", text) return text.strip()清洗规则要根据你的文档特征来调整，这里只是一个通用示例。5.2 切片策略切片是整个 RAG 链路里“看起来简单、实际上影响最大”的环节之一。如果切片太大，一个 chunk 里可能混杂多个主题，检索时模型会把不相关内容一起带进 Prompt，干扰生成；如果切片太小，可能切断了完整的语义表达，导致每个 chunk 都信息不足。常见的切片策略有：固定字符数切片：每 500 个字符切一段，带少量重叠；按段落切片：优先按照文档的自然段落进行切割，保留语义单元；按标题层级切片：Markdown 文档按标题维度切，适合结构化文档；递归字符切片：框架内置的方式，先按段落切，再按句子切。实际项目中，我建议“按段落优先，控制最大长度”。下面是一个相对通用的示例：# 文件路径：rag_pipeline/text_splitter.py import tiktoken def split_text_by_size(text: str, max_chunk_size: int = 500, overlap: int = 50) -> list[str]: encoder = tiktoken.get_encoding("cl100k_base") paragraphs = text.split("\n") chunks = [] current_chunk = "" current_tokens = 0 for para in paragraphs: para_token_count = len(encoder.encode(para)) if current_tokens + para_token_count > max_chunk_size: if current_chunk: chunks.append(current_chunk) overlap_text = current_chunk[-overlap:] if overlap > 0 else "" current_chunk = overlap_text + para current_tokens = len(encoder.encode(current_chunk)) else: current_chunk += "\n" + para current_tokens += para_token_count if current_chunk: chunks.append(current_chunk) return chunks这里用 token 数量而不是字符数量来控制长度，是因为嵌入模型和大模型的上下文限制通常按 token 计算。overlap 参数的意义是让相邻 chunk 之间保留一部分重复内容，避免因为硬切导致一句话前后断开。切片策略没有绝对最优，需要结合你的文档类型做实验。后面第 7 节会讲如何根据效果反向调整。5.3 批量向量化并写入向量数据库切片完成后，使用嵌入模型将每个 chunk 转为向量，然后写入 Qdrant。# 文件路径：rag_pipeline/index_builder.py from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct from openai import OpenAI client = OpenAI() def get_embedding(text: str) -> list[float]: # 这里替换为你选择的嵌入模型；本地模型则修改 base_url resp = client.embeddings.create( model="text-embedding-3-small", input=text ) return resp.data[0].embedding def build_index(chunks: list[str], collection_name: str = "knowledge_base") -> None: qdrant = QdrantClient(host="localhost", port=6333) # 如果集合不存在，先创建；已存在则跳过 existing_collections = [c.name for c in qdrant.get_collections().collections] if collection_name not in existing_collections: qdrant.create_collection( collection_name=collection_name, vectors_config=VectorParams(size=1536, distance=Distance.COSINE) ) points = [] for idx, chunk in enumerate(chunks): vector = get_embedding(chunk) points.append( PointStruct( id=idx, vector=vector, payload={"text": chunk} ) ) # 分批写入，避免一次性提交过多数据 BATCH_SIZE = 100 for i in range(0, len(points), BATCH_SIZE): qdrant.upsert( collection_name=collection_name, points=points[i:i + BATCH_SIZE] ) if __name__ == "__main__": from document_loader import load_pdf_text from text_cleaner import clean_text from text_splitter import split_text_by_size raw_text = load_pdf_text("sample_docs/product_manual.pdf") cleaned_text = clean_text(raw_text) chunks = split_text_by_size(cleaned_text) build_index(chunks) print(f"索引完成，共写入 {len(chunks)} 个切片")向量维度 1536 对应 text-embedding-3-small 模型的输出维度。如果你换了嵌入模型，这里的 size 需要同步修改。5.4 构造检索函数检索函数负责把用户问题向量化，然后去向量数据库里找最相关的切片。# 文件路径：rag_pipeline/retriever.py from qdrant_client import QdrantClient def retrieve_relevant_chunks(query: str, top_k: int = 4) -> list[str]: qdrant = QdrantClient(host="localhost", port=6333) query_vector = get_embedding(query) search_result = qdrant.search( collection_name="knowledge_base", query_vector=query_vector, limit=top_k ) return [hit.payload["text"] for hit in search_result]注意这里 get_embedding 函数和索引构建时用的是同一个嵌入模型。如果检索和索引用了不同的嵌入模型，向量空间不一致，召回效果会直接崩溃。5.5 构造 Prompt 并调用大模型检索结果拿到后，需要组织成 Prompt。这里的关键是让大模型只依据提供的资料回答，不要自行发挥。# 文件路径：rag_pipeline/generator.py from openai import OpenAI client = OpenAI() def generate_answer(query: str, retrieved_chunks: list[str]) -> str: context = "\n\n".join(retrieved_chunks) prompt = f"""请根据以下参考资料回答用户问题。 参考资料： {context} 用户问题：{query} 回答要求： 1. 如果参考资料中包含答案，请基于参考资料回答； 2. 如果参考资料中找不到答案，明确说明“资料中未找到相关信息”； 3. 不要编造资料中不存在的内容； 4. 回答时尽量引用原文关键信息。 回答：""" resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.2 ) return resp.choices[0].message.contenttemperature 设为较低值，是为了让模型更忠实于资料，减少自由发挥。如果希望答案更有创造性，可以适当调高，但在知识库问答场景里，低温度通常是安全选择。5.6 组装完整问答流程把前面所有模块串起来，形成一个完整的 ask 函数。# 文件路径：rag_pipeline/rag_engine.py from document_loader import load_pdf_text from text_cleaner import clean_text from text_splitter import split_text_by_size from index_builder import build_index from retriever import retrieve_relevant_chunks from generator import generate_answer def init_knowledge_base(pdf_path: str) -> None: raw_text = load_pdf_text(pdf_path) cleaned_text = clean_text(raw_text) chunks = split_text_by_size(cleaned_text) build_index(chunks) def ask(question: str) -> str: chunks = retrieve_relevant_chunks(question, top_k=4) answer = generate_answer(question, chunks) return answer if __name__ == "__main__": # 首次运行先初始化知识库；如果已经建立索引，可以注释掉 # init_knowledge_base("sample_docs/product_manual.pdf") while True: user_input = input("请输入问题（输入 exit 退出）：") if user_input.strip().lower() == "exit": break answer = ask(user_input) print("答案：", answer) print("-" * 50)到这里，一个最小可用的 RAG 知识库问答系统就完成了。6. 运行效果与召回质量验证代码写完后，需要验证系统是否真正工作正常。这一步很多新手会直接跳过，导致后续排查问题时没有方向。6.1 验证流程按顺序执行以下检查：输入一个与文档内容高度相关的问题，确认能返回合理回答；输入一个与文档内容无关的问题，确认模型没有强行编造；直接查看检索返回的 top_k 切片，判断召回内容是否真的和问题语义相关；对比“仅调用大模型”和“RAG 完整流程”的答案，确认 RAG 带来的信息增量。# 在项目根目录下运行 python rag_pipeline/rag_engine.py运行后示例输出：请输入问题（输入 exit 退出）：产品支持的售后响应时间是多久？ 答案：根据产品手册中的说明，售后响应时间分为两个等级：基础支持服务为 2 个工作日，高级支持服务为 4 个工作小时内响应。 -------------------------------------------------- 请输入问题（输入 exit 退出）：公司今年的财务报表是多少？ 答案：资料中未找到相关信息。 --------------------------------------------------判断标准很简单：能在文档中找到明确答案的问题，答案应该有据可查；不在文档中的问题，应该明确说“未找到”，而不是给你编一个数字。6.2 检索效果的具体检查方法如果回答效果不理想，不要急着改 Prompt，先检查检索环节。方法是在 ask 函数中临时打印检索到的切片内容。def ask_with_debug(question: str) -> str: chunks = retrieve_relevant_chunks(question, top_k=4) print("=== 检索到的切片 ===") for i, chunk in enumerate(chunks): print(f"[{i}] {chunk[:150]}") print("-" * 30) answer = generate_answer(question, chunks) return answer输出中可以看到，系统到底有没有把正确答案所在的切片召回出来。判断逻辑如下：如果正确答案对应的切片没有被召回，问题出在“检索”而非“生成”；如果正确切片被召回了，但模型回答错误，问题出在“Prompt 组织”或“模型选择”；如果检索结果中相关和不相关的内容混杂，问题出在“切片策略”或“嵌入模型”。这个调试思路可以覆盖 RAG 80% 以上的问题定位场景。6.3 用评测问题集量化效果做一次认真的验证时，建议准备一个评测问题集。每个问题包含三个字段：问题文本、参考答案、预期来源位置。然后对每个问题运行系统，用下面的标准打分：结果类型说明评分完全正确答案准确且来源正确1.0部分正确答案方向对，但关键信息缺失或偏差0.5检索正确但生成错误检索切片正确，但模型没答好0.3检索错误正确答案切片没有被召回0理想情况下，10 个问题中至少应该有 6 到 7 个达到“完全正确”。如果都达不到，先按 6.2 的方法定位问题，不要继续叠加复杂功能。7. RAG 常见问题与排查方法下面这张表整理了 RAG 项目中最常遇到的现象和排查思路。它比单看某一步报错更有用，因为很多问题不是编译错误，而是“效果不对、说不出哪里不对”的状态。问题现象可能原因排查方式解决方案文档中文解析为空或乱码PDF 是扫描件或图片格式用 OCR 提取文本，或检查 PDF 文本层引入 PaddleOCR 方案，替换文档源为可编辑格式回答内容明显超出资料范围Prompt 没有限制模型基于资料回答检查 Prompt 模板是否有明确约束增加“若资料中无答案，请说明未找到”的指令相关答案明明存在，却召回不到切片粒度过大或过小打印检索切片，确认正确内容是否被召回调整切片大小和 overlap，或切换嵌入模型检索结果中内容主题混杂文档结构复杂，直接按字符硬切观察 chunk 内容是否完整按标题层级切分，保留文档结构上传新文档后检索结果没变化新文档未写入向量数据库确认索引构建流程是否执行重新执行索引脚本，检查 upsert 返回检索性能越来越差集合中数据量大且未优化索引查看向量数据库索引类型和查询耗时调整索引参数，或分库分集合调用大模型接口超时检索返回的切片过多，Prompt 过长查看请求量和上下文 Token 数降低 top_k，或对检索切片做重排截断从这张表可以总结出一个重要的经验：RAG 调试的正确顺序是“先查检索，再查生成”。很多团队浪费大量时间在调 Prompt 上，最后发现根因是切片策略导致正确答案根本没被召回。8. 召回优化：从基础版到可用版前面实现的 RAG 是最基础的版本。真实项目里，问题往往不会那么单纯：用户问法和文档原文表述不同、答案分散在多个切片、检索出来的内容噪音比例较高。这一节讲几个工程上常用的优化手段，按实施成本从低到高排列。8.1 混合检索与稀疏向量纯向量检索擅长语义匹配，但对精确关键词不敏感。比如用户搜索“ERROR_CODE_403”，向量检索可能返回一堆“错误码”相关的泛泛内容，而基于关键词的 BM25 检索却能精确命中目标代码片段。实践中，可以将向量检索与传统关键词检索结果融合。Qdrant 支持同时挂载稠密向量和稀疏向量，也可以用独立的关键词检索服务，然后做结果合并。合并策略可以是简单的分数加权：final_score = alpha * vector_score + (1 - alpha) * bm25_scorealpha 需要根据数据集调试。混合检索对“代码、编号、专业术语”较多的知识库提升明显。8.2 重新排序向量检索的目标是“快速召回”，而不是“精确排序”。即使是同一个知识库，向量检索返回的前几结果也可能有噪声。重新排序的思路是：先用向量检索召回 20 到 50 条候选切片，再用一个更强的重排模型（Reranker）对候选做精排，只取前 3 到 5 条进入 Prompt。Reranker 与普通嵌入模型的区别在于：普通嵌入模型把文本变成向量后比较相似度，而 Reranker 直接计算“问题+文档片段”的相关性打分，效果更准但速度更慢。因为只对几十条候选做精排，性能开销通常可以接受。如果没有专门的重排模型，可以用大模型做一次粗排，让模型按相关性对切片打分，再取 TopK。8.3 查询改写用户的问题往往是碎片化的，尤其在多轮对话里。比如用户先问“公司年假政策是什么”，接着问“那加班呢”，如果直接把第二个问题拿去检索，向量数据库根本不知道该找什么。查询改写要解决的就是这个问题。常用做法是：调用大模型，结合对话历史，把当前问题改写成一个独立的、完整的检索 query。def rewrite_query(history: list[dict], current_query: str) -> str: history_text = "\n".join([f"{item['role']}: {item['content']}" for item in history[-4:]]) prompt = f"""根据对话历史，将最新用户问题改写为一个适合检索的独立问题。 对话历史： {history_text} 最新问题：{current_query} 改写后的问题：""" resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0 ) return resp.choices[0].message.content改写后的 query 只用于检索，不直接用于生成回答。这样既保证检索能命中，又不会让改写过程丢失原始问题的意图。8.4 结构化的分层检索如果文档有清晰的结构，比如产品手册包含章节编号，可以考虑两层检索：第一层根据用户问题定位相关章节，第二层只在相关章节中做向量检索。这种方式大幅缩小检索范围，精度更高。一个更简单的实现是：在切片的 payload 中带上章节信息，在检索层面把用户的 query 先做一次分类，然后给向量检索增加 filter 条件。8.5 去重与上下文压缩召回切片之间经常有大量重叠内容，直接全部拼进 Prompt 不仅浪费 Token，而且可能让模型被重复信息干扰。可以在拼接前做简易去重，比如对切片摘要计算哈希，删除相似度超过阈值的重复项。上下文压缩解决的是另一个问题：切片里只有一小段和问题相关，但整块内容都很长。压缩组件可以抽取与问题高度相关的句子，替代完整切片进入 Prompt。实现时可借助大模型做抽取式压缩，虽然会增加一次模型调用，但能显著改善长文档场景下的回答质量。这里需要强调一个原则：召回优化没有“放之四海而皆准”的组合，必须基于你自己的文档类型和失败案例来取舍。比较稳妥的路径是：先建一个评测问题集，再逐项加入上述优化，每次只改一个变量，对比评测分数。9. RAG 系统生产化落地的工程建议从 Demo 到生产环境，差的不是代码量，而是工程细节。这一节给出几个面向真实项目的建议。9.1 文档更新与索引生命周期管理知识库内容不会一成不变。生产系统中需要设计文档的新增、更新、删除流程。建议为每个文档分配唯一 doc_id，向量数据写入时把 doc_id 和 chunk_id 一起存进 payload。当文档更新时，先按 doc_id 删除旧切片，再写入新切片。避免因为重复导入导致同一文档的旧版本污染检索结果。Qdrant 中按 payload 条件删除的示例：qdrant.delete( collection_name="knowledge_base", points_selector=models.FilterSelector( filter=models.Filter( must=[models.FieldCondition(key="doc_id", match=models.MatchValue(value="doc_2024_001"))] ) ) )更新操作最好做成异步任务，并加入失败重试机制，避免文档处理到一半系统崩溃造成索引不完整。9.2 API 服务化与并发控制生产环境不能只有一个命令行脚本。建议使用 FastAPI 将问答能力封装成 HTTP 接口，同时做以下处理：请求超时控制，避免大模型接口慢导致整个服务雪崩；基于用户的访问频率限流；将知识库集合按业务线拆分，不同业务使用不同的 collection；对问答请求做日志记录，方便后续分析用户提问趋势和回答质量。9.3 安全与合规边界知识库问答系统的核心资产是私有文档。在设计和部署时必须考虑：向量数据库不应直接暴露在公网，应只允许应用服务器访问；如果使用云 API 嵌入模型或大模型，要确认数据出境是否符合合规要求，敏感数据优先采用本地部署模型；对用户的访问权限做隔离，不要让低权限用户检索到高权限文档；对敏感文档的切片进行脱敏处理后再入向量库。涉及权限控制的场景，可以采用“文档元数据过滤 + 用户权限标签”的方式，在检索时通过 filter 条件限定可见范围。9.4 评估与持续优化机制生产环境必须建立持续评估机制。建议做三件事：收集线上用户问题，每周挑出回答质量差的案例，归类到“检索问题”还是“生成问题”；维护一个回归评测集，每次修改切片策略或 Prompt 后，先跑一遍评测集，防止改好一个问题却带崩另外几个问题；记录回答的引用来源，用户可点击查看答案出处，这既能提升可信度，也方便人工审核。9.5 成本控制RAG 的成本集中在向量化和大模型生成两方面。降低成本的常见手段：对知识库做增量索引，只对新文档向量化，而不是全量重建；检索时合理设置 top_k，避免大量不相关切片浪费 Token；对常见问题设置缓存，完全相同的用户问题直接命中缓存答案；高频简单问题可以先用小模型回答，只有复杂问题才调用大模型。成本不是越低越好，关键是建立“回答质量与成本挂钩”的评估意识，在整体效率维度上做权衡。10. 总结与下一步实践建议回到文章开头的问题：RAG 适合做什么、不适合做什么，现在应该已经有了更具体的答案。它不会让大模型“无所不知”，但能让大模型在“有资料可依”的场景里回答得更可靠。它的核心价值是把私有知识、动态知识注入到生成链路里，而且不需要修改模型参数。真正决定 RAG 项目成败的，不是代码量多大，而是五个维度：文档解析质量、切片策略、向量召回效果、Prompt 约束、工程运维能力。建议你在动手写项目时，把时间分配大致调整为：40% 花在文档处理和切片优化上，30% 花在检索效果验证上，剩下的时间才留给代码框架和接口封装。下一步的实践路径可以这样安排：找一份你熟悉的业务文档，用本文示例代码建立索引；准备 10 到 20 个有标准答案的评测问题；跑通基线版本，记录评测分数；依次加入混合检索、查询改写、重排，观察每一步的分数变化；将问答封装成 HTTP 服务，再考虑文档更新流程和权限控制。学习过程中最容易踩的坑是“过早引入框架”。如果一上来就用 LangChain 或 LlamaIndex，你会被各种抽象类、回调、链式调用绕晕，出了问题也不知道该看哪个模块。先手写一遍流程，再用框架封装，认识的深度完全不同。如果后续想深入，可以继续学习这些方向：文本切分的自适应算法、向量索引参数的调优原理、Reranker 模型训练数据的构造、Agent 与 RAG 的结合方式，以及多路召回后的精排策略。建议现在就把这篇教程收藏，然后选一份自己的文档，按第 5 节的步骤跑通第一个知识库问答。只有亲手走完一遍，你才会真正理解为什么说“RAG 的难点不在原理，而在每一个细节”。
1. RAG 到底解决了什么问题，为什么现在非学不可最近两年，大模型相关的技术方向非常多，微调、Agent、多模态、RAG……如果不深入去看，很容易被各种名词绕晕。尤其是 RAG（Retrieval-Augmented Generation，检索增强生成），被很多文章说成“解决大模型幻觉的银弹”，导致一批人带着过高预期来学，结果做完一个 Demo 后又发现效果并没有想象中那么理想。先给一个清晰判断：RAG 不是用来解决“模型能力不足”的，它解决的是“模型不知道、记不住、容易编造”的问题。如果你希望模型回答你私有文档里的内容，或者回答最近才发生、不在模型训练数据里的事情，RAG 是目前工程落地性价比最高的方案。它不需要重新训练模型，不需要昂贵的 GPU 集群，也不需要你掌握从零预训练的语言模型知识。一个普通后端工程师，只要能写接口、会用数据库，就能在一到两周内做出一个可用版本的知识库问答系统。那为什么很多人做出来的 RAG 效果很差？因为 RAG 的链路比表面上看到的要长。大多数新手把 RAG 简单理解成“先把文档切碎，再往向量数据库里存，然后检索出来拼进 Prompt 交给大模型回答”。这个流程没有错，但只做对了一半。真正的工程难点集中在四个地方：文档解析质量、切片策略、召回精度、以及答案生成时的 Prompt 组织。任何一个环节粗糙处理，最后的效果都会大打折扣。这篇文章就是围绕这些关键环节来写的。我会从 RAG 的工作原理讲起，然后逐步覆盖向量数据库选型、环境搭建、完整代码实现、召回优化、效果验证、常见问题排查和工程落地建议。内容主线基于一个典型的“知识库问答”项目，不涉及过于底层的机器学习推导，尽量用“能直接照着做”的方式来组织。如果你是以下三类读者，这篇文章的收益最大：刚接触大模型应用开发，想用最快路径做出一个知识库问答系统的新手；已经做过简单 RAG Demo，但觉得回答质量不稳定、不知道从哪里优化的人；团队准备把 RAG 接入生产环境，需要一份相对完整的方案参考的技术负责人。不需要太多前置知识。有一点 Python 基础、会使用命令行，理解基本 HTTP 接口的概念，就足够跟完本文的实操部分。2. RAG 的核心概念与适用场景2.1 什么是 RAGRAG 的全称是 Retrieval-Augmented Generation，中文通常叫“检索增强生成”。它的基本思路是：在大模型回答问题之前，先从外部知识库中检索出与问题相关的资料片段，把这些片段作为上下文拼接到 Prompt 中，再让大模型根据这些资料生成回答。用一句话概括：先给模型开卷考试的参考资料，再让它答题。对比一下两种传统方案，能更清楚地看出 RAG 的定位：直接让大模型回答（无检索）：模型靠内部参数里的记忆来回答，对于私有知识、新知识无能为力，而且容易一本正经地编答案；微调模型：用特定数据进一步训练模型，让模型“记住”知识，但成本高、周期长，知识更新时需要重新训练；RAG：不改模型参数，只改“输入给模型的外部资料”，所以知识可以实时更新，维护成本低。这三者的选择并不互斥，但在大多数“知识库问答”场景里，RAG 是投入产出比最高的。2.2 RAG 的完整工作链路一个标准的 RAG 系统由两条链路组成：离线索引链路和在线问答链路。离线索引链路负责把原始文档处理成可检索的向量和文本片段，主要步骤如下：文档加载：读取 PDF、Word、Markdown、HTML 等格式的源文件；文本清洗：去掉页眉页脚、多余空行、特殊符号等噪声；切片处理：将长文本切分成多个 chunk（文本块），每个 chunk 是一个独立检索单元；向量化：用嵌入模型（Embedding Model）把每个 chunk 转换成向量；存入向量数据库：向量与原始文本一起存储，并建立索引。在线问答链路则面向用户查询，主要步骤如下：接收用户问题；用同一个嵌入模型把问题转换成向量；在向量数据库中执行相似度检索，召回 TopK 个最相关的文本片段；将用户问题和检索到的片段拼接到 Prompt 模板中；调用大模型生成答案；返回答案给用户，并可选附上引用来源。这个过程看起来不复杂，但每一步都有很多细节会直接影响最终效果。下面几节会逐一展开。2.3 适用场景与不适合场景RAG 最适合的场景有这些：企业内部知识库问答：比如员工手册、规章制度、产品文档、客服知识库；私有化资料问答：合同、研究报告、技术文档、历史档案等不便公开训练的内容；动态内容问答：模型训练截止日期之后的新文档、新公告、新政策；跨文档信息聚合：问题答案分散在多个文档中，需要先检索再综合回答。同时也要清醒地看到 RAG 不适合什么场景：对答案的数学推理、逻辑推理要求极高的场景，RAG 帮助有限；问题非常简短且依赖多轮上下文时，单纯 RAG 效果不稳定，需要做会话历史和意图改写；实时性要求极高、毫秒级响应的场景，RAG 链路本身有检索耗时，需要额外优化；需要模型“创造性地续写代码”等任务时，RAG 只是辅助，不能替代训练阶段的代码能力。换句话说，RAG 适合“从资料里找答案”的问题，不适合“凭空推理”的问题。理解这个边界，比多跑几个 Demo 更重要。3. RAG 技术选型：嵌入模型、向量数据库与大模型在动手写代码之前，先把关键组件选好。很多新手在环境搭建阶段就被各种库和框架搞混，核心原因是没搞清楚每一层承担什么职责。3.1 嵌入模型（Embedding Model）嵌入模型负责把文本转换成向量。这是 RAG 召回效果的基石，因为“检索得准不准”很大程度上取决于嵌入模型对语义的理解能力。选嵌入模型时，注意几个维度：中文支持能力：如果语料以中文为主，优先选择中文效果好的模型；向量维度：维度越高通常表示能力越强，但存储和计算成本也越高；最大输入长度：切片时不能让单个 chunk 超出模型的长度上限；是否可本地部署：数据敏感场景下，不能调用云 API 的模型，需要本地部署；与检索算法的兼容性：不同向量数据库对向量距离算法支持不同，一般默认余弦相似度即可。常见选择包括 OpenAI 的 text-embedding-ada-002、text-embedding-3-small，以及开源的中文嵌入模型如 BGE（BAAI General Embedding）系列、M3E 等。本地部署一般推荐 BGE 或 M3E，它们在中文语义检索上表现不错。3.2 向量数据库向量数据库负责存储向量，并提供基于向量相似度的检索能力。它和传统关系型数据库的区别在于：传统数据库擅长精确匹配和范围查询，向量数据库擅长“找语义上最接近的内容”。当前常用的向量数据库和方案主要有：方案类型特点Milvus分布式向量数据库适合大规模数据，支持多种索引，生产级能力强Qdrant向量数据库轻量易用，Rust 编写，性能好Chroma嵌入式向量数据库适合学习和原型验证，安装简单FAISS向量检索库不是完整数据库，只负责检索计算Elasticsearch + 向量插件搜索引擎扩展适合已有 ES 体系的团队关系型数据库 + pgvectorPostgreSQL 扩展复用现有 PG，适合中小数据量对于新手，我建议分两个阶段：学习阶段用 Chroma 或 Qdrant，快速跑通流程；进入生产阶段再根据数据量和并发需求迁移到 Milvus 或 Qdrant 集群。本文示例采用 Qdrant，因为它安装简单、有完整的 Python SDK，性能和功能也能覆盖真实项目。3.3 大模型RAG 的生成环节需要一个可调用的大模型。常见选择是 OpenAI 系列，或者通过国内云服务商提供的模型 API。如果要求数据不出内网，可以使用 Ollama、vLLM 等方式本地部署开源模型。这里要特别提醒：在 RAG 系统里，大模型的选择更多影响“表达质量和逻辑能力”，不影响“知识覆盖范围”。也就是说，模型再强，如果检索到的内容不对，答案也不会对。因此不要一上来就追求最强模型，先用一个性价比适中的模型跑通链路，把精力放在召回环节更划算。3.4 框架选择：LangChain、LlamaIndex 还是自己写很多教程一上来就引导你用 LangChain，但 LangChain 抽象层级较多，新手在报错时很难判断问题出在哪一层。相比之下，我更推荐新手先“裸写”一遍 RAG 流程：直接用 Embedding API、向量数据库 SDK 和 LLM API 分别调用，手工把链路串起来。这样你才能真正理解每一步。如果输入材料较少、想快速构建应用，也可以用 LlamaIndex，它对文档处理做了更好的封装。但无论如何，理解手写流程的价值不会因为使用框架而消失。本文示例将采用接近原生调用的方式，方便你看到每一步到底做了什么。4. 项目环境准备与前置条件开始写代码之前，先把环境准备好。本文的示例代码基于 Python 3，使用 Qdrant 作为向量数据库，嵌入模型和大模型接口可以按你的实际情况替换。4.1 安装 Python 与虚拟环境建议使用 Python 3.9 以上版本。创建一个新的虚拟环境，避免和本机其他项目依赖冲突。python3 -m venv rag_env source rag_env/bin/activateWindows 环境下激活命令略有区别：rag_env\Scripts\activate4.2 安装依赖包本文示例需要以下依赖：pip install qdrant-client pip install openai pip install pypdf pip install python-docx pip install beautifulsoup4 pip install tiktoken如果使用本地嵌入模型，可以根据所选模型额外安装 sentence-transformers：pip install sentence-transformers没有把握的版本号不要写死，直接用 pip 安装最新稳定版即可。如果网络环境无法直接安装，可以配置国内 pip 镜像源。4.3 启动向量数据库使用 Docker 启动 Qdrant 是最快的方式：docker run -d --name qdrant -p 6333:6333 -v $(pwd)/qdrant_storage:/qdrant/storage qdrant/qdrant启动后，通过如下方式检查是否正常运行：curl http://localhost:6333/collections正常情况下会返回一个 JSON 对象，其中 collections 字段是一个列表。此时本地向量数据库已经就绪。4.4 准备 OpenAI 密钥或本地模型如果使用 OpenAI API，需要设置环境变量：export OPENAI_API_KEY="your-api-key"如果使用本地模型，需要先启动一个兼容 OpenAI 格式的服务，例如通过 Ollama 或 vLLM。这样在代码里只需要修改 base_url，就能复用同一套调用逻辑。5. 知识库问答完整实现：从文档加载到答案生成现在开始实现一个最小的知识库问答系统。为了让你看清完整链路，我将流程拆成六个步骤，每个步骤对应一个可独立运行的模块。5.1 文档加载与文本清洗先把一个 PDF 文档加载进来。这里使用 pypdf 库读取文本内容。# 文件路径：rag_pipeline/document_loader.py from pypdf import PdfReader def load_pdf_text(pdf_path: str) -> str: reader = PdfReader(pdf_path) pages_text = [] for page in reader.pages: page_text = page.extract_text() if page_text: pages_text.append(page_text) return "\n".join(pages_text) if __name__ == "__main__": text = load_pdf_text("sample_docs/product_manual.pdf") print(f"读取字符数: {len(text)}") # 简单输出前 200 个字符，确认文本解析正常 print(text[:200])这一步看起来简单，但实际项目中 PDF 解析最容易出问题。很多 PDF 的文本层是图片或扫描件，pypdf 提取出来全是乱码或空白。遇到这种情况，需要改用 OCR 方案，例如 PaddleOCR 或 Tesseract。如果文档是 Word 格式，用 python-docx；如果是网页，用 BeautifulSoup 解析 HTML。文本清洗的主要任务是去掉噪声。页眉页脚、页码、版权信息、连续换行等都会干扰后续的切片和向量化，建议做一个统一的清洗函数。# 文件路径：rag_pipeline/text_cleaner.py import re def clean_text(text: str) -> str: # 去掉多余空行 text = re.sub(r"\n\s*\n+", "\n", text) # 去掉页眉页脚常见模式，例如 "第 1 页" text = re.sub(r"第\s*\d+\s*页", "", text) # 去掉多余空白字符 text = re.sub(r"[ \t]+", " ", text) return text.strip()清洗规则要根据你的文档特征来调整，这里只是一个通用示例。5.2 切片策略切片是整个 RAG 链路里“看起来简单、实际上影响最大”的环节之一。如果切片太大，一个 chunk 里可能混杂多个主题，检索时模型会把不相关内容一起带进 Prompt，干扰生成；如果切片太小，可能切断了完整的语义表达，导致每个 chunk 都信息不足。常见的切片策略有：固定字符数切片：每 500 个字符切一段，带少量重叠；按段落切片：优先按照文档的自然段落进行切割，保留语义单元；按标题层级切片：Markdown 文档按标题维度切，适合结构化文档；递归字符切片：框架内置的方式，先按段落切，再按句子切。实际项目中，我建议“按段落优先，控制最大长度”。下面是一个相对通用的示例：# 文件路径：rag_pipeline/text_splitter.py import tiktoken def split_text_by_size(text: str, max_chunk_size: int = 500, overlap: int = 50) -> list[str]: encoder = tiktoken.get_encoding("cl100k_base") paragraphs = text.split("\n") chunks = [] current_chunk = "" current_tokens = 0 for para in paragraphs: para_token_count = len(encoder.encode(para)) if current_tokens + para_token_count > max_chunk_size: if current_chunk: chunks.append(current_chunk) overlap_text = current_chunk[-overlap:] if overlap > 0 else "" current_chunk = overlap_text + para current_tokens = len(encoder.encode(current_chunk)) else: current_chunk += "\n" + para current_tokens += para_token_count if current_chunk: chunks.append(current_chunk) return chunks这里用 token 数量而不是字符数量来控制长度，是因为嵌入模型和大模型的上下文限制通常按 token 计算。overlap 参数的意义是让相邻 chunk 之间保留一部分重复内容，避免因为硬切导致一句话前后断开。切片策略没有绝对最优，需要结合你的文档类型做实验。后面第 7 节会讲如何根据效果反向调整。5.3 批量向量化并写入向量数据库切片完成后，使用嵌入模型将每个 chunk 转为向量，然后写入 Qdrant。# 文件路径：rag_pipeline/index_builder.py from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct from openai import OpenAI client = OpenAI() def get_embedding(text: str) -> list[float]: # 这里替换为你选择的嵌入模型；本地模型则修改 base_url resp = client.embeddings.create( model="text-embedding-3-small", input=text ) return resp.data[0].embedding def build_index(chunks: list[str], collection_name: str = "knowledge_base") -> None: qdrant = QdrantClient(host="localhost", port=6333) # 如果集合不存在，先创建；已存在则跳过 existing_collections = [c.name for c in qdrant.get_collections().collections] if collection_name not in existing_collections: qdrant.create_collection( collection_name=collection_name, vectors_config=VectorParams(size=1536, distance=Distance.COSINE) ) points = [] for idx, chunk in enumerate(chunks): vector = get_embedding(chunk) points.append( PointStruct( id=idx, vector=vector, payload={"text": chunk} ) ) # 分批写入，避免一次性提交过多数据 BATCH_SIZE = 100 for i in range(0, len(points), BATCH_SIZE): qdrant.upsert( collection_name=collection_name, points=points[i:i + BATCH_SIZE] ) if __name__ == "__main__": from document_loader import load_pdf_text from text_cleaner import clean_text from text_splitter import split_text_by_size raw_text = load_pdf_text("sample_docs/product_manual.pdf") cleaned_text = clean_text(raw_text) chunks = split_text_by_size(cleaned_text) build_index(chunks) print(f"索引完成，共写入 {len(chunks)} 个切片")向量维度 1536 对应 text-embedding-3-small 模型的输出维度。如果你换了嵌入模型，这里的 size 需要同步修改。5.4 构造检索函数检索函数负责把用户问题向量化，然后去向量数据库里找最相关的切片。# 文件路径：rag_pipeline/retriever.py from qdrant_client import QdrantClient def retrieve_relevant_chunks(query: str, top_k: int = 4) -> list[str]: qdrant = QdrantClient(host="localhost", port=6333) query_vector = get_embedding(query) search_result = qdrant.search( collection_name="knowledge_base", query_vector=query_vector, limit=top_k ) return [hit.payload["text"] for hit in search_result]注意这里 get_embedding 函数和索引构建时用的是同一个嵌入模型。如果检索和索引用了不同的嵌入模型，向量空间不一致，召回效果会直接崩溃。5.5 构造 Prompt 并调用大模型检索结果拿到后，需要组织成 Prompt。这里的关键是让大模型只依据提供的资料回答，不要自行发挥。# 文件路径：rag_pipeline/generator.py from openai import OpenAI client = OpenAI() def generate_answer(query: str, retrieved_chunks: list[str]) -> str: context = "\n\n".join(retrieved_chunks) prompt = f"""请根据以下参考资料回答用户问题。 参考资料： {context} 用户问题：{query} 回答要求： 1. 如果参考资料中包含答案，请基于参考资料回答； 2. 如果参考资料中找不到答案，明确说明“资料中未找到相关信息”； 3. 不要编造资料中不存在的内容； 4. 回答时尽量引用原文关键信息。 回答：""" resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.2 ) return resp.choices[0].message.contenttemperature 设为较低值，是为了让模型更忠实于资料，减少自由发挥。如果希望答案更有创造性，可以适当调高，但在知识库问答场景里，低温度通常是安全选择。5.6 组装完整问答流程把前面所有模块串起来，形成一个完整的 ask 函数。# 文件路径：rag_pipeline/rag_engine.py from document_loader import load_pdf_text from text_cleaner import clean_text from text_splitter import split_text_by_size from index_builder import build_index from retriever import retrieve_relevant_chunks from generator import generate_answer def init_knowledge_base(pdf_path: str) -> None: raw_text = load_pdf_text(pdf_path) cleaned_text = clean_text(raw_text) chunks = split_text_by_size(cleaned_text) build_index(chunks) def ask(question: str) -> str: chunks = retrieve_relevant_chunks(question, top_k=4) answer = generate_answer(question, chunks) return answer if __name__ == "__main__": # 首次运行先初始化知识库；如果已经建立索引，可以注释掉 # init_knowledge_base("sample_docs/product_manual.pdf") while True: user_input = input("请输入问题（输入 exit 退出）：") if user_input.strip().lower() == "exit": break answer = ask(user_input) print("答案：", answer) print("-" * 50)到这里，一个最小可用的 RAG 知识库问答系统就完成了。6. 运行效果与召回质量验证代码写完后，需要验证系统是否真正工作正常。这一步很多新手会直接跳过，导致后续排查问题时没有方向。6.1 验证流程按顺序执行以下检查：输入一个与文档内容高度相关的问题，确认能返回合理回答；输入一个与文档内容无关的问题，确认模型没有强行编造；直接查看检索返回的 top_k 切片，判断召回内容是否真的和问题语义相关；对比“仅调用大模型”和“RAG 完整流程”的答案，确认 RAG 带来的信息增量。# 在项目根目录下运行 python rag_pipeline/rag_engine.py运行后示例输出：请输入问题（输入 exit 退出）：产品支持的售后响应时间是多久？ 答案：根据产品手册中的说明，售后响应时间分为两个等级：基础支持服务为 2 个工作日，高级支持服务为 4 个工作小时内响应。 -------------------------------------------------- 请输入问题（输入 exit 退出）：公司今年的财务报表是多少？ 答案：资料中未找到相关信息。 --------------------------------------------------判断标准很简单：能在文档中找到明确答案的问题，答案应该有据可查；不在文档中的问题，应该明确说“未找到”，而不是给你编一个数字。6.2 检索效果的具体检查方法如果回答效果不理想，不要急着改 Prompt，先检查检索环节。方法是在 ask 函数中临时打印检索到的切片内容。def ask_with_debug(question: str) -> str: chunks = retrieve_relevant_chunks(question, top_k=4) print("=== 检索到的切片 ===") for i, chunk in enumerate(chunks): print(f"[{i}] {chunk[:150]}") print("-" * 30) answer = generate_answer(question, chunks) return answer输出中可以看到，系统到底有没有把正确答案所在的切片召回出来。判断逻辑如下：如果正确答案对应的切片没有被召回，问题出在“检索”而非“生成”；如果正确切片被召回了，但模型回答错误，问题出在“Prompt 组织”或“模型选择”；如果检索结果中相关和不相关的内容混杂，问题出在“切片策略”或“嵌入模型”。这个调试思路可以覆盖 RAG 80% 以上的问题定位场景。6.3 用评测问题集量化效果做一次认真的验证时，建议准备一个评测问题集。每个问题包含三个字段：问题文本、参考答案、预期来源位置。然后对每个问题运行系统，用下面的标准打分：结果类型说明评分完全正确答案准确且来源正确1.0部分正确答案方向对，但关键信息缺失或偏差0.5检索正确但生成错误检索切片正确，但模型没答好0.3检索错误正确答案切片没有被召回0理想情况下，10 个问题中至少应该有 6 到 7 个达到“完全正确”。如果都达不到，先按 6.2 的方法定位问题，不要继续叠加复杂功能。7. RAG 常见问题与排查方法下面这张表整理了 RAG 项目中最常遇到的现象和排查思路。它比单看某一步报错更有用，因为很多问题不是编译错误，而是“效果不对、说不出哪里不对”的状态。问题现象可能原因排查方式解决方案文档中文解析为空或乱码PDF 是扫描件或图片格式用 OCR 提取文本，或检查 PDF 文本层引入 PaddleOCR 方案，替换文档源为可编辑格式回答内容明显超出资料范围Prompt 没有限制模型基于资料回答检查 Prompt 模板是否有明确约束增加“若资料中无答案，请说明未找到”的指令相关答案明明存在，却召回不到切片粒度过大或过小打印检索切片，确认正确内容是否被召回调整切片大小和 overlap，或切换嵌入模型检索结果中内容主题混杂文档结构复杂，直接按字符硬切观察 chunk 内容是否完整按标题层级切分，保留文档结构上传新文档后检索结果没变化新文档未写入向量数据库确认索引构建流程是否执行重新执行索引脚本，检查 upsert 返回检索性能越来越差集合中数据量大且未优化索引查看向量数据库索引类型和查询耗时调整索引参数，或分库分集合调用大模型接口超时检索返回的切片过多，Prompt 过长查看请求量和上下文 Token 数降低 top_k，或对检索切片做重排截断从这张表可以总结出一个重要的经验：RAG 调试的正确顺序是“先查检索，再查生成”。很多团队浪费大量时间在调 Prompt 上，最后发现根因是切片策略导致正确答案根本没被召回。8. 召回优化：从基础版到可用版前面实现的 RAG 是最基础的版本。真实项目里，问题往往不会那么单纯：用户问法和文档原文表述不同、答案分散在多个切片、检索出来的内容噪音比例较高。这一节讲几个工程上常用的优化手段，按实施成本从低到高排列。8.1 混合检索与稀疏向量纯向量检索擅长语义匹配，但对精确关键词不敏感。比如用户搜索“ERROR_CODE_403”，向量检索可能返回一堆“错误码”相关的泛泛内容，而基于关键词的 BM25 检索却能精确命中目标代码片段。实践中，可以将向量检索与传统关键词检索结果融合。Qdrant 支持同时挂载稠密向量和稀疏向量，也可以用独立的关键词检索服务，然后做结果合并。合并策略可以是简单的分数加权：final_score = alpha * vector_score + (1 - alpha) * bm25_scorealpha 需要根据数据集调试。混合检索对“代码、编号、专业术语”较多的知识库提升明显。8.2 重新排序向量检索的目标是“快速召回”，而不是“精确排序”。即使是同一个知识库，向量检索返回的前几结果也可能有噪声。重新排序的思路是：先用向量检索召回 20 到 50 条候选切片，再用一个更强的重排模型（Reranker）对候选做精排，只取前 3 到 5 条进入 Prompt。Reranker 与普通嵌入模型的区别在于：普通嵌入模型把文本变成向量后比较相似度，而 Reranker 直接计算“问题+文档片段”的相关性打分，效果更准但速度更慢。因为只对几十条候选做精排，性能开销通常可以接受。如果没有专门的重排模型，可以用大模型做一次粗排，让模型按相关性对切片打分，再取 TopK。8.3 查询改写用户的问题往往是碎片化的，尤其在多轮对话里。比如用户先问“公司年假政策是什么”，接着问“那加班呢”，如果直接把第二个问题拿去检索，向量数据库根本不知道该找什么。查询改写要解决的就是这个问题。常用做法是：调用大模型，结合对话历史，把当前问题改写成一个独立的、完整的检索 query。def rewrite_query(history: list[dict], current_query: str) -> str: history_text = "\n".join([f"{item['role']}: {item['content']}" for item in history[-4:]]) prompt = f"""根据对话历史，将最新用户问题改写为一个适合检索的独立问题。 对话历史： {history_text} 最新问题：{current_query} 改写后的问题：""" resp = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0 ) return resp.choices[0].message.content改写后的 query 只用于检索，不直接用于生成回答。这样既保证检索能命中，又不会让改写过程丢失原始问题的意图。8.4 结构化的分层检索如果文档有清晰的结构，比如产品手册包含章节编号，可以考虑两层检索：第一层根据用户问题定位相关章节，第二层只在相关章节中做向量检索。这种方式大幅缩小检索范围，精度更高。一个更简单的实现是：在切片的 payload 中带上章节信息，在检索层面把用户的 query 先做一次分类，然后给向量检索增加 filter 条件。8.5 去重与上下文压缩召回切片之间经常有大量重叠内容，直接全部拼进 Prompt 不仅浪费 Token，而且可能让模型被重复信息干扰。可以在拼接前做简易去重，比如对切片摘要计算哈希，删除相似度超过阈值的重复项。上下文压缩解决的是另一个问题：切片里只有一小段和问题相关，但整块内容都很长。压缩组件可以抽取与问题高度相关的句子，替代完整切片进入 Prompt。实现时可借助大模型做抽取式压缩，虽然会增加一次模型调用，但能显著改善长文档场景下的回答质量。这里需要强调一个原则：召回优化没有“放之四海而皆准”的组合，必须基于你自己的文档类型和失败案例来取舍。比较稳妥的路径是：先建一个评测问题集，再逐项加入上述优化，每次只改一个变量，对比评测分数。9. RAG 系统生产化落地的工程建议从 Demo 到生产环境，差的不是代码量，而是工程细节。这一节给出几个面向真实项目的建议。9.1 文档更新与索引生命周期管理知识库内容不会一成不变。生产系统中需要设计文档的新增、更新、删除流程。建议为每个文档分配唯一 doc_id，向量数据写入时把 doc_id 和 chunk_id 一起存进 payload。当文档更新时，先按 doc_id 删除旧切片，再写入新切片。避免因为重复导入导致同一文档的旧版本污染检索结果。Qdrant 中按 payload 条件删除的示例：qdrant.delete( collection_name="knowledge_base", points_selector=models.FilterSelector( filter=models.Filter( must=[models.FieldCondition(key="doc_id", match=models.MatchValue(value="doc_2024_001"))] ) ) )更新操作最好做成异步任务，并加入失败重试机制，避免文档处理到一半系统崩溃造成索引不完整。9.2 API 服务化与并发控制生产环境不能只有一个命令行脚本。建议使用 FastAPI 将问答能力封装成 HTTP 接口，同时做以下处理：请求超时控制，避免大模型接口慢导致整个服务雪崩；基于用户的访问频率限流；将知识库集合按业务线拆分，不同业务使用不同的 collection；对问答请求做日志记录，方便后续分析用户提问趋势和回答质量。9.3 安全与合规边界知识库问答系统的核心资产是私有文档。在设计和部署时必须考虑：向量数据库不应直接暴露在公网，应只允许应用服务器访问；如果使用云 API 嵌入模型或大模型，要确认数据出境是否符合合规要求，敏感数据优先采用本地部署模型；对用户的访问权限做隔离，不要让低权限用户检索到高权限文档；对敏感文档的切片进行脱敏处理后再入向量库。涉及权限控制的场景，可以采用“文档元数据过滤 + 用户权限标签”的方式，在检索时通过 filter 条件限定可见范围。9.4 评估与持续优化机制生产环境必须建立持续评估机制。建议做三件事：收集线上用户问题，每周挑出回答质量差的案例，归类到“检索问题”还是“生成问题”；维护一个回归评测集，每次修改切片策略或 Prompt 后，先跑一遍评测集，防止改好一个问题却带崩另外几个问题；记录回答的引用来源，用户可点击查看答案出处，这既能提升可信度，也方便人工审核。9.5 成本控制RAG 的成本集中在向量化和大模型生成两方面。降低成本的常见手段：对知识库做增量索引，只对新文档向量化，而不是全量重建；检索时合理设置 top_k，避免大量不相关切片浪费 Token；对常见问题设置缓存，完全相同的用户问题直接命中缓存答案；高频简单问题可以先用小模型回答，只有复杂问题才调用大模型。成本不是越低越好，关键是建立“回答质量与成本挂钩”的评估意识，在整体效率维度上做权衡。10. 总结与下一步实践建议回到文章开头的问题：RAG 适合做什么、不适合做什么，现在应该已经有了更具体的答案。它不会让大模型“无所不知”，但能让大模型在“有资料可依”的场景里回答得更可靠。它的核心价值是把私有知识、动态知识注入到生成链路里，而且不需要修改模型参数。真正决定 RAG 项目成败的，不是代码量多大，而是五个维度：文档解析质量、切片策略、向量召回效果、Prompt 约束、工程运维能力。建议你在动手写项目时，把时间分配大致调整为：40% 花在文档处理和切片优化上，30% 花在检索效果验证上，剩下的时间才留给代码框架和接口封装。下一步的实践路径可以这样安排：找一份你熟悉的业务文档，用本文示例代码建立索引；准备 10 到 20 个有标准答案的评测问题；跑通基线版本，记录评测分数；依次加入混合检索、查询改写、重排，观察每一步的分数变化；将问答封装成 HTTP 服务，再考虑文档更新流程和权限控制。学习过程中最容易踩的坑是“过早引入框架”。如果一上来就用 LangChain 或 LlamaIndex，你会被各种抽象类、回调、链式调用绕晕，出了问题也不知道该看哪个模块。先手写一遍流程，再用框架封装，认识的深度完全不同。如果后续想深入，可以继续学习这些方向：文本切分的自适应算法、向量索引参数的调优原理、Reranker 模型训练数据的构造、Agent 与 RAG 的结合方式，以及多路召回后的精排策略。建议现在就把这篇教程收藏，然后选一份自己的文档，按第 5 节的步骤跑通第一个知识库问答。只有亲手走完一遍，你才会真正理解为什么说“RAG 的难点不在原理，而在每一个细节”。