这次我们来看一个非常硬核又几乎是现阶段大模型应用落地绕不开的方向RAG 知识库问答系统。都知道大模型本身有幻觉问题很多私有知识也不在训练数据里。直接对话问不出来微调成本又太高。RAG 的解决思路很简单先从外部知识库里检索出相关内容再把这些内容拼到 Prompt 里让模型基于检索结果回答。这样回答的可信度、时效性和私有化程度都会明显提升。这篇内容不是只讲概念而是按一条完整的落地链路走一遍从环境准备、文档加载、文本切分、Embedding 向量化、向量库存储到检索、组装 Prompt、接入 DeepSeek 大模型、启动 Web API 服务最后再加上检索优化、常见问题排查和工程化建议。整个流程可以直接对照源码复现。无论你是刚接触 RAG 的初学者还是想在公司内部快速搭一套私有知识库问答系统的工程师这篇文章都值得收藏按步骤走一遍。核心代码用 Python 编写依赖开源组件生成模型接口以 DeepSeek 为例也可以替换成其他 OpenAI 兼容接口。1. 核心能力速览先把这一套系统涉及的核心能力列出来方便快速判断你要不要继续看。能力项说明项目类型RAG 知识库问答系统完整源码 原理讲解大模型接入DeepSeek API 调用兼容 OpenAI SDK 风格向量检索支持常见开源向量库如 Chroma、Qdrant、FAISS文档格式Markdown、TXT、PDF、Word、HTML 等文本类文档核心流程文档加载 → 文本切分 → Embedding → 向量存储 → 相似度检索 → Prompt 组装 → 生成回答是否需要 GPU不需要Embedding 与生成均走 API 或轻量本地组件显存占用不依赖本地大模型推理常规开发机即可运行启动方式Python 脚本启动 / FastAPI 接口服务是否支持 API支持可封装为 HTTP 接口供外部系统调用是否支持批量任务支持批量文档入库、批量问答测试均可脚本化适用场景私有文档问答、企业知识库、客服辅助、内部 QA 系统合规提醒私有文档需确认版权和隐私边界商用前需做内容复核有一点要提前说清楚DeepSeek API 本身主要提供文本生成能力不直接提供向量化接口。所以这套方案里Embedding 向量化通常用独立的 Embedding 模型完成比如 BGE、M3E、text-embedding 系列。下面会给出可替换的实现方式。2. 适用场景与使用边界RAG 知识库问答系统出现在什么场景下先弄明白这几个问题。适合谁想给团队内部文档做智能问答不想全文喂给大模型也不想做高成本微调。需要对模型回答标注来源回答完能跳回原文。需要经常更新知识内容比如产品文档、规章制度、FAQ、技术手册。想以较低成本体验“私有知识库 大模型”的开发过程。能解决什么问题大模型不了解的内部知识通过检索补全上下文。减少模型编造答案的情况回答有依据。知识更新无需重新训练模型替换文档重新索引即可。不适合什么场景需要模型掌握复杂推理逻辑且知识库内容本身没有对应文本。知识库文档质量很差、错别字多、格式混乱检索效果会很差。对响应速度要求极高RAG 流程比直接对话慢因为多了检索步骤。没有合法授权的文档不要做入库和商用。使用边界一定要明确RAG 不会让模型“凭空知道”你的私有内容也不会替代数据权限管理。它只是通过检索把相关内容临时注入上下文。如果文档本身包含敏感信息你必须自己处理权限、离线部署和访问控制。涉及他人版权内容、个人隐私、肖像声音等信息必须确认授权后方可使用。3. 环境准备与前置条件这套系统的部署门槛不高常规开发机就可以跑通。下面列一下基础环境清单。3.1 运行环境操作系统Windows、macOS、Linux 都支持。Python 版本建议 3.10 及以上避免部分依赖出现兼容问题。网络环境需要能访问 DeepSeek API 服务。磁盘空间文本和向量库占用不大预留 5GB 以上比较稳妥。开发工具VS Code、PyCharm 均可或者直接用命令行。3.2 依赖组件项目核心依赖包括大模型 SDKOpenAI SDK 或 DeepSeek 官方 SDK。向量库Chroma、Qdrant、FAISS 三选一新手推荐 Chroma零配置上手。Embedding 模型可以调用在线 Embedding API也可以本地跑轻量 Embedding 模型。文档解析PDF、Word 等格式处理相关库。API 服务FastAPI Uvicorn。建议使用虚拟环境隔离依赖避免污染全局 Python 环境。3.3 DeepSeek API Key去 DeepSeek 开放平台注册账号并创建 API Key。创建后妥善保存不要提交到公共仓库。可以配置在.env文件中方便代码读取。DEEPSEEK_API_KEYsk-xxxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat4. 系统架构与 RAG 核心原理在写代码之前先把 RAG 的流程压缩成一张图用文字描述出来。一个标准的 RAG 系统分为三个阶段阶段作用关键操作索引阶段把文档变成可检索的数据文档加载、文本清洗、切分、Embedding、存入向量库检索阶段找到与问题最相关的片段查询向量化、相似度检索、重排序生成阶段基于检索结果生成回答组装 Prompt、调用大模型、输出回答很多入门者容易踩一个大坑直接拿原始文档灌进向量库检索时发现什么都搜不到。问题的根源往往不在模型而在索引阶段。文档切分太大会让检索结果不精准切分太小又会丢失上下文。Embedding 模型选择不合适也会让相似度计算效果变差。这套系统里的增强链路是用户提问 → 问题向量化 → 向量库相似度检索 → 取 TopK 文档片段 → 拼进 Prompt → DeepSeek 模型生成回答 → 返回结果并附上来源片段。理解了这个链路后面所有代码就都围绕这几个环节展开。5. 从零搭建 RAG 知识库核心代码实现这一节直接上代码。先搭一个最小可运行的版本再逐步扩展 API 和优化。5.1 初始化项目结构建议按下面目录组织rag-demo/ ├── .env ├── requirements.txt ├── config.py ├── data/ │ └── knowledge/ ├── src/ │ ├── loader.py │ ├── splitter.py │ ├── embedder.py │ ├── store.py │ ├── retriever.py │ ├── llm.py │ └── pipeline.py └── app.py5.2 安装依赖先创建一个虚拟环境python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate然后安装依赖pip install openai langchain langchain-community langchain-text-splitters chromadb fastapi uvicorn python-dotenv pypdf python-docx这里要说明一下LangChain 的版本更新很快不同版本的 API 有差异。下面代码按较常见的方式编写如果版本差异导致报错优先检查导入路径和类名。5.3 配置文件# config.py import os from dotenv import load_dotenv load_dotenv() DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) DEEPSEEK_BASE_URL os.getenv(DEEPSEEK_BASE_URL, https://api.deepseek.com) DEEPSEEK_MODEL os.getenv(DEEPSEEK_MODEL, deepseek-chat) EMBEDDING_MODEL BAAI/bge-small-zh-v1.5 VECTOR_DB_DIR ./vector_db KNOWLEDGE_DIR ./data/knowledge CHUNK_SIZE 500 CHUNK_OVERLAP 50 TOP_K 55.4 文档加载文档加载要做的事情很简单把各种格式的文档统一转化成原始文本。这里支持 TXT、Markdown、PDF、Word。# src/loader.py from pathlib import Path from typing import List def load_documents(directory: str) - List[str]: texts [] base_dir Path(directory) for file_path in base_dir.rglob(*): if file_path.suffix.lower() in [.txt, .md]: texts.append(file_path.read_text(encodingutf-8)) elif file_path.suffix.lower() .pdf: from pypdf import PdfReader reader PdfReader(str(file_path)) text \n.join(page.extract_text() or for page in reader.pages) texts.append(text) elif file_path.suffix.lower() .docx: from docx import Document doc Document(str(file_path)) text \n.join(p.text for p in doc.paragraphs) texts.append(text) return texts文档加载不是越复杂越好关键是稳定。PDF 提取出来的文字经常有空白换行问题建议后续做一下正则清洗。5.5 文本切分切分文本是整个 RAG 流程里最容易影响效果的一步。切分策略要兼顾两个方向太大则检索粒度粗可能引入无关内容太小则片段缺乏上下文模型也没有足够信息。常用做法是按段落和长度结合保留一定的重叠区间。# src/splitter.py from langchain_text_splitters import RecursiveCharacterTextSplitter from typing import List def split_texts(texts: List[str], chunk_size: int 500, chunk_overlap: int 50) - List[str]: splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , ], ) chunks splitter.split_text(\n\n.join(texts)) return chunks这里的separators是中英文标点结合目的是尽量在语义完整的地方断开。5.6 Embedding 向量化Embedding 是 RAG 系统里信息密度的关键。同一个问题用不同的 Embedding 模型检索结果可能差很远。在线 API 方案直接调用 Embedding 接口。本地方案可以用 HuggingFace 的轻量模型这里以 BGE 为例。# src/embedder.py from sentence_transformers import SentenceTransformer from typing import List model None def get_model(): global model if model is None: model SentenceTransformer(BAAI/bge-small-zh-v1.5) return model def embed_texts(texts: List[str]) - List[List[float]]: model get_model() embeddings model.encode(texts, normalize_embeddingsTrue) return embeddings.tolist() def embed_query(query: str) - List[float]: return embed_texts([query])[0]BGE 系列模型建议在 encode 时对中文查询加一句指令这里简化处理。如果你本地运行遇到模型下载问题也可以替换为在线 Embedding API。5.7 向量库存储Chroma 是本地向量库里最省心的一个运行后会在本地目录生成向量数据文件。# src/store.py import chromadb from typing import List def get_collection(persist_dir: str, collection_name: str knowledge_base): client chromadb.PersistentClient(pathpersist_dir) collection client.get_or_create_collection(namecollection_name) return collection def add_documents(collection, ids: List[str], texts: List[str], embeddings: List[List[float]]): collection.add( idsids, documentstexts, embeddingsembeddings, )5.8 检索器检索器负责把用户问题向量化然后在向量库中找最相似的 TopK 片段。# src/retriever.py import chromadb from typing import List def retrieve( persist_dir: str, query: str, query_embedding: List[float], top_k: int 5, ) - List[str]: client chromadb.PersistentClient(pathpersist_dir) collection client.get_collection(knowledge_base) result collection.query( query_embeddings[query_embedding], n_resultstop_k, ) documents result.get(documents, [[]])[0] return documents这一步检索出的片段就是大模型回答问题的“参考阅读材料”。5.9 组装 Prompt 并调用 DeepSeek这是把 RAG 链路串起来的关键步骤。# src/llm.py from openai import OpenAI from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, DEEPSEEK_MODEL client OpenAI( api_keyDEEPSEEK_API_KEY, base_urlDEEPSEEK_BASE_URL, ) def build_prompt(query: str, context_docs: List[str]) - str: context \n\n.join( f【资料片段 {idx 1}】\n{doc} for idx, doc in enumerate(context_docs) ) prompt f请根据以下资料片段回答问题。如果资料中没有相关内容请直接说明资料中未覆盖不要编造答案。 【资料片段】 {context} 【问题】 {query} return prompt def generate_answer(prompt: str) - str: response client.chat.completions.create( modelDEEPSEEK_MODEL, messages[ {role: system, content: 你是一个严谨的知识库问答助手回答时必须基于给定资料。}, {role: user, content: prompt}, ], temperature0.3, max_tokens1024, ) return response.choices[0].message.content这里用了 OpenAI SDK 来调用 DeepSeek因为 DeepSeek 提供兼容接口改下base_url就能连通。如果你换用其他模型平台只要也兼容 OpenAI 接口代码可以直接复用。5.10 完整流程串联# src/pipeline.py from src.loader import load_documents from src.splitter import split_texts from src.embedder import embed_texts, embed_query from src.store import get_collection, add_documents from src.retriever import retrieve from src.llm import build_prompt, generate_answer from config import KNOWLEDGE_DIR, VECTOR_DB_DIR, CHUNK_SIZE, CHUNK_OVERLAP, TOP_K import uuid def build_knowledge_base(): texts load_documents(KNOWLEDGE_DIR) chunks split_texts(texts, CHUNK_SIZE, CHUNK_OVERLAP) embeddings embed_texts(chunks) collection get_collection(VECTOR_DB_DIR) ids [str(uuid.uuid4()) for _ in range(len(chunks))] add_documents(collection, ids, chunks, embeddings) print(f知识库构建完成共 {len(chunks)} 个片段) def ask(query: str) - str: query_embedding embed_query(query) docs retrieve(VECTOR_DB_DIR, query, query_embedding, TOP_K) prompt build_prompt(query, docs) answer generate_answer(prompt) return answer到这里一个最小可运行的 RAG 知识库问答系统就完成了。调用方式from src.pipeline import build_knowledge_base, ask build_knowledge_base() answer ask(公司请假的审批流程是什么) print(answer)6. 把 RAG 封装成 HTTP API 服务实际工程中知识库提问不是总在 Python 脚本里进行。更常见的是通过 HTTP 接口调用方便前端页面、机器人、其他后端系统接入。这里直接基于 FastAPI 封装。# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from src.pipeline import build_knowledge_base, ask app FastAPI(titleRAG 知识库问答服务) class AskRequest(BaseModel): query: str class AskResponse(BaseModel): answer: str app.post(/api/rebuild) def rebuild_knowledge_base(): try: build_knowledge_base() return {message: 知识库重建完成} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/api/ask, response_modelAskResponse) def ask_question(req: AskRequest): if not req.query.strip(): raise HTTPException(status_code400, detailquery 不能为空) try: answer ask(req.query) return AskResponse(answeranswer) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)启动服务python app.py接口说明接口方法作用/api/rebuildPOST重新加载文档并重建向量库/api/askPOST传入问题返回知识库回答用 curl 测试curl -X POST http://127.0.0.1:8000/api/ask \ -H Content-Type: application/json \ -d {query: 介绍一下内部项目的部署流程}用 Python requests 调用import requests url http://127.0.0.1:8000/api/ask payload {query: 公司请假的审批流程是什么} response requests.post(url, jsonpayload, timeout60) print(response.json())返回结果示例{ answer: 根据资料请假审批流程是员工提交申请直属上级审批人事部门备案。 }有了接口以后前端页面、钉钉机器人、企业微信应用都可以接。前端页面就是一个输入框加一个按钮逻辑很直接。7. 批量文档入库与批量问答测试RAG 系统在实际使用中文档数量不会只有几篇。批量处理是必须考虑的问题。批量入库的思路很简单把所有文档放入data/knowledge目录脚本遍历加载统一切分统一写入向量库。# scripts/batch_build.py from src.pipeline import build_knowledge_base if __name__ __main__: build_knowledge_base()这里要提醒一下如果你的文档量很大一次性全部分片并 Embedding内存占用会明显上升。更稳妥的做法是分批处理每处理完一批写入向量库然后释放内存。伪代码思路def batch_build_with_limit(directory, batch_size100): texts load_documents(directory) chunks split_texts(texts) collection get_collection(VECTOR_DB_DIR) for i in range(0, len(chunks), batch_size): batch chunks[i:i batch_size] embeddings embed_texts(batch) ids [str(uuid.uuid4()) for _ in range(len(batch))] add_documents(collection, ids, batch, embeddings) print(f已处理 {i len(batch)} / {len(chunks)} 片段)批量问答测试可以用一个 QA 测试集文件每条记录包含问题、标准答案、期望包含的关键词跑完后自动统计命中率。# scripts/batch_eval.py import json from src.pipeline import ask def evaluate(test_file: str): with open(test_file, r, encodingutf-8) as f: cases json.load(f) total len(cases) hit 0 for case in cases: answer ask(case[query]) contains all(kw in answer for kw in case.get(keywords, [])) hit 1 if contains else 0 print(f问题: {case[query]}) print(f回答: {answer[:100]}...) print(---) print(f命中率: {hit}/{total} {hit / total:.2%})这种批量测试可以量化观察切分策略和 Embedding 模型调优前后的效果差异而不是靠感觉判断。8. 检索效果优化实践很多人在跑通基础版本后发现回答效果一般。这是正常的。基础链路只能保证“能跑”要让效果真正可用需要从下面几个方向优化。8.1 优化文档切分策略不同文档类型要选择不同切分策略。技术文档、规章制度按标题层级切分更合理比如按 Markdown 的#、##标记切分。长段落文本使用重叠切分避免切断关键信息。FAQ 类文档最好按问答对整体存入检索时一个条目就是完整上下文。推荐一个做法先人工看几篇典型文档手动标出哪些位置是最小语义单元再决定切分参数。8.2 混合检索向量检索擅长语义模糊匹配但对精确关键词处理一般。例如“服务器 IP 是多少”这种问题精确数字匹配反而更像关键词检索。可以加入 BM25 关键词检索然后把向量检索和关键词检索结果做融合比如分数倒序排列或加权求和。# 伪代码混合检索流程 vector_results vector_search(query) keyword_results bm25_search(query) merged merge_by_rrf(vector_results, keyword_results, top_k10)RRF 是 Reciprocal Rank Fusion简单说就是对多个检索结果的排名取倒数分数再求和是一种稳定且不依赖权重调节的融合方式。8.3 重排序先召回 TopK 再精排是 RAG 提升精度的常见手段。召回阶段可以多召回一些比如 20 个片段。然后用重排序模型统一打分取前 5 个进入 Prompt。这样既保证了召回率又避免了 TopK 过大导致 Prompt 太长、垃圾信息太多。8.4 Prompt 模板调优Prompt 对回答质量影响很大。建议调整的方向要求模型只能依据资料回答。让模型在资料不足时明确说不清楚。要求回答中标注引用片段编号。示例请严格根据【资料片段】回答问题。回答必须基于资料原文不得自行扩展。如果资料中没有相关信息请直接说明“资料中未覆盖该问题”。回答末尾标注提供了主要依据的片段编号。8.5 元数据过滤当知识库包含多种类型的文档时比如制度文件、产品手册、故障手册给每个文本片段打上文档来源和类型标签检索时先按标签过滤效果会更好。collection.add( idsids, documentstexts, embeddingsembeddings, metadatas[{source: policy, file_name: 考勤制度.md}], )检索时指定where条件result collection.query( query_embeddings[query_embedding], n_resultstop_k, where{source: policy}, )9. 资源占用与性能观察RAG 系统相比本地大模型推理最大的优势是资源开销低。整个流程里最重的计算是 Embedding 模型推理如果本地跑 BGE 小模型CPU 也能撑住。生成回答部分完全由 DeepSeek API 完成本地只负责组装请求和接收结果。如果使用在线 Embedding API本地资源占用可以进一步降低。需要重点观察的指标有这几个指标说明入库耗时文档切分 Embedding 的总时间文档多时需要批量写入单次问答首字延迟从提交问题到第一个字输出的时间主要受网络和 DeepSeek API 影响单次问答总耗时检索耗时 Prompt 构建 大模型生成耗时向量库大小由文本片段数量和 Embedding 维度决定API 并发限制DeepSeek API 有速率限制高并发场景需要加任务队列降低延迟的建议Embedding 模型选计算量更小的版本。检索时关闭离线统计等无关操作。大模型回答使用流式输出首字延迟体验更好。对高频问题做缓存命中缓存直接返回。流式接口是一个比较明显的体验提升点。DeepSeek 接口支持streamTruedef generate_answer_stream(prompt: str): response client.chat.completions.create( modelDEEPSEEK_MODEL, messages[ {role: system, content: 你是一个严谨的知识库问答助手回答时必须基于给定资料。}, {role: user, content: prompt}, ], temperature0.3, max_tokens1024, streamTrue, ) for chunk in response: content chunk.choices[0].delta.content if content: yield contentFastAPI 可以通过StreamingResponse输出流式响应。这样做问答体验更接近原生对话。10. 常见问题与排查方法部署和使用过程中大概率会遇到以下几个问题整理成排查表。问题现象可能原因排查方式解决方案DeepSeek 调用报错API Key 错误、余额不足或网络问题检查返回错误码和日志确认 Key 正确检查网络连通性向量库查询结果为空未先构建知识库或 collection 名不一致确认向量库目录下是否有数据文件先执行build_knowledge_base()再提问检索结果与问题完全无关文档切分粒度不合理、Embedding 模型不适合打印检索片段人工检查调整切分参数更换 Embedding 模型回答存在明显编造检索片段未包含答案Prompt 约束不足检查送入 Prompt 的片段内容优化检索增加重排序强化“没有就直说”文档加载输出乱码PDF 为扫描件或编码不一致检查 extract_text 输出扫描件需要 OCR普通文本转 UTF-8API 服务启动失败端口被占用或依赖缺失查看错误日志换端口补装依赖批量导入内存过大一次性处理大量文档观察内存占用曲线分批处理每批完成后释放变量输出返回很慢网络延迟或 API 服务负载高测试纯 API 调用延迟开启流式输出加缓存遇到问题时最直接的排查方法是把 RAG 流程拆开分别打印中间结果文档加载后看文本、切分后看片段、Embedding 后看向量、检索后看召回内容。哪一步结果不对就定位到哪一步。11. 最佳实践与工程化建议最后给几条工程化建议这决定了整套系统能不能从 demo 走向稳定运行。第一先小参数跑通再上全量数据。文档先放两三篇确认切分、Embedding、检索、生成整个链路都能跑通再批量入库。一上来就全量灌入出了问题很难定位。第二保留一套最小可运行配置。把配置文件、依赖版本、启动命令记录下来遇到环境问题可以快速回到可用状态。第三文档输入、临时文件、向量库、输出结果分目录管理。目录结构清晰后续做增量更新、测试回溯、数据备份都方便。第四批量任务必须加日志和失败重试。入库、问答测试、接口调用都需要记录执行时间、成功失败状态、错误信息。对失败的批次实现重试逻辑。第五接口服务要控制访问范围和并发。服务默认绑定的127.0.0.1只允许本机访问部署到服务器时要通过反向代理校验调用方身份。API Key 不要明文写在前端代码里应该由后端统一持有和调用。第六涉及版权、隐私、敏感数据的文档入库前先走合规确认流程。RAG 本身不增加信息但也不会自动过滤数据权限。不同用户能问什么应该由你的权限系统控制而不是由模型判断。第七商用或正式发布前做一轮效果测试。准备一套覆盖不同难度的测试问题集核对回答准确性、来源引用、拒绝回答边界。12. 总结与下一步这一套 RAG 知识库问答系统核心链路并不复杂文档向量化、检索、拼 Prompt、交给大模型回答。真正决定系统好不好的往往不是大模型本身而是文档处理和检索质量。拿到代码后最先建议验证三件事第一把自己的文档放入data/knowledge目录跑通建库和提问第二打印检索片段确认召回内容是否贴合问题第三用几个刁钻问题测试模型是否会编造如果会就加强 Prompt 约束和重排序。最容易踩的坑有两个一个是文本切分不合理导致检索内容太碎或太杂另一个是拿原始长文档直接建库没有清洗和切分。这两点解决掉整套系统的体验会提升一大截。后续可以继续扩展的方向很多接入语音转写作为知识入口、增加多人权限隔离、加入增量更新策略、集成 Agent 工具调用、迁移到更大量级的向量库。RAG 是一个值得长期投入的方向因为知识管理是每个企业都会遇到的真实问题。建议把这套基础链路保存好后续做扩展时直接复用。
RAG知识库问答系统实战：向量检索+DeepSeek接入全流程
这次我们来看一个非常硬核又几乎是现阶段大模型应用落地绕不开的方向RAG 知识库问答系统。都知道大模型本身有幻觉问题很多私有知识也不在训练数据里。直接对话问不出来微调成本又太高。RAG 的解决思路很简单先从外部知识库里检索出相关内容再把这些内容拼到 Prompt 里让模型基于检索结果回答。这样回答的可信度、时效性和私有化程度都会明显提升。这篇内容不是只讲概念而是按一条完整的落地链路走一遍从环境准备、文档加载、文本切分、Embedding 向量化、向量库存储到检索、组装 Prompt、接入 DeepSeek 大模型、启动 Web API 服务最后再加上检索优化、常见问题排查和工程化建议。整个流程可以直接对照源码复现。无论你是刚接触 RAG 的初学者还是想在公司内部快速搭一套私有知识库问答系统的工程师这篇文章都值得收藏按步骤走一遍。核心代码用 Python 编写依赖开源组件生成模型接口以 DeepSeek 为例也可以替换成其他 OpenAI 兼容接口。1. 核心能力速览先把这一套系统涉及的核心能力列出来方便快速判断你要不要继续看。能力项说明项目类型RAG 知识库问答系统完整源码 原理讲解大模型接入DeepSeek API 调用兼容 OpenAI SDK 风格向量检索支持常见开源向量库如 Chroma、Qdrant、FAISS文档格式Markdown、TXT、PDF、Word、HTML 等文本类文档核心流程文档加载 → 文本切分 → Embedding → 向量存储 → 相似度检索 → Prompt 组装 → 生成回答是否需要 GPU不需要Embedding 与生成均走 API 或轻量本地组件显存占用不依赖本地大模型推理常规开发机即可运行启动方式Python 脚本启动 / FastAPI 接口服务是否支持 API支持可封装为 HTTP 接口供外部系统调用是否支持批量任务支持批量文档入库、批量问答测试均可脚本化适用场景私有文档问答、企业知识库、客服辅助、内部 QA 系统合规提醒私有文档需确认版权和隐私边界商用前需做内容复核有一点要提前说清楚DeepSeek API 本身主要提供文本生成能力不直接提供向量化接口。所以这套方案里Embedding 向量化通常用独立的 Embedding 模型完成比如 BGE、M3E、text-embedding 系列。下面会给出可替换的实现方式。2. 适用场景与使用边界RAG 知识库问答系统出现在什么场景下先弄明白这几个问题。适合谁想给团队内部文档做智能问答不想全文喂给大模型也不想做高成本微调。需要对模型回答标注来源回答完能跳回原文。需要经常更新知识内容比如产品文档、规章制度、FAQ、技术手册。想以较低成本体验“私有知识库 大模型”的开发过程。能解决什么问题大模型不了解的内部知识通过检索补全上下文。减少模型编造答案的情况回答有依据。知识更新无需重新训练模型替换文档重新索引即可。不适合什么场景需要模型掌握复杂推理逻辑且知识库内容本身没有对应文本。知识库文档质量很差、错别字多、格式混乱检索效果会很差。对响应速度要求极高RAG 流程比直接对话慢因为多了检索步骤。没有合法授权的文档不要做入库和商用。使用边界一定要明确RAG 不会让模型“凭空知道”你的私有内容也不会替代数据权限管理。它只是通过检索把相关内容临时注入上下文。如果文档本身包含敏感信息你必须自己处理权限、离线部署和访问控制。涉及他人版权内容、个人隐私、肖像声音等信息必须确认授权后方可使用。3. 环境准备与前置条件这套系统的部署门槛不高常规开发机就可以跑通。下面列一下基础环境清单。3.1 运行环境操作系统Windows、macOS、Linux 都支持。Python 版本建议 3.10 及以上避免部分依赖出现兼容问题。网络环境需要能访问 DeepSeek API 服务。磁盘空间文本和向量库占用不大预留 5GB 以上比较稳妥。开发工具VS Code、PyCharm 均可或者直接用命令行。3.2 依赖组件项目核心依赖包括大模型 SDKOpenAI SDK 或 DeepSeek 官方 SDK。向量库Chroma、Qdrant、FAISS 三选一新手推荐 Chroma零配置上手。Embedding 模型可以调用在线 Embedding API也可以本地跑轻量 Embedding 模型。文档解析PDF、Word 等格式处理相关库。API 服务FastAPI Uvicorn。建议使用虚拟环境隔离依赖避免污染全局 Python 环境。3.3 DeepSeek API Key去 DeepSeek 开放平台注册账号并创建 API Key。创建后妥善保存不要提交到公共仓库。可以配置在.env文件中方便代码读取。DEEPSEEK_API_KEYsk-xxxxx DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat4. 系统架构与 RAG 核心原理在写代码之前先把 RAG 的流程压缩成一张图用文字描述出来。一个标准的 RAG 系统分为三个阶段阶段作用关键操作索引阶段把文档变成可检索的数据文档加载、文本清洗、切分、Embedding、存入向量库检索阶段找到与问题最相关的片段查询向量化、相似度检索、重排序生成阶段基于检索结果生成回答组装 Prompt、调用大模型、输出回答很多入门者容易踩一个大坑直接拿原始文档灌进向量库检索时发现什么都搜不到。问题的根源往往不在模型而在索引阶段。文档切分太大会让检索结果不精准切分太小又会丢失上下文。Embedding 模型选择不合适也会让相似度计算效果变差。这套系统里的增强链路是用户提问 → 问题向量化 → 向量库相似度检索 → 取 TopK 文档片段 → 拼进 Prompt → DeepSeek 模型生成回答 → 返回结果并附上来源片段。理解了这个链路后面所有代码就都围绕这几个环节展开。5. 从零搭建 RAG 知识库核心代码实现这一节直接上代码。先搭一个最小可运行的版本再逐步扩展 API 和优化。5.1 初始化项目结构建议按下面目录组织rag-demo/ ├── .env ├── requirements.txt ├── config.py ├── data/ │ └── knowledge/ ├── src/ │ ├── loader.py │ ├── splitter.py │ ├── embedder.py │ ├── store.py │ ├── retriever.py │ ├── llm.py │ └── pipeline.py └── app.py5.2 安装依赖先创建一个虚拟环境python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate然后安装依赖pip install openai langchain langchain-community langchain-text-splitters chromadb fastapi uvicorn python-dotenv pypdf python-docx这里要说明一下LangChain 的版本更新很快不同版本的 API 有差异。下面代码按较常见的方式编写如果版本差异导致报错优先检查导入路径和类名。5.3 配置文件# config.py import os from dotenv import load_dotenv load_dotenv() DEEPSEEK_API_KEY os.getenv(DEEPSEEK_API_KEY) DEEPSEEK_BASE_URL os.getenv(DEEPSEEK_BASE_URL, https://api.deepseek.com) DEEPSEEK_MODEL os.getenv(DEEPSEEK_MODEL, deepseek-chat) EMBEDDING_MODEL BAAI/bge-small-zh-v1.5 VECTOR_DB_DIR ./vector_db KNOWLEDGE_DIR ./data/knowledge CHUNK_SIZE 500 CHUNK_OVERLAP 50 TOP_K 55.4 文档加载文档加载要做的事情很简单把各种格式的文档统一转化成原始文本。这里支持 TXT、Markdown、PDF、Word。# src/loader.py from pathlib import Path from typing import List def load_documents(directory: str) - List[str]: texts [] base_dir Path(directory) for file_path in base_dir.rglob(*): if file_path.suffix.lower() in [.txt, .md]: texts.append(file_path.read_text(encodingutf-8)) elif file_path.suffix.lower() .pdf: from pypdf import PdfReader reader PdfReader(str(file_path)) text \n.join(page.extract_text() or for page in reader.pages) texts.append(text) elif file_path.suffix.lower() .docx: from docx import Document doc Document(str(file_path)) text \n.join(p.text for p in doc.paragraphs) texts.append(text) return texts文档加载不是越复杂越好关键是稳定。PDF 提取出来的文字经常有空白换行问题建议后续做一下正则清洗。5.5 文本切分切分文本是整个 RAG 流程里最容易影响效果的一步。切分策略要兼顾两个方向太大则检索粒度粗可能引入无关内容太小则片段缺乏上下文模型也没有足够信息。常用做法是按段落和长度结合保留一定的重叠区间。# src/splitter.py from langchain_text_splitters import RecursiveCharacterTextSplitter from typing import List def split_texts(texts: List[str], chunk_size: int 500, chunk_overlap: int 50) - List[str]: splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , ], ) chunks splitter.split_text(\n\n.join(texts)) return chunks这里的separators是中英文标点结合目的是尽量在语义完整的地方断开。5.6 Embedding 向量化Embedding 是 RAG 系统里信息密度的关键。同一个问题用不同的 Embedding 模型检索结果可能差很远。在线 API 方案直接调用 Embedding 接口。本地方案可以用 HuggingFace 的轻量模型这里以 BGE 为例。# src/embedder.py from sentence_transformers import SentenceTransformer from typing import List model None def get_model(): global model if model is None: model SentenceTransformer(BAAI/bge-small-zh-v1.5) return model def embed_texts(texts: List[str]) - List[List[float]]: model get_model() embeddings model.encode(texts, normalize_embeddingsTrue) return embeddings.tolist() def embed_query(query: str) - List[float]: return embed_texts([query])[0]BGE 系列模型建议在 encode 时对中文查询加一句指令这里简化处理。如果你本地运行遇到模型下载问题也可以替换为在线 Embedding API。5.7 向量库存储Chroma 是本地向量库里最省心的一个运行后会在本地目录生成向量数据文件。# src/store.py import chromadb from typing import List def get_collection(persist_dir: str, collection_name: str knowledge_base): client chromadb.PersistentClient(pathpersist_dir) collection client.get_or_create_collection(namecollection_name) return collection def add_documents(collection, ids: List[str], texts: List[str], embeddings: List[List[float]]): collection.add( idsids, documentstexts, embeddingsembeddings, )5.8 检索器检索器负责把用户问题向量化然后在向量库中找最相似的 TopK 片段。# src/retriever.py import chromadb from typing import List def retrieve( persist_dir: str, query: str, query_embedding: List[float], top_k: int 5, ) - List[str]: client chromadb.PersistentClient(pathpersist_dir) collection client.get_collection(knowledge_base) result collection.query( query_embeddings[query_embedding], n_resultstop_k, ) documents result.get(documents, [[]])[0] return documents这一步检索出的片段就是大模型回答问题的“参考阅读材料”。5.9 组装 Prompt 并调用 DeepSeek这是把 RAG 链路串起来的关键步骤。# src/llm.py from openai import OpenAI from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL, DEEPSEEK_MODEL client OpenAI( api_keyDEEPSEEK_API_KEY, base_urlDEEPSEEK_BASE_URL, ) def build_prompt(query: str, context_docs: List[str]) - str: context \n\n.join( f【资料片段 {idx 1}】\n{doc} for idx, doc in enumerate(context_docs) ) prompt f请根据以下资料片段回答问题。如果资料中没有相关内容请直接说明资料中未覆盖不要编造答案。 【资料片段】 {context} 【问题】 {query} return prompt def generate_answer(prompt: str) - str: response client.chat.completions.create( modelDEEPSEEK_MODEL, messages[ {role: system, content: 你是一个严谨的知识库问答助手回答时必须基于给定资料。}, {role: user, content: prompt}, ], temperature0.3, max_tokens1024, ) return response.choices[0].message.content这里用了 OpenAI SDK 来调用 DeepSeek因为 DeepSeek 提供兼容接口改下base_url就能连通。如果你换用其他模型平台只要也兼容 OpenAI 接口代码可以直接复用。5.10 完整流程串联# src/pipeline.py from src.loader import load_documents from src.splitter import split_texts from src.embedder import embed_texts, embed_query from src.store import get_collection, add_documents from src.retriever import retrieve from src.llm import build_prompt, generate_answer from config import KNOWLEDGE_DIR, VECTOR_DB_DIR, CHUNK_SIZE, CHUNK_OVERLAP, TOP_K import uuid def build_knowledge_base(): texts load_documents(KNOWLEDGE_DIR) chunks split_texts(texts, CHUNK_SIZE, CHUNK_OVERLAP) embeddings embed_texts(chunks) collection get_collection(VECTOR_DB_DIR) ids [str(uuid.uuid4()) for _ in range(len(chunks))] add_documents(collection, ids, chunks, embeddings) print(f知识库构建完成共 {len(chunks)} 个片段) def ask(query: str) - str: query_embedding embed_query(query) docs retrieve(VECTOR_DB_DIR, query, query_embedding, TOP_K) prompt build_prompt(query, docs) answer generate_answer(prompt) return answer到这里一个最小可运行的 RAG 知识库问答系统就完成了。调用方式from src.pipeline import build_knowledge_base, ask build_knowledge_base() answer ask(公司请假的审批流程是什么) print(answer)6. 把 RAG 封装成 HTTP API 服务实际工程中知识库提问不是总在 Python 脚本里进行。更常见的是通过 HTTP 接口调用方便前端页面、机器人、其他后端系统接入。这里直接基于 FastAPI 封装。# app.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from src.pipeline import build_knowledge_base, ask app FastAPI(titleRAG 知识库问答服务) class AskRequest(BaseModel): query: str class AskResponse(BaseModel): answer: str app.post(/api/rebuild) def rebuild_knowledge_base(): try: build_knowledge_base() return {message: 知识库重建完成} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.post(/api/ask, response_modelAskResponse) def ask_question(req: AskRequest): if not req.query.strip(): raise HTTPException(status_code400, detailquery 不能为空) try: answer ask(req.query) return AskResponse(answeranswer) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)启动服务python app.py接口说明接口方法作用/api/rebuildPOST重新加载文档并重建向量库/api/askPOST传入问题返回知识库回答用 curl 测试curl -X POST http://127.0.0.1:8000/api/ask \ -H Content-Type: application/json \ -d {query: 介绍一下内部项目的部署流程}用 Python requests 调用import requests url http://127.0.0.1:8000/api/ask payload {query: 公司请假的审批流程是什么} response requests.post(url, jsonpayload, timeout60) print(response.json())返回结果示例{ answer: 根据资料请假审批流程是员工提交申请直属上级审批人事部门备案。 }有了接口以后前端页面、钉钉机器人、企业微信应用都可以接。前端页面就是一个输入框加一个按钮逻辑很直接。7. 批量文档入库与批量问答测试RAG 系统在实际使用中文档数量不会只有几篇。批量处理是必须考虑的问题。批量入库的思路很简单把所有文档放入data/knowledge目录脚本遍历加载统一切分统一写入向量库。# scripts/batch_build.py from src.pipeline import build_knowledge_base if __name__ __main__: build_knowledge_base()这里要提醒一下如果你的文档量很大一次性全部分片并 Embedding内存占用会明显上升。更稳妥的做法是分批处理每处理完一批写入向量库然后释放内存。伪代码思路def batch_build_with_limit(directory, batch_size100): texts load_documents(directory) chunks split_texts(texts) collection get_collection(VECTOR_DB_DIR) for i in range(0, len(chunks), batch_size): batch chunks[i:i batch_size] embeddings embed_texts(batch) ids [str(uuid.uuid4()) for _ in range(len(batch))] add_documents(collection, ids, batch, embeddings) print(f已处理 {i len(batch)} / {len(chunks)} 片段)批量问答测试可以用一个 QA 测试集文件每条记录包含问题、标准答案、期望包含的关键词跑完后自动统计命中率。# scripts/batch_eval.py import json from src.pipeline import ask def evaluate(test_file: str): with open(test_file, r, encodingutf-8) as f: cases json.load(f) total len(cases) hit 0 for case in cases: answer ask(case[query]) contains all(kw in answer for kw in case.get(keywords, [])) hit 1 if contains else 0 print(f问题: {case[query]}) print(f回答: {answer[:100]}...) print(---) print(f命中率: {hit}/{total} {hit / total:.2%})这种批量测试可以量化观察切分策略和 Embedding 模型调优前后的效果差异而不是靠感觉判断。8. 检索效果优化实践很多人在跑通基础版本后发现回答效果一般。这是正常的。基础链路只能保证“能跑”要让效果真正可用需要从下面几个方向优化。8.1 优化文档切分策略不同文档类型要选择不同切分策略。技术文档、规章制度按标题层级切分更合理比如按 Markdown 的#、##标记切分。长段落文本使用重叠切分避免切断关键信息。FAQ 类文档最好按问答对整体存入检索时一个条目就是完整上下文。推荐一个做法先人工看几篇典型文档手动标出哪些位置是最小语义单元再决定切分参数。8.2 混合检索向量检索擅长语义模糊匹配但对精确关键词处理一般。例如“服务器 IP 是多少”这种问题精确数字匹配反而更像关键词检索。可以加入 BM25 关键词检索然后把向量检索和关键词检索结果做融合比如分数倒序排列或加权求和。# 伪代码混合检索流程 vector_results vector_search(query) keyword_results bm25_search(query) merged merge_by_rrf(vector_results, keyword_results, top_k10)RRF 是 Reciprocal Rank Fusion简单说就是对多个检索结果的排名取倒数分数再求和是一种稳定且不依赖权重调节的融合方式。8.3 重排序先召回 TopK 再精排是 RAG 提升精度的常见手段。召回阶段可以多召回一些比如 20 个片段。然后用重排序模型统一打分取前 5 个进入 Prompt。这样既保证了召回率又避免了 TopK 过大导致 Prompt 太长、垃圾信息太多。8.4 Prompt 模板调优Prompt 对回答质量影响很大。建议调整的方向要求模型只能依据资料回答。让模型在资料不足时明确说不清楚。要求回答中标注引用片段编号。示例请严格根据【资料片段】回答问题。回答必须基于资料原文不得自行扩展。如果资料中没有相关信息请直接说明“资料中未覆盖该问题”。回答末尾标注提供了主要依据的片段编号。8.5 元数据过滤当知识库包含多种类型的文档时比如制度文件、产品手册、故障手册给每个文本片段打上文档来源和类型标签检索时先按标签过滤效果会更好。collection.add( idsids, documentstexts, embeddingsembeddings, metadatas[{source: policy, file_name: 考勤制度.md}], )检索时指定where条件result collection.query( query_embeddings[query_embedding], n_resultstop_k, where{source: policy}, )9. 资源占用与性能观察RAG 系统相比本地大模型推理最大的优势是资源开销低。整个流程里最重的计算是 Embedding 模型推理如果本地跑 BGE 小模型CPU 也能撑住。生成回答部分完全由 DeepSeek API 完成本地只负责组装请求和接收结果。如果使用在线 Embedding API本地资源占用可以进一步降低。需要重点观察的指标有这几个指标说明入库耗时文档切分 Embedding 的总时间文档多时需要批量写入单次问答首字延迟从提交问题到第一个字输出的时间主要受网络和 DeepSeek API 影响单次问答总耗时检索耗时 Prompt 构建 大模型生成耗时向量库大小由文本片段数量和 Embedding 维度决定API 并发限制DeepSeek API 有速率限制高并发场景需要加任务队列降低延迟的建议Embedding 模型选计算量更小的版本。检索时关闭离线统计等无关操作。大模型回答使用流式输出首字延迟体验更好。对高频问题做缓存命中缓存直接返回。流式接口是一个比较明显的体验提升点。DeepSeek 接口支持streamTruedef generate_answer_stream(prompt: str): response client.chat.completions.create( modelDEEPSEEK_MODEL, messages[ {role: system, content: 你是一个严谨的知识库问答助手回答时必须基于给定资料。}, {role: user, content: prompt}, ], temperature0.3, max_tokens1024, streamTrue, ) for chunk in response: content chunk.choices[0].delta.content if content: yield contentFastAPI 可以通过StreamingResponse输出流式响应。这样做问答体验更接近原生对话。10. 常见问题与排查方法部署和使用过程中大概率会遇到以下几个问题整理成排查表。问题现象可能原因排查方式解决方案DeepSeek 调用报错API Key 错误、余额不足或网络问题检查返回错误码和日志确认 Key 正确检查网络连通性向量库查询结果为空未先构建知识库或 collection 名不一致确认向量库目录下是否有数据文件先执行build_knowledge_base()再提问检索结果与问题完全无关文档切分粒度不合理、Embedding 模型不适合打印检索片段人工检查调整切分参数更换 Embedding 模型回答存在明显编造检索片段未包含答案Prompt 约束不足检查送入 Prompt 的片段内容优化检索增加重排序强化“没有就直说”文档加载输出乱码PDF 为扫描件或编码不一致检查 extract_text 输出扫描件需要 OCR普通文本转 UTF-8API 服务启动失败端口被占用或依赖缺失查看错误日志换端口补装依赖批量导入内存过大一次性处理大量文档观察内存占用曲线分批处理每批完成后释放变量输出返回很慢网络延迟或 API 服务负载高测试纯 API 调用延迟开启流式输出加缓存遇到问题时最直接的排查方法是把 RAG 流程拆开分别打印中间结果文档加载后看文本、切分后看片段、Embedding 后看向量、检索后看召回内容。哪一步结果不对就定位到哪一步。11. 最佳实践与工程化建议最后给几条工程化建议这决定了整套系统能不能从 demo 走向稳定运行。第一先小参数跑通再上全量数据。文档先放两三篇确认切分、Embedding、检索、生成整个链路都能跑通再批量入库。一上来就全量灌入出了问题很难定位。第二保留一套最小可运行配置。把配置文件、依赖版本、启动命令记录下来遇到环境问题可以快速回到可用状态。第三文档输入、临时文件、向量库、输出结果分目录管理。目录结构清晰后续做增量更新、测试回溯、数据备份都方便。第四批量任务必须加日志和失败重试。入库、问答测试、接口调用都需要记录执行时间、成功失败状态、错误信息。对失败的批次实现重试逻辑。第五接口服务要控制访问范围和并发。服务默认绑定的127.0.0.1只允许本机访问部署到服务器时要通过反向代理校验调用方身份。API Key 不要明文写在前端代码里应该由后端统一持有和调用。第六涉及版权、隐私、敏感数据的文档入库前先走合规确认流程。RAG 本身不增加信息但也不会自动过滤数据权限。不同用户能问什么应该由你的权限系统控制而不是由模型判断。第七商用或正式发布前做一轮效果测试。准备一套覆盖不同难度的测试问题集核对回答准确性、来源引用、拒绝回答边界。12. 总结与下一步这一套 RAG 知识库问答系统核心链路并不复杂文档向量化、检索、拼 Prompt、交给大模型回答。真正决定系统好不好的往往不是大模型本身而是文档处理和检索质量。拿到代码后最先建议验证三件事第一把自己的文档放入data/knowledge目录跑通建库和提问第二打印检索片段确认召回内容是否贴合问题第三用几个刁钻问题测试模型是否会编造如果会就加强 Prompt 约束和重排序。最容易踩的坑有两个一个是文本切分不合理导致检索内容太碎或太杂另一个是拿原始长文档直接建库没有清洗和切分。这两点解决掉整套系统的体验会提升一大截。后续可以继续扩展的方向很多接入语音转写作为知识入口、增加多人权限隔离、加入增量更新策略、集成 Agent 工具调用、迁移到更大量级的向量库。RAG 是一个值得长期投入的方向因为知识管理是每个企业都会遇到的真实问题。建议把这套基础链路保存好后续做扩展时直接复用。
