当前笔记围绕企业级 RAG检索增强生成知识库展开从架构原理到检索、召回、重排、生成与工程化落地完整走一遍。适合正在做大模型应用、知识库问答、私有化部署的开发者参考也适合想系统理解 RAG 全链路的新手。1. RAG 是什么为什么企业知识库需要它1.1 先理解大模型的两大短板大模型虽然能聊天、能写代码、能做摘要但在企业知识问答场景里有两个明显问题第一知识有时效性。大模型的训练数据有截止日期之后发生的事情它不知道。公司内部新发布的制度、刚上线的产品规格、最新的法务条款模型训练时根本没见过。第二生成内容不可控。模型回答问题时会把训练时学到的“常识”和“编造内容”混在一起出现一本正经的胡说八道。这个问题在业内叫“幻觉”。对于企业知识库幻觉是致命的法务、财务、医疗、制造等场景尤其不能接受。1.2 RAG 的核心思路先检索再生成RAGRetrieval-Augmented Generation检索增强生成的思路很直接让大模型在回答前先“查资料”。整个链路可以概括为用户输入一个问题。系统从企业知识库中检索与问题最相关的内容片段。将这些内容片段拼接到 Prompt 中一起交给大模型。大模型基于“检索到的内容 用户问题”生成答案。这意味着模型不需要“记住”企业内部知识它只需要“读懂”检索到的资料。知识的新增、更新、删除都发生在知识库侧模型侧不需要重训练维护成本大幅降低。1.3 RAG 与模型微调的区别很多人会把 RAG 和模型微调Fine-tuning放在一起比较。两者的定位完全不同维度RAG模型微调知识更新改知识库即可成本低需要重新训练周期长硬件要求普通服务器可运行主要消耗检索资源需要 GPU 训练成本高回答准确性基于检索证据可控性较好依赖训练数据质量适用场景知识频繁变化、对正确性要求高模型能力增强、固定格式输出、风格调整幻觉风险可通过检索内容约束仍然存在实际项目中两者可以结合。比如先用 RAG 解决知识来源问题再用微调让模型学会特定的输出格式或行业表达习惯。但大多数知识库项目RAG 是主链路。2. RAG 全链路架构拆解RAG 不是“单个模型”或者“单个 API”而是一条完整的数据流水线。只有把整条链路拆清楚才能定位问题、优化效果。2.1 离线索引链路离线链路负责把企业文档变成可检索的结构化数据。文档接入 → 格式解析 → 文档清洗 → 文本切分 → 向量化 → 写入向量数据库 ↓ 建立关键词索引核心环节包括文档接入支持 PDF、Word、Markdown、TXT、HTML、Excel 等格式。格式解析PDF 需要抽取文本和表格Word 需要保留标题层级扫描件需要 OCR。文档清洗去掉页眉页脚、水印、无效换行、乱码字符。文本切分Chunking这是全链路里最容易影响效果的一步。向量化Embedding将文本转换成向量。索引写入向量入库同时保留原文、元数据、文档来源。2.2 在线推理链路在线链路处理用户请求。用户提问 → 查询改写 → 混合检索 → 重排 → 构造 Prompt → 大模型生成 ↓ 生成引用来源每一步都有优化空间查询改写把模糊问题改写成适合检索的表达比如把“上次说的那个方案”改写成具体名词。混合检索同时使用关键词检索和向量检索再合并结果。重排Rerank对检索结果做精细排序把最相关的片段排到前面。Prompt 构造在 Prompt 中限制模型只能使用检索内容作答。生成大模型输出答案并附上引用来源。2.3 完整流程图用户问题 ↓ [查询预处理] → 纠错、改写、意图识别 ↓ [混合检索] → BM25 关键词检索 → 向量检索KNN ↓ [结果融合] → 去重、分数归一化、取 TopN ↓ [重排] → Cross-Encoder / Rerank 模型 ↓ [Prompt 构造] → 注入检索片段、设定回答边界 ↓ [大模型生成] → 输出答案 引用来源 ↓ [后处理] → 格式校验、敏感词过滤、记录日志3. 环境准备与版本选型RAG 项目涉及的组件比较多。为了避免“本地能跑生产就崩”的问题需要一开始就做好技术选型。3.1 核心组件清单组件作用选型建议向量数据库存储向量、执行相似度检索Milvus、Qdrant、Elasticsearch 8.x、Chroma搜索引擎关键词检索Elasticsearch、OpenSearchEmbedding 模型文本转向量BGE、M3E、OpenAI Embedding API重排模型精细化排序bge-reranker、cross-encoder 类模型大模型最终答案生成开源模型或商业 API应用框架编排链路LangChain、LlamaIndex 或自研 Python 服务任务队列异步处理文档Celery、Arq、Redis Stream缓存降低重复请求延迟Redis3.2 版本注意事项技术版本更新很快所以这里不给写死的版本号大家按自己项目的实际情况选择。需要特别注意几个原则向量数据库版本与 SDK 版本必须匹配。客户端 SDK 大版本升级后连接方式、索引参数都可能有变化。Embedding 模型维度必须与索引配置一致。如果先建了 768 维的集合换模型后向量变成 1024 维就会写入失败。Python 版本建议使用 3.9。新版 LangChain 和向量数据库 SDK 对旧版本支持越来越差。生产环境避免使用 SQLite 类向量方案。Chroma 的原生持久化适合 Demo高并发场景需要独立的向量数据库服务。3.3 示例项目结构rag-project/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置文件 │ ├── ingestion/ │ │ ├── loader.py # 文档加载 │ │ ├── splitter.py # 文本切分 │ │ └── vectorize.py # 向量化与入库 │ ├── retrieval/ │ │ ├── keyword_search.py # 关键词检索 │ │ ├── vector_search.py # 向量检索 │ │ ├── hybrid_search.py # 混合检索与融合 │ │ └── rerank.py # 重排模块 │ ├── generation/ │ │ └── llm.py # Prompt 构造与生成 │ └── schemas.py # 请求响应模型 ├── data/ # 原始文档 ├── tests/ # 测试用例 └── requirements.txt如果是 Java 技术栈可以类比拆包用 Spring Boot 作为 Web 层用 langchain4j 做链路编排用 Milvus Java SDK 做向量检索。4. 完整实战从零搭建一个 RAG 知识库下面用一个可运行的 Python 示例把 RAG 全链路走一遍。这里使用 FastAPI Chroma HuggingFace Embedding 和 Rerank 模型重点演示架构思路。生产环境可以替换成 Milvus 或 Elasticsearch。4.1 安装依赖pip install fastapi uvicorn chromadb sentence-transformers pip install pypdf python-docx markdown pip install openai # 或者对接本地模型服务 pip install bm25s # 轻量 BM25 工具一些库可能在不同版本中 API 有变化安装时如果遇到兼容问题可以把版本固定下来。4.2 定义配置文件# 文件路径app/config.py import os EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, BAAI/bge-large-zh-v1.5) RERANK_MODEL os.getenv(RERANK_MODEL, BAAI/bge-reranker-v2-m3) LLM_BASE_URL os.getenv(LLM_BASE_URL, http://localhost:8000/v1) LLM_API_KEY os.getenv(LLM_API_KEY, EMPTY) LLM_MODEL os.getenv(LLM_MODEL, qwen2.5-7b-instruct) CHUNK_SIZE 512 CHUNK_OVERLAP 64 TOP_K 20 # 检索阶段取回的数量 RERANK_TOP_N 5 # 重排后保留的数量4.3 文档加载与清洗不同格式的文档需要不同的解析方式。这里先写一个统一的加载函数。# 文件路径app/ingestion/loader.py from pathlib import Path def load_document(file_path: str) - str: path Path(file_path) suffix path.suffix.lower() if suffix .txt: return path.read_text(encodingutf-8, errorsignore) elif suffix .md: return path.read_text(encodingutf-8, errorsignore) elif suffix .pdf: from pypdf import PdfReader reader PdfReader(str(path)) return \n.join(page.extract_text() or for page in reader.pages) elif suffix .docx: from docx import Document doc Document(str(path)) return \n.join(p.text for p in doc.paragraphs) else: raise ValueError(f暂不支持的文件格式{suffix})文档清洗时要注意去除页面页脚中的重复内容。将全角字符统一转半角。删除连续空行。保留 Markdown 标题结构方便后续切分。4.4 文本切分切分是 RAG 链路中最“细节”的环节。切太大检索精度会下降切太小片段缺乏上下文语义不完整。推荐使用“递归字符切分 固定窗口重叠”的思路# 文件路径app/ingestion/splitter.py from typing import List def recursive_split(text: str, chunk_size: int 512, chunk_overlap: int 64) - List[str]: separators [\n\n, \n, 。, , , , , , ] chunks [] current for para in text.split(\n): if len(current) len(para) chunk_size: current para \n else: if current: chunks.append(current.strip()) current para \n if current.strip(): chunks.append(current.strip()) # 对过长的片段做二次切分并保留重叠 final_chunks [] for chunk in chunks: if len(chunk) chunk_size: final_chunks.append(chunk) continue start 0 while start len(chunk): end start chunk_size final_chunks.append(chunk[start:end]) if end len(chunk): break start end - chunk_overlap return final_chunks这个实现保证以段落为基本单位优先合并。过长段落按固定长度切分。相邻 chunk 之间保留重叠避免重要信息被截断。实际项目里还可以结合标题层级切分让同一章节的内容尽量在同一个 chunk 中。4.5 向量化与入库使用 sentence-transformers 加载 Embedding 模型将文本转为向量后写入 Chroma。# 文件路径app/ingestion/vectorize.py import chromadb from sentence_transformers import SentenceTransformer from app.config import EMBEDDING_MODEL, CHUNK_SIZE, CHUNK_OVERLAP from app.ingestion.loader import load_document from app.ingestion.splitter import recursive_split client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(nameenterprise_rag) embedder SentenceTransformer(EMBEDDING_MODEL) def index_document(file_path: str, doc_id: str, metadata: dict None): text load_document(file_path) chunks recursive_split(text, CHUNK_SIZE, CHUNK_OVERLAP) embeddings embedder.encode(chunks).tolist() ids [f{doc_id}_{i} for i in range(len(chunks))] metadatas [ { doc_id: doc_id, chunk_index: i, source: file_path, **(metadata or {}), } for i in range(len(chunks)) ] collection.add( idsids, documentschunks, embeddingsembeddings, metadatasmetadatas, ) return len(chunks)注意Chromat 的add方法在重复执行同一文档时会报 ID 冲突所以更新文档时可以先collection.delete(where{doc_id: doc_id})再重新写入。4.6 混合检索向量检索 关键词检索只用向量检索的问题在于一些精确的专有名词、编号、公式、报错代码向量模型理解得并不好。比如搜索“BGP 邻居状态 Establised”关键词匹配能精准命中向量检索可能召回到语义相近但不够准确的内容。所以企业级 RAG 通常使用混合检索。向量检索代码# 文件路径app/retrieval/vector_search.py from sentence_transformers import SentenceTransformer from app.config import EMBEDDING_MODEL embedder SentenceTransformer(EMBEDDING_MODEL) def vector_search(query: str, top_k: int 20): query_embedding embedder.encode([query]).tolist()[0] results collection.query( query_embeddings[query_embedding], n_resultstop_k, include[documents, metadatas, distances], ) return results关键词检索使用 BM25 算法。这是一个经典的概率检索模型对专有名词匹配非常有效。# 文件路径app/retrieval/keyword_search.py from bm25s import BM25 from app.config import TOP_K # 全局维护一个语料库生产环境建议用 Elasticsearch bm25_index None def build_bm25_index(corpus: list[str], doc_ids: list[str]): global bm25_index tokenized [doc.split() for doc in corpus] bm25_index BM25() bm25_index.index(tokenized, doc_idsdoc_ids) def keyword_search(query: str, top_k: int TOP_K): if bm25_index is None: return [] tokenized_query query.split() results bm25_index.retrieve(tokenized_query, ktop_k) return results混合检索的融合策略有很多种。最简单的是 “Score 加权求和”# 文件路径app/retrieval/hybrid_search.py def hybrid_search(query: str, top_k: int TOP_K, alpha: float 0.5): vector_results vector_search(query, top_k) keyword_results keyword_search(query, top_k) # 统一分数区间后加权 scores {} docs {} for i, doc in enumerate(vector_results[documents][0]): score 1 / (1 vector_results[distances][0][i]) docs[doc] vector_results[metadatas][0][i] scores[doc] scores.get(doc, 0) alpha * score for doc_id, score in keyword_results.items(): if score 0: continue doc_text doc_id_to_text.get(doc_id, ) if not doc_text: continue scores[doc_text] scores.get(doc_text, 0) (1 - alpha) * score docs.setdefault(doc_text, {doc_id: doc_id}) sorted_docs sorted(scores.items(), keylambda x: x[1], reverseTrue) return sorted_docs[:top_k]注意alpha表示向量检索的权重。不同业务需要调参代码、编号、合同类内容关键词权重应更高语义问答、综述类内容向量权重应更高。4.7 重排解决“召回多但排不准”的问题向量检索和关键词检索都属于“粗召回”它们的优势是效率高但排序质量不够精细。重排Rerank的作用是把召回的 TopK 片段用更精确的模型重新打分排序。核心思路是使用 Cross-Encoder 模型把“问题”和“候选片段”拼在一起输入模型模型输出一个相关性分数。这种方式比双塔向量模型更精确但速度慢所以只对前 20 或前 50 条候选结果做重排。# 文件路径app/retrieval/rerank.py from sentence_transformers import CrossEncoder from app.config import RERANK_MODEL reranker CrossEncoder(RERANK_MODEL) def rerank(query: str, candidates: list[dict], top_n: int 5): if not candidates: return [] pairs [(query, candidate[text]) for candidate in candidates] scores reranker.predict(pairs) for candidate, score in zip(candidates, scores): candidate[rerank_score] float(score) ranked sorted(candidates, keylambda x: x[rerank_score], reverseTrue) return ranked[:top_n]重排的效果提升非常明显。以相同召回集合为前提使用 bge-reranker 类模型重排后正确答案排进前三的概率通常比向量打分排序高 10 到 20 个百分点。不过具体数字与数据集有关这里只强调这一环节的必要性。4.8 构造 Prompt 与生成重排后的片段要作为“参考资料”注入到 Prompt 中。Prompt 设计要满足三个要求说明资料的内容边界。禁止模型使用资料之外的知识。要求模型在信息不足时明确回答“不知道”。# 文件路径app/generation/llm.py from openai import OpenAI from app.config import LLM_BASE_URL, LLM_API_KEY, LLM_MODEL client OpenAI(base_urlLLM_BASE_URL, api_keyLLM_API_KEY) SYSTEM_PROMPT 你是一个企业知识助手。你的回答必须严格基于提供的参考资料不得使用参考资料之外的信息。 如果参考资料无法回答问题请回答“根据现有资料无法回答该问题”。 回答时请用编号标注引用来源例如[1][2]。 def build_prompt(query: str, contexts: list[dict]) - str: context_text \n\n.join( [ f[{i 1}] {item[text]}\n来源{item.get(source, 未知)} for i, item in enumerate(contexts) ] ) return f用户问题{query}\n\n参考资料\n{context_text} def generate_answer(query: str, contexts: list[dict]): prompt build_prompt(query, contexts) response client.chat.completions.create( modelLLM_MODEL, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: prompt}, ], temperature0.2, max_tokens1024, ) return response.choices[0].message.contenttemperature设置为 0.2 左右可以显著降低模型随意发挥的概率。如果业务要求严格还可以设为 0。4.9 完整 API 服务将上述模块串成 FastAPI 接口。# 文件路径app/main.py from fastapi import FastAPI from pydantic import BaseModel from app.retrieval.hybrid_search import hybrid_search from app.retrieval.rerank import rerank from app.generation.llm import generate_answer app FastAPI(titleEnterprise RAG API) class QueryRequest(BaseModel): query: str top_k: int 20 rerank_top_n: int 5 class QueryResponse(BaseModel): answer: str sources: list[dict] app.post(/api/query, response_modelQueryResponse) def query(request: QueryRequest): # 1. 混合检索 candidates hybrid_search(request.query, request.top_k) # 2. 重排 reranked rerank(request.query, candidates, request.rerank_top_n) # 3. 生成 answer generate_answer(request.query, reranked) return QueryResponse( answeranswer, sources[ { text: item[text][:200], source: item.get(source, ), score: item.get(rerank_score, 0), } for item in reranked ], ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000请求示例curl -X POST http://localhost:8000/api/query \ -H Content-Type: application/json \ -d {query: 公司请休假制度有哪些规定}4.10 运行结果说明预期返回结果类似{ answer: 根据制度文件[1]员工累计工作满1年不满10年的年休假5天已满10年不满20年的年休假10天。, sources: [ { text: 员工累计工作已满1年不满10年的年休假5天……, source: data/员工考勤管理制度.pdf, score: 8.23 } ] }到这里一个最小可运行的 RAG 知识库已经完成。接下来要考虑的是如何把这条链路做成企业级系统。5. RAG 知识库质量的度量指标很多项目上线后说不清“效果好不好”这就是缺少指标定义。建议从以下维度建立评估体系。5.1 检索质量指标指标含义说明RecallK前 K 条结果中包含正确答案的比例召回是否足够全面MRR正确答案在结果中排位的倒数均值排序位置是否靠前NDCG考虑相关性强弱的排序质量可以用于多级相关性标注命中率检索结果中至少有一条被最终回答引用占的比例更贴近实际问答在没有标注数据的情况下也可以用“回答引用率”来粗测把线上日志中的 query 与最终引用来源做统计看有多少请求使用了检索结果。5.2 生成质量指标指标含义评估方式忠实度答案是否严格基于参考资料人工抽样 LLM 评估答案相关度是否回答了用户问题人工评分幻觉率答案中出现资料中没有的信息的比例人工抽检引用准确率引用来源是否真实对应答案内容人工抽检线上建议建立“用户反馈”机制答案下方放“有帮助 / 无帮助”按钮无帮助的样本定期进入标注池反哺检索与重排优化。6. 工程化落地从 Demo 到生产环境Demo 能跑和项目能上线之间隔着很多工程问题。6.1 文档更新与增量索引企业文档每天都在变。文档更新要分成全量索引和增量索引全量索引每天或每周固定时间把所有文档重新处理一遍。增量索引监控文件系统、OA 系统或数据库表变化新增和变更的文档实时处理。增量索引要注意“删除”逻辑。文档变更后旧向量仍然存在会导致检索到过期内容所以先删除旧 ID再写入新向量。6.2 并发与性能在线链路的瓶颈通常不在大模型而在检索和重排。向量检索使用近似最近邻索引控制查询超时。重排使用 GPU 或更强的 CPU 实例。对高频重复问题增加缓存缓存 key 可以用 query 的向量距离判断相似度。大模型生成环节设置合理的max_tokens上限避免慢请求占满连接池。6.3 安全与权限企业知识库的内容往往涉及权限分级。如果没有权限控制用户可能会通过知识库问答间接读取到无权访问的文档。工程上至少要做到文档级权限向量库的 metadata 中记录文档权限组。检索阶段过滤查询时携带用户身份检索时按权限组过滤。生成后校验答案中的引用来源必须是当前用户可见的文档。访问审计记录每次问答的 user、query、引用的文档 ID、时间戳。6.4 可观测性线上系统不能只靠“感觉”。需要监控检索耗时、重排耗时、生成耗时。检索召回数量、重排后 Top1 分数。大模型 API 的 token 消耗。用户反馈数据。建议将每个环节的指标写入日志系统配合 Prometheus 和 Grafana 做大盘展示。7. 常见问题与排查思路7.1 检索结果不相关问题现象常见原因解决思路回答明显是编的检索内容没有进入 Prompt检查检索结果是否为空检查 Prompt 是否拼接问题换个说法就检索不到语义向量模型效果不足换更大的 Embedding 模型增加混合检索专有名词总是匹配不准切分把专有名词截断了切分时加入自定义词典做分词后再切检索到的都是重复内容多份文档内容高度相似检索后去重按文档来源分散召回排查顺序建议先看召回结果再看重排结果。如果召回结果里没有正确答案问题在索引链路如果召回有但重排后没了问题在重排。7.2 向量库写入失败错误现象常见原因解决思路Dimension mismatchEmbedding 模型输出维度变了删除旧集合重新建索引ID already exists重复写入相同 ID写入前先 delete 对应 doc_id连接超时向量数据库负载过高增加超时时间检查服务状态7.3 大模型回答幻觉问题现象常见原因解决思路回答内容与资料不一致Prompt 没有严格限制强化 System Prompt要求只基于资料回答资料能回答但模型说不知道检索片段过少或上下文不全提高 top_k 或调大 chunk_size回答添油加醋temperature 设置过高将 temperature 调低到 0 或 0.27.4 重排速度太慢Cross-Encoder 重排的复杂度是 O(N) 次前向推理。当 N 50 时每个请求需要 50 次模型推理。优化方向先用向量和关键词粗召回只保留前 20 条进重排。重排服务单独部署独立扩缩容。如果重排模型过大可以尝试量化版本或更小的 distill 版本。8. 最佳实践与工程建议8.1 从“检索正确”而不是“生成正确”入手RAG 的上限由检索决定。大模型再强喂进去的资料不对答案也不可能对。所以效果优化应该把重心放在先人工检查召回结果。再做重排优化。最后才调整 Prompt。很多团队一上来就调 Prompt结果发现调来调去回答还是不准确原因就是检索环节根本没召回相关内容。8.2 建立一个评测集知识库项目上线前必须准备一个评测集。评测集不需要很大50 到 100 条真实问题即可但要有标准答案和来源文档。每次修改切分策略、Embedding 模型、重排模型、检索参数后都跑一遍评测集对比指标变化。这样可以避免“改一个参数好了这个问题坏了另一个问题”的情况。8.3 保留原文溯源向量库中除了存向量一定要存原文 chunk、文档路径、页码、章节标题。这样生成答案时可以附带引用来源方便用户核对也方便事后审计。8.4 用日志反哺优化线上日志是免费的金矿。定期分析用户点了“有帮助”的 query 和引用的文档。用户点了“无帮助”的 query。多轮对话中重复提问的内容。这些数据可以帮助识别知识库中的盲区比如某类问题永远检索不到对应文档那么很可能知识库里根本没有相关资料或者有但切分和索引方式不合理。9. 总结与学习路线RAG 知识库并不是一个大模型 API 就能解决的问题而是一条包含文档处理、文本切分、向量化、混合检索、重排、Prompt 工程和工程化部署的完整链路。本文把全链路拆解成了离线索引和在线推理两条流程并动手实现了一个可运行的 FastAPI 服务。你可以在这个基础上继续学习Embedding 模型微调用领域数据微调让向量检索更懂业务语言。** Query 改写**引入大模型做查询改写解决口语化表达和指代问题。Agent 化 RAG让模型在检索不到答案时主动查询数据库、调用 API 或追问用户。GraphRAG / Ontology RAG引入知识图谱结构处理多跳问答和复杂关系推理。分布式向量数据库在 Milvus、Elasticsearch 上做大规模索引和部署。如果你正在做企业级项目建议依次关注评测集建设、权限控制、日志监控和文档增量更新这四件事。它们不一定能让 Demo 跑得更快但决定了系统能不能真正投入生产。动手搭一个最小版本跑通全链路然后把你不满意的环节逐一替换成更优方案——这是学习 RAG 最直接的方式。 企业级RAG知识库全链路解析：从架构原理到工程化落地
当前笔记围绕企业级 RAG检索增强生成知识库展开从架构原理到检索、召回、重排、生成与工程化落地完整走一遍。适合正在做大模型应用、知识库问答、私有化部署的开发者参考也适合想系统理解 RAG 全链路的新手。1. RAG 是什么为什么企业知识库需要它1.1 先理解大模型的两大短板大模型虽然能聊天、能写代码、能做摘要但在企业知识问答场景里有两个明显问题第一知识有时效性。大模型的训练数据有截止日期之后发生的事情它不知道。公司内部新发布的制度、刚上线的产品规格、最新的法务条款模型训练时根本没见过。第二生成内容不可控。模型回答问题时会把训练时学到的“常识”和“编造内容”混在一起出现一本正经的胡说八道。这个问题在业内叫“幻觉”。对于企业知识库幻觉是致命的法务、财务、医疗、制造等场景尤其不能接受。1.2 RAG 的核心思路先检索再生成RAGRetrieval-Augmented Generation检索增强生成的思路很直接让大模型在回答前先“查资料”。整个链路可以概括为用户输入一个问题。系统从企业知识库中检索与问题最相关的内容片段。将这些内容片段拼接到 Prompt 中一起交给大模型。大模型基于“检索到的内容 用户问题”生成答案。这意味着模型不需要“记住”企业内部知识它只需要“读懂”检索到的资料。知识的新增、更新、删除都发生在知识库侧模型侧不需要重训练维护成本大幅降低。1.3 RAG 与模型微调的区别很多人会把 RAG 和模型微调Fine-tuning放在一起比较。两者的定位完全不同维度RAG模型微调知识更新改知识库即可成本低需要重新训练周期长硬件要求普通服务器可运行主要消耗检索资源需要 GPU 训练成本高回答准确性基于检索证据可控性较好依赖训练数据质量适用场景知识频繁变化、对正确性要求高模型能力增强、固定格式输出、风格调整幻觉风险可通过检索内容约束仍然存在实际项目中两者可以结合。比如先用 RAG 解决知识来源问题再用微调让模型学会特定的输出格式或行业表达习惯。但大多数知识库项目RAG 是主链路。2. RAG 全链路架构拆解RAG 不是“单个模型”或者“单个 API”而是一条完整的数据流水线。只有把整条链路拆清楚才能定位问题、优化效果。2.1 离线索引链路离线链路负责把企业文档变成可检索的结构化数据。文档接入 → 格式解析 → 文档清洗 → 文本切分 → 向量化 → 写入向量数据库 ↓ 建立关键词索引核心环节包括文档接入支持 PDF、Word、Markdown、TXT、HTML、Excel 等格式。格式解析PDF 需要抽取文本和表格Word 需要保留标题层级扫描件需要 OCR。文档清洗去掉页眉页脚、水印、无效换行、乱码字符。文本切分Chunking这是全链路里最容易影响效果的一步。向量化Embedding将文本转换成向量。索引写入向量入库同时保留原文、元数据、文档来源。2.2 在线推理链路在线链路处理用户请求。用户提问 → 查询改写 → 混合检索 → 重排 → 构造 Prompt → 大模型生成 ↓ 生成引用来源每一步都有优化空间查询改写把模糊问题改写成适合检索的表达比如把“上次说的那个方案”改写成具体名词。混合检索同时使用关键词检索和向量检索再合并结果。重排Rerank对检索结果做精细排序把最相关的片段排到前面。Prompt 构造在 Prompt 中限制模型只能使用检索内容作答。生成大模型输出答案并附上引用来源。2.3 完整流程图用户问题 ↓ [查询预处理] → 纠错、改写、意图识别 ↓ [混合检索] → BM25 关键词检索 → 向量检索KNN ↓ [结果融合] → 去重、分数归一化、取 TopN ↓ [重排] → Cross-Encoder / Rerank 模型 ↓ [Prompt 构造] → 注入检索片段、设定回答边界 ↓ [大模型生成] → 输出答案 引用来源 ↓ [后处理] → 格式校验、敏感词过滤、记录日志3. 环境准备与版本选型RAG 项目涉及的组件比较多。为了避免“本地能跑生产就崩”的问题需要一开始就做好技术选型。3.1 核心组件清单组件作用选型建议向量数据库存储向量、执行相似度检索Milvus、Qdrant、Elasticsearch 8.x、Chroma搜索引擎关键词检索Elasticsearch、OpenSearchEmbedding 模型文本转向量BGE、M3E、OpenAI Embedding API重排模型精细化排序bge-reranker、cross-encoder 类模型大模型最终答案生成开源模型或商业 API应用框架编排链路LangChain、LlamaIndex 或自研 Python 服务任务队列异步处理文档Celery、Arq、Redis Stream缓存降低重复请求延迟Redis3.2 版本注意事项技术版本更新很快所以这里不给写死的版本号大家按自己项目的实际情况选择。需要特别注意几个原则向量数据库版本与 SDK 版本必须匹配。客户端 SDK 大版本升级后连接方式、索引参数都可能有变化。Embedding 模型维度必须与索引配置一致。如果先建了 768 维的集合换模型后向量变成 1024 维就会写入失败。Python 版本建议使用 3.9。新版 LangChain 和向量数据库 SDK 对旧版本支持越来越差。生产环境避免使用 SQLite 类向量方案。Chroma 的原生持久化适合 Demo高并发场景需要独立的向量数据库服务。3.3 示例项目结构rag-project/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置文件 │ ├── ingestion/ │ │ ├── loader.py # 文档加载 │ │ ├── splitter.py # 文本切分 │ │ └── vectorize.py # 向量化与入库 │ ├── retrieval/ │ │ ├── keyword_search.py # 关键词检索 │ │ ├── vector_search.py # 向量检索 │ │ ├── hybrid_search.py # 混合检索与融合 │ │ └── rerank.py # 重排模块 │ ├── generation/ │ │ └── llm.py # Prompt 构造与生成 │ └── schemas.py # 请求响应模型 ├── data/ # 原始文档 ├── tests/ # 测试用例 └── requirements.txt如果是 Java 技术栈可以类比拆包用 Spring Boot 作为 Web 层用 langchain4j 做链路编排用 Milvus Java SDK 做向量检索。4. 完整实战从零搭建一个 RAG 知识库下面用一个可运行的 Python 示例把 RAG 全链路走一遍。这里使用 FastAPI Chroma HuggingFace Embedding 和 Rerank 模型重点演示架构思路。生产环境可以替换成 Milvus 或 Elasticsearch。4.1 安装依赖pip install fastapi uvicorn chromadb sentence-transformers pip install pypdf python-docx markdown pip install openai # 或者对接本地模型服务 pip install bm25s # 轻量 BM25 工具一些库可能在不同版本中 API 有变化安装时如果遇到兼容问题可以把版本固定下来。4.2 定义配置文件# 文件路径app/config.py import os EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, BAAI/bge-large-zh-v1.5) RERANK_MODEL os.getenv(RERANK_MODEL, BAAI/bge-reranker-v2-m3) LLM_BASE_URL os.getenv(LLM_BASE_URL, http://localhost:8000/v1) LLM_API_KEY os.getenv(LLM_API_KEY, EMPTY) LLM_MODEL os.getenv(LLM_MODEL, qwen2.5-7b-instruct) CHUNK_SIZE 512 CHUNK_OVERLAP 64 TOP_K 20 # 检索阶段取回的数量 RERANK_TOP_N 5 # 重排后保留的数量4.3 文档加载与清洗不同格式的文档需要不同的解析方式。这里先写一个统一的加载函数。# 文件路径app/ingestion/loader.py from pathlib import Path def load_document(file_path: str) - str: path Path(file_path) suffix path.suffix.lower() if suffix .txt: return path.read_text(encodingutf-8, errorsignore) elif suffix .md: return path.read_text(encodingutf-8, errorsignore) elif suffix .pdf: from pypdf import PdfReader reader PdfReader(str(path)) return \n.join(page.extract_text() or for page in reader.pages) elif suffix .docx: from docx import Document doc Document(str(path)) return \n.join(p.text for p in doc.paragraphs) else: raise ValueError(f暂不支持的文件格式{suffix})文档清洗时要注意去除页面页脚中的重复内容。将全角字符统一转半角。删除连续空行。保留 Markdown 标题结构方便后续切分。4.4 文本切分切分是 RAG 链路中最“细节”的环节。切太大检索精度会下降切太小片段缺乏上下文语义不完整。推荐使用“递归字符切分 固定窗口重叠”的思路# 文件路径app/ingestion/splitter.py from typing import List def recursive_split(text: str, chunk_size: int 512, chunk_overlap: int 64) - List[str]: separators [\n\n, \n, 。, , , , , , ] chunks [] current for para in text.split(\n): if len(current) len(para) chunk_size: current para \n else: if current: chunks.append(current.strip()) current para \n if current.strip(): chunks.append(current.strip()) # 对过长的片段做二次切分并保留重叠 final_chunks [] for chunk in chunks: if len(chunk) chunk_size: final_chunks.append(chunk) continue start 0 while start len(chunk): end start chunk_size final_chunks.append(chunk[start:end]) if end len(chunk): break start end - chunk_overlap return final_chunks这个实现保证以段落为基本单位优先合并。过长段落按固定长度切分。相邻 chunk 之间保留重叠避免重要信息被截断。实际项目里还可以结合标题层级切分让同一章节的内容尽量在同一个 chunk 中。4.5 向量化与入库使用 sentence-transformers 加载 Embedding 模型将文本转为向量后写入 Chroma。# 文件路径app/ingestion/vectorize.py import chromadb from sentence_transformers import SentenceTransformer from app.config import EMBEDDING_MODEL, CHUNK_SIZE, CHUNK_OVERLAP from app.ingestion.loader import load_document from app.ingestion.splitter import recursive_split client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(nameenterprise_rag) embedder SentenceTransformer(EMBEDDING_MODEL) def index_document(file_path: str, doc_id: str, metadata: dict None): text load_document(file_path) chunks recursive_split(text, CHUNK_SIZE, CHUNK_OVERLAP) embeddings embedder.encode(chunks).tolist() ids [f{doc_id}_{i} for i in range(len(chunks))] metadatas [ { doc_id: doc_id, chunk_index: i, source: file_path, **(metadata or {}), } for i in range(len(chunks)) ] collection.add( idsids, documentschunks, embeddingsembeddings, metadatasmetadatas, ) return len(chunks)注意Chromat 的add方法在重复执行同一文档时会报 ID 冲突所以更新文档时可以先collection.delete(where{doc_id: doc_id})再重新写入。4.6 混合检索向量检索 关键词检索只用向量检索的问题在于一些精确的专有名词、编号、公式、报错代码向量模型理解得并不好。比如搜索“BGP 邻居状态 Establised”关键词匹配能精准命中向量检索可能召回到语义相近但不够准确的内容。所以企业级 RAG 通常使用混合检索。向量检索代码# 文件路径app/retrieval/vector_search.py from sentence_transformers import SentenceTransformer from app.config import EMBEDDING_MODEL embedder SentenceTransformer(EMBEDDING_MODEL) def vector_search(query: str, top_k: int 20): query_embedding embedder.encode([query]).tolist()[0] results collection.query( query_embeddings[query_embedding], n_resultstop_k, include[documents, metadatas, distances], ) return results关键词检索使用 BM25 算法。这是一个经典的概率检索模型对专有名词匹配非常有效。# 文件路径app/retrieval/keyword_search.py from bm25s import BM25 from app.config import TOP_K # 全局维护一个语料库生产环境建议用 Elasticsearch bm25_index None def build_bm25_index(corpus: list[str], doc_ids: list[str]): global bm25_index tokenized [doc.split() for doc in corpus] bm25_index BM25() bm25_index.index(tokenized, doc_idsdoc_ids) def keyword_search(query: str, top_k: int TOP_K): if bm25_index is None: return [] tokenized_query query.split() results bm25_index.retrieve(tokenized_query, ktop_k) return results混合检索的融合策略有很多种。最简单的是 “Score 加权求和”# 文件路径app/retrieval/hybrid_search.py def hybrid_search(query: str, top_k: int TOP_K, alpha: float 0.5): vector_results vector_search(query, top_k) keyword_results keyword_search(query, top_k) # 统一分数区间后加权 scores {} docs {} for i, doc in enumerate(vector_results[documents][0]): score 1 / (1 vector_results[distances][0][i]) docs[doc] vector_results[metadatas][0][i] scores[doc] scores.get(doc, 0) alpha * score for doc_id, score in keyword_results.items(): if score 0: continue doc_text doc_id_to_text.get(doc_id, ) if not doc_text: continue scores[doc_text] scores.get(doc_text, 0) (1 - alpha) * score docs.setdefault(doc_text, {doc_id: doc_id}) sorted_docs sorted(scores.items(), keylambda x: x[1], reverseTrue) return sorted_docs[:top_k]注意alpha表示向量检索的权重。不同业务需要调参代码、编号、合同类内容关键词权重应更高语义问答、综述类内容向量权重应更高。4.7 重排解决“召回多但排不准”的问题向量检索和关键词检索都属于“粗召回”它们的优势是效率高但排序质量不够精细。重排Rerank的作用是把召回的 TopK 片段用更精确的模型重新打分排序。核心思路是使用 Cross-Encoder 模型把“问题”和“候选片段”拼在一起输入模型模型输出一个相关性分数。这种方式比双塔向量模型更精确但速度慢所以只对前 20 或前 50 条候选结果做重排。# 文件路径app/retrieval/rerank.py from sentence_transformers import CrossEncoder from app.config import RERANK_MODEL reranker CrossEncoder(RERANK_MODEL) def rerank(query: str, candidates: list[dict], top_n: int 5): if not candidates: return [] pairs [(query, candidate[text]) for candidate in candidates] scores reranker.predict(pairs) for candidate, score in zip(candidates, scores): candidate[rerank_score] float(score) ranked sorted(candidates, keylambda x: x[rerank_score], reverseTrue) return ranked[:top_n]重排的效果提升非常明显。以相同召回集合为前提使用 bge-reranker 类模型重排后正确答案排进前三的概率通常比向量打分排序高 10 到 20 个百分点。不过具体数字与数据集有关这里只强调这一环节的必要性。4.8 构造 Prompt 与生成重排后的片段要作为“参考资料”注入到 Prompt 中。Prompt 设计要满足三个要求说明资料的内容边界。禁止模型使用资料之外的知识。要求模型在信息不足时明确回答“不知道”。# 文件路径app/generation/llm.py from openai import OpenAI from app.config import LLM_BASE_URL, LLM_API_KEY, LLM_MODEL client OpenAI(base_urlLLM_BASE_URL, api_keyLLM_API_KEY) SYSTEM_PROMPT 你是一个企业知识助手。你的回答必须严格基于提供的参考资料不得使用参考资料之外的信息。 如果参考资料无法回答问题请回答“根据现有资料无法回答该问题”。 回答时请用编号标注引用来源例如[1][2]。 def build_prompt(query: str, contexts: list[dict]) - str: context_text \n\n.join( [ f[{i 1}] {item[text]}\n来源{item.get(source, 未知)} for i, item in enumerate(contexts) ] ) return f用户问题{query}\n\n参考资料\n{context_text} def generate_answer(query: str, contexts: list[dict]): prompt build_prompt(query, contexts) response client.chat.completions.create( modelLLM_MODEL, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: prompt}, ], temperature0.2, max_tokens1024, ) return response.choices[0].message.contenttemperature设置为 0.2 左右可以显著降低模型随意发挥的概率。如果业务要求严格还可以设为 0。4.9 完整 API 服务将上述模块串成 FastAPI 接口。# 文件路径app/main.py from fastapi import FastAPI from pydantic import BaseModel from app.retrieval.hybrid_search import hybrid_search from app.retrieval.rerank import rerank from app.generation.llm import generate_answer app FastAPI(titleEnterprise RAG API) class QueryRequest(BaseModel): query: str top_k: int 20 rerank_top_n: int 5 class QueryResponse(BaseModel): answer: str sources: list[dict] app.post(/api/query, response_modelQueryResponse) def query(request: QueryRequest): # 1. 混合检索 candidates hybrid_search(request.query, request.top_k) # 2. 重排 reranked rerank(request.query, candidates, request.rerank_top_n) # 3. 生成 answer generate_answer(request.query, reranked) return QueryResponse( answeranswer, sources[ { text: item[text][:200], source: item.get(source, ), score: item.get(rerank_score, 0), } for item in reranked ], ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动服务uvicorn app.main:app --host 0.0.0.0 --port 8000请求示例curl -X POST http://localhost:8000/api/query \ -H Content-Type: application/json \ -d {query: 公司请休假制度有哪些规定}4.10 运行结果说明预期返回结果类似{ answer: 根据制度文件[1]员工累计工作满1年不满10年的年休假5天已满10年不满20年的年休假10天。, sources: [ { text: 员工累计工作已满1年不满10年的年休假5天……, source: data/员工考勤管理制度.pdf, score: 8.23 } ] }到这里一个最小可运行的 RAG 知识库已经完成。接下来要考虑的是如何把这条链路做成企业级系统。5. RAG 知识库质量的度量指标很多项目上线后说不清“效果好不好”这就是缺少指标定义。建议从以下维度建立评估体系。5.1 检索质量指标指标含义说明RecallK前 K 条结果中包含正确答案的比例召回是否足够全面MRR正确答案在结果中排位的倒数均值排序位置是否靠前NDCG考虑相关性强弱的排序质量可以用于多级相关性标注命中率检索结果中至少有一条被最终回答引用占的比例更贴近实际问答在没有标注数据的情况下也可以用“回答引用率”来粗测把线上日志中的 query 与最终引用来源做统计看有多少请求使用了检索结果。5.2 生成质量指标指标含义评估方式忠实度答案是否严格基于参考资料人工抽样 LLM 评估答案相关度是否回答了用户问题人工评分幻觉率答案中出现资料中没有的信息的比例人工抽检引用准确率引用来源是否真实对应答案内容人工抽检线上建议建立“用户反馈”机制答案下方放“有帮助 / 无帮助”按钮无帮助的样本定期进入标注池反哺检索与重排优化。6. 工程化落地从 Demo 到生产环境Demo 能跑和项目能上线之间隔着很多工程问题。6.1 文档更新与增量索引企业文档每天都在变。文档更新要分成全量索引和增量索引全量索引每天或每周固定时间把所有文档重新处理一遍。增量索引监控文件系统、OA 系统或数据库表变化新增和变更的文档实时处理。增量索引要注意“删除”逻辑。文档变更后旧向量仍然存在会导致检索到过期内容所以先删除旧 ID再写入新向量。6.2 并发与性能在线链路的瓶颈通常不在大模型而在检索和重排。向量检索使用近似最近邻索引控制查询超时。重排使用 GPU 或更强的 CPU 实例。对高频重复问题增加缓存缓存 key 可以用 query 的向量距离判断相似度。大模型生成环节设置合理的max_tokens上限避免慢请求占满连接池。6.3 安全与权限企业知识库的内容往往涉及权限分级。如果没有权限控制用户可能会通过知识库问答间接读取到无权访问的文档。工程上至少要做到文档级权限向量库的 metadata 中记录文档权限组。检索阶段过滤查询时携带用户身份检索时按权限组过滤。生成后校验答案中的引用来源必须是当前用户可见的文档。访问审计记录每次问答的 user、query、引用的文档 ID、时间戳。6.4 可观测性线上系统不能只靠“感觉”。需要监控检索耗时、重排耗时、生成耗时。检索召回数量、重排后 Top1 分数。大模型 API 的 token 消耗。用户反馈数据。建议将每个环节的指标写入日志系统配合 Prometheus 和 Grafana 做大盘展示。7. 常见问题与排查思路7.1 检索结果不相关问题现象常见原因解决思路回答明显是编的检索内容没有进入 Prompt检查检索结果是否为空检查 Prompt 是否拼接问题换个说法就检索不到语义向量模型效果不足换更大的 Embedding 模型增加混合检索专有名词总是匹配不准切分把专有名词截断了切分时加入自定义词典做分词后再切检索到的都是重复内容多份文档内容高度相似检索后去重按文档来源分散召回排查顺序建议先看召回结果再看重排结果。如果召回结果里没有正确答案问题在索引链路如果召回有但重排后没了问题在重排。7.2 向量库写入失败错误现象常见原因解决思路Dimension mismatchEmbedding 模型输出维度变了删除旧集合重新建索引ID already exists重复写入相同 ID写入前先 delete 对应 doc_id连接超时向量数据库负载过高增加超时时间检查服务状态7.3 大模型回答幻觉问题现象常见原因解决思路回答内容与资料不一致Prompt 没有严格限制强化 System Prompt要求只基于资料回答资料能回答但模型说不知道检索片段过少或上下文不全提高 top_k 或调大 chunk_size回答添油加醋temperature 设置过高将 temperature 调低到 0 或 0.27.4 重排速度太慢Cross-Encoder 重排的复杂度是 O(N) 次前向推理。当 N 50 时每个请求需要 50 次模型推理。优化方向先用向量和关键词粗召回只保留前 20 条进重排。重排服务单独部署独立扩缩容。如果重排模型过大可以尝试量化版本或更小的 distill 版本。8. 最佳实践与工程建议8.1 从“检索正确”而不是“生成正确”入手RAG 的上限由检索决定。大模型再强喂进去的资料不对答案也不可能对。所以效果优化应该把重心放在先人工检查召回结果。再做重排优化。最后才调整 Prompt。很多团队一上来就调 Prompt结果发现调来调去回答还是不准确原因就是检索环节根本没召回相关内容。8.2 建立一个评测集知识库项目上线前必须准备一个评测集。评测集不需要很大50 到 100 条真实问题即可但要有标准答案和来源文档。每次修改切分策略、Embedding 模型、重排模型、检索参数后都跑一遍评测集对比指标变化。这样可以避免“改一个参数好了这个问题坏了另一个问题”的情况。8.3 保留原文溯源向量库中除了存向量一定要存原文 chunk、文档路径、页码、章节标题。这样生成答案时可以附带引用来源方便用户核对也方便事后审计。8.4 用日志反哺优化线上日志是免费的金矿。定期分析用户点了“有帮助”的 query 和引用的文档。用户点了“无帮助”的 query。多轮对话中重复提问的内容。这些数据可以帮助识别知识库中的盲区比如某类问题永远检索不到对应文档那么很可能知识库里根本没有相关资料或者有但切分和索引方式不合理。9. 总结与学习路线RAG 知识库并不是一个大模型 API 就能解决的问题而是一条包含文档处理、文本切分、向量化、混合检索、重排、Prompt 工程和工程化部署的完整链路。本文把全链路拆解成了离线索引和在线推理两条流程并动手实现了一个可运行的 FastAPI 服务。你可以在这个基础上继续学习Embedding 模型微调用领域数据微调让向量检索更懂业务语言。** Query 改写**引入大模型做查询改写解决口语化表达和指代问题。Agent 化 RAG让模型在检索不到答案时主动查询数据库、调用 API 或追问用户。GraphRAG / Ontology RAG引入知识图谱结构处理多跳问答和复杂关系推理。分布式向量数据库在 Milvus、Elasticsearch 上做大规模索引和部署。如果你正在做企业级项目建议依次关注评测集建设、权限控制、日志监控和文档增量更新这四件事。它们不一定能让 Demo 跑得更快但决定了系统能不能真正投入生产。动手搭一个最小版本跑通全链路然后把你不满意的环节逐一替换成更优方案——这是学习 RAG 最直接的方式。