前阵子在给部门搭企业级知识库问答系统时踩了不少 RAG 的坑。从文档解析乱码、切块策略选错导致检索结果稀碎到向量召回准确率低、引用溯源无法落地几乎把 RAG 从入门到放弃的弯路都走了一遍。市面上讲 RAG 概念的资料很多但真正能照着从零搭出一套可运行、可优化、接近生产环境的知识库问答系统的完整教程却很少。这篇文章我会基于实际项目落地经验从 RAG 基础原理讲起带你完整体验一遍企业级知识库搭建全流程环境准备、文档加载与切块、向量化与检索、重排序、LLM 生成、FastAPI 接口封装以及生产环境必须注意的引用溯源、权限、评估与排查方案。整个教程以代码为主线每个环节都会解释“为什么这么做”保证零基础的同学能跟着搭起来有经验的开发者也能直接复用工程思路。1. RAG 是什么为什么要用 RAG1.1 大模型的知识瓶颈先从一个非常直观的问题说起你直接问 GPT 类大模型“我们公司内部上个月的销售数据是多少”它大概率答不上来。原因很简单大模型的知识截止于训练数据企业内部文档、数据库、实时资讯并不在它的记忆范围内。你可以通过微调让模型记住新知识但每次业务变动都要重新训练成本高、周期长而且无法覆盖不断更新的私有数据。RAGRetrieval-Augmented Generation检索增强生成就是用来解决这个问题的。它的思路很朴素不逼模型“记住”所有知识而是在每次回答前先从外部知识库中检索出与问题相关的文档片段把这些片段和问题一起组装成提示词交给大模型生成答案。模型不需要知道全部内容只需要基于给定的材料进行归纳和回答。1.2 RAG 的核心流程一个标准的 RAG 系统本质上由两条链路组成离线索引链路文档加载 - 文本清洗 - 分块Chunking- 向量化Embedding- 写入向量数据库。在线问答链路用户提问 - 问题向量化 - 向量相似度检索 - 可选重排序 - 组装提示词 - LLM 生成答案 - 返回答案与引用来源。用大白话讲离线阶段是“把书拆开、编号、做成索引卡片”在线阶段是“根据问题找卡片、把相关卡片递给学霸让他组织语言回答”。这样一来每次回答用的都是最新的文档内容既不需要重新训练模型也能在答案中附上引用来源方便人工核对。1.3 RAG 的经典应用场景RAG 在企业里的落地场景非常多常见的包括企业知识库问答把内部制度、技术文档、产品手册导入系统员工可以通过对话快速查询。政务与法律文书辅助从大量法规、判例、政策文件中检索相关条款辅助起草和审查。客服智能助手基于产品 FAQ、售后工单、操作手册回答用户问题。研发效能助手让大模型基于内部 API 文档、代码规范回答开发问题。金融研报分析从研报、公告、新闻中检索信息并生成分析摘要。1.4 为什么不能只靠向量检索这里需要澄清一个容易误解的概念RAG 不是“向量数据库的另一种叫法”向量检索只是 RAG 中间的一环。很多初学者把文档全部向量化存入 Milvus 或 Qdrant 后就认为搭建好了知识库结果问答效果很差。原因在于向量检索负责“找可能相关的片段”但片段是否真正回答了用户问题还需要重排序和 LLM 生成来把关。此外切块策略、向量模型质量、检索 TopK 设置、提示词模板都会直接影响最终效果。所以完整的企业级 RAG 项目必须把整条链路都做扎实而不是只停留在“能搜索”的阶段。2. 环境准备与项目结构2.1 技术选型说明RAG 的技术栈有非常多的选择比如 LangChain、LlamaIndex、Haystack或者直接用向量数据库 SDK 手写流程。本文选择一套轻量但完整的方案编程语言Python 3.10。Web 框架FastAPI用于提供问答 API。文档加载pypdf、python-docx分别处理 PDF 和 Word 文档。文本切块使用类似 LangChain 的 RecursiveCharacterTextSplitter 思路但为了减少框架黑盒我会在示例中手写一个递归字符切块器方便你理解切块原理。向量化模型sentence-transformers 加载开源 Embedding 模型例如 BAAI/bge-small-zh-v1.5。如果你有 OpenAI 或其他云端 Embedding 接口也可以替换。向量数据库Qdrant支持 Docker 部署轻量且 Python SDK 易用。大模型兼容 OpenAI API 格式的模型接口可以是 OpenAI、DeepSeek、通义千问、本地 vLLM 等只需要配置 Base URL 和 API Key。重排序可选使用 bge-reranker 或 Cohere Rerank用于提升召回精度。版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。如果你用的是更新的 Python 或库版本个别 API 可能略有变化建议以官方文档为准。2.2 安装依赖首先创建项目虚拟环境并安装基础依赖mkdir rag-knowledge-base cd rag-knowledge-base python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate然后安装 Python 依赖pip install fastapi uvicorn pypdf python-docx sentence-transformers qdrant-client openai python-dotenv如果你要用重排序模型还需要安装pip install FlagEmbedding2.3 启动 Qdrant 向量数据库Qdrant 有两种常见使用方式一种是 Docker 启动服务端另一种是嵌入式本地模式。生产环境推荐 Docker 方式docker run -d --name qdrant \ -p 6333:6333 \ -p 6334:6334 \ -v $(pwd)/qdrant_storage:/qdrant/storage \ qdrant/qdrant启动后Qdrant 默认在 6333 端口提供 HTTP API6334 是 gRPC 端口。你可以通过http://localhost:6333/dashboard访问 Web 控制台方便查看集合和向量数据。2.4 项目结构规划为了让代码清晰可维护建议按下面的结构组织工程rag-knowledge-base/ ├── app.py # FastAPI 入口 ├── config.py # 全局配置 ├── ingest.py # 离线索引脚本加载、切块、向量化、写入 ├── rag_pipeline.py # 在线问答链路 ├── text_splitter.py # 文本切块器 ├── docs/ # 原始文档目录 ├── data/chunks.json # 切块后的文本可按需保留 └── requirements.txt # 依赖清单下面我们从核心组件开始逐个实现。3. 核心原理与关键代码拆解3.1 文档加载与清洗企业知识库里的文档格式五花八门最常见的是 PDF、Word、Markdown、TXT。文档加载的核心目标是“把文件里的内容提取成纯文本”但这一步往往是坑最多的地方。PDF 常见的坑包括扫描件没有文字层、表格解析错乱、多栏排版文字顺序混乱。Word 文档相对好处理但也要注意提取出的内容可能包含页眉页脚。所以加载后必须做清洗比如去除多余空行、去掉页眉页脚特征文字、统一换行符。下面是一个简单的文档加载工具函数支持 PDF 和 TXT 文本提取# 文件路径ingest.py片段 from pypdf import PdfReader def load_pdf(file_path: str) - str: 提取 PDF 文件中的文本内容。 注意扫描版 PDF 需要 OCR 处理这里只处理带文字层的文档。 reader PdfReader(file_path) pages [] for page in reader.pages: text page.extract_text() if text: pages.append(text) return \n.join(pages) def load_txt(file_path: str) - str: 读取纯文本文件统一编码为 UTF-8。 with open(file_path, r, encodingutf-8) as f: return f.read()这里需要特别提醒如果你发现 PDF 提取出的文本顺序错乱或者缺少部分内容可能是 PDF 本身使用了复杂的排版结构。建议先人工确认几页内容再决定是否引入 OCR 流程。3.2 文本切块RAG 效果的关键文本切块Chunking是 RAG 中影响效果最明显的环节之一。切块太大向量包含的噪音信息多检索精度下降而且可能超出模型上下文窗口切块太小片段缺乏上下文语义召回结果容易被割裂同时增加向量数量。业界常见的切块策略主要有三类固定大小切块按固定的字符数切简单易实现但容易切断语义完整的句子或段落。递归字符切块按不同层级的分隔符段落 - 句号 - 逗号依次尝试切分尽量保证块内语义完整。这是目前最常用的方案。语义切块利用 Embedding 判断句子间语义差异在语义变化处切分效果更好但计算开销更大。下面实现一个基于递归思想的切块器输入是一段长文本输出是若干带序号的文本块# 文件路径text_splitter.py from typing import List class RecursiveTextSplitter: 递归字符切块器 优先按段落分隔符切如果切出的块仍然过大再按句号、逗号等次级分隔符切。 这样可以尽可能保证每个块内语义完整。 def __init__( self, chunk_size: int 500, chunk_overlap: int 80, separators: List[str] None, ): self.chunk_size chunk_size self.chunk_overlap chunk_overlap self.separators separators or [\n\n, \n, 。, , , , , , ] def split_text(self, text: str) - List[str]: 递归切分文本返回文本块列表。 text text.strip() if not text: return [] chunks self._split_with_separator(text, self.separators) # 合并过小的块避免向量库中碎片过多 merged self._merge_chunks(chunks) return merged def _split_with_separator(self, text: str, separators: List[str]) - List[str]: if not text: return [] if len(text) self.chunk_size: return [text] sep None for s in separators: if s in text: sep s break if sep is None: # 没有任何分隔符时硬切 return [ text[i: i self.chunk_size] for i in range(0, len(text), self.chunk_size) ] segments text.split(sep) result [] current for segment in segments: if not segment: continue if len(current) len(segment) len(sep) self.chunk_size: current current sep segment if current else segment else: if current: result.append(current) current segment if current: result.append(current) # 对仍然过大的块递归切分 final_result [] for item in result: if len(item) self.chunk_size: final_result.extend(self._split_with_separator(item, separators[1:])) else: final_result.append(item) return final_result def _merge_chunks(self, chunks: List[str]) - List[str]: 将过小的相邻块合并减少碎片数量。 if not chunks: return [] merged [] buffer for chunk in chunks: if len(buffer) len(chunk) self.chunk_size // 2: buffer chunk else: if buffer: merged.append(buffer) buffer chunk if buffer: merged.append(buffer) return merged切块参数怎么定经验上面向中文文档chunk_size 取 300~500 字符、chunk_overlap 取 50~100 是比较常见的起点。overlap 的作用是让相邻块之间有重叠信息避免一个完整知识点刚好被切断后前后两块都缺少上下文。具体数值需要根据你的文档类型和模型上下文窗口调整。3.3 Embedding 向量化与向量数据库写入文本切块完成后下一步需要把每个块转成向量。Embedding 模型的选择直接影响检索效果。中英文混合场景推荐 BGE 系列比如BAAI/bge-small-zh-v1.5它对中文支持较好且模型体积小适合 CPU 部署。如果你的文档全是英文可以考虑BAAI/bge-base-en-v1.5或 OpenAI 的text-embedding-3-small。向量化代码实现如下# 文件路径ingest.py片段 from sentence_transformers import SentenceTransformer # 加载 Embedding 模型首次运行会下载模型权重 embedding_model SentenceTransformer(BAAI/bge-small-zh-v1.5) def embed_texts(texts): 将文本列表转换为向量列表。 embeddings embedding_model.encode(texts, normalize_embeddingsTrue) return embeddings.tolist()normalize_embeddingsTrue会把向量归一化为单位向量这样后续用余弦相似度计算时内积等价于余弦相似度检索效果更稳定。接下来把文本块和向量写入 Qdrant。Qdrant 是专门为向量检索设计的数据库核心概念包括 Collection集合、Point数据点包含向量和 payload、Payload附加元数据例如来源文件名、页码、原始文本。# 文件路径ingest.py完整离线索引流程 import json import os from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct from text_splitter import RecursiveTextSplitter # 1. 连接 Qdrant qdrant QdrantClient(urlhttp://localhost:6333) COLLECTION_NAME enterprise_kb EMBEDDING_DIM 512 # 根据你的 Embedding 模型维度调整 # 2. 创建集合如果不存在 collections qdrant.get_collections().collections if not any(c.name COLLECTION_NAME for c in collections): qdrant.create_collection( collection_nameCOLLECTION_NAME, vectors_configVectorParams(sizeEMBEDDING_DIM, distanceDistance.COSINE), ) # 3. 加载文档并切块 def load_and_split(file_path: str, splitter: RecursiveTextSplitter): text load_pdf(file_path) if file_path.endswith(.pdf) else load_txt(file_path) chunks splitter.split_text(text) return chunks splitter RecursiveTextSplitter(chunk_size400, chunk_overlap80) all_points [] point_id 0 for filename in os.listdir(docs): file_path os.path.join(docs, filename) if not os.path.isfile(file_path): continue chunks load_and_split(file_path, splitter) embeddings embed_texts(chunks) for i, (chunk_text, vector) in enumerate(zip(chunks, embeddings)): all_points.append( PointStruct( idpoint_id, vectorvector, payload{ text: chunk_text, source: filename, chunk_index: i, }, ) ) point_id 1 # 4. 批量写入 Qdrant qdrant.upsert(collection_nameCOLLECTION_NAME, pointsall_points) print(f成功写入 {len(all_points)} 个文本块)运行这个脚本就会把 docs 目录下所有文档切块、向量化并写入 Qdrant。注意这里向量维度EMBEDDING_DIM 512要和实际模型输出维度一致bge-small-zh-v1.5的输出维度是 512 维。如果你换成了其他模型一定要同步调整这个值。3.4 检索与重排序离线索引完成后就可以实现在线检索了。基础检索逻辑比较简单用户问题向量化后在 Qdrant 中查询最相似的 TopK 个文本块。# 文件路径rag_pipeline.py片段 def search(query: str, top_k: int 5): 向量检索查找与用户问题最相似的文本块。 query_vector embedding_model.encode(query, normalize_embeddingsTrue).tolist() hits qdrant.search( collection_nameCOLLECTION_NAME, query_vectorquery_vector, limittop_k, ) results [] for hit in hits: results.append( { text: hit.payload[text], source: hit.payload[source], score: hit.score, } ) return results单纯靠向量检索有一个高频问题query 和文档块的语义相似但文档块并不是真正能回答问题的最佳片段。比如用户问“怎么申请年假”检索出的前三块都在介绍“年假天数规定”只有第五块真正写了“申请流程”。如果只取 TopK3正确答案就丢了。解决思路是引入重排序Rerank。先用向量召回较多数量的候选块比如召回 20 个再用重排序模型或 LLM 对候选块重新打分选出最相关的 5 个送入生成环节。重排序模型通常比向量检索更精准但计算成本更高所以采用“粗召回 精排序”的两阶段方案是更均衡的做法。下面是一个使用 bge-reranker 的示例# 文件路径rag_pipeline.py片段 from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-base, use_fp16True) def search_with_rerank(query: str, recall_k: int 20, top_k: int 5): 两阶段检索向量召回 重排序精排。 # 向量召回 query_vector embedding_model.encode(query, normalize_embeddingsTrue).tolist() hits qdrant.search( collection_nameCOLLECTION_NAME, query_vectorquery_vector, limitrecall_k, ) candidates [] for hit in hits: candidates.append( { text: hit.payload[text], source: hit.payload[source], score: hit.score, } ) # 使用重排序模型重新打分 pairs [(query, item[text]) for item in candidates] rerank_scores reranker.compute_score(pairs) for item, score in zip(candidates, rerank_scores): item[rerank_score] float(score) candidates.sort(keylambda x: x[rerank_score], reverseTrue) return candidates[:top_k]关于 bge-reranker 的补充说明compute_score返回的分值不一定在 0~1 之间不同模型的范围可能不同。这里我们不依赖绝对分值只看相对排序所以直接排序取 TopK 即可。3.5 提示词组装与 LLM 生成检索出相关片段之后最关键的一步是把这些片段“投喂”给大模型。提示词模板直接影响回答风格和可信度。一个企业级可用的提示词模板至少需要包含三个要素角色设定、检索材料和引用标注、输出约束。注意模板中必须明确要求模型“只能基于给定材料回答如果材料中没有相关内容请诚实说明”这是避免大模型幻觉的核心手段之一。# 文件路径rag_pipeline.py提示词模板 SYSTEM_PROMPT 你是一个专业的企业知识库问答助手。请基于用户问题和检索到的参考资料回答问题。 要求 1. 只能使用参考资料中的内容作答不得编造材料中不存在的信息。 2. 如果参考资料不足请明确回答“根据现有资料无法回答”。 3. 回答时尽量分点说明便于用户阅读。 4. 在回答最后用“参考来源”列出引用到的文档名称。 def build_prompt(query: str, search_results: list) - str: 根据检索结果组装提示词。 每个片段前加上 [1] [2] 这样的编号方便后续生成引用标注。 context_blocks [] for idx, item in enumerate(search_results, start1): source item.get(source, 未知来源) text item[text].strip() context_blocks.append(f[{idx}] 来源: {source}\n{text}) context \n\n.join(context_blocks) user_prompt f请基于以下参考资料回答问题。 参考资料 {context} 问题{query} 请给出答案并在句末使用[编号]标注引用来源。 return user_promptLLM 调用部分我统一使用 OpenAI 的 SDK 格式这样无论是 DeepSeek、通义千问、Moonshot 还是本地 vLLM 启动的模型服务都可以通过配置 Base URL 来接入不需要改代码逻辑。# 文件路径rag_pipeline.pyLLM 调用 from openai import OpenAI import config llm_client OpenAI( api_keyconfig.LLM_API_KEY, base_urlconfig.LLM_BASE_URL, ) def generate_answer(query: str, search_results: list) - dict: 调用大模型生成答案并返回答案、引用的来源列表。 prompt build_prompt(query, search_results) response llm_client.chat.completions.create( modelconfig.LLM_MODEL, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: prompt}, ], temperature0.2, max_tokens1024, ) answer response.choices[0].message.content # 按索引顺序去重提取引用来源 sources [] for item in search_results: source item.get(source, ) if source and source not in sources: sources.append(source) return { answer: answer, sources: sources, }这里temperature0.2是刻意调低的目的是减少模型自由发挥的空间让回答更贴合检索材料。如果生成结果仍然很发散可以考虑降到 0.1 甚至 0。3.6 引用溯源与诚实回答前面提到 RAG 的 Groundedness接地性也就是回答必须基于检索材料。这里再展开说一个企业场景特别看重的点引用溯源。企业内部使用知识库问答时答案必须能追根溯源。员工看到回答后需要能点击跳转到原始文档位置进行核实。实现方法其实不复杂提示词中要求模型在句子末尾标注[编号]我们在代码中将编号映射到对应的文档名和文本块。如果模型没有按格式标注至少也要返回检索命中的来源列表让前端可以展示“参考文档”区域。# 文件路径rag_pipeline.py引用溯源增强 def format_answer_with_citations(answer: str, search_results: list) - str: 简单后处理如果模型没有正确输出[编号]可以追加来源列表兜底。 if [ not in answer: answer \n\n参考来源 seen set() for item in search_results: src item.get(source, 未知来源) if src not in seen: seen.add(src) answer f\n- {src} return answer诚实回答指的则是当检索内容不足以回答问题时模型应该明确说“不知道”而不是强行编造。要做到这一点除了提示词约束还需要在工程上设置一个检索阈值。如果最高相似度分数过低直接返回“没有检索到足够相关信息”不再调用 LLM。这在企业场景中非常重要因为编造的答案一旦流入业务决策风险极大。配置示例# 文件路径config.py RETRIEVAL_MIN_SCORE 0.35 # 检索最低相似度阈值低于该值不触发生成4. 完整实战基于 FastAPI 搭建知识库问答服务在上一节中我们已经把检索和生成的核心函数写好了。现在把它串成一个完整的 FastAPI 服务并提供两个接口POST /ingest手动触发文档索引方便新增文档后增量更新。POST /query用户提问返回答案和引用来源。4.1 FastAPI 应用入口# 文件路径app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field import os import config import rag_pipeline import ingest app FastAPI(title企业级 RAG 知识库问答系统) class QueryRequest(BaseModel): question: str Field(..., max_length500, description用户问题) top_k: int Field(5, ge1, le10, description返回的检索片段数量) class QueryResponse(BaseModel): answer: str sources: list search_results: list class IngestResponse(BaseModel): message: str chunk_count: int app.post(/ingest, response_modelIngestResponse) def run_ingest(): 重新扫描 docs 目录并写入向量数据库。 生产环境建议改为增量更新并加权限控制。 try: chunk_count ingest.run() return IngestResponse(message索引完成, chunk_countchunk_count) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/query, response_modelQueryResponse) def query_kb(req: QueryRequest): 知识库问答接口。 流程向量检索 - 重排序 - 组装提示词 - LLM 生成。 if not req.question.strip(): raise HTTPException(status_code400, detail问题不能为空) # 判断是否启用重排序 use_rerank True if use_rerank: search_results rag_pipeline.search_with_rerank(req.question, recall_k20, top_kreq.top_k) else: search_results rag_pipeline.search(req.question, top_kreq.top_k) # 检索分数过低时不调用 LLM防止幻觉 if not search_results or search_results[0].get(score, 0) config.RETRIEVAL_MIN_SCORE: return QueryResponse( answer抱歉根据现有知识库内容没有检索到足够相关的信息。请尝试换个问法或补充相关文档。, sources[], search_results[], ) result rag_pipeline.generate_answer(req.question, search_results) return QueryResponse( answerresult[answer], sourcesresult[sources], search_resultssearch_results, )4.2 统一配置管理配置文件单独放一个config.py方便在不同环境开发、测试、生产之间切换# 文件路径config.py import os from dotenv import load_dotenv load_dotenv() # Qdrant 配置 QDRANT_URL os.getenv(QDRANT_URL, http://localhost:6333) COLLECTION_NAME os.getenv(COLLECTION_NAME, enterprise_kb) # Embedding 模型配置 EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, BAAI/bge-small-zh-v1.5) EMBEDDING_DIM int(os.getenv(EMBEDDING_DIM, 512)) # LLM 配置 LLM_API_KEY os.getenv(LLM_API_KEY, your-api-key) LLM_BASE_URL os.getenv(LLM_BASE_URL, https://api.openai.com/v1) LLM_MODEL os.getenv(LLM_MODEL, gpt-4o-mini) # 检索配置 RETRIEVAL_MIN_SCORE float(os.getenv(RETRIEVAL_MIN_SCORE, 0.35)) # 文本切块配置 CHUNK_SIZE int(os.getenv(CHUNK_SIZE, 400)) CHUNK_OVERLAP int(os.getenv(CHUNK_OVERLAP, 80))使用.env文件管理敏感配置时注意不要把密钥提交到 Git 仓库生产环境建议用配置中心或密钥管理服务。4.3 启动服务并验证效果启动 FastAPI 服务uvicorn app:app --host 0.0.0.0 --port 8000首次启动时会加载 Embedding 模型如果你的机器上没有提前下载它会自动从 Hugging Face 拉取模型权重耗时取决于网络环境。国内网络如果下载慢可以设置环境变量HF_ENDPOINThttps://hf-mirror.com切换为镜像源。启动成功后先调用/ingest对文档建立索引curl -X POST http://localhost:8000/ingest预期返回类似{ message: 索引完成, chunk_count: 156 }然后调用/query测试问答curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 员工年假怎么申请}返回示例{ answer: 根据员工手册员工年假申请需要提前三天在 OA 系统中提交申请经部门负责人审批后方可生效。[1]\n\n参考来源\n- 员工手册.pdf, sources: [员工手册.pdf], search_results: [ { text: 员工年假申请流程员工需提前三天在 OA 系统中提交年假申请……, source: 员工手册.pdf, score: 0.62 } ] }到这里一个最小可用的 RAG 知识库问答系统就跑通了。4.4 验证检索质量的小技巧当你测试时发现回答不理想不要急着调大模型先看看检索结果对不对。一个很实用的技巧是直接打印search_results的前几项问问自己“如果我是人光看这些片段能不能回答这个问题”。如果检索片段本身就没有正确答案那么换再强的 LLM 也没用。我建议在开发阶段增加一个调试接口或者在/query的响应中临时返回search_results这样可以快速定位问题到底出在检索阶段还是生成阶段避免盲目优化。5. 企业级 RAG 常见问题与排查思路下面是搭建 RAG 知识库过程中最常遇到的问题和排查方向。这些内容都是实战中踩过坑后总结出来的建议收藏备用。问题现象常见原因解决思路回答内容完全是编的和文档无关检索结果为空或相关性太低LLM 自由发挥增加检索最低分数阈值检查文档是否成功切片入库降低 temperature检索结果不准答非所问Embedding 模型不适合你的语言或领域换成中文优化的 BGE 模型引入重排序微调切块参数答案内容分散缺乏完整上下文切块过小语义被切断调大 chunk_size增加 chunk_overlap改用递归切块PDF 文档提取的文本乱序PDF 是多栏排版或扫描件使用 pdfplumber 做版面分析扫描件走 OCR 流程向量库写入报维度错误Embedding 模型输出维度与集合配置不一致删除集合并重新创建确认 EMBEDDING_DIM 与模型一致问答接口耗时过高召回数量大 重排序 LLM 生成串行执行重排序候选集控制在 20 条以内LLM 生成改为流式输出新增文档后查询不到只更新了文档目录没有重新执行索引建立文档变更监听或定时增量索引机制重复内容特别多回答啰嗦多个检索块内容高度重叠增加去重逻辑相同或相似 chunk 只保留一条降低 overlap关于“文本切块策略怎么选”如果再具体一点我的建议是文档结构规范有清晰段落、标题优先按 Markdown 标题层级切块。文档是长段落叙述使用递归字符切块chunk_size 400~600。文档是表格为主需要先转成 Markdown 表格再切块否则表格信息严重丢失。代码文档按代码块边界切块避免把函数定义和实现切到不同块。6. 生产环境落地最佳实践6.1 权限与安全边界企业知识库往往包含敏感内部信息。直接把这个服务暴露到公网非常危险。生产环境至少要做到接口鉴权在 FastAPI 前增加网关或中间件校验 Token。文档级权限索引时在 payload 中标记文档的可见部门或密级检索后按当前用户角色过滤。敏感信息脱敏索引前先扫描并脱敏身份证号、手机号、银行卡等信息。最小权限原则给知识库 API 分配独立服务账号避免使用最高权限凭据。6.2 增量更新与数据一致性真实业务中文档是不断变化的不能每次全量重建索引。常用的做法是为文档维护一个状态记录表记录文件名、文件 Hash、最后更新时间。定时或通过消息队列监听文件变更变更时只更新该文件对应的向量。删除文档时同步删除对应的向量数据避免脏数据残留。Qdrant 支持按 payload 过滤删除例如删除某个文件名下的所有点# 文件路径ingest.py增量删除片段 from qdrant_client.models import Filter, FieldCondition, MatchValue qdrant.delete( collection_nameCOLLECTION_NAME, points_selectorFilter( must[FieldCondition(keysource, matchMatchValue(value员工手册.pdf))] ), )6.3 效果评估没有评测就没有优化很多团队把 RAG 系统上线后面对用户反馈“回答不准”却不知道从哪里优化。原因是没有建立评测集。建议在项目初期就准备 50~100 条高质量的“问题-标准答案-参考文档”三元组作为回归评测集。每次修改切块策略、模型、检索参数后跑一遍评测集对比三个指标检索命中率标准答案所在的文档块是否出现在 TopK 结果中。忠实度GroundednessLLM 回答中的陈述是否都能在检索材料中找到依据。用户满意度抽样人工打分重点关注事实准确性和引用正确性。评测是 RAG 工程中最容易被忽视、但回报率最高的环节。没有评测集后面所有优化都是盲目的。6.4 日志与可观测性生产环境建议至少记录以下信息用户问题、检索 TopK 结果及分数。是否触发重排序、重排序后结果变化。LLM 输入提示词与输出回答。各阶段耗时检索耗时、重排序耗时、生成耗时。异常情况检索为空、LLM 超时、向量库连接失败。有了这些日志线上问题排查就有了抓手。否则出了问题你连是检索环节挂了还是生成环节挂了都分不清。6.5 服务稳定性与性能LLM 接口调用是企业级 RAG 中最不稳定的环节。建议在代码中增加超时控制给 OpenAI SDK 设置 timeout默认值可以设为 30~60 秒。重试机制对临时网络错误做指数退避重试。熔断降级连续失败 N 次后直接返回检索到的片段原文避免系统不可用。异步与流式交互式场景使用流式输出减少首字延迟。下面是一个带超时和重试的调用示例# 文件路径rag_pipeline.py稳定性增强片段 from openai import OpenAI import time def call_llm_with_retry(messages, max_retries3, timeout60): 带重试的 LLM 调用。 生产环境建议使用更成熟的重试库例如 tenacity。 for attempt in range(max_retries): try: response llm_client.chat.completions.create( modelconfig.LLM_MODEL, messagesmessages, temperature0.2, max_tokens1024, timeouttimeout, ) return response except Exception as e: if attempt max_retries - 1: raise e time.sleep(2 ** attempt) # 指数退避1s, 2s, 4s return None7. 从 RAG 到 Agentic RAG下一步学习方向当你把基础 RAG 链路跑通并上线后可以尝试向 Agentic RAG智能体 RAG方向演进。所谓 Agentic RAG不是简单地把向量库工具暴露给大模型而是让大模型在回答过程中自主规划检索动作。举个例子用户问“对比一下 A 产品和 B 产品在价格、性能和售后服务上的差异”。普通 RAG 只会做一次检索然后把检索到的片段一股脑塞给模型而 Agentic RAG 会先规划出三个子问题依次检索价格、性能、售后相关的文档最后汇总生成对比表格。实现 Agentic RAG 的常见方案包括ReAct 模式让 LLM 循环执行“思考 - 调用工具 - 观察结果 - 再思考”直至完成。多工具路由LLM 判断当前问题属于知识问答、数据库查询还是图表分析自动路由到不同工具。自问自答LLM 把大问题拆成小问题逐个子检索后合并答案。对于大多数企业场景我的建议是先不要盲目上 Agentic RAG。先把基础 RAG 的检索准确率、切块策略、评测机制做到位再引入 Agent 能力。否则复杂链路带来的不稳定因素会成倍放大排查问题更困难。另外一个明确的学习方向是多路召回与知识图谱增强。你可以把向量检索、关键词检索BM25、知识图谱查询的结果做融合排序进一步提升复杂问题的召回率。这个方向在金融、法律等需要精确条款定位的场景非常有用。文章写到这里一套企业级 RAG 知识库问答系统从原理、代码到生产实践就完整梳理了一遍。如果你在搭建过程中遇到新的问题建议先回到检索结果环节去确认而不是直接怀疑大模型的能力——大多数 RAG 效果问题根源都出在数据清洗和分段上。动手搭一套自己的知识库跑通之后再逐步优化这条路走完远比看十篇教程收获大。如果这篇文章对你有帮助可以收藏备用也欢迎在评论区交流实战中遇到的问题。 企业级RAG知识库问答系统搭建实战：从原理到代码
前阵子在给部门搭企业级知识库问答系统时踩了不少 RAG 的坑。从文档解析乱码、切块策略选错导致检索结果稀碎到向量召回准确率低、引用溯源无法落地几乎把 RAG 从入门到放弃的弯路都走了一遍。市面上讲 RAG 概念的资料很多但真正能照着从零搭出一套可运行、可优化、接近生产环境的知识库问答系统的完整教程却很少。这篇文章我会基于实际项目落地经验从 RAG 基础原理讲起带你完整体验一遍企业级知识库搭建全流程环境准备、文档加载与切块、向量化与检索、重排序、LLM 生成、FastAPI 接口封装以及生产环境必须注意的引用溯源、权限、评估与排查方案。整个教程以代码为主线每个环节都会解释“为什么这么做”保证零基础的同学能跟着搭起来有经验的开发者也能直接复用工程思路。1. RAG 是什么为什么要用 RAG1.1 大模型的知识瓶颈先从一个非常直观的问题说起你直接问 GPT 类大模型“我们公司内部上个月的销售数据是多少”它大概率答不上来。原因很简单大模型的知识截止于训练数据企业内部文档、数据库、实时资讯并不在它的记忆范围内。你可以通过微调让模型记住新知识但每次业务变动都要重新训练成本高、周期长而且无法覆盖不断更新的私有数据。RAGRetrieval-Augmented Generation检索增强生成就是用来解决这个问题的。它的思路很朴素不逼模型“记住”所有知识而是在每次回答前先从外部知识库中检索出与问题相关的文档片段把这些片段和问题一起组装成提示词交给大模型生成答案。模型不需要知道全部内容只需要基于给定的材料进行归纳和回答。1.2 RAG 的核心流程一个标准的 RAG 系统本质上由两条链路组成离线索引链路文档加载 - 文本清洗 - 分块Chunking- 向量化Embedding- 写入向量数据库。在线问答链路用户提问 - 问题向量化 - 向量相似度检索 - 可选重排序 - 组装提示词 - LLM 生成答案 - 返回答案与引用来源。用大白话讲离线阶段是“把书拆开、编号、做成索引卡片”在线阶段是“根据问题找卡片、把相关卡片递给学霸让他组织语言回答”。这样一来每次回答用的都是最新的文档内容既不需要重新训练模型也能在答案中附上引用来源方便人工核对。1.3 RAG 的经典应用场景RAG 在企业里的落地场景非常多常见的包括企业知识库问答把内部制度、技术文档、产品手册导入系统员工可以通过对话快速查询。政务与法律文书辅助从大量法规、判例、政策文件中检索相关条款辅助起草和审查。客服智能助手基于产品 FAQ、售后工单、操作手册回答用户问题。研发效能助手让大模型基于内部 API 文档、代码规范回答开发问题。金融研报分析从研报、公告、新闻中检索信息并生成分析摘要。1.4 为什么不能只靠向量检索这里需要澄清一个容易误解的概念RAG 不是“向量数据库的另一种叫法”向量检索只是 RAG 中间的一环。很多初学者把文档全部向量化存入 Milvus 或 Qdrant 后就认为搭建好了知识库结果问答效果很差。原因在于向量检索负责“找可能相关的片段”但片段是否真正回答了用户问题还需要重排序和 LLM 生成来把关。此外切块策略、向量模型质量、检索 TopK 设置、提示词模板都会直接影响最终效果。所以完整的企业级 RAG 项目必须把整条链路都做扎实而不是只停留在“能搜索”的阶段。2. 环境准备与项目结构2.1 技术选型说明RAG 的技术栈有非常多的选择比如 LangChain、LlamaIndex、Haystack或者直接用向量数据库 SDK 手写流程。本文选择一套轻量但完整的方案编程语言Python 3.10。Web 框架FastAPI用于提供问答 API。文档加载pypdf、python-docx分别处理 PDF 和 Word 文档。文本切块使用类似 LangChain 的 RecursiveCharacterTextSplitter 思路但为了减少框架黑盒我会在示例中手写一个递归字符切块器方便你理解切块原理。向量化模型sentence-transformers 加载开源 Embedding 模型例如 BAAI/bge-small-zh-v1.5。如果你有 OpenAI 或其他云端 Embedding 接口也可以替换。向量数据库Qdrant支持 Docker 部署轻量且 Python SDK 易用。大模型兼容 OpenAI API 格式的模型接口可以是 OpenAI、DeepSeek、通义千问、本地 vLLM 等只需要配置 Base URL 和 API Key。重排序可选使用 bge-reranker 或 Cohere Rerank用于提升召回精度。版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。如果你用的是更新的 Python 或库版本个别 API 可能略有变化建议以官方文档为准。2.2 安装依赖首先创建项目虚拟环境并安装基础依赖mkdir rag-knowledge-base cd rag-knowledge-base python -m venv venv source venv/bin/activate # Windows 使用 venv\Scripts\activate然后安装 Python 依赖pip install fastapi uvicorn pypdf python-docx sentence-transformers qdrant-client openai python-dotenv如果你要用重排序模型还需要安装pip install FlagEmbedding2.3 启动 Qdrant 向量数据库Qdrant 有两种常见使用方式一种是 Docker 启动服务端另一种是嵌入式本地模式。生产环境推荐 Docker 方式docker run -d --name qdrant \ -p 6333:6333 \ -p 6334:6334 \ -v $(pwd)/qdrant_storage:/qdrant/storage \ qdrant/qdrant启动后Qdrant 默认在 6333 端口提供 HTTP API6334 是 gRPC 端口。你可以通过http://localhost:6333/dashboard访问 Web 控制台方便查看集合和向量数据。2.4 项目结构规划为了让代码清晰可维护建议按下面的结构组织工程rag-knowledge-base/ ├── app.py # FastAPI 入口 ├── config.py # 全局配置 ├── ingest.py # 离线索引脚本加载、切块、向量化、写入 ├── rag_pipeline.py # 在线问答链路 ├── text_splitter.py # 文本切块器 ├── docs/ # 原始文档目录 ├── data/chunks.json # 切块后的文本可按需保留 └── requirements.txt # 依赖清单下面我们从核心组件开始逐个实现。3. 核心原理与关键代码拆解3.1 文档加载与清洗企业知识库里的文档格式五花八门最常见的是 PDF、Word、Markdown、TXT。文档加载的核心目标是“把文件里的内容提取成纯文本”但这一步往往是坑最多的地方。PDF 常见的坑包括扫描件没有文字层、表格解析错乱、多栏排版文字顺序混乱。Word 文档相对好处理但也要注意提取出的内容可能包含页眉页脚。所以加载后必须做清洗比如去除多余空行、去掉页眉页脚特征文字、统一换行符。下面是一个简单的文档加载工具函数支持 PDF 和 TXT 文本提取# 文件路径ingest.py片段 from pypdf import PdfReader def load_pdf(file_path: str) - str: 提取 PDF 文件中的文本内容。 注意扫描版 PDF 需要 OCR 处理这里只处理带文字层的文档。 reader PdfReader(file_path) pages [] for page in reader.pages: text page.extract_text() if text: pages.append(text) return \n.join(pages) def load_txt(file_path: str) - str: 读取纯文本文件统一编码为 UTF-8。 with open(file_path, r, encodingutf-8) as f: return f.read()这里需要特别提醒如果你发现 PDF 提取出的文本顺序错乱或者缺少部分内容可能是 PDF 本身使用了复杂的排版结构。建议先人工确认几页内容再决定是否引入 OCR 流程。3.2 文本切块RAG 效果的关键文本切块Chunking是 RAG 中影响效果最明显的环节之一。切块太大向量包含的噪音信息多检索精度下降而且可能超出模型上下文窗口切块太小片段缺乏上下文语义召回结果容易被割裂同时增加向量数量。业界常见的切块策略主要有三类固定大小切块按固定的字符数切简单易实现但容易切断语义完整的句子或段落。递归字符切块按不同层级的分隔符段落 - 句号 - 逗号依次尝试切分尽量保证块内语义完整。这是目前最常用的方案。语义切块利用 Embedding 判断句子间语义差异在语义变化处切分效果更好但计算开销更大。下面实现一个基于递归思想的切块器输入是一段长文本输出是若干带序号的文本块# 文件路径text_splitter.py from typing import List class RecursiveTextSplitter: 递归字符切块器 优先按段落分隔符切如果切出的块仍然过大再按句号、逗号等次级分隔符切。 这样可以尽可能保证每个块内语义完整。 def __init__( self, chunk_size: int 500, chunk_overlap: int 80, separators: List[str] None, ): self.chunk_size chunk_size self.chunk_overlap chunk_overlap self.separators separators or [\n\n, \n, 。, , , , , , ] def split_text(self, text: str) - List[str]: 递归切分文本返回文本块列表。 text text.strip() if not text: return [] chunks self._split_with_separator(text, self.separators) # 合并过小的块避免向量库中碎片过多 merged self._merge_chunks(chunks) return merged def _split_with_separator(self, text: str, separators: List[str]) - List[str]: if not text: return [] if len(text) self.chunk_size: return [text] sep None for s in separators: if s in text: sep s break if sep is None: # 没有任何分隔符时硬切 return [ text[i: i self.chunk_size] for i in range(0, len(text), self.chunk_size) ] segments text.split(sep) result [] current for segment in segments: if not segment: continue if len(current) len(segment) len(sep) self.chunk_size: current current sep segment if current else segment else: if current: result.append(current) current segment if current: result.append(current) # 对仍然过大的块递归切分 final_result [] for item in result: if len(item) self.chunk_size: final_result.extend(self._split_with_separator(item, separators[1:])) else: final_result.append(item) return final_result def _merge_chunks(self, chunks: List[str]) - List[str]: 将过小的相邻块合并减少碎片数量。 if not chunks: return [] merged [] buffer for chunk in chunks: if len(buffer) len(chunk) self.chunk_size // 2: buffer chunk else: if buffer: merged.append(buffer) buffer chunk if buffer: merged.append(buffer) return merged切块参数怎么定经验上面向中文文档chunk_size 取 300~500 字符、chunk_overlap 取 50~100 是比较常见的起点。overlap 的作用是让相邻块之间有重叠信息避免一个完整知识点刚好被切断后前后两块都缺少上下文。具体数值需要根据你的文档类型和模型上下文窗口调整。3.3 Embedding 向量化与向量数据库写入文本切块完成后下一步需要把每个块转成向量。Embedding 模型的选择直接影响检索效果。中英文混合场景推荐 BGE 系列比如BAAI/bge-small-zh-v1.5它对中文支持较好且模型体积小适合 CPU 部署。如果你的文档全是英文可以考虑BAAI/bge-base-en-v1.5或 OpenAI 的text-embedding-3-small。向量化代码实现如下# 文件路径ingest.py片段 from sentence_transformers import SentenceTransformer # 加载 Embedding 模型首次运行会下载模型权重 embedding_model SentenceTransformer(BAAI/bge-small-zh-v1.5) def embed_texts(texts): 将文本列表转换为向量列表。 embeddings embedding_model.encode(texts, normalize_embeddingsTrue) return embeddings.tolist()normalize_embeddingsTrue会把向量归一化为单位向量这样后续用余弦相似度计算时内积等价于余弦相似度检索效果更稳定。接下来把文本块和向量写入 Qdrant。Qdrant 是专门为向量检索设计的数据库核心概念包括 Collection集合、Point数据点包含向量和 payload、Payload附加元数据例如来源文件名、页码、原始文本。# 文件路径ingest.py完整离线索引流程 import json import os from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct from text_splitter import RecursiveTextSplitter # 1. 连接 Qdrant qdrant QdrantClient(urlhttp://localhost:6333) COLLECTION_NAME enterprise_kb EMBEDDING_DIM 512 # 根据你的 Embedding 模型维度调整 # 2. 创建集合如果不存在 collections qdrant.get_collections().collections if not any(c.name COLLECTION_NAME for c in collections): qdrant.create_collection( collection_nameCOLLECTION_NAME, vectors_configVectorParams(sizeEMBEDDING_DIM, distanceDistance.COSINE), ) # 3. 加载文档并切块 def load_and_split(file_path: str, splitter: RecursiveTextSplitter): text load_pdf(file_path) if file_path.endswith(.pdf) else load_txt(file_path) chunks splitter.split_text(text) return chunks splitter RecursiveTextSplitter(chunk_size400, chunk_overlap80) all_points [] point_id 0 for filename in os.listdir(docs): file_path os.path.join(docs, filename) if not os.path.isfile(file_path): continue chunks load_and_split(file_path, splitter) embeddings embed_texts(chunks) for i, (chunk_text, vector) in enumerate(zip(chunks, embeddings)): all_points.append( PointStruct( idpoint_id, vectorvector, payload{ text: chunk_text, source: filename, chunk_index: i, }, ) ) point_id 1 # 4. 批量写入 Qdrant qdrant.upsert(collection_nameCOLLECTION_NAME, pointsall_points) print(f成功写入 {len(all_points)} 个文本块)运行这个脚本就会把 docs 目录下所有文档切块、向量化并写入 Qdrant。注意这里向量维度EMBEDDING_DIM 512要和实际模型输出维度一致bge-small-zh-v1.5的输出维度是 512 维。如果你换成了其他模型一定要同步调整这个值。3.4 检索与重排序离线索引完成后就可以实现在线检索了。基础检索逻辑比较简单用户问题向量化后在 Qdrant 中查询最相似的 TopK 个文本块。# 文件路径rag_pipeline.py片段 def search(query: str, top_k: int 5): 向量检索查找与用户问题最相似的文本块。 query_vector embedding_model.encode(query, normalize_embeddingsTrue).tolist() hits qdrant.search( collection_nameCOLLECTION_NAME, query_vectorquery_vector, limittop_k, ) results [] for hit in hits: results.append( { text: hit.payload[text], source: hit.payload[source], score: hit.score, } ) return results单纯靠向量检索有一个高频问题query 和文档块的语义相似但文档块并不是真正能回答问题的最佳片段。比如用户问“怎么申请年假”检索出的前三块都在介绍“年假天数规定”只有第五块真正写了“申请流程”。如果只取 TopK3正确答案就丢了。解决思路是引入重排序Rerank。先用向量召回较多数量的候选块比如召回 20 个再用重排序模型或 LLM 对候选块重新打分选出最相关的 5 个送入生成环节。重排序模型通常比向量检索更精准但计算成本更高所以采用“粗召回 精排序”的两阶段方案是更均衡的做法。下面是一个使用 bge-reranker 的示例# 文件路径rag_pipeline.py片段 from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-base, use_fp16True) def search_with_rerank(query: str, recall_k: int 20, top_k: int 5): 两阶段检索向量召回 重排序精排。 # 向量召回 query_vector embedding_model.encode(query, normalize_embeddingsTrue).tolist() hits qdrant.search( collection_nameCOLLECTION_NAME, query_vectorquery_vector, limitrecall_k, ) candidates [] for hit in hits: candidates.append( { text: hit.payload[text], source: hit.payload[source], score: hit.score, } ) # 使用重排序模型重新打分 pairs [(query, item[text]) for item in candidates] rerank_scores reranker.compute_score(pairs) for item, score in zip(candidates, rerank_scores): item[rerank_score] float(score) candidates.sort(keylambda x: x[rerank_score], reverseTrue) return candidates[:top_k]关于 bge-reranker 的补充说明compute_score返回的分值不一定在 0~1 之间不同模型的范围可能不同。这里我们不依赖绝对分值只看相对排序所以直接排序取 TopK 即可。3.5 提示词组装与 LLM 生成检索出相关片段之后最关键的一步是把这些片段“投喂”给大模型。提示词模板直接影响回答风格和可信度。一个企业级可用的提示词模板至少需要包含三个要素角色设定、检索材料和引用标注、输出约束。注意模板中必须明确要求模型“只能基于给定材料回答如果材料中没有相关内容请诚实说明”这是避免大模型幻觉的核心手段之一。# 文件路径rag_pipeline.py提示词模板 SYSTEM_PROMPT 你是一个专业的企业知识库问答助手。请基于用户问题和检索到的参考资料回答问题。 要求 1. 只能使用参考资料中的内容作答不得编造材料中不存在的信息。 2. 如果参考资料不足请明确回答“根据现有资料无法回答”。 3. 回答时尽量分点说明便于用户阅读。 4. 在回答最后用“参考来源”列出引用到的文档名称。 def build_prompt(query: str, search_results: list) - str: 根据检索结果组装提示词。 每个片段前加上 [1] [2] 这样的编号方便后续生成引用标注。 context_blocks [] for idx, item in enumerate(search_results, start1): source item.get(source, 未知来源) text item[text].strip() context_blocks.append(f[{idx}] 来源: {source}\n{text}) context \n\n.join(context_blocks) user_prompt f请基于以下参考资料回答问题。 参考资料 {context} 问题{query} 请给出答案并在句末使用[编号]标注引用来源。 return user_promptLLM 调用部分我统一使用 OpenAI 的 SDK 格式这样无论是 DeepSeek、通义千问、Moonshot 还是本地 vLLM 启动的模型服务都可以通过配置 Base URL 来接入不需要改代码逻辑。# 文件路径rag_pipeline.pyLLM 调用 from openai import OpenAI import config llm_client OpenAI( api_keyconfig.LLM_API_KEY, base_urlconfig.LLM_BASE_URL, ) def generate_answer(query: str, search_results: list) - dict: 调用大模型生成答案并返回答案、引用的来源列表。 prompt build_prompt(query, search_results) response llm_client.chat.completions.create( modelconfig.LLM_MODEL, messages[ {role: system, content: SYSTEM_PROMPT}, {role: user, content: prompt}, ], temperature0.2, max_tokens1024, ) answer response.choices[0].message.content # 按索引顺序去重提取引用来源 sources [] for item in search_results: source item.get(source, ) if source and source not in sources: sources.append(source) return { answer: answer, sources: sources, }这里temperature0.2是刻意调低的目的是减少模型自由发挥的空间让回答更贴合检索材料。如果生成结果仍然很发散可以考虑降到 0.1 甚至 0。3.6 引用溯源与诚实回答前面提到 RAG 的 Groundedness接地性也就是回答必须基于检索材料。这里再展开说一个企业场景特别看重的点引用溯源。企业内部使用知识库问答时答案必须能追根溯源。员工看到回答后需要能点击跳转到原始文档位置进行核实。实现方法其实不复杂提示词中要求模型在句子末尾标注[编号]我们在代码中将编号映射到对应的文档名和文本块。如果模型没有按格式标注至少也要返回检索命中的来源列表让前端可以展示“参考文档”区域。# 文件路径rag_pipeline.py引用溯源增强 def format_answer_with_citations(answer: str, search_results: list) - str: 简单后处理如果模型没有正确输出[编号]可以追加来源列表兜底。 if [ not in answer: answer \n\n参考来源 seen set() for item in search_results: src item.get(source, 未知来源) if src not in seen: seen.add(src) answer f\n- {src} return answer诚实回答指的则是当检索内容不足以回答问题时模型应该明确说“不知道”而不是强行编造。要做到这一点除了提示词约束还需要在工程上设置一个检索阈值。如果最高相似度分数过低直接返回“没有检索到足够相关信息”不再调用 LLM。这在企业场景中非常重要因为编造的答案一旦流入业务决策风险极大。配置示例# 文件路径config.py RETRIEVAL_MIN_SCORE 0.35 # 检索最低相似度阈值低于该值不触发生成4. 完整实战基于 FastAPI 搭建知识库问答服务在上一节中我们已经把检索和生成的核心函数写好了。现在把它串成一个完整的 FastAPI 服务并提供两个接口POST /ingest手动触发文档索引方便新增文档后增量更新。POST /query用户提问返回答案和引用来源。4.1 FastAPI 应用入口# 文件路径app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field import os import config import rag_pipeline import ingest app FastAPI(title企业级 RAG 知识库问答系统) class QueryRequest(BaseModel): question: str Field(..., max_length500, description用户问题) top_k: int Field(5, ge1, le10, description返回的检索片段数量) class QueryResponse(BaseModel): answer: str sources: list search_results: list class IngestResponse(BaseModel): message: str chunk_count: int app.post(/ingest, response_modelIngestResponse) def run_ingest(): 重新扫描 docs 目录并写入向量数据库。 生产环境建议改为增量更新并加权限控制。 try: chunk_count ingest.run() return IngestResponse(message索引完成, chunk_countchunk_count) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/query, response_modelQueryResponse) def query_kb(req: QueryRequest): 知识库问答接口。 流程向量检索 - 重排序 - 组装提示词 - LLM 生成。 if not req.question.strip(): raise HTTPException(status_code400, detail问题不能为空) # 判断是否启用重排序 use_rerank True if use_rerank: search_results rag_pipeline.search_with_rerank(req.question, recall_k20, top_kreq.top_k) else: search_results rag_pipeline.search(req.question, top_kreq.top_k) # 检索分数过低时不调用 LLM防止幻觉 if not search_results or search_results[0].get(score, 0) config.RETRIEVAL_MIN_SCORE: return QueryResponse( answer抱歉根据现有知识库内容没有检索到足够相关的信息。请尝试换个问法或补充相关文档。, sources[], search_results[], ) result rag_pipeline.generate_answer(req.question, search_results) return QueryResponse( answerresult[answer], sourcesresult[sources], search_resultssearch_results, )4.2 统一配置管理配置文件单独放一个config.py方便在不同环境开发、测试、生产之间切换# 文件路径config.py import os from dotenv import load_dotenv load_dotenv() # Qdrant 配置 QDRANT_URL os.getenv(QDRANT_URL, http://localhost:6333) COLLECTION_NAME os.getenv(COLLECTION_NAME, enterprise_kb) # Embedding 模型配置 EMBEDDING_MODEL os.getenv(EMBEDDING_MODEL, BAAI/bge-small-zh-v1.5) EMBEDDING_DIM int(os.getenv(EMBEDDING_DIM, 512)) # LLM 配置 LLM_API_KEY os.getenv(LLM_API_KEY, your-api-key) LLM_BASE_URL os.getenv(LLM_BASE_URL, https://api.openai.com/v1) LLM_MODEL os.getenv(LLM_MODEL, gpt-4o-mini) # 检索配置 RETRIEVAL_MIN_SCORE float(os.getenv(RETRIEVAL_MIN_SCORE, 0.35)) # 文本切块配置 CHUNK_SIZE int(os.getenv(CHUNK_SIZE, 400)) CHUNK_OVERLAP int(os.getenv(CHUNK_OVERLAP, 80))使用.env文件管理敏感配置时注意不要把密钥提交到 Git 仓库生产环境建议用配置中心或密钥管理服务。4.3 启动服务并验证效果启动 FastAPI 服务uvicorn app:app --host 0.0.0.0 --port 8000首次启动时会加载 Embedding 模型如果你的机器上没有提前下载它会自动从 Hugging Face 拉取模型权重耗时取决于网络环境。国内网络如果下载慢可以设置环境变量HF_ENDPOINThttps://hf-mirror.com切换为镜像源。启动成功后先调用/ingest对文档建立索引curl -X POST http://localhost:8000/ingest预期返回类似{ message: 索引完成, chunk_count: 156 }然后调用/query测试问答curl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {question: 员工年假怎么申请}返回示例{ answer: 根据员工手册员工年假申请需要提前三天在 OA 系统中提交申请经部门负责人审批后方可生效。[1]\n\n参考来源\n- 员工手册.pdf, sources: [员工手册.pdf], search_results: [ { text: 员工年假申请流程员工需提前三天在 OA 系统中提交年假申请……, source: 员工手册.pdf, score: 0.62 } ] }到这里一个最小可用的 RAG 知识库问答系统就跑通了。4.4 验证检索质量的小技巧当你测试时发现回答不理想不要急着调大模型先看看检索结果对不对。一个很实用的技巧是直接打印search_results的前几项问问自己“如果我是人光看这些片段能不能回答这个问题”。如果检索片段本身就没有正确答案那么换再强的 LLM 也没用。我建议在开发阶段增加一个调试接口或者在/query的响应中临时返回search_results这样可以快速定位问题到底出在检索阶段还是生成阶段避免盲目优化。5. 企业级 RAG 常见问题与排查思路下面是搭建 RAG 知识库过程中最常遇到的问题和排查方向。这些内容都是实战中踩过坑后总结出来的建议收藏备用。问题现象常见原因解决思路回答内容完全是编的和文档无关检索结果为空或相关性太低LLM 自由发挥增加检索最低分数阈值检查文档是否成功切片入库降低 temperature检索结果不准答非所问Embedding 模型不适合你的语言或领域换成中文优化的 BGE 模型引入重排序微调切块参数答案内容分散缺乏完整上下文切块过小语义被切断调大 chunk_size增加 chunk_overlap改用递归切块PDF 文档提取的文本乱序PDF 是多栏排版或扫描件使用 pdfplumber 做版面分析扫描件走 OCR 流程向量库写入报维度错误Embedding 模型输出维度与集合配置不一致删除集合并重新创建确认 EMBEDDING_DIM 与模型一致问答接口耗时过高召回数量大 重排序 LLM 生成串行执行重排序候选集控制在 20 条以内LLM 生成改为流式输出新增文档后查询不到只更新了文档目录没有重新执行索引建立文档变更监听或定时增量索引机制重复内容特别多回答啰嗦多个检索块内容高度重叠增加去重逻辑相同或相似 chunk 只保留一条降低 overlap关于“文本切块策略怎么选”如果再具体一点我的建议是文档结构规范有清晰段落、标题优先按 Markdown 标题层级切块。文档是长段落叙述使用递归字符切块chunk_size 400~600。文档是表格为主需要先转成 Markdown 表格再切块否则表格信息严重丢失。代码文档按代码块边界切块避免把函数定义和实现切到不同块。6. 生产环境落地最佳实践6.1 权限与安全边界企业知识库往往包含敏感内部信息。直接把这个服务暴露到公网非常危险。生产环境至少要做到接口鉴权在 FastAPI 前增加网关或中间件校验 Token。文档级权限索引时在 payload 中标记文档的可见部门或密级检索后按当前用户角色过滤。敏感信息脱敏索引前先扫描并脱敏身份证号、手机号、银行卡等信息。最小权限原则给知识库 API 分配独立服务账号避免使用最高权限凭据。6.2 增量更新与数据一致性真实业务中文档是不断变化的不能每次全量重建索引。常用的做法是为文档维护一个状态记录表记录文件名、文件 Hash、最后更新时间。定时或通过消息队列监听文件变更变更时只更新该文件对应的向量。删除文档时同步删除对应的向量数据避免脏数据残留。Qdrant 支持按 payload 过滤删除例如删除某个文件名下的所有点# 文件路径ingest.py增量删除片段 from qdrant_client.models import Filter, FieldCondition, MatchValue qdrant.delete( collection_nameCOLLECTION_NAME, points_selectorFilter( must[FieldCondition(keysource, matchMatchValue(value员工手册.pdf))] ), )6.3 效果评估没有评测就没有优化很多团队把 RAG 系统上线后面对用户反馈“回答不准”却不知道从哪里优化。原因是没有建立评测集。建议在项目初期就准备 50~100 条高质量的“问题-标准答案-参考文档”三元组作为回归评测集。每次修改切块策略、模型、检索参数后跑一遍评测集对比三个指标检索命中率标准答案所在的文档块是否出现在 TopK 结果中。忠实度GroundednessLLM 回答中的陈述是否都能在检索材料中找到依据。用户满意度抽样人工打分重点关注事实准确性和引用正确性。评测是 RAG 工程中最容易被忽视、但回报率最高的环节。没有评测集后面所有优化都是盲目的。6.4 日志与可观测性生产环境建议至少记录以下信息用户问题、检索 TopK 结果及分数。是否触发重排序、重排序后结果变化。LLM 输入提示词与输出回答。各阶段耗时检索耗时、重排序耗时、生成耗时。异常情况检索为空、LLM 超时、向量库连接失败。有了这些日志线上问题排查就有了抓手。否则出了问题你连是检索环节挂了还是生成环节挂了都分不清。6.5 服务稳定性与性能LLM 接口调用是企业级 RAG 中最不稳定的环节。建议在代码中增加超时控制给 OpenAI SDK 设置 timeout默认值可以设为 30~60 秒。重试机制对临时网络错误做指数退避重试。熔断降级连续失败 N 次后直接返回检索到的片段原文避免系统不可用。异步与流式交互式场景使用流式输出减少首字延迟。下面是一个带超时和重试的调用示例# 文件路径rag_pipeline.py稳定性增强片段 from openai import OpenAI import time def call_llm_with_retry(messages, max_retries3, timeout60): 带重试的 LLM 调用。 生产环境建议使用更成熟的重试库例如 tenacity。 for attempt in range(max_retries): try: response llm_client.chat.completions.create( modelconfig.LLM_MODEL, messagesmessages, temperature0.2, max_tokens1024, timeouttimeout, ) return response except Exception as e: if attempt max_retries - 1: raise e time.sleep(2 ** attempt) # 指数退避1s, 2s, 4s return None7. 从 RAG 到 Agentic RAG下一步学习方向当你把基础 RAG 链路跑通并上线后可以尝试向 Agentic RAG智能体 RAG方向演进。所谓 Agentic RAG不是简单地把向量库工具暴露给大模型而是让大模型在回答过程中自主规划检索动作。举个例子用户问“对比一下 A 产品和 B 产品在价格、性能和售后服务上的差异”。普通 RAG 只会做一次检索然后把检索到的片段一股脑塞给模型而 Agentic RAG 会先规划出三个子问题依次检索价格、性能、售后相关的文档最后汇总生成对比表格。实现 Agentic RAG 的常见方案包括ReAct 模式让 LLM 循环执行“思考 - 调用工具 - 观察结果 - 再思考”直至完成。多工具路由LLM 判断当前问题属于知识问答、数据库查询还是图表分析自动路由到不同工具。自问自答LLM 把大问题拆成小问题逐个子检索后合并答案。对于大多数企业场景我的建议是先不要盲目上 Agentic RAG。先把基础 RAG 的检索准确率、切块策略、评测机制做到位再引入 Agent 能力。否则复杂链路带来的不稳定因素会成倍放大排查问题更困难。另外一个明确的学习方向是多路召回与知识图谱增强。你可以把向量检索、关键词检索BM25、知识图谱查询的结果做融合排序进一步提升复杂问题的召回率。这个方向在金融、法律等需要精确条款定位的场景非常有用。文章写到这里一套企业级 RAG 知识库问答系统从原理、代码到生产实践就完整梳理了一遍。如果你在搭建过程中遇到新的问题建议先回到检索结果环节去确认而不是直接怀疑大模型的能力——大多数 RAG 效果问题根源都出在数据清洗和分段上。动手搭一套自己的知识库跑通之后再逐步优化这条路走完远比看十篇教程收获大。如果这篇文章对你有帮助可以收藏备用也欢迎在评论区交流实战中遇到的问题。