Hello大家好。最近大模型应用开发越来越热尤其是 RAGRetrieval-Augmented Generation检索增强生成几乎成了企业知识库项目的标配方案。很多同学私下问我网上的 RAG 教程要么只讲概念要么只贴几个零散的代码片段真正能落地的“检索→召回→重排→生成”全链路项目实战却很少。这篇文章我结合近期做企业级知识库项目的经验从零开始手把手带大家实现一个完整的 RAG 系统。文章不追求“花哨”而是尽量还原工程落地时的完整流程包括文档解析、文本切分、向量化、向量库存储、混合检索、重排优化、LLM 生成以及最后的评估与工程化部署。无论你是刚接触大模型的新手还是已经在做应用开发的工程师都能从里面找到可以直接复用的思路和代码。开始之前先说明本文以常见的开源技术栈为主代码示例以 Python LangChain FAISS/Milvus Qwen/Ollama 为例。版本和接口会随社区更新变化遇到报错时优先对照你的实际版本来调整。1. RAG 到底是什么为什么企业知识库离不开它1.1 从一个业务场景说起假设你现在要给公司做一个内部知识库助手员工可以提问“我们公司的年假制度是什么新员工入职后多久可以转正”如果直接把这类问题丢给 ChatGPT 或开源大模型模型大概率会按照训练数据里的通用知识来回答而不会知道你公司的具体制度。原因很简单大模型的参数里只能保存训练阶段见过的知识无法实时、无损地学习你内部的文档。这时有两个选择微调Fine-tuning把公司文档整理成问答对微调模型参数。RAG不改变模型参数先把文档切分、向量化存入知识库用户提问时先从知识库里检索最相关的片段再把“问题片段”一起交给大模型生成答案。在企业内部知识库场景里RAG 通常是性价比更高、落地更快的方案。理由后面会详细讲。1.2 RAG 的专业定义RAG 的全称是 Retrieval-Augmented Generation即检索增强生成。它的核心思想是在文本生成过程中引入一个外部知识检索模块让大模型在作答时不仅依赖自身参数记忆还能参考检索出来的真实文档内容。这样做的好处非常明显回答更准确能覆盖企业内部、私有领域的知识。答案可溯源可以给用户展示“这一段回答来自哪份文档”。知识更新成本低文档改了重新索引一遍即可不需要重新训练模型。1.3 RAG 和微调的区别很多初学者会把 RAG 和微调混在一起。这里做一个简单区分对比项RAG微调是否需要 GPU 训练通常不需要推理时用大模型即可需要 GPU 训练知识更新成本低重新切分向量化即可高每次更新要重新训练是否改变模型参数不改变改变适合场景企业知识库、实时信息问答、私有文档问答特定风格输出、领域术语固定、强化模型能力可解释性强可以追溯到原始文档弱难以解释生成依据实际项目中两者也可以结合例如先用微调让模型适配行业术语再用 RAG 挂载最新的企业知识文档。但如果你刚起步优先把 RAG 链路做好收益往往更明显。1.4 常见应用场景企业内部制度问答人事、财务、行政产品说明书/维修手册智能客服科研文献检索问答医疗、法律、教育等垂直领域的知识助手基于私有数据的 BI 分析辅助Agent 工具调用中的长期记忆和外部检索2. 环境准备与技术选型在动手写代码之前建议先把环境梳理清楚。RAG 项目的技术栈涉及大模型推理、向量数据库、文档处理等多个模块版本不一致很容易出问题。本文示例环境如下组件推荐选型说明操作系统Windows 10/11、macOS、Linux 均可建议 16G 以上内存Python3.9 - 3.11不要直接用 3.13很多依赖尚未完全兼容大模型推理Ollama Qwen2.5-7B-Instruct本地离线运行也可以换成 OpenAI/DeepSeek APIEmbedding 模型BAAI/bge-m3中文效果较好也可用 text2vec 系列向量数据库FAISS入门/ Milvus生产看数据量先 FAISS 后 Milvus编排框架LangChain 0.2/0.3也可以不用框架自己写 Pipeline文档解析PyMuPDF / unstructuredPDF、Word、Markdown 解析安装核心依赖pip install langchain langchain-community langchain-openai pip install faiss-cpu pip install pymupdf pip install bs4 pip install ollama pip install flask # 后面做 API 服务用 pip install pymilvus # 如果使用 Milvus如果你本地没有足够资源跑大模型可以改用在线 API例如 DeepSeek、通义千问的 API代码只需要修改模型名和 Base URL。这里有一个需要注意的点不同 Embedding 模型的向量维度不同向量库建好后不要随意更换 Embedding 模型否则检索会失效。from langchain_community.embeddings import HuggingFaceBgeEmbeddings model_name BAAI/bge-m3 encode_kwargs {normalize_embeddings: True} embeddings HuggingFaceBgeEmbeddings( model_namemodel_name, encode_kwargsencode_kwargs, query_instruction为这个句子生成表示以用于检索相关文章 )如果你首次运行HuggingFace 会自动下载模型到本地缓存目录。国内网络环境下建议设置镜像export HF_ENDPOINThttps://hf-mirror.com3. RAG 全链路核心原理拆解有了环境之后我们来把 RAG 的完整链路拆开。很多人一上来就写代码但代码写好了检索效果却很差回头排查又无从下手就是因为对链路里的“坑”不够了解。一个完整的 RAG 系统通常包含以下阶段文档加载与解析文本切分Chunking向量化Embedding向量存储用户查询处理检索召回Recall重排Rerank生成Generation下面挨个讲。3.1 文档加载与解析数据源通常是 PDF、Word、Markdown、HTML 或数据库记录。解析环节要保证文字不乱码、表格不丢失、章节结构尽量保留。简单场景下可以用PyMuPDF提取 PDF 文本import fitz # PyMuPDF def extract_text_from_pdf(pdf_path): doc fitz.open(pdf_path) texts [] for page_num in range(doc.page_count): page doc[page_num] texts.append(page.get_text()) return \n.join(texts)如果是扫描版 PDF图片格式需要 OCR常用方案是PaddleOCR或RapidOCR。OCR 是另一个大坑如果文档清晰不一定需要上 OCR可以先人工确认再决定。3.2 文本切分是 RAG 效果的第一道关卡很多人忽略切分的重要性默认按固定长度切 500 个字结果检索到的片段要么太碎、要么把两个不相关的主题拼在一起。切分的目标是尽可能让每个 chunk 保持语义完整并且包含足够的上下文信息。推荐做法先按文档结构切优先保留标题、段落。再按句子边界或固定滑动窗口切。设置重叠overlap避免关键句被拦腰截断。LangChain 提供了多种切分器。做中文知识库我推荐先试RecursiveCharacterTextSplitterfrom langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap80, separators[\n\n, \n, 。, , , , , , ], length_functionlen, ) chunks text_splitter.split_text(document_text) print(f切分后共有 {len(chunks)} 个片段)注意separators的优先级越靠前的分隔符越优先使用。中文文本中\n\n通常表示段落分隔语义完整性最好所以放在最前面其次按句子标点符号切分。如果文档有明确的 Markdown 标题结构可以先用MarkdownHeaderTextSplitter按标题切再细切这样携带的上下文更完整。3.3 Embedding让计算机理解“语义相近”文本不能直接丢给向量库搜索。我们需要把每个 chunk 转换成一个高维向量例如 1024 维浮点数转换为法就是 Embedding 模型。Embedding 模型的核心能力是语义相近的文本在向量空间中的距离也相近。例如“如何申请年假”和“休年假的流程”两个句子虽然字面不同但向量距离应该很近。中文场景下常用的 Embedding 模型模型向量维度特点BAAI/bge-m31024中文效果优秀也支持多语言moka-ai/m3e-base768轻量适合入门text2vec-large-chinese1024纯中文优化OpenAI text-embedding-3-small1536需要调用 API选择 Embedding 模型时建议拿自己的领域数据做一个小测试集对比不同模型在检索召回中的表现不要只看榜单分数。3.4 向量存储与检索召回向量化的 chunks 需要存储到向量数据库中。入门阶段用 FAISS 就够了数据量到百万级别再迁移 Milvus。FAISS 本地索引实现from langchain_community.vectorstores import FAISS vectorstore FAISS.from_documents( documentsdocuments, # 这里是 LangChain Document 对象列表 embeddingembeddings ) # 保存到本地 vectorstore.save_local(faiss_index) # 加载 vectorstore FAISS.load_local( faiss_index, embeddings, allow_dangerous_deserializationTrue # 仅限可信数据源 )检索时最基本的方法是向量相似度检索ANNdocs vectorstore.similarity_search_with_score(query, k10) for doc, score in docs: print(f相似度: {score}) print(doc.page_content) print(---)但这里有一个常见的性能陷阱单纯用向量检索往往只关注语义相似容易忽略关键词精确匹配。例如用户输入一个产品型号“ABC-2000”Embedding 模型可能把它和“设备型号、产品规格”的语义混在一起而精确的关键词匹配反而更有效。因此企业级 RAG 一般会做“混合检索”把向量检索和全文检索关键词匹配结合下面会重点讲。3.5 重排让前几名更精准向量检索召回 Top 20 甚至 Top 50 片段后如果直接全塞给大模型有两个问题大模型输入长度有限不能塞太多。相似的并不代表真正相关低质量的片段会干扰生成。重排Rerank模型的作用是对召回结果进行精细化排序把最可能满足当前问题的片段排到最前。通常我们会召回 Top 20~50重排后取 Top 3~5再送进大模型。常用的重排方案bge-reranker-v2-m3中英文效果都不错推荐先用它。Cohere Rerank API商业服务。跨编码器Cross-Encoder模型直接对“问题文档片段”打分。示例代码基于 FlagEmbeddingpip install FlagEmbeddingfrom FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-v2-m3) pairs [[query, doc.page_content] for doc in docs] scores reranker.compute_score(pairs, normalizeTrue) # 按分数排序 sorted_docs [doc for _, doc in sorted(zip(scores, docs), keylambda x: x[0], reverseTrue)]重排后的片段顺序需要保持得分从高到低。这一步虽然增加了耗时但对生成质量的提升非常明显尤其是文档数量多、主题分散的场景。3.6 生成把检索结果组装成 Prompt最后一步是把重排后的片段和用户问题拼成一个 Prompt交给大模型生成答案。这里要注意两点片段不能“只塞给模型”最好加上来源引用便于溯源。Prompt 中要明确告诉模型“只能根据提供的文档回答如果文档中没有答案直接说明不知道不要编造。”一个可用的 Prompt 模板你是一个企业知识库问答助手。请你根据以下资料片段回答问题。 资料片段 {context} 问题 {question} 要求 1. 如果资料中没有足够信息直接回答“根据现有资料无法回答”。 2. 在回答末尾标注依据的资料来源编号例如 [1]。 3. 回答要简洁、准确。4. 完整实战从零搭建企业知识库 RAG 系统这一节我们来完成一个完整的可运行项目。项目结构如下rag_project/ ├── data/ # 存放原始文档 ├── src/ │ ├── config.py # 全局配置 │ ├── document_loader.py # 文档加载与解析 │ ├── text_splitter.py # 文本切分 │ ├── vectorstore.py # 向量库构建与检索 │ ├── reranker.py # 重排模块 │ ├── generator.py # 大模型生成模块 │ └── rag_pipeline.py # 完整 RAG 流程编排 ├── app.py # Flask API 服务 └── requirements.txt4.1 全局配置模块# src/config.py import os BASE_DIR os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DATA_DIR os.path.join(BASE_DIR, data) FAISS_INDEX_DIR os.path.join(BASE_DIR, faiss_index) # Embedding 模型 EMBEDDING_MODEL BAAI/bge-m3 # 重排模型 RERANK_MODEL BAAI/bge-reranker-v2-m3 # 大模型配置以 Ollama 为例 LLM_BASE_URL http://localhost:11434 LLM_MODEL qwen2.5:7b # 检索策略 TOP_K_RECALL 20 # 召回数量 TOP_K_RERANK 5 # 重排后保留数量 CHUNK_SIZE 512 CHUNK_OVERLAP 804.2 文档加载与解析# src/document_loader.py import os import fitz from langchain.schema import Document def load_documents_from_dir(data_dir: str): 加载 data 目录下的所有文档支持 pdf / md / txt docs [] for root, _, files in os.walk(data_dir): for file in files: file_path os.path.join(root, file) ext file.lower().rsplit(., 1)[-1] if ext pdf: text _extract_pdf(file_path) elif ext in (md, txt): with open(file_path, r, encodingutf-8) as f: text f.read() else: continue # 记录来源 doc Document(page_contenttext, metadata{source: file_path}) docs.append(doc) return docs def _extract_pdf(file_path: str) - str: doc fitz.open(file_path) text_parts [] for page in doc: text_parts.append(page.get_text()) return \n.join(text_parts)4.3 文本切分# src/text_splitter.py from langchain.text_splitter import RecursiveCharacterTextSplitter from src.config import CHUNK_SIZE, CHUNK_OVERLAP def split_documents(documents): text_splitter RecursiveCharacterTextSplitter( chunk_sizeCHUNK_SIZE, chunk_overlapCHUNK_OVERLAP, separators[\n\n, \n, 。, , , , , , ], length_functionlen, ) split_docs text_splitter.split_documents(documents) print(f切分完成共得到 {len(split_docs)} 个片段) return split_docs4.4 向量库构建# src/vectorstore.py from langchain_community.embeddings import HuggingFaceBgeEmbeddings from langchain_community.vectorstores import FAISS from src.config import EMBEDDING_MODEL, FAISS_INDEX_DIR def build_vectorstore(documents): embeddings HuggingFaceBgeEmbeddings( model_nameEMBEDDING_MODEL, encode_kwargs{normalize_embeddings: True}, ) vectorstore FAISS.from_documents(documents, embeddings) vectorstore.save_local(FAISS_INDEX_DIR) return vectorstore def load_vectorstore(): embeddings HuggingFaceBgeEmbeddings( model_nameEMBEDDING_MODEL, encode_kwargs{normalize_embeddings: True}, ) return FAISS.load_local( FAISS_INDEX_DIR, embeddings, allow_dangerous_deserializationTrue, )我们把“构建索引”和“加载索引”分成了两个函数。实际项目中索引构建通常由离线任务完成在线服务只负责加载索引。4.5 重排模块# src/reranker.py from FlagEmbedding import FlagReranker from src.config import RERANK_MODEL, TOP_K_RERANK class Reranker: def __init__(self): self.model FlagReranker(RERANK_MODEL) def rerank(self, query, documents, top_kTOP_K_RERANK): pairs [[query, doc.page_content] for doc in documents] scores self.model.compute_score(pairs, normalizeTrue) ranked_pairs sorted(zip(scores, documents), keylambda x: x[0], reverseTrue) return ranked_pairs[:top_k]4.6 大模型生成模块这里以 Ollama 本地模型为例通过 OpenAI 兼容接口调用# src/generator.py from openai import OpenAI from src.config import LLM_BASE_URL, LLM_MODEL class Generator: def __init__(self): self.client OpenAI(base_urlLLM_BASE_URL, api_keyollama) def generate(self, question, context_docs): context_text \n\n.join( f[{idx 1}] {doc.page_content} for idx, doc in enumerate(context_docs) ) prompt f你是一个企业知识库问答助手。请你根据以下资料片段回答问题。 资料片段 {context_text} 问题 {question} 要求 1. 如果资料中没有足够信息直接回答根据现有资料无法回答。 2. 回答末尾需要标注依据的资料编号例如 [1][2]。 3. 回答要简洁、准确。 response self.client.chat.completions.create( modelLLM_MODEL, messages[{role: user, content: prompt}], temperature0.2, ) return response.choices[0].message.content4.7 完整 RAG Pipeline# src/rag_pipeline.py from src.document_loader import load_documents_from_dir from src.text_splitter import split_documents from src.vectorstore import build_vectorstore, load_vectorstore from src.reranker import Reranker from src.generator import Generator from src.config import DATA_DIR, TOP_K_RECALL class RAGPipeline: def __init__(self, use_cacheTrue): # 优先加载已有向量库减少重建耗时 try: self.vectorstore load_vectorstore() except Exception as e: print(f加载向量库失败开始重建: {e}) self._build_index() self.reranker Reranker() self.generator Generator() def _build_index(self): documents load_documents_from_dir(DATA_DIR) split_docs split_documents(documents) build_vectorstore(split_docs) self.vectorstore load_vectorstore() def answer(self, question): # 第一步向量检索召回 recalled_docs self.vectorstore.similarity_search(question, kTOP_K_RECALL) print(f召回片段数{len(recalled_docs)}) # 第二步重排 ranked_results self.reranker.rerank(question, recalled_docs) final_docs [doc for _, doc in ranked_results] print(f重排后保留片段数{len(final_docs)}) # 第三步生成 answer self.generator.generate(question, final_docs) return answer, final_docs4.8 Flask API 服务为了让项目能对外提供服务我们写一个简单 API# app.py from flask import Flask, request, jsonify from src.rag_pipeline import RAGPipeline app Flask(__name__) pipeline RAGPipeline() app.route(/chat, methods[POST]) def chat(): data request.get_json() question data.get(question, ) if not question: return jsonify({error: question 不能为空}), 400 answer, source_docs pipeline.answer(question) sources [doc.metadata.get(source, ) for doc in source_docs] return jsonify({ answer: answer, sources: sources, }) if __name__ __main__: app.run(host0.0.0.0, port8000)启动服务python app.py测试请求curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {question: 公司年假制度是什么}预期返回 JSON其中包含模型生成的答案和命中文档来源列表。5. 进阶优化混合检索与重排实战上文示例使用了“向量检索 重排”的方案。但真实企业环境中文档数量大、主题分散只用向量检索往往不够。5.1 什么是混合检索混合检索Hybrid Search是指同时使用向量相似度检索和基于关键词的全文检索再把两部分结果合并排序。为什么需要混合检索因为向量检索擅长语义匹配例如“我想休假”和“年假申请流程”关联。关键词检索擅长精确匹配例如型号、编号、专有名词。LangChain 中可以把 BM25 检索器和 FAISS 检索器组合起来from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever from langchain_community.vectorstores import FAISS def build_hybrid_retriever(split_docs, vectorstore): # BM25 关键词检索器 bm25_retriever BM25Retriever.from_documents(split_docs, k10) # 向量检索器 vector_retriever vectorstore.as_retriever(search_kwargs{k: 10}) # 组合检索器 hybrid_retriever EnsembleRetriever( retrievers[bm25_retriever, vector_retriever], weights[0.3, 0.7], ) return hybrid_retrieverweights控制两种检索结果的权重。经验上如果领域术语较多可以适当提高 BM25 的权重如果问答以自然语言为主提高向量检索权重。5.2 RRF 分数融合混合检索的核心问题是如何合并两个排名列表。常用方法是 RRFReciprocal Rank Fusion倒数排名融合公式为score(d) Σ 1 / (k rank_i(d))其中rank_i(d)是文档 d 在第 i 个检索结果中的排名k是常数通常取 60。LangChain 的EnsembleRetriever内部已经实现了类似融合逻辑。如果想自己实现可以参考下面的核心思路def rrf_fusion(ranked_results, k60): scores {} for result_list in ranked_results: for rank, doc in enumerate(result_list): doc_id doc.metadata.get(source, ) doc.page_content[:50] scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)5.3 重排策略的选择重排模型不是越贵越好要结合延迟和资源来选。数据量小、对延迟不敏感使用bge-reranker-v2-m3准确率高。高并发、低延迟场景可以先向量召回 10~20 条重排只处理 Top 5减少计算量。纯英文场景可以尝试cohere-rerank-english-v3.0等商业化 API不要照搬中文经验。另外重排分数需要归一化方便设置阈值过滤低质量片段。例如分数低于 0.3 的片段直接丢弃。5.4 查询改写与多路召回用户在真实业务中的问题往往很口语化例如“咱们公司那个休假咋弄来着”直接检索很难命中“员工休假管理办法”。常见做法是引入查询改写Query Rewriting用 LLM 把口语化问题改写成更正式、更完整的检索式句子。对原问题抽取关键词生成多个查询词分别检索后再融合。示例 Prompt请将用户问题改写为更正式的知识库检索查询语句只输出改写后的查询不要多余解释。 用户问题咱们公司那个休假咋弄来着 改写查询6. RAG 知识库指标与评估很多同学做完 RAG 系统后面对“效果怎么样”的问题很难回答。这里梳理几个核心指标和评估思路。6.1 检索质量指标指标英文含义召回率RecallK前 K 个结果中命中了多少正确答案命中率Hit RateK前 K 个结果中是否至少有一个正确答案MRRMean Reciprocal Rank第一个正确答案排在第几位越大越好NDCGNormalized Discounted Cumulative Gain考虑排序位置的综合质量指标精确率PrecisionK前 K 个结果中真正相关的占比实际项目中常用 Hit Rate 和 MRR 来快速评估检索效果因为它们直观、便于解释。6.2 生成质量指标生成质量评估相对主观常用方法人工评估非常耗时但最可靠。LLM-as-a-Judge用 GPT-4 等强模型给答案打分评估维度包括忠实度、相关性和完整性。RAGAS 框架开源评估框架自动化计算忠实度Faithfulness和答案相关性Answer Relevance。RAGAS 中两个关键概念忠实度Faithfulness生成的答案是否严格基于检索到的文档有没有编造。答案相关性Answer Relevance答案是否对用户的问题有针对性而不是泛泛而谈。对于企业知识库来说忠实度比答案相关性更重要——知识库助手最怕一本正经地胡说八道。建议在 RAG 上线前准备 100 条左右的标准测试问题覆盖正常情况、边界情况和文档中找不到答案的情况手工标注后持续回归评估。6.3 检索质量差先查哪个环节如果你的 RAG 效果差建议按以下顺序排查数据质量问题源文档格式混乱、扫描件 OCR 错误、表格被拆散都会导致上游信息错误。切分策略问题chunk 太大语义噪声多chunk 太小上下文不足。Embedding 模型问题通用模型对专业领域文本不敏感可以采样对比不同模型。检索策略问题是否试过混合检索召回的数量是否太少重排阈值问题是否引入了大量低分片段Prompt 问题模型是否被允许说“不知道”上下文格式是否清晰很多时候问题不在某一个环节而是多个环节叠加。建议每个环节单独做 A/B 测试不要同时修改多个变量。7. 工程化落地从 Demo 到企业级系统7.1 离线索引与在线查询分离第一版 RAG 系统常常是“启动时先构建索引”数据量小还可以数据量大了会有几个问题启动时间过长。文档更新时不能动态生效。在线服务崩溃后重建索引成本高。工程上应该做索引构建索引写入和查询服务索引读取分离。我的建议是增加一个 build 命令和 incremental update 能力。文档第一次全量构建后续增量更新。离线任务可以每天或每小时跑一次更新向量库。如果是 Milvus支持动态插入数据可以在文档变更后直接 insert 新数据同时删除旧数据。7.2 向量数据库选型方案数据量级适用场景FAISS百万以下单机原型、低成本Milvus百万以上分布式生产、高并发Chroma小规模快速开发调试Elasticsearch文本搜索为主需要全文检索、聚合分析pgvector中等已有 PostgreSQL 环境实际项目如果已经有 Elasticsearch也可以直接把向量写入 ES避免多引入一个中间件。但要注意 ES 的向量检索性能和 Milvus 相比在高并发场景下有差距。7.3 缓存设计大模型推理成本高、延迟高。对重复问题或相似问题可以加一层语义缓存把用户查询向量化在缓存库中查找相似度高于阈值的向量直接返回历史答案。或者用简单的文本归一化缓存去除标点、大小写归一化。简单缓存实现class SemanticCache: def __init__(self, threshold0.92): self.cache [] # 元素为 (query_vector, query, answer, sources) self.threshold threshold def get(self, query_vector, query): for vec, q, answer, sources in self.cache: similarity cosine_similarity(query_vector, vec) if similarity self.threshold: return answer, sources return None def set(self, query_vector, query, answer, sources): self.cache.append((query_vector, query, answer, sources)) if len(self.cache) 1000: self.cache.pop(0) # 简单 LRU7.4 异步化与流式输出对于企业级 API建议文档解析、切分、向量化放到异步任务队列Celery Redis执行。大模型生成接口开启流式输出SSE提升用户体验。检索和生成链路中增加 Trace ID便于排查问题。使用 FastAPI 实现流式输出示例from fastapi import FastAPI from fastapi.responses import StreamingResponse import json app FastAPI() app.post(/chat/stream) async def chat_stream(question: dict): def generate(): answer pipeline.answer_stream(question[question]) for token in answer: yield fdata: {json.dumps({token: token}, ensure_asciiFalse)}\n\n return StreamingResponse(generate(), media_typetext/event-stream)7.5 安全与权限企业知识库最大的风险是权限绕过。例如一个普通员工可以通过构造 Prompt 获取 HR 系统中的敏感信息。在设计 RAG 系统时务必考虑以下几点文档级别权限控制不同用户只能检索其有权访问的文档。对检索结果进行权限过滤而不是只对最终答案过滤。对 Prompt 注入攻击做好防护如果文档中包含“忽略以上指令”等文本模型可能被诱导。审计日志记录每个用户的问题、检索到的文档、生成的答案。简单做法是在文档 metadata 中加入权限标签检索后过滤allowed_doc_ids get_user_permissions(user_id) filtered_docs [ doc for doc in retrieved_docs if doc.metadata.get(doc_id) in allowed_doc_ids ]如果使用 Milvus可以通过 Partition Key 或 Filter 条件实现粗粒度的权限过滤。8. 常见问题与排查思路下面整理 RAG 项目里最常遇到的几个问题。8.1 检索不到相关内容问题现象常见原因解决思路检索结果与问题完全无关Embedding 模型效果差更换中文领域模型测试不同模型检索结果太少召回 K 值设置太小调大 TOP_K_RECALL例如从 5 调到 20关键词匹配失效没有使用混合检索增加 BM25 检索并融合新文档检索不到向量库没有更新确认文档已切分并写入向量库检索到的片段太碎chunk_size 过小增大 chunk_size 并设置 overlap8.2 生成答案出现幻觉问题现象常见原因解决思路回答内容文档中没有Prompt 没有限制在 Prompt 中明确禁止编造引用来源错误没有关联 chunk 归属在切分时保留文档 ID 和标题信息答案过于发散temperature 过高将 temperature 调低至 0~0.2上下文不够重排后只剩 1~2 个片段增加 TOP_K_RERANK保证证据充分模型被文档中的恶意 Prompt 干扰未处理注入攻击对文档做 Prompt 注入清洗或增加隔离分隔符8.3 性能问题问题现象常见原因解决思路每个问题响应超过 10 秒重排模型计算量大缩小重排输入或对重排结果做缓存向量检索耗时高数据量太大FAISS 单机内存不足迁移 Milvus并添加索引参数并发一高就超时没有限流和队列增加异步队列、限流、服务扩容大模型推理慢模型参数量太大量化模型GGUF 格式 Q4或换小参数量模型8.4 环境与依赖问题Python 版本导致 LangChain 组件不兼容是很常见的问题。推荐使用虚拟环境python -m venv venv source venv/bin/activate # Windows 是 venv\Scripts\activate pip install -r requirements.txt如果遇到ModuleNotFoundError: No module named langchain_community说明 LangChain 版本太老或没安装 community 包执行pip install -U langchain-community如果 FAISS 加载时出现allow_dangerous_deserialization报错这是 LangChain 新版本的安全限制。只在你信任索引文件来源时才设置allow_dangerous_deserializationTrue不要对来自外部系统或用户的文件放松校验。9. 最佳实践与工程建议9.1 数据与索引规范文档必须先清洗再切分去掉页眉页脚、水印、重复内容。维护文档版本号保证向量库和源文档的映射关系清晰。chunk 的 metadata 中至少包含来源文件、文档 ID、标题、页码。索引构建任务要有幂等性重复执行不应产生脏数据。9.2 检索策略建议优先使用“混合检索 RRF”。向量召回数量建议 20~50重排后再截断到 3~5 个片段。基于领域数据测试不同切分参数不要盲目套用默认值。对高频问题使用缓存减少大模型调用。9.3 Prompt 工程建议提示词要明确告诉模型“基于给定文档回答”。给模型一个回答不了的出口例如“根据现有资料无法回答”。建议输出中带引用编号保证可溯源性。评估不同 Prompt 时固定测试集逐条对比输出不要凭感觉调整。9.4 安全与权限建议最小权限原则每个用户只能检索到其权限范围内的文档。不要直接拼接用户输入到检索语句注意查询注入风险。对文档中的恶意内容例如“忽略以上指令”做检测和过滤。敏感知识库建议本地部署大模型避免数据外泄。10. 最后RAG 学习路线与项目迭代方向如果本文的内容你已经全部走通恭喜你你已经具备从零搭建一套 RAG 知识库系统的基础能力。接下来可以沿着以下几个方向继续深入Agentic RAG让 RAG 系统具备自主规划能力可以多轮追问、自主决定是否需要检索、调用多个工具。Graph RAG用知识图谱增强 RAG适合多跳问答和关系型知识。更细粒度的权限控制实现基于用户的文档级检索权限隔离。多模态 RAG扩展支持图片、表格、语音等多类型数据的知识库。评估自动化搭建一套可持续运行的回归测试平台每次修改后自动评测 RAG 质量。整体来看RAG 项目并不难入门难的是“调优”和“落地”。一个生产级 RAG 系统花在数据清洗、效果评估和权限处理上的时间往往会超过写核心代码的时间。建议你先跑通一个小型数据集再逐步扩大规模和优化这样排查问题会更从容。如果本文对你有帮助欢迎收藏备用。遇到问题也可以在评论区交流我会尽量回复。动手搭一套自己的企业级 RAG 知识库系统其实没有想象中那么复杂关键是先把链路走通再一步步打磨细节。 RAG全链路实战：从零搭建企业知识库问答系统
Hello大家好。最近大模型应用开发越来越热尤其是 RAGRetrieval-Augmented Generation检索增强生成几乎成了企业知识库项目的标配方案。很多同学私下问我网上的 RAG 教程要么只讲概念要么只贴几个零散的代码片段真正能落地的“检索→召回→重排→生成”全链路项目实战却很少。这篇文章我结合近期做企业级知识库项目的经验从零开始手把手带大家实现一个完整的 RAG 系统。文章不追求“花哨”而是尽量还原工程落地时的完整流程包括文档解析、文本切分、向量化、向量库存储、混合检索、重排优化、LLM 生成以及最后的评估与工程化部署。无论你是刚接触大模型的新手还是已经在做应用开发的工程师都能从里面找到可以直接复用的思路和代码。开始之前先说明本文以常见的开源技术栈为主代码示例以 Python LangChain FAISS/Milvus Qwen/Ollama 为例。版本和接口会随社区更新变化遇到报错时优先对照你的实际版本来调整。1. RAG 到底是什么为什么企业知识库离不开它1.1 从一个业务场景说起假设你现在要给公司做一个内部知识库助手员工可以提问“我们公司的年假制度是什么新员工入职后多久可以转正”如果直接把这类问题丢给 ChatGPT 或开源大模型模型大概率会按照训练数据里的通用知识来回答而不会知道你公司的具体制度。原因很简单大模型的参数里只能保存训练阶段见过的知识无法实时、无损地学习你内部的文档。这时有两个选择微调Fine-tuning把公司文档整理成问答对微调模型参数。RAG不改变模型参数先把文档切分、向量化存入知识库用户提问时先从知识库里检索最相关的片段再把“问题片段”一起交给大模型生成答案。在企业内部知识库场景里RAG 通常是性价比更高、落地更快的方案。理由后面会详细讲。1.2 RAG 的专业定义RAG 的全称是 Retrieval-Augmented Generation即检索增强生成。它的核心思想是在文本生成过程中引入一个外部知识检索模块让大模型在作答时不仅依赖自身参数记忆还能参考检索出来的真实文档内容。这样做的好处非常明显回答更准确能覆盖企业内部、私有领域的知识。答案可溯源可以给用户展示“这一段回答来自哪份文档”。知识更新成本低文档改了重新索引一遍即可不需要重新训练模型。1.3 RAG 和微调的区别很多初学者会把 RAG 和微调混在一起。这里做一个简单区分对比项RAG微调是否需要 GPU 训练通常不需要推理时用大模型即可需要 GPU 训练知识更新成本低重新切分向量化即可高每次更新要重新训练是否改变模型参数不改变改变适合场景企业知识库、实时信息问答、私有文档问答特定风格输出、领域术语固定、强化模型能力可解释性强可以追溯到原始文档弱难以解释生成依据实际项目中两者也可以结合例如先用微调让模型适配行业术语再用 RAG 挂载最新的企业知识文档。但如果你刚起步优先把 RAG 链路做好收益往往更明显。1.4 常见应用场景企业内部制度问答人事、财务、行政产品说明书/维修手册智能客服科研文献检索问答医疗、法律、教育等垂直领域的知识助手基于私有数据的 BI 分析辅助Agent 工具调用中的长期记忆和外部检索2. 环境准备与技术选型在动手写代码之前建议先把环境梳理清楚。RAG 项目的技术栈涉及大模型推理、向量数据库、文档处理等多个模块版本不一致很容易出问题。本文示例环境如下组件推荐选型说明操作系统Windows 10/11、macOS、Linux 均可建议 16G 以上内存Python3.9 - 3.11不要直接用 3.13很多依赖尚未完全兼容大模型推理Ollama Qwen2.5-7B-Instruct本地离线运行也可以换成 OpenAI/DeepSeek APIEmbedding 模型BAAI/bge-m3中文效果较好也可用 text2vec 系列向量数据库FAISS入门/ Milvus生产看数据量先 FAISS 后 Milvus编排框架LangChain 0.2/0.3也可以不用框架自己写 Pipeline文档解析PyMuPDF / unstructuredPDF、Word、Markdown 解析安装核心依赖pip install langchain langchain-community langchain-openai pip install faiss-cpu pip install pymupdf pip install bs4 pip install ollama pip install flask # 后面做 API 服务用 pip install pymilvus # 如果使用 Milvus如果你本地没有足够资源跑大模型可以改用在线 API例如 DeepSeek、通义千问的 API代码只需要修改模型名和 Base URL。这里有一个需要注意的点不同 Embedding 模型的向量维度不同向量库建好后不要随意更换 Embedding 模型否则检索会失效。from langchain_community.embeddings import HuggingFaceBgeEmbeddings model_name BAAI/bge-m3 encode_kwargs {normalize_embeddings: True} embeddings HuggingFaceBgeEmbeddings( model_namemodel_name, encode_kwargsencode_kwargs, query_instruction为这个句子生成表示以用于检索相关文章 )如果你首次运行HuggingFace 会自动下载模型到本地缓存目录。国内网络环境下建议设置镜像export HF_ENDPOINThttps://hf-mirror.com3. RAG 全链路核心原理拆解有了环境之后我们来把 RAG 的完整链路拆开。很多人一上来就写代码但代码写好了检索效果却很差回头排查又无从下手就是因为对链路里的“坑”不够了解。一个完整的 RAG 系统通常包含以下阶段文档加载与解析文本切分Chunking向量化Embedding向量存储用户查询处理检索召回Recall重排Rerank生成Generation下面挨个讲。3.1 文档加载与解析数据源通常是 PDF、Word、Markdown、HTML 或数据库记录。解析环节要保证文字不乱码、表格不丢失、章节结构尽量保留。简单场景下可以用PyMuPDF提取 PDF 文本import fitz # PyMuPDF def extract_text_from_pdf(pdf_path): doc fitz.open(pdf_path) texts [] for page_num in range(doc.page_count): page doc[page_num] texts.append(page.get_text()) return \n.join(texts)如果是扫描版 PDF图片格式需要 OCR常用方案是PaddleOCR或RapidOCR。OCR 是另一个大坑如果文档清晰不一定需要上 OCR可以先人工确认再决定。3.2 文本切分是 RAG 效果的第一道关卡很多人忽略切分的重要性默认按固定长度切 500 个字结果检索到的片段要么太碎、要么把两个不相关的主题拼在一起。切分的目标是尽可能让每个 chunk 保持语义完整并且包含足够的上下文信息。推荐做法先按文档结构切优先保留标题、段落。再按句子边界或固定滑动窗口切。设置重叠overlap避免关键句被拦腰截断。LangChain 提供了多种切分器。做中文知识库我推荐先试RecursiveCharacterTextSplitterfrom langchain.text_splitter import RecursiveCharacterTextSplitter text_splitter RecursiveCharacterTextSplitter( chunk_size512, chunk_overlap80, separators[\n\n, \n, 。, , , , , , ], length_functionlen, ) chunks text_splitter.split_text(document_text) print(f切分后共有 {len(chunks)} 个片段)注意separators的优先级越靠前的分隔符越优先使用。中文文本中\n\n通常表示段落分隔语义完整性最好所以放在最前面其次按句子标点符号切分。如果文档有明确的 Markdown 标题结构可以先用MarkdownHeaderTextSplitter按标题切再细切这样携带的上下文更完整。3.3 Embedding让计算机理解“语义相近”文本不能直接丢给向量库搜索。我们需要把每个 chunk 转换成一个高维向量例如 1024 维浮点数转换为法就是 Embedding 模型。Embedding 模型的核心能力是语义相近的文本在向量空间中的距离也相近。例如“如何申请年假”和“休年假的流程”两个句子虽然字面不同但向量距离应该很近。中文场景下常用的 Embedding 模型模型向量维度特点BAAI/bge-m31024中文效果优秀也支持多语言moka-ai/m3e-base768轻量适合入门text2vec-large-chinese1024纯中文优化OpenAI text-embedding-3-small1536需要调用 API选择 Embedding 模型时建议拿自己的领域数据做一个小测试集对比不同模型在检索召回中的表现不要只看榜单分数。3.4 向量存储与检索召回向量化的 chunks 需要存储到向量数据库中。入门阶段用 FAISS 就够了数据量到百万级别再迁移 Milvus。FAISS 本地索引实现from langchain_community.vectorstores import FAISS vectorstore FAISS.from_documents( documentsdocuments, # 这里是 LangChain Document 对象列表 embeddingembeddings ) # 保存到本地 vectorstore.save_local(faiss_index) # 加载 vectorstore FAISS.load_local( faiss_index, embeddings, allow_dangerous_deserializationTrue # 仅限可信数据源 )检索时最基本的方法是向量相似度检索ANNdocs vectorstore.similarity_search_with_score(query, k10) for doc, score in docs: print(f相似度: {score}) print(doc.page_content) print(---)但这里有一个常见的性能陷阱单纯用向量检索往往只关注语义相似容易忽略关键词精确匹配。例如用户输入一个产品型号“ABC-2000”Embedding 模型可能把它和“设备型号、产品规格”的语义混在一起而精确的关键词匹配反而更有效。因此企业级 RAG 一般会做“混合检索”把向量检索和全文检索关键词匹配结合下面会重点讲。3.5 重排让前几名更精准向量检索召回 Top 20 甚至 Top 50 片段后如果直接全塞给大模型有两个问题大模型输入长度有限不能塞太多。相似的并不代表真正相关低质量的片段会干扰生成。重排Rerank模型的作用是对召回结果进行精细化排序把最可能满足当前问题的片段排到最前。通常我们会召回 Top 20~50重排后取 Top 3~5再送进大模型。常用的重排方案bge-reranker-v2-m3中英文效果都不错推荐先用它。Cohere Rerank API商业服务。跨编码器Cross-Encoder模型直接对“问题文档片段”打分。示例代码基于 FlagEmbeddingpip install FlagEmbeddingfrom FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-v2-m3) pairs [[query, doc.page_content] for doc in docs] scores reranker.compute_score(pairs, normalizeTrue) # 按分数排序 sorted_docs [doc for _, doc in sorted(zip(scores, docs), keylambda x: x[0], reverseTrue)]重排后的片段顺序需要保持得分从高到低。这一步虽然增加了耗时但对生成质量的提升非常明显尤其是文档数量多、主题分散的场景。3.6 生成把检索结果组装成 Prompt最后一步是把重排后的片段和用户问题拼成一个 Prompt交给大模型生成答案。这里要注意两点片段不能“只塞给模型”最好加上来源引用便于溯源。Prompt 中要明确告诉模型“只能根据提供的文档回答如果文档中没有答案直接说明不知道不要编造。”一个可用的 Prompt 模板你是一个企业知识库问答助手。请你根据以下资料片段回答问题。 资料片段 {context} 问题 {question} 要求 1. 如果资料中没有足够信息直接回答“根据现有资料无法回答”。 2. 在回答末尾标注依据的资料来源编号例如 [1]。 3. 回答要简洁、准确。4. 完整实战从零搭建企业知识库 RAG 系统这一节我们来完成一个完整的可运行项目。项目结构如下rag_project/ ├── data/ # 存放原始文档 ├── src/ │ ├── config.py # 全局配置 │ ├── document_loader.py # 文档加载与解析 │ ├── text_splitter.py # 文本切分 │ ├── vectorstore.py # 向量库构建与检索 │ ├── reranker.py # 重排模块 │ ├── generator.py # 大模型生成模块 │ └── rag_pipeline.py # 完整 RAG 流程编排 ├── app.py # Flask API 服务 └── requirements.txt4.1 全局配置模块# src/config.py import os BASE_DIR os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DATA_DIR os.path.join(BASE_DIR, data) FAISS_INDEX_DIR os.path.join(BASE_DIR, faiss_index) # Embedding 模型 EMBEDDING_MODEL BAAI/bge-m3 # 重排模型 RERANK_MODEL BAAI/bge-reranker-v2-m3 # 大模型配置以 Ollama 为例 LLM_BASE_URL http://localhost:11434 LLM_MODEL qwen2.5:7b # 检索策略 TOP_K_RECALL 20 # 召回数量 TOP_K_RERANK 5 # 重排后保留数量 CHUNK_SIZE 512 CHUNK_OVERLAP 804.2 文档加载与解析# src/document_loader.py import os import fitz from langchain.schema import Document def load_documents_from_dir(data_dir: str): 加载 data 目录下的所有文档支持 pdf / md / txt docs [] for root, _, files in os.walk(data_dir): for file in files: file_path os.path.join(root, file) ext file.lower().rsplit(., 1)[-1] if ext pdf: text _extract_pdf(file_path) elif ext in (md, txt): with open(file_path, r, encodingutf-8) as f: text f.read() else: continue # 记录来源 doc Document(page_contenttext, metadata{source: file_path}) docs.append(doc) return docs def _extract_pdf(file_path: str) - str: doc fitz.open(file_path) text_parts [] for page in doc: text_parts.append(page.get_text()) return \n.join(text_parts)4.3 文本切分# src/text_splitter.py from langchain.text_splitter import RecursiveCharacterTextSplitter from src.config import CHUNK_SIZE, CHUNK_OVERLAP def split_documents(documents): text_splitter RecursiveCharacterTextSplitter( chunk_sizeCHUNK_SIZE, chunk_overlapCHUNK_OVERLAP, separators[\n\n, \n, 。, , , , , , ], length_functionlen, ) split_docs text_splitter.split_documents(documents) print(f切分完成共得到 {len(split_docs)} 个片段) return split_docs4.4 向量库构建# src/vectorstore.py from langchain_community.embeddings import HuggingFaceBgeEmbeddings from langchain_community.vectorstores import FAISS from src.config import EMBEDDING_MODEL, FAISS_INDEX_DIR def build_vectorstore(documents): embeddings HuggingFaceBgeEmbeddings( model_nameEMBEDDING_MODEL, encode_kwargs{normalize_embeddings: True}, ) vectorstore FAISS.from_documents(documents, embeddings) vectorstore.save_local(FAISS_INDEX_DIR) return vectorstore def load_vectorstore(): embeddings HuggingFaceBgeEmbeddings( model_nameEMBEDDING_MODEL, encode_kwargs{normalize_embeddings: True}, ) return FAISS.load_local( FAISS_INDEX_DIR, embeddings, allow_dangerous_deserializationTrue, )我们把“构建索引”和“加载索引”分成了两个函数。实际项目中索引构建通常由离线任务完成在线服务只负责加载索引。4.5 重排模块# src/reranker.py from FlagEmbedding import FlagReranker from src.config import RERANK_MODEL, TOP_K_RERANK class Reranker: def __init__(self): self.model FlagReranker(RERANK_MODEL) def rerank(self, query, documents, top_kTOP_K_RERANK): pairs [[query, doc.page_content] for doc in documents] scores self.model.compute_score(pairs, normalizeTrue) ranked_pairs sorted(zip(scores, documents), keylambda x: x[0], reverseTrue) return ranked_pairs[:top_k]4.6 大模型生成模块这里以 Ollama 本地模型为例通过 OpenAI 兼容接口调用# src/generator.py from openai import OpenAI from src.config import LLM_BASE_URL, LLM_MODEL class Generator: def __init__(self): self.client OpenAI(base_urlLLM_BASE_URL, api_keyollama) def generate(self, question, context_docs): context_text \n\n.join( f[{idx 1}] {doc.page_content} for idx, doc in enumerate(context_docs) ) prompt f你是一个企业知识库问答助手。请你根据以下资料片段回答问题。 资料片段 {context_text} 问题 {question} 要求 1. 如果资料中没有足够信息直接回答根据现有资料无法回答。 2. 回答末尾需要标注依据的资料编号例如 [1][2]。 3. 回答要简洁、准确。 response self.client.chat.completions.create( modelLLM_MODEL, messages[{role: user, content: prompt}], temperature0.2, ) return response.choices[0].message.content4.7 完整 RAG Pipeline# src/rag_pipeline.py from src.document_loader import load_documents_from_dir from src.text_splitter import split_documents from src.vectorstore import build_vectorstore, load_vectorstore from src.reranker import Reranker from src.generator import Generator from src.config import DATA_DIR, TOP_K_RECALL class RAGPipeline: def __init__(self, use_cacheTrue): # 优先加载已有向量库减少重建耗时 try: self.vectorstore load_vectorstore() except Exception as e: print(f加载向量库失败开始重建: {e}) self._build_index() self.reranker Reranker() self.generator Generator() def _build_index(self): documents load_documents_from_dir(DATA_DIR) split_docs split_documents(documents) build_vectorstore(split_docs) self.vectorstore load_vectorstore() def answer(self, question): # 第一步向量检索召回 recalled_docs self.vectorstore.similarity_search(question, kTOP_K_RECALL) print(f召回片段数{len(recalled_docs)}) # 第二步重排 ranked_results self.reranker.rerank(question, recalled_docs) final_docs [doc for _, doc in ranked_results] print(f重排后保留片段数{len(final_docs)}) # 第三步生成 answer self.generator.generate(question, final_docs) return answer, final_docs4.8 Flask API 服务为了让项目能对外提供服务我们写一个简单 API# app.py from flask import Flask, request, jsonify from src.rag_pipeline import RAGPipeline app Flask(__name__) pipeline RAGPipeline() app.route(/chat, methods[POST]) def chat(): data request.get_json() question data.get(question, ) if not question: return jsonify({error: question 不能为空}), 400 answer, source_docs pipeline.answer(question) sources [doc.metadata.get(source, ) for doc in source_docs] return jsonify({ answer: answer, sources: sources, }) if __name__ __main__: app.run(host0.0.0.0, port8000)启动服务python app.py测试请求curl -X POST http://localhost:8000/chat \ -H Content-Type: application/json \ -d {question: 公司年假制度是什么}预期返回 JSON其中包含模型生成的答案和命中文档来源列表。5. 进阶优化混合检索与重排实战上文示例使用了“向量检索 重排”的方案。但真实企业环境中文档数量大、主题分散只用向量检索往往不够。5.1 什么是混合检索混合检索Hybrid Search是指同时使用向量相似度检索和基于关键词的全文检索再把两部分结果合并排序。为什么需要混合检索因为向量检索擅长语义匹配例如“我想休假”和“年假申请流程”关联。关键词检索擅长精确匹配例如型号、编号、专有名词。LangChain 中可以把 BM25 检索器和 FAISS 检索器组合起来from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever from langchain_community.vectorstores import FAISS def build_hybrid_retriever(split_docs, vectorstore): # BM25 关键词检索器 bm25_retriever BM25Retriever.from_documents(split_docs, k10) # 向量检索器 vector_retriever vectorstore.as_retriever(search_kwargs{k: 10}) # 组合检索器 hybrid_retriever EnsembleRetriever( retrievers[bm25_retriever, vector_retriever], weights[0.3, 0.7], ) return hybrid_retrieverweights控制两种检索结果的权重。经验上如果领域术语较多可以适当提高 BM25 的权重如果问答以自然语言为主提高向量检索权重。5.2 RRF 分数融合混合检索的核心问题是如何合并两个排名列表。常用方法是 RRFReciprocal Rank Fusion倒数排名融合公式为score(d) Σ 1 / (k rank_i(d))其中rank_i(d)是文档 d 在第 i 个检索结果中的排名k是常数通常取 60。LangChain 的EnsembleRetriever内部已经实现了类似融合逻辑。如果想自己实现可以参考下面的核心思路def rrf_fusion(ranked_results, k60): scores {} for result_list in ranked_results: for rank, doc in enumerate(result_list): doc_id doc.metadata.get(source, ) doc.page_content[:50] scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)5.3 重排策略的选择重排模型不是越贵越好要结合延迟和资源来选。数据量小、对延迟不敏感使用bge-reranker-v2-m3准确率高。高并发、低延迟场景可以先向量召回 10~20 条重排只处理 Top 5减少计算量。纯英文场景可以尝试cohere-rerank-english-v3.0等商业化 API不要照搬中文经验。另外重排分数需要归一化方便设置阈值过滤低质量片段。例如分数低于 0.3 的片段直接丢弃。5.4 查询改写与多路召回用户在真实业务中的问题往往很口语化例如“咱们公司那个休假咋弄来着”直接检索很难命中“员工休假管理办法”。常见做法是引入查询改写Query Rewriting用 LLM 把口语化问题改写成更正式、更完整的检索式句子。对原问题抽取关键词生成多个查询词分别检索后再融合。示例 Prompt请将用户问题改写为更正式的知识库检索查询语句只输出改写后的查询不要多余解释。 用户问题咱们公司那个休假咋弄来着 改写查询6. RAG 知识库指标与评估很多同学做完 RAG 系统后面对“效果怎么样”的问题很难回答。这里梳理几个核心指标和评估思路。6.1 检索质量指标指标英文含义召回率RecallK前 K 个结果中命中了多少正确答案命中率Hit RateK前 K 个结果中是否至少有一个正确答案MRRMean Reciprocal Rank第一个正确答案排在第几位越大越好NDCGNormalized Discounted Cumulative Gain考虑排序位置的综合质量指标精确率PrecisionK前 K 个结果中真正相关的占比实际项目中常用 Hit Rate 和 MRR 来快速评估检索效果因为它们直观、便于解释。6.2 生成质量指标生成质量评估相对主观常用方法人工评估非常耗时但最可靠。LLM-as-a-Judge用 GPT-4 等强模型给答案打分评估维度包括忠实度、相关性和完整性。RAGAS 框架开源评估框架自动化计算忠实度Faithfulness和答案相关性Answer Relevance。RAGAS 中两个关键概念忠实度Faithfulness生成的答案是否严格基于检索到的文档有没有编造。答案相关性Answer Relevance答案是否对用户的问题有针对性而不是泛泛而谈。对于企业知识库来说忠实度比答案相关性更重要——知识库助手最怕一本正经地胡说八道。建议在 RAG 上线前准备 100 条左右的标准测试问题覆盖正常情况、边界情况和文档中找不到答案的情况手工标注后持续回归评估。6.3 检索质量差先查哪个环节如果你的 RAG 效果差建议按以下顺序排查数据质量问题源文档格式混乱、扫描件 OCR 错误、表格被拆散都会导致上游信息错误。切分策略问题chunk 太大语义噪声多chunk 太小上下文不足。Embedding 模型问题通用模型对专业领域文本不敏感可以采样对比不同模型。检索策略问题是否试过混合检索召回的数量是否太少重排阈值问题是否引入了大量低分片段Prompt 问题模型是否被允许说“不知道”上下文格式是否清晰很多时候问题不在某一个环节而是多个环节叠加。建议每个环节单独做 A/B 测试不要同时修改多个变量。7. 工程化落地从 Demo 到企业级系统7.1 离线索引与在线查询分离第一版 RAG 系统常常是“启动时先构建索引”数据量小还可以数据量大了会有几个问题启动时间过长。文档更新时不能动态生效。在线服务崩溃后重建索引成本高。工程上应该做索引构建索引写入和查询服务索引读取分离。我的建议是增加一个 build 命令和 incremental update 能力。文档第一次全量构建后续增量更新。离线任务可以每天或每小时跑一次更新向量库。如果是 Milvus支持动态插入数据可以在文档变更后直接 insert 新数据同时删除旧数据。7.2 向量数据库选型方案数据量级适用场景FAISS百万以下单机原型、低成本Milvus百万以上分布式生产、高并发Chroma小规模快速开发调试Elasticsearch文本搜索为主需要全文检索、聚合分析pgvector中等已有 PostgreSQL 环境实际项目如果已经有 Elasticsearch也可以直接把向量写入 ES避免多引入一个中间件。但要注意 ES 的向量检索性能和 Milvus 相比在高并发场景下有差距。7.3 缓存设计大模型推理成本高、延迟高。对重复问题或相似问题可以加一层语义缓存把用户查询向量化在缓存库中查找相似度高于阈值的向量直接返回历史答案。或者用简单的文本归一化缓存去除标点、大小写归一化。简单缓存实现class SemanticCache: def __init__(self, threshold0.92): self.cache [] # 元素为 (query_vector, query, answer, sources) self.threshold threshold def get(self, query_vector, query): for vec, q, answer, sources in self.cache: similarity cosine_similarity(query_vector, vec) if similarity self.threshold: return answer, sources return None def set(self, query_vector, query, answer, sources): self.cache.append((query_vector, query, answer, sources)) if len(self.cache) 1000: self.cache.pop(0) # 简单 LRU7.4 异步化与流式输出对于企业级 API建议文档解析、切分、向量化放到异步任务队列Celery Redis执行。大模型生成接口开启流式输出SSE提升用户体验。检索和生成链路中增加 Trace ID便于排查问题。使用 FastAPI 实现流式输出示例from fastapi import FastAPI from fastapi.responses import StreamingResponse import json app FastAPI() app.post(/chat/stream) async def chat_stream(question: dict): def generate(): answer pipeline.answer_stream(question[question]) for token in answer: yield fdata: {json.dumps({token: token}, ensure_asciiFalse)}\n\n return StreamingResponse(generate(), media_typetext/event-stream)7.5 安全与权限企业知识库最大的风险是权限绕过。例如一个普通员工可以通过构造 Prompt 获取 HR 系统中的敏感信息。在设计 RAG 系统时务必考虑以下几点文档级别权限控制不同用户只能检索其有权访问的文档。对检索结果进行权限过滤而不是只对最终答案过滤。对 Prompt 注入攻击做好防护如果文档中包含“忽略以上指令”等文本模型可能被诱导。审计日志记录每个用户的问题、检索到的文档、生成的答案。简单做法是在文档 metadata 中加入权限标签检索后过滤allowed_doc_ids get_user_permissions(user_id) filtered_docs [ doc for doc in retrieved_docs if doc.metadata.get(doc_id) in allowed_doc_ids ]如果使用 Milvus可以通过 Partition Key 或 Filter 条件实现粗粒度的权限过滤。8. 常见问题与排查思路下面整理 RAG 项目里最常遇到的几个问题。8.1 检索不到相关内容问题现象常见原因解决思路检索结果与问题完全无关Embedding 模型效果差更换中文领域模型测试不同模型检索结果太少召回 K 值设置太小调大 TOP_K_RECALL例如从 5 调到 20关键词匹配失效没有使用混合检索增加 BM25 检索并融合新文档检索不到向量库没有更新确认文档已切分并写入向量库检索到的片段太碎chunk_size 过小增大 chunk_size 并设置 overlap8.2 生成答案出现幻觉问题现象常见原因解决思路回答内容文档中没有Prompt 没有限制在 Prompt 中明确禁止编造引用来源错误没有关联 chunk 归属在切分时保留文档 ID 和标题信息答案过于发散temperature 过高将 temperature 调低至 0~0.2上下文不够重排后只剩 1~2 个片段增加 TOP_K_RERANK保证证据充分模型被文档中的恶意 Prompt 干扰未处理注入攻击对文档做 Prompt 注入清洗或增加隔离分隔符8.3 性能问题问题现象常见原因解决思路每个问题响应超过 10 秒重排模型计算量大缩小重排输入或对重排结果做缓存向量检索耗时高数据量太大FAISS 单机内存不足迁移 Milvus并添加索引参数并发一高就超时没有限流和队列增加异步队列、限流、服务扩容大模型推理慢模型参数量太大量化模型GGUF 格式 Q4或换小参数量模型8.4 环境与依赖问题Python 版本导致 LangChain 组件不兼容是很常见的问题。推荐使用虚拟环境python -m venv venv source venv/bin/activate # Windows 是 venv\Scripts\activate pip install -r requirements.txt如果遇到ModuleNotFoundError: No module named langchain_community说明 LangChain 版本太老或没安装 community 包执行pip install -U langchain-community如果 FAISS 加载时出现allow_dangerous_deserialization报错这是 LangChain 新版本的安全限制。只在你信任索引文件来源时才设置allow_dangerous_deserializationTrue不要对来自外部系统或用户的文件放松校验。9. 最佳实践与工程建议9.1 数据与索引规范文档必须先清洗再切分去掉页眉页脚、水印、重复内容。维护文档版本号保证向量库和源文档的映射关系清晰。chunk 的 metadata 中至少包含来源文件、文档 ID、标题、页码。索引构建任务要有幂等性重复执行不应产生脏数据。9.2 检索策略建议优先使用“混合检索 RRF”。向量召回数量建议 20~50重排后再截断到 3~5 个片段。基于领域数据测试不同切分参数不要盲目套用默认值。对高频问题使用缓存减少大模型调用。9.3 Prompt 工程建议提示词要明确告诉模型“基于给定文档回答”。给模型一个回答不了的出口例如“根据现有资料无法回答”。建议输出中带引用编号保证可溯源性。评估不同 Prompt 时固定测试集逐条对比输出不要凭感觉调整。9.4 安全与权限建议最小权限原则每个用户只能检索到其权限范围内的文档。不要直接拼接用户输入到检索语句注意查询注入风险。对文档中的恶意内容例如“忽略以上指令”做检测和过滤。敏感知识库建议本地部署大模型避免数据外泄。10. 最后RAG 学习路线与项目迭代方向如果本文的内容你已经全部走通恭喜你你已经具备从零搭建一套 RAG 知识库系统的基础能力。接下来可以沿着以下几个方向继续深入Agentic RAG让 RAG 系统具备自主规划能力可以多轮追问、自主决定是否需要检索、调用多个工具。Graph RAG用知识图谱增强 RAG适合多跳问答和关系型知识。更细粒度的权限控制实现基于用户的文档级检索权限隔离。多模态 RAG扩展支持图片、表格、语音等多类型数据的知识库。评估自动化搭建一套可持续运行的回归测试平台每次修改后自动评测 RAG 质量。整体来看RAG 项目并不难入门难的是“调优”和“落地”。一个生产级 RAG 系统花在数据清洗、效果评估和权限处理上的时间往往会超过写核心代码的时间。建议你先跑通一个小型数据集再逐步扩大规模和优化这样排查问题会更从容。如果本文对你有帮助欢迎收藏备用。遇到问题也可以在评论区交流我会尽量回复。动手搭一套自己的企业级 RAG 知识库系统其实没有想象中那么复杂关键是先把链路走通再一步步打磨细节。