RAG实战：从零搭建大模型知识库问答系统

这次我们来看一个 RAG 实战项目。RAG 全称 Retrieval-Augmented Generation检索增强生成是目前大模型应用落地最主流的技术路线之一。它解决的核心问题很明确让大模型回答私有知识、最新资讯、企业内部文档时不再凭空捏造而是先检索相关资料再基于资料生成答案。如果你在做知识库问答、企业文档助手、智能客服这类场景RAG 基本是绕不开的方案。本文会从零开始拆解一个完整 RAG 项目的搭建过程覆盖文档加载、文本切分、向量化、向量检索、重排、生成、接口封装和批量任务。重点不是讲概念而是给出可以直接照着跑的代码和验证流程。无论你是刚接触大模型的新手还是准备在企业项目里落地 RAG 的开发者都可以按这篇文章的思路走一遍。整个技术栈是常见且成熟的Python LangChain/LlamaIndex 向量数据库 Ollama 本地模型。这样搭配的好处是不依赖商业 API 也能跑通整个流程适合学习也方便后续切换到企业级组件。1. RAG 核心能力速览先给一张速览表方便你判断这个项目是否符合需求。能力项说明项目类型基于大模型的检索增强生成RAG系统覆盖知识库问答全流程核心技术文档加载、文本切分、Embedding 向量化、向量检索、Rerank 重排、LLM 生成数据源Markdown、TXT、PDF、Word、HTML 等常见文档格式检索方式向量相似度检索 可选 Rerank 重排可扩展多路召回生成模型可接入 Ollama 本地模型也可替换为 OpenAI 兼容 API向量数据库Chroma / FAISS / Milvus / Qdrant 等支持平台Windows / Linux / macOSCPU 可运行有 GPU 效果更好启动方式命令行脚本 FastAPI 接口服务是否支持 API支持封装为 HTTP 接口是否支持批量任务支持可对多文档批量建索引、批量问答适合场景本地知识库问答、企业内部文档检索、客服助手、学习 RAG 原理需要注意RAG 项目本身没有固定的“一键启动包”它更像是一个技术框架。你可以用 LangChain 快速验证也可以换成 LlamaIndex甚至手写检索逻辑。下面文章按“能跑通、能验证、能扩展”的思路来写。2. RAG 工作原理拆解RAG 的全流程可以分为四个阶段索引构建、检索、重排、生成。理解这四个阶段遇到问题才能快速定位。第一步索引构建。原始文档是杂乱无章的不能直接丢给模型。需要先做三件事文档加载把 PDF、Word、Markdown 等内容读取为纯文本。文本切分把长文本切成块每块 300 到 800 字左右。切分太大会导致检索不精准切分太碎会丢失上下文。向量化用 Embedding 模型把文本块转换成向量并写入向量数据库。这一步的输出是一个“检索索引”。以后每次查询系统都是在索引里找相关内容。第二步检索。用户输入问题后用同一个 Embedding 模型把问题向量化然后在向量数据库里做相似度搜索常见的算法是余弦相似度或内积。这里有一个关键点Embedding 模型必须统一。建索引时用什么模型检索时也必须用同一个否则向量空间不一致检索结果会完全不可用。第三步重排可选但推荐。向量检索返回的是 Top K 候选文本块但排序质量不一定好。Rerank 模型可以对候选结果重新打分把最相关的文本排到前面。对于企业级 RAG重排这一步通常不能省。第四步生成。系统把检索到的文本块和用户问题拼装成 Prompt交给大模型生成答案。模型看到的是请根据以下资料回答问题 资料1... 资料2... 问题...如果检索结果为空系统应该回答“知识库中没有相关信息”而不是强行编造。从整体来看RAG 的核心价值在于把模型的“记忆力”问题转化为“检索力”问题。模型不需要记住所有知识只需要理解检索到的内容并组织语言输出。这也是 RAG 相比微调在知识更新、可追溯性方面更有优势的原因。3. RAG 知识库适用场景与使用边界RAG 系统最合适的场景有几类企业内部知识库问答制度文档、技术文档、产品手册、FAQ 等。个人知识库助手把笔记、书摘、论文转换为可检索问答。特定垂直领域问答法律条文、医疗指南、政策文件、设备说明书。客服系统辅助基于历史工单和知识库内容给客服提供回答建议。不适合的场景也要说清楚需要模型具备复杂推理、多步计算的场景RAG 单独撑不起来需要配合 Agent 或工具调用。知识之间强关联、上下文跨度很大的文档简单切分后检索效果会下降需要做父子块切分或重排优化。对答案格式要求极高的场景需要额外做输出约束和后处理。合规和安全边界必须重视企业文档中可能包含敏感信息、商业机密、个人隐私部署前要做权限控制接口不能裸奔在公网。涉及人脸、声音、个人信息的内容要确认有合法处理和使用的授权。知识产权问题上传到知识库的资料、用于生成答案的模型权重都要确认版权合规。RAG 系统局部幻觉仍然存在尤其是检索到了无关或错误内容时生成结果可能“自信地胡说”。面向用户输出时要有免责声明和人工复核机制。4. RAG 本地部署环境准备开始搭建之前先检查本机环境。下面是一份通用检查清单具体版本以你自己的项目要求为准。检查项要求参考说明操作系统Windows 10/11、Ubuntu 20.04、macOS 12本文命令以 Linux/macOS 为主Python3.9 或 3.10 以上检查python --versionpip最新版检查pip --version内存8GB 以上推荐 16GB加载模型和向量数据库都会占内存GPU可选CPU 可以跑完整个流程GPU 只影响推理速度磁盘空间至少 10GB模型文件、依赖库、测试数据网络需要下载依赖和模型如果离线环境需要预先下载好模型包Ollama用于管理本地 LLM 模型也可以不用 Ollama直接接 OpenAI 兼容 API建议先建一个独立的目录和 Python 虚拟环境避免依赖冲突mkdir rag-project cd rag-project python -m venv venv source venv/bin/activate # Windows 下为 venv\Scripts\activate然后安装基础依赖pip install langchain langchain-community langchain-chroma chromadb pip install pypdf python-docx markdown unstructured pip install sentence-transformers pip install fastapi uvicorn requests pip install ollama如果你使用 Ollama还需要先在 Ollama 中拉取需要的模型。以常见的 7B 模型为例命令如下实际模型名称以 Ollama 仓库为准ollama pull qwen2.5:7b ollama pull nomic-embed-text这里建议准备两个模型一个用于生成回答的 LLM一个用于文本向量化的 Embedding 模型。两者不是同一个东西不能混用。5. RAG 系统架构与目录设计一个适合学习和扩展的 RAG 项目建议按下面的目录结构组织rag-project/ ├── data/ │ ├── raw_docs/ # 原始文档 │ └── processed/ # 切分后的文本 ├── index/ │ └── chroma_db/ # 向量数据库存储目录 ├── config/ │ └── config.yaml # 项目配置 ├── src/ │ ├── loader.py # 文档加载与切分 │ ├── embedder.py # 向量化与索引构建 │ ├── retriever.py # 检索器 │ ├── generator.py # 生成器 │ ├── api.py # FastAPI 服务 │ └── batch.py # 批量任务 ├── tests/ │ └── test_query.py # 验证脚本 └── requirements.txt这种分层设计的好处是每个模块只做一件事。后续要换 Embedding 模型、换向量数据库、换 LLM只需要修改对应模块不会牵一发动全身。下面写一个通用配置模板# config/config.yaml embedding: model: nomic-embed-text # 按实际使用的 embedding 模型替换 device: cpu # 可选 cpu / cuda vectorstore: type: chroma persist_dir: ./index/chroma_db retriever: top_k: 5 # 初始检索数量 score_threshold: 0.3 # 相似度阈值低于阈值视为无答案 reranker: model: bge-reranker-base # 可选按实际可用模型替换 top_n: 3 # 重排后保留数量 generator: model: qwen2.5:7b # 按 Ollama 中实际模型替换 temperature: 0.2 max_tokens: 1024 api: host: 127.0.0.1 port: 8000注意上面的模型名称只是示例。实际使用时要先确认你本机 Ollama 中已经拉取了哪些模型用ollama list查看。6. 核心代码实现文档加载与索引构建6.1 文档加载与文本切分文档加载这一步看起来简单实际容易踩坑。PDF 有扫描版和文字版Word 有 doc 和 docxMarkdown 有代码块和表格。没有一个万能加载器能完美处理所有格式所以先用少量文件测试再扩展到全量。这里演示一个通用流程# src/loader.py import os from langchain_community.document_loaders import ( PyPDFLoader, TextLoader, UnstructuredMarkdownLoader, Docx2txtLoader, ) def load_document(filepath: str): ext os.path.splitext(filepath)[-1].lower() if ext .pdf: loader PyPDFLoader(filepath) elif ext .md: loader UnstructuredMarkdownLoader(filepath) elif ext .txt: loader TextLoader(filepath, encodingutf-8) elif ext .docx: loader Docx2txtLoader(filepath) else: raise ValueError(f不支持的格式: {ext}) return loader.load()文本切分是 RAG 检索质量的关键。推荐使用递归字符切分器并设置chunk_size和chunk_overlap# src/loader.py from langchain.text_splitter import RecursiveCharacterTextSplitter def split_documents(docs, chunk_size500, chunk_overlap50): splitter RecursiveCharacterTextSplitter( chunk_sizechunk_size, chunk_overlapchunk_overlap, separators[\n\n, \n, 。, , , , , , ], ) chunks splitter.split_documents(docs) print(f切分完成共得到 {len(chunks)} 个文本块) return chunks切分参数的调整原则如果文档是段落式文章500 字搭配 50 字重叠通常效果不错。如果是代码文档或表格型内容需要根据语义边界调整分隔符。6.2 Embedding 向量化与索引写入向量化的作用是把文本映射到高维空间。这里以 Ollama 提供的 Embedding 接口为例# src/embedder.py import requests class OllamaEmbedder: def __init__(self, base_url: str http://localhost:11434, model: str nomic-embed-text): self.base_url base_url self.model model def embed_texts(self, texts: list[str]) - list[list[float]]: vectors [] for text in texts: resp requests.post( f{self.base_url}/api/embeddings, json{model: self.model, prompt: text}, ) resp.raise_for_status() vectors.append(resp.json()[embedding]) return vectors把切分后的文本块写入 Chroma 向量数据库# src/embedder.py from langchain_chroma import Chroma def build_index(chunks, embedder, persist_dir: str ./index/chroma_db): vectorstore Chroma.from_documents( documentschunks, embeddingCustomLangChainEmbedding(embedder), persist_directorypersist_dir, ) return vectorstoreCustomLangChainEmbedding是把上面的OllamaEmbedder适配成 LangChain Embedding 接口的类核心是继承langchain_core.embeddings.Embeddings并实现embed_documents和embed_query两个方法。代码结构如下from langchain_core.embeddings import Embeddings class CustomLangChainEmbedding(Embeddings): def __init__(self, embedder: OllamaEmbedder): self.embedder embedder def embed_documents(self, texts): return self.embedder.embed_texts(texts) def embed_query(self, text): return self.embedder.embed_texts([text])[0]这样一个最小可用的索引构建流程就完成了。生成index/chroma_db目录后后续检索不需要重新构建直接加载即可。7. 检索器与生成器实现7.1 向量检索检索器负责接收问题、返回相关文本块# src/retriever.py from langchain_chroma import Chroma def load_vectorstore(persist_dir: str ./index/chroma_db): return Chroma( persist_directorypersist_dir, embeddingCustomLangChainEmbedding(embedder), ) def search(vectorstore, query: str, top_k: int 5): results vectorstore.similarity_search_with_score(query, ktop_k) return resultssimilarity_search_with_score返回的 score 是距离值值越小表示越相似。不同向量数据库的 score 定义不同Chromabase 的 score 是距离不能直接作为置信度。建议先打印几条结果观察分数范围再设置阈值。观察方法for doc, score in search(vectorstore, 你的测试问题, top_k5): print(fscore: {score:.4f}) print(doc.page_content[:200]) print(---)7.2 RAG 答案生成检索到相关内容后拼接 Prompt 并调用 LLM# src/generator.py import requests class OllamaGenerator: def __init__(self, base_url: str http://localhost:11434, model: str qwen2.5:7b): self.base_url base_url self.model model def generate(self, context: str, question: str) - str: prompt f请基于以下资料回答问题。 ## 资料 {context} ## 问题 {question} 要求 1. 只使用资料中的信息回答问题。 2. 如果资料中没有答案请直接说“知识库中暂无相关信息”。 3. 回答要简洁不超过300字。 resp requests.post( f{self.base_url}/api/generate, json{ model: self.model, prompt: prompt, stream: False, options: { temperature: 0.2, num_predict: 1024, }, }, ) resp.raise_for_status() return resp.json()[response]这里stream: False表示等待完整结果返回。生产环境如果响应时间长建议改用流式输出提升体验。7.3 完整问答链路把检索和生成串起来# src/rag_chain.py def ask(question: str, vectorstore, generator): docs_with_score search(vectorstore, question, top_k5) if not docs_with_score: return 知识库中暂无相关信息 context \n\n.join([doc.page_content for doc, _ in docs_with_score]) answer generator.generate(context, question) return { answer: answer, source_docs: [ {content: doc.page_content[:200], score: round(score, 4)} for doc, score in docs_with_score ], }返回source_docs非常关键。企业级 RAG 必须让用户看到答案依据是什么否则无法判断答案是否可信。这也是 RAG 相比直接调用 LLM 的一个重要优势可追溯。8. 功能测试与效果验证系统搭建完成后不要急着加功能先跑一组标准测试。8.1 索引构建测试准备一批测试文档例如 3 到 5 篇 Markdown 技术文章内容与你后续要问答的知识相关。执行索引构建python -c from src.loader import load_document, split_documents; from src.embedder import build_index; docs load_document(data/raw_docs/test.md); chunks split_documents(docs); build_index(chunks, embedder)判断成功的标准日志输出切分了多少个文本块。index/chroma_db目录下出现向量数据文件。再次运行不报重复写入错误。8.2 检索质量测试用 5 个问题分别测试问题类型示例期望结果原文直接覆盖文档标题或小标题类问题检索到对应文本块score 足够小表述改写换一种说法问同一件事仍能检索到相似内容跨文档组合答案分散在多个文档各文档相关文本块都返回无答案问题与知识库完全无关返回空或低分结果生成器给出“暂无相关信息”相似但错误容易混淆的术语重排后应该把最相关内容排到前面如果检索结果不理想优先调整切分参数和top_k不要先去换模型。很多 RAG 效果差不是模型不行是文本切分和检索粒度不合适。8.3 问答效果测试调用ask函数from src.rag_chain import ask result ask(你的测试问题, vectorstore, generator) print(result[answer]) print(--- 来源文档 ---) for doc in result[source_docs]: print(doc[content])判断标准答案内容是否来自检索结果而不是模型自身知识。答案是否准确是否包含编造的细节。无答案问题时是否拒绝回答而不是硬编。回答延迟是否可接受。8.4 RAG 系统评估思路对于刚上手的项目建议用“人工判断题”快速评估 20 到 30 个问题分三个维度打分检索命中率正确答案是否出现在检索返回的 Top 5 中。答案正确率生成答案是否与检索内容一致。拒绝准确率无答案问题时是否正确拒绝。更正式的评估可以用 RAGAS 框架它包含忠实度、答案相关性、上下文相关性等指标。不过刚开始不要过度依赖指标先把检索结果打印出来看问题出在检索还是生成一目了然。9. 接口 API 与批量任务实战RAG 系统最终要提供服务不能每次都在命令行跑脚本。用 FastAPI 封装是最常见的做法。9.1 启动 API 服务# src/api.py from fastapi import FastAPI from pydantic import BaseModel app FastAPI(titleRAG API Service) class QueryRequest(BaseModel): question: str top_k: int 5 class QueryResponse(BaseModel): answer: str source_docs: list app.post(/ask, response_modelQueryResponse) def ask_question(req: QueryRequest): result ask(req.question, vectorstore, generator) return result app.get(/health) def health(): return {status: ok}启动服务uvicorn src.api:app --host 127.0.0.1 --port 8000启动后打开浏览器访问http://127.0.0.1:8000/docs可以看到自动生成的 Swagger 文档可以直接在页面上测试接口。9.2 接口调用示例用 curl 测试curl -X POST http://127.0.0.1:8000/ask \ -H Content-Type: application/json \ -d {question: 你的测试问题, top_k: 5}用 Python 调用import requests url http://127.0.0.1:8000/ask payload { question: RAG 的核心流程是什么, top_k: 5 } resp requests.post(url, jsonpayload, timeout120) print(resp.json())接口能通之后就可以接到前端页面、企业微信机器人、飞书机器人或者自己的自动化工具里。9.3 批量任务设计批量问答的典型场景用 Excel 或 CSV 导入一批问题批量生成答案并导出结果。实现思路如下# src/batch.py import csv import time def batch_ask(input_file: str, output_file: str): with open(input_file, r, encodingutf-8) as f: reader csv.DictReader(f) rows list(reader) results [] for i, row in enumerate(rows): question row[question] try: result ask(question, vectorstore, generator) results.append({ question: question, answer: result[answer], sources: | .join([d[content] for d in result[source_docs]]), }) except Exception as exc: results.append({ question: question, answer: fERROR: {exc}, sources: , }) print(f[{i 1}/{len(rows)}] 已完成: {question[:30]}) time.sleep(1) with open(output_file, w, encodingutf-8, newline) as f: writer csv.DictWriter(f, fieldnames[question, answer, sources]) writer.writeheader() writer.writerows(results) print(f批量任务完成结果写入 {output_file})批量任务最容易出问题的点是长文本超时和接口限流。建议每次遍历之间加time.sleep并把单条超时时间设置得宽松一些。10. 资源占用与性能观察RAG 系统运行时的资源消耗主要来自三部分Embedding 模型、向量数据库、LLM 生成模型。其中 LLM 生成模型通常是内存或显存消耗最大的部分。如果你在本地用 Ollama 运行 7B 级别的模型需要观察内存和显存占用。观察方法NVIDIA GPU 环境用nvidia-smi查看显存占用。CPU 环境用top或htop查看内存占用。Ollama 加载多个模型时可以设置OLLAMA_MAX_LOADED_MODELS环境变量控制同时加载的模型数量。影响性能的主要因素因素影响文本块总数数量越多索引构建越慢检索耗时越长top_k数值越大传给 LLM 的上下文越长生成耗时越长chunk_size太小导致上下文不足太大会稀释检索精度LLM 模型参数量模型越大生成速度越慢显存占用越高是否使用 GPUCPU 可以跑但生成速度明显变慢Prompt 长度检索结果过多会让每次请求变慢成本也变高如果想降低资源占用优先做三件事把 Embedding 模型换为更小的模型本地 CPU 推理也能接受。控制检索返回数量top_k从 10 降到 5再配合重排只保留 3 条。减少 LLM 的num_predict答案限制在 512 或 1024 token 内。还有一个容易被忽略的点向量数据库如果长期不清洗会积累大量无用的文本块。重新构建索引比增量清理更简单批量更新文档后建议重建一次索引。11. 常见问题与排查方法RAG 项目报错集中在几个地方依赖安装、模型名称错误、向量数据库冲突、端口占用、检索结果为空。下面是一份排查表。问题现象可能原因排查方式解决方案安装依赖失败Python 版本不匹配或依赖冲突查看安装日志确认 Python 版本升级 Python 到 3.10新建虚拟环境重装启动时找不到 OllamaOllama 服务未启动curl http://localhost:11434测试启动 Ollama 后再运行项目调用模型报 404模型未拉取或名称错误ollama list查看已安装模型修正模型名称或重新拉取向量数据库重复写入使用了相同的persist_dir且未清理检查索引目录内容删除旧的索引目录重新构建检索结果为空未设置 score 阈值或阈值过高打印原始 score观察数值范围调整score_threshold或检查查询文本检索结果不相关切分过大/过小或 Embedding 模型与查询不匹配打印检索到的文本块内容调整chunk_size和top_kAPI 端口被占用8000 端口已被其他程序占用lsof -i:8000或netstat -ano更换端口如--port 8001LLM 回答明显错误检索到无关内容或 Prompt 约束不足检查source_docs内容增加重排或调整 Prompt 要求“仅基于资料回答”批量任务卡住单条请求超时异常未被捕获在批量脚本中添加超时和日志设置timeout120记录失败条目并重试首次生成速度很慢模型需要加载到内存/显存观察 Ollama 日志和nvidia-smi预热模型或用更小量化模型一个排查技巧遇到问题时把用户的问题和检索返回的文本块先打印出来人工判断文本块是否包含答案。如果文本块里没有答案问题出在检索如果文本块里有答案但模型答错问题出在生成。这个二分法能快速缩小范围。12. 进阶方向多路召回与 Agentic RAG基础 RAG 跑通之后可以往两个方向扩展。多路召回。只用向量检索会有局限尤其是专有名词、缩写、编号类问题单纯靠语义相似度不一定召回正确结果。企业级 RAG 经常采用多路召回策略向量召回负责语义相似内容。BM25 / 关键词召回负责精确匹配适合代码、编号、专有名词。元数据过滤限定时间范围、来源、部门等条件。重排融合将多路召回结果合并用 Rerank 模型统一打分。实现多路召回时要注意去重和归一化。不同召回策略返回的候选可能有重叠需要按分数融合或直接让 Rerank 模型重新排序。Agentic RAG。这是近期热度很高的方向。它不是一次性检索然后生成而是让大模型判断“需要哪些信息、是否需要多次检索、是否需要调用工具”然后把检索当作工具使用多轮迭代直到获得足够信息。典型的扩展思路问题 → Agent 判断意图 → 选择检索策略 → 执行检索 → 判断是否满足 → 不满足则改写问题再检索 → 满足则生成答案这种方式适合复杂问题比如“去年 Q3 的销售数据和不含税收入的差异原因是什么”涉及多表查询和多步推理。但工程复杂度也会显著上升需要处理 Agent 的循环控制、超时、误判和 token 消耗。建议初学者先把基础 RAG 做扎实再考虑 Agentic 扩展。基础流程跑不稳加再多 Agent 也是空中楼阁。13. 企业级落地最佳实践与合规建议最后总结几条工程化建议这些内容在我看过的 RAG 项目踩坑案例里反复出现。第一次先小参数测试。先用 3 到 5 篇文档、top_k3跑通全流程再扩展到全量数据。不要一上来就处理 1 万篇文档排错成本会很高。建索引和检索必须使用同一个 Embedding 模型。换模型必须重建索引否则检索结果不可用。输出结果必须带来源。企业场景下用户需要看到答案依据哪份文档这既是信任问题也是排查问题的关键入口。批量任务必须做失败重试和日志。推荐每处理一条就记录一条崩溃时能从断点继续而不是全部重跑。接口服务要限制访问范围。至少要绑定内网地址必要时加 API Key 鉴权不能在公网开放/ask接口。涉及企业内部数据、个人隐私、人脸、声音或版权资料的必须先确认授权和合规边界。RAG 系统的价值是提高效率不是绕过授权。发布前做效果复核。生成结果要抽检特别是面向客户或对外发布的内容不能直接完全自动化输出。Prompt 模板要持续迭代。建议把 Prompt 独立成配置文件方便调整不要硬编码在业务代码里。从学习路径来看建议按照“最小链路跑通 → 单文档调优 → 多文档评估 → API 封装 → 批量任务 → 高级检索”的顺序推进每一步都留出验证节点。这套流程掌握之后你已经具备自己搭建一套 RAG 知识库问答系统的基本能力。接下来可以尝试替换不同的向量数据库、接入不同的 LLM、扩展多路召回或者把它接到实际业务系统里。最值得先验证的永远是你在真实场景中遇到的那批问题把问题列表准备好跑一轮测试看检索命中率和答案准确率然后再决定下一步优化方向。