这次我们把大模型 RAG 从原理到落地完整过一遍。RAG 即 Retrieval-Augmented Generation检索增强生成是目前把大模型接入私有数据最常用的一套技术方案。它的核心思路很朴素先检索资料再让大模型基于资料回答。解决的核心问题是大模型不知道你企业内部文档、产品手册、个人知识库里的内容直接提问很容易出现编造和答非所问。通过 RAG可以把用户问题转换成查询从知识库中检索相关片段拼进提示词让大模型基于这些片段生成有依据的回答而不是凭空瞎猜。这篇文章不是只画概念图。我们从原理讲起依次覆盖 Embedding 模型与向量库选型、本地环境搭建、最小可运行代码、召回优化最后落到接口服务、前端界面和批量入库。你可以把它当成一份 RAG 项目的从零到一工程笔记。如果你正准备做知识库问答、内部文档检索、客服助手或者只是想把 RAG 完整流程弄明白这篇文章值得收藏。文中所有代码选用常见开源组件尽量做到能复制、能跑通。涉及调用外部大模型 API 时私有数据需要先做脱敏和授权评估不要随意提交到第三方服务。下面进入正文。1. 核心内容速览覆盖模块本文会讲什么你能得到什么RAG 原理为什么大模型需要先检索再生成从概念上理解 RAG 的三段式流程完整链路离线入库、在线问答、评估迭代三条流水线对项目全貌有清晰认识组件选型Embedding 模型、向量库、检索器选择思路根据数据量和场景选型环境准备Python 依赖与本地推理服务接入搭出一套可运行的实验环境最小实现无框架依赖的 RAG Demo 代码真正跑通一个知识库问答召回优化切分策略、混合检索、重排序、查询改写解决检索不准导致回答错误的问题项目落地FastAPI、Streamlit、批量入库把 Demo 升级成可用的服务2. RAG 核心原理为什么大模型要先查后答大模型的能力来自预训练时见过的语料知识范围受训练数据和截止时间限制。你问一个模型“我们公司上季度的退款流程是什么”如果这句话不在预训练语料里模型只能根据泛化经验去组合答案组合得再流畅也可能是错的这就是幻觉的来源。RAG 相当于给大模型开卷考试。系统先去外部知识库检索候选资料再把资料和问题一起交给模型模型只能在资料范围内组织答案。这样一来回答就不是模型凭记忆生成而是有据可依。一个完整的 RAG 问答流程通常包含下面几个环节用户输入 Query。可选步骤查询改写。用大模型把“问题不完整、指代不清”的情况纠正或者把一个问题拆成多个子问题分别检索。检索。通过向量召回、关键词召回或混合召回从知识库找出候选文档片段。重排序。如果候选结果多用精排模型或分数融合策略把最相关的片段排在前面。上下文拼装。把筛出的片段按顺序组装成 Prompt 中的“资料区”。大模型生成。模型阅读资料并回答问题必要时返回引用来源。后处理。在多轮对话场景中当前答案还会和对话历史拼接。RAG 和微调的使用边界需要分清。如果要做频繁更新的知识库问答、需要答案可追溯来源RAG 明显更合适。如果要让模型固定输出某一类格式、学习某种语气微调更合适。两者也可以组合但新手不必一开始就上微调先把 RAG 链路跑通成本更低。3. 完整链路拆解离线入库、在线问答与评估迭代RAG 项目不只是“写个检索函数”那么简单。为了工程化我们通常把它拆成三条流水线。3.1 离线文档入库 Pipeline这个阶段负责把原始文档变成向量库里的数据不依赖用户提问是 RAG 的地基。文档会经过加载、清洗、切分、向量化、入库五个步骤。加载要兼容 txt、PDF、Markdown、Word 等格式清洗要去掉页眉页脚、无效空行处理乱码切分决定检索的最小粒度直接影响召回质量向量化把文本变成稠密向量最后连同元数据一起写入向量库。元数据建议认真维护。文件名、章节标题、更新时间、来源链接这些信息既能用于检索结果过滤也能在下游回答里做引用来源。3.2 在线问答 Pipeline用户发起提问后进入这个流程。系统处理 Query、调用检索器获取候选片段、重排序、拼装 Prompt、调用大模型生成回答。在线链路对性能更敏感。检索一般控制在几百毫秒以内大模型生成是主要耗时环节。如果检索本身就很慢要考虑向量索引类型、候选数量、是否把检索结果缓存。3.3 评估迭代 Pipeline很多 RAG 项目没有评估这一步上线后靠用户反馈发现问题成本很高。更推荐的做法是准备几十到上百条测试问题每条问题标注标准答案和期望检索到的文档片段把结果跑完后统计命中率和答案准确率。评估结果引导优化方向。如果召回阶段没召回正确内容先看切分和检索策略如果召回了但答案还是错的再看 Prompt 设计和模型生成参数如果答案本身质量平平再考虑换更强的大模型或加入重排序。这一步是 RAG 持续迭代的闭环。4. Embedding、向量库与检索组件选型参考4.1 Embedding 模型选择文本向量化的质量是 RAG 的上限。选型时可以参考这几个维度维度说明中英文支持中文为主的场景优先选中文语料优化过的模型向量维度维度越高通常表达越细但占空间和计算也越多最大输入长度决定了单个 chunk 能编码多长文本部署成本本地开源模型不额外收费但占资源商用 API 更省事但产生费用更新维护新版本是否持续维护对长期项目很重要常见方案包括本地部署开源 Embedding 模型以及使用成熟商用 API。实际效果建议直接拿自己的领域文档测试不要只看公开榜单。4.2 向量库选择向量库用于存储和检索向量选型要看数据量、部署复杂度和是否已有基础设施。向量库定位适用场景Chroma轻量级嵌入式向量库本地 Demo、小规模知识库起步FAISSFacebook 开源的向量检索库以计算为主适合中等规模快速检索Milvus分布式向量数据库大规模、生产级、需要高可用Elasticsearch/OpenSearch搜索引擎增强向量能力已有 ES 基础设施需要全文和向量混合检索数据量小的时候不需要一上来就上重组件。几百上千个文档片段用 Chroma 或 FAISS 都能很好跑起来。数据量到百万级以上再评估 Milvus 这类分布式系统。4.3 检索方式向量检索引近似语义关键词检索引精确匹配两者各有优势。向量检索对“换个说法问同样的问题”更友好关键词检索对“型号编号、专有名词、精确值”更可靠。生产的常用做法是混合检索一路向量召回一路 BM25 关键词召回再做结果融合最后用重排序模型精排。5. 本地实验环境准备RAG 项目需要 Python 环境、Embedding 模型和向量库依赖、大模型接入渠道。下面给出一套通用实验环境配置。建议使用 Python 3.10 左右版本并创建独立虚拟环境避免和系统依赖冲突。核心依赖可以这样安装# 创建虚拟环境 python3 -m venv rag_env source rag_env/bin/activate # 安装核心依赖 pip install chromadb sentence-transformers openai fastapi uvicorn streamlit rank-bm25如果想测试重排序可以另外安装 FlagEmbeddingpip install FlagEmbedding这里说明一下各部分的作用chromadb轻量向量库负责数据持久化和检索。sentence-transformers加载本地 Embedding 模型并生成向量。openai调用兼容 OpenAI 格式的大模型接口。fastapi uvicorn把 RAG 包装成 HTTP 服务。streamlit快速搭建测试界面。rank-bm25关键词检索用于混合召回。FlagEmbedding用于 Cross-Encoder 重排序。大模型的接入渠道建议先统一到 openai 兼容接口。本地可以用 vLLM、Ollama 等推理服务暴露一个 HTTP 端口线上则直接用云服务商提供的接口。两种方式对本文代码来说基本通用只需要改 base_url 和 api_key。Embedding 模型可以从 Hugging Face 下载下载慢时可以配置镜像环境变量。具体命令以实际环境为准。6. 最小可用的 RAG 实现文档入库与检索问答这一章我们写一个不依赖 LangChain 的最小 RAG 实现这样对新手来说依赖少、问题更好排查。项目结构可以是rag_demo/ ├── data/ │ └── knowledge.txt ├── rag_build.py # 文档入库 └── rag_query.py # 检索问答6.1 准备演示文档产品A 支持批量导入知识库文档。 产品A 的问答接口地址是 /api/ask。 产品A 本地部署建议使用 Linux 服务器。 产品A 支持常见文本格式导入包括 txt 和 markdown。这里的内容只是用来演示流程实际使用时替换成自己的业务资料。6.2 文档切分与向量化入库# rag_build.py from sentence_transformers import SentenceTransformer import chromadb # 1. 读取文档并按简单策略切分 text open(data/knowledge.txt, encodingutf-8).read() def split_text(text, chunk_size100, chunk_overlap20): chunks [] for i in range(0, len(text), chunk_size - chunk_overlap): chunks.append(text[i:i chunk_size]) return chunks chunks split_text(text, chunk_size100, chunk_overlap20) # 2. 加载本地 Embedding 模型 model SentenceTransformer(BAAI/bge-small-zh-v1.5) # 3. 初始化向量库 client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(nameknowledge_demo) # 4. 向量化并写入 ids [fchunk_{i} for i in range(len(chunks))] embeddings model.encode(chunks).tolist() collection.add(documentschunks, embeddingsembeddings, idsids) print(f入库完成共 {len(chunks)} 个片段)这个切分函数是按字符硬切的训练演示足够生产环境还需要考虑语义边界和文档结构。6.3 检索上下文并调用大模型生成回答# rag_query.py from sentence_transformers import SentenceTransformer from openai import OpenAI import chromadb # 1. 重新加载 Embedding 模型和向量库 model SentenceTransformer(BAAI/bge-small-zh-v1.5) client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(nameknowledge_demo) # 2. 用户问题 query 产品A 怎么批量导入文档 # 3. 向量召回 query_embedding model.encode([query]).tolist() results collection.query( query_embeddingsquery_embedding, n_results3 ) documents results[documents][0] context \n.join(documents) print(检索到的上下文) print(context) # 4. 拼装 Prompt prompt f你是知识库问答助手。请只根据下面的资料回答问题。 如果资料中没有相关信息请直接回答“知识库中没有找到相关信息”不要编造。 资料 {context} 问题{query} 回答 # 5. 调用大模型 client OpenAI( base_urlhttp://127.0.0.1:8000/v1, api_keyEMPTY ) response client.chat.completions.create( modelyour-local-model, # 改成你实际部署的模型名 messages[{role: user, content: prompt}], temperature0.2 ) print(最终回答) print(response.choices[0].message.content)如果你用的是在线大模型 API把 base_url 改成服务商地址、api_key 换成真实密钥、model 换成实际模型名即可。这一步跑通后你已经拥有了一个最简知识库问答链路。判断成功的标准是输入一个问题返回的 answer 直接来自资料内容而不是模型自由发挥。第 6.3 节代码出现了两个client变量一个用于向量库、一个用于大模型 API命名容易混淆。出于可读性可以改成vector_client和llm_client。实际执行时也要注意大模型服务的 context length 是否足够。7. 召回优化实战切分策略、混合检索与重排序RAG 效果不好一半以上问题出在召回阶段。这一章是全文重点按顺序讲切分、查询改写、混合检索和重排序。7.1 切分策略chunk 粒度直接决定检索质量。切太大一个块里包含多个主题语义被稀释切太小单个块承载的信息量不足检索命中但上下文不完整。比较实用的起步配置是 chunk_size 300 到 500overlap 控制在 10% 到 20%。切分要尽量避开句子中间保持语义完整。对于有明确结构的文档优先按标题层级先切再对长段落二次切分。更进阶的做法是父子切分把父块用于上下文理解子块用于精确匹配。实现时可以给每个 chunk 绑定父块 ID检索到子块后把父块一并拼进 Prompt。7.2 查询改写用户问题经常很短比如“它支持吗”“价格是多少”。原样拿去检索召回质量很低。常见的查询改写方案有两种。第一种是 Multi-Query让大模型把原问题改写成多个子问题分别检索后合并结果适合复杂问题。第二种是 HyDE让大模型先根据问题写一段假设性回答再用这段假设回答去检索间接提升语义匹配度。查询改写会多一次大模型调用成本有所增加但很多场景下收益明显。7.3 混合检索与 RRF 融合向量召回擅长语义匹配BM25 擅长精确匹配。混合检索把两条召回链路的结果合并再通过 RRF 算法排序能同时兼顾两种优势。下面给出一个简化示例from rank_bm25 import BM25Okapi # 假设 corpus 是全部 chunk tokenized_corpus [list(chunk) for chunk in corpus] bm25 BM25Okapi(tokenized_corpus) # 向量召回得到 candidate_ids vector_ids [chunk_1, chunk_2] # 示例 # BM25 召回得到 bm25_ids bm25_ids [chunk_2, chunk_3] # 示例 # RRF 融合 def rrf(rank_lists, k60): scores {} for rank_list in rank_lists: for rank, doc_id in enumerate(rank_list): scores[doc_id] scores.get(doc_id, 0) 1.0 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue) final_ranking rrf([vector_ids, bm25_ids])实际项目中直接把 collection.query 返回的文档 ID 列表作为 vector_ids 传入即可。7.4 重排序向量检索得到的 Top-K 候选并不是最终顺序。更稳妥的做法是先用向量或关键词召回 10 到 20 个候选再用 Cross-Encoder 重排序模型逐对计算 query 和文档的相关性最后取前 3 到 5 个拼进 Prompt。from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-base, use_fp16False) candidates [产品A 支持批量导入知识库文档。, 产品A 的问答接口地址是 /api/ask。] pairs [(query, doc) for doc in candidates] scores reranker.compute_score(pairs) # 按分数降序重排 ordered sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue) for doc, score in ordered: print(score, doc)注意Cross-Encoder 重排序本质是让模型逐对打分候选数越多耗时越长一般只对 Top 20 以内的候选精排。7.5 元数据过滤知识库规模变大后所有文档检索一遍会引入大量无关内容。建议在入库时为每个 chunk 绑定来源、部门、文档类型、更新时间等元数据检索时按条件过滤。比如客服场景只检索“客服知识库”来源的文档避免把产品研发文档混进来。优化时不要一上来把所有技巧全叠上。第一步先看坏案例Query 到底是什么检索到的 top-1 chunk 是否包含答案。如果不包含优先调切分如果包含但排名靠后优先加重排序如果是追问和指代类问题优先做查询改写。8. 项目落地接口 API、前端界面与批量入库跑通最小 RAG 后下一步是把它工程化。日常使用不能每次跑 Python 脚本需要用 HTTP 接口暴露服务加上一个简单的 Web 界面并让文档入库支持批量和增量。8.1 FastAPI 包装 RAG 服务# app.py from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class AskRequest(BaseModel): question: str def run_rag(question: str): # 这里调用之前 rag_query.py 中的检索和生成逻辑 # 返回答案和来源 answer 根据检索结果生成的回答 sources [chunk_1, chunk_2] return answer, sources app.post(/api/ask) def ask(request: AskRequest): answer, sources run_rag(request.question) return {answer: answer, sources: sources}启动服务uvicorn app:app --host 0.0.0.0 --port 8000接口服务一旦对外开放需要加上鉴权和访问控制。建议先绑定127.0.0.1或内网地址不要直接暴露到公网。8.2 Streamlit 搭建测试界面# web.py import streamlit as st import requests API_URL http://127.0.0.1:8000/api/ask st.title(RAG 知识库问答 Demo) question st.text_input(请输入问题) if st.button(提问): resp requests.post(API_URL, json{question: question}, timeout60) if resp.status_code 200: st.info(resp.json()[answer]) else: st.error(请求失败请检查后端服务)启动前端streamlit run web.py --server.port 8501浏览器访问http://127.0.0.1:8501就能看到问答页面。8.3 批量入库与增量更新当知识库文档数量变大后手动一条条入库不可行需要用脚本监听输入目录并新增到向量库。import json from pathlib import Path PROCESSED_FILE processed.json def load_processed(): if Path(PROCESSED_FILE).exists(): return set(json.loads(Path(PROCESSED_FILE).read_text(encodingutf-8))) return set() def save_processed(file_path): processed load_processed() processed.add(str(file_path)) Path(PROCESSED_FILE).write_text( json.dumps(list(processed), ensure_asciiFalse, indent2), encodingutf-8 ) def ingest_directory(input_dir: Path): processed load_processed() for file_path in sorted(input_dir.glob(*.txt)): if str(file_path) in processed: continue try: text file_path.read_text(encodingutf-8) chunks split_text(text) add_to_vector_db(chunks, sourcestr(file_path)) save_processed(file_path) print(f[OK] {file_path}) except Exception as e: print(f[FAIL] {file_path} - {e})建议在批量入库时记录文件 hash 或修改时间而不是只记路径避免同名文件更新后漏处理。8.4 数据安全与合规提醒知识库里的文档往往涉及内部流程、客户资料和版权内容。在构建和部署时要注意私域数据不要直接提交到未授权的第三方 API。对外提供问答服务前确认资料使用范围和查看权限。文档内容涉及个人隐私、人脸、声音等敏感信息时必须获得合法授权并做好脱敏。接口服务要限制来源 IP增加访问令牌或登录校验。回答内容可能包含敏感信息时增加输出过滤和人工复核环节。9. 常见问题与排查方法RAG 项目第一次跑通容易但要稳定好用一定会遇到各种问题。下面整理一份高频排查清单问题现象可能原因排查方式解决方案检索结果不相关切分粒度过大或过小、Embedding 不适配打印检索到的 chunk 内容对照问题和答案调整 chunk_size换中文效果更好的 Embedding 模型增加查询改写回答明显在编造上下文没有召回到关键信息或大模型忽略资料打印最终 Prompt检查 context 是否包含答案增大 top_k加强 Prompt 约束要求“无资料必须说明不知道”加入重排序中文场景效果差使用了英文为主的 Embedding 模型对比不同模型在中文测试集上的命中率替换成中文优化模型启动慢或内存高本地大模型和 Embedding 模型同时占用资源查看进程内存和显存占用使用 API 接入大模型对 Embedding 计算做 batch换更小模型导入文件乱码文件编码不是 UTF-8 或 PDF 是扫描件查看原始文件编码统一转 UTF-8扫描件先走 OCR 再入库批量入库中断单个文件格式异常或网络超时查看日志定位失败文件用 try/except 隔离异常支持断点续跑API 请求超时大模型生成时间过长观察单次请求耗时调大 timeout考虑异步接口降低生成长度向量库新增报错向量维度或模型不一致检查传入 embedding 的维度同一项目固定 Embedding 模型版本检索结果重复严重切分 overlap 过大或向量库存在重复入库检查相似 chunk 是否来自同一文档入库时做去重维护 doc id 和 chunk id排查时最常用的一个手段是打印中间结果把 query、检索到的 chunks、最终 Prompt 都打出来看问题到底出在哪一环。不要直接盲目调大模型RAG 的线上问题很多在检索阶段就已经定了。10. 新手学习路线与避坑建议完整跑一遍 RAG 项目后可以按下面这个路径继续深入。第一阶段先跑通最小 RAG 链路熟悉文档入库和检索回答的基础流程。第二阶段用 FastAPI 和 Streamlit 把 Demo 服务化掌握批量入库和增量更新。第三阶段围绕坏案例做召回优化优先调切分和加混合检索再考虑重排序。第四阶段建立自有测试集固定评估指标每次改动后跑一遍对比效果。第五阶段再评估是否需要引入 LangChain、LlamaIndex 等框架来提升开发效率。新手最容易踩的坑有四个一是一上来就堆重排序、查询改写、复杂 Prompt结果问题定位不清二是不做评估靠感觉调参数三是盲目追求大向量库和分布式数据量不大没必要四是把外部文档直接灌进知识库忽视版权和隐私边界。对大多数知识库问答场景来说RAG 的效果瓶颈不在模型参数大小而在召回质量。文本切分的颗粒度、检索到的片段是否命中答案、Prompt 是否让模型学会说“不知道”这三件事决定了最终体验。建议先动手跑通这里的第一个版本再围绕你的失败案例做针对性优化。如果项目遇到问题回到第 9 节排查表对照处理。 
长河君 · 2026/10/20 2:02:22 · 阅读 1.2k RAG从原理到落地：检索增强生成与知识库问答系统实践指南
这次我们把大模型 RAG 从原理到落地完整过一遍。RAG 即 Retrieval-Augmented Generation检索增强生成是目前把大模型接入私有数据最常用的一套技术方案。它的核心思路很朴素先检索资料再让大模型基于资料回答。解决的核心问题是大模型不知道你企业内部文档、产品手册、个人知识库里的内容直接提问很容易出现编造和答非所问。通过 RAG可以把用户问题转换成查询从知识库中检索相关片段拼进提示词让大模型基于这些片段生成有依据的回答而不是凭空瞎猜。这篇文章不是只画概念图。我们从原理讲起依次覆盖 Embedding 模型与向量库选型、本地环境搭建、最小可运行代码、召回优化最后落到接口服务、前端界面和批量入库。你可以把它当成一份 RAG 项目的从零到一工程笔记。如果你正准备做知识库问答、内部文档检索、客服助手或者只是想把 RAG 完整流程弄明白这篇文章值得收藏。文中所有代码选用常见开源组件尽量做到能复制、能跑通。涉及调用外部大模型 API 时私有数据需要先做脱敏和授权评估不要随意提交到第三方服务。下面进入正文。1. 核心内容速览覆盖模块本文会讲什么你能得到什么RAG 原理为什么大模型需要先检索再生成从概念上理解 RAG 的三段式流程完整链路离线入库、在线问答、评估迭代三条流水线对项目全貌有清晰认识组件选型Embedding 模型、向量库、检索器选择思路根据数据量和场景选型环境准备Python 依赖与本地推理服务接入搭出一套可运行的实验环境最小实现无框架依赖的 RAG Demo 代码真正跑通一个知识库问答召回优化切分策略、混合检索、重排序、查询改写解决检索不准导致回答错误的问题项目落地FastAPI、Streamlit、批量入库把 Demo 升级成可用的服务2. RAG 核心原理为什么大模型要先查后答大模型的能力来自预训练时见过的语料知识范围受训练数据和截止时间限制。你问一个模型“我们公司上季度的退款流程是什么”如果这句话不在预训练语料里模型只能根据泛化经验去组合答案组合得再流畅也可能是错的这就是幻觉的来源。RAG 相当于给大模型开卷考试。系统先去外部知识库检索候选资料再把资料和问题一起交给模型模型只能在资料范围内组织答案。这样一来回答就不是模型凭记忆生成而是有据可依。一个完整的 RAG 问答流程通常包含下面几个环节用户输入 Query。可选步骤查询改写。用大模型把“问题不完整、指代不清”的情况纠正或者把一个问题拆成多个子问题分别检索。检索。通过向量召回、关键词召回或混合召回从知识库找出候选文档片段。重排序。如果候选结果多用精排模型或分数融合策略把最相关的片段排在前面。上下文拼装。把筛出的片段按顺序组装成 Prompt 中的“资料区”。大模型生成。模型阅读资料并回答问题必要时返回引用来源。后处理。在多轮对话场景中当前答案还会和对话历史拼接。RAG 和微调的使用边界需要分清。如果要做频繁更新的知识库问答、需要答案可追溯来源RAG 明显更合适。如果要让模型固定输出某一类格式、学习某种语气微调更合适。两者也可以组合但新手不必一开始就上微调先把 RAG 链路跑通成本更低。3. 完整链路拆解离线入库、在线问答与评估迭代RAG 项目不只是“写个检索函数”那么简单。为了工程化我们通常把它拆成三条流水线。3.1 离线文档入库 Pipeline这个阶段负责把原始文档变成向量库里的数据不依赖用户提问是 RAG 的地基。文档会经过加载、清洗、切分、向量化、入库五个步骤。加载要兼容 txt、PDF、Markdown、Word 等格式清洗要去掉页眉页脚、无效空行处理乱码切分决定检索的最小粒度直接影响召回质量向量化把文本变成稠密向量最后连同元数据一起写入向量库。元数据建议认真维护。文件名、章节标题、更新时间、来源链接这些信息既能用于检索结果过滤也能在下游回答里做引用来源。3.2 在线问答 Pipeline用户发起提问后进入这个流程。系统处理 Query、调用检索器获取候选片段、重排序、拼装 Prompt、调用大模型生成回答。在线链路对性能更敏感。检索一般控制在几百毫秒以内大模型生成是主要耗时环节。如果检索本身就很慢要考虑向量索引类型、候选数量、是否把检索结果缓存。3.3 评估迭代 Pipeline很多 RAG 项目没有评估这一步上线后靠用户反馈发现问题成本很高。更推荐的做法是准备几十到上百条测试问题每条问题标注标准答案和期望检索到的文档片段把结果跑完后统计命中率和答案准确率。评估结果引导优化方向。如果召回阶段没召回正确内容先看切分和检索策略如果召回了但答案还是错的再看 Prompt 设计和模型生成参数如果答案本身质量平平再考虑换更强的大模型或加入重排序。这一步是 RAG 持续迭代的闭环。4. Embedding、向量库与检索组件选型参考4.1 Embedding 模型选择文本向量化的质量是 RAG 的上限。选型时可以参考这几个维度维度说明中英文支持中文为主的场景优先选中文语料优化过的模型向量维度维度越高通常表达越细但占空间和计算也越多最大输入长度决定了单个 chunk 能编码多长文本部署成本本地开源模型不额外收费但占资源商用 API 更省事但产生费用更新维护新版本是否持续维护对长期项目很重要常见方案包括本地部署开源 Embedding 模型以及使用成熟商用 API。实际效果建议直接拿自己的领域文档测试不要只看公开榜单。4.2 向量库选择向量库用于存储和检索向量选型要看数据量、部署复杂度和是否已有基础设施。向量库定位适用场景Chroma轻量级嵌入式向量库本地 Demo、小规模知识库起步FAISSFacebook 开源的向量检索库以计算为主适合中等规模快速检索Milvus分布式向量数据库大规模、生产级、需要高可用Elasticsearch/OpenSearch搜索引擎增强向量能力已有 ES 基础设施需要全文和向量混合检索数据量小的时候不需要一上来就上重组件。几百上千个文档片段用 Chroma 或 FAISS 都能很好跑起来。数据量到百万级以上再评估 Milvus 这类分布式系统。4.3 检索方式向量检索引近似语义关键词检索引精确匹配两者各有优势。向量检索对“换个说法问同样的问题”更友好关键词检索对“型号编号、专有名词、精确值”更可靠。生产的常用做法是混合检索一路向量召回一路 BM25 关键词召回再做结果融合最后用重排序模型精排。5. 本地实验环境准备RAG 项目需要 Python 环境、Embedding 模型和向量库依赖、大模型接入渠道。下面给出一套通用实验环境配置。建议使用 Python 3.10 左右版本并创建独立虚拟环境避免和系统依赖冲突。核心依赖可以这样安装# 创建虚拟环境 python3 -m venv rag_env source rag_env/bin/activate # 安装核心依赖 pip install chromadb sentence-transformers openai fastapi uvicorn streamlit rank-bm25如果想测试重排序可以另外安装 FlagEmbeddingpip install FlagEmbedding这里说明一下各部分的作用chromadb轻量向量库负责数据持久化和检索。sentence-transformers加载本地 Embedding 模型并生成向量。openai调用兼容 OpenAI 格式的大模型接口。fastapi uvicorn把 RAG 包装成 HTTP 服务。streamlit快速搭建测试界面。rank-bm25关键词检索用于混合召回。FlagEmbedding用于 Cross-Encoder 重排序。大模型的接入渠道建议先统一到 openai 兼容接口。本地可以用 vLLM、Ollama 等推理服务暴露一个 HTTP 端口线上则直接用云服务商提供的接口。两种方式对本文代码来说基本通用只需要改 base_url 和 api_key。Embedding 模型可以从 Hugging Face 下载下载慢时可以配置镜像环境变量。具体命令以实际环境为准。6. 最小可用的 RAG 实现文档入库与检索问答这一章我们写一个不依赖 LangChain 的最小 RAG 实现这样对新手来说依赖少、问题更好排查。项目结构可以是rag_demo/ ├── data/ │ └── knowledge.txt ├── rag_build.py # 文档入库 └── rag_query.py # 检索问答6.1 准备演示文档产品A 支持批量导入知识库文档。 产品A 的问答接口地址是 /api/ask。 产品A 本地部署建议使用 Linux 服务器。 产品A 支持常见文本格式导入包括 txt 和 markdown。这里的内容只是用来演示流程实际使用时替换成自己的业务资料。6.2 文档切分与向量化入库# rag_build.py from sentence_transformers import SentenceTransformer import chromadb # 1. 读取文档并按简单策略切分 text open(data/knowledge.txt, encodingutf-8).read() def split_text(text, chunk_size100, chunk_overlap20): chunks [] for i in range(0, len(text), chunk_size - chunk_overlap): chunks.append(text[i:i chunk_size]) return chunks chunks split_text(text, chunk_size100, chunk_overlap20) # 2. 加载本地 Embedding 模型 model SentenceTransformer(BAAI/bge-small-zh-v1.5) # 3. 初始化向量库 client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(nameknowledge_demo) # 4. 向量化并写入 ids [fchunk_{i} for i in range(len(chunks))] embeddings model.encode(chunks).tolist() collection.add(documentschunks, embeddingsembeddings, idsids) print(f入库完成共 {len(chunks)} 个片段)这个切分函数是按字符硬切的训练演示足够生产环境还需要考虑语义边界和文档结构。6.3 检索上下文并调用大模型生成回答# rag_query.py from sentence_transformers import SentenceTransformer from openai import OpenAI import chromadb # 1. 重新加载 Embedding 模型和向量库 model SentenceTransformer(BAAI/bge-small-zh-v1.5) client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection(nameknowledge_demo) # 2. 用户问题 query 产品A 怎么批量导入文档 # 3. 向量召回 query_embedding model.encode([query]).tolist() results collection.query( query_embeddingsquery_embedding, n_results3 ) documents results[documents][0] context \n.join(documents) print(检索到的上下文) print(context) # 4. 拼装 Prompt prompt f你是知识库问答助手。请只根据下面的资料回答问题。 如果资料中没有相关信息请直接回答“知识库中没有找到相关信息”不要编造。 资料 {context} 问题{query} 回答 # 5. 调用大模型 client OpenAI( base_urlhttp://127.0.0.1:8000/v1, api_keyEMPTY ) response client.chat.completions.create( modelyour-local-model, # 改成你实际部署的模型名 messages[{role: user, content: prompt}], temperature0.2 ) print(最终回答) print(response.choices[0].message.content)如果你用的是在线大模型 API把 base_url 改成服务商地址、api_key 换成真实密钥、model 换成实际模型名即可。这一步跑通后你已经拥有了一个最简知识库问答链路。判断成功的标准是输入一个问题返回的 answer 直接来自资料内容而不是模型自由发挥。第 6.3 节代码出现了两个client变量一个用于向量库、一个用于大模型 API命名容易混淆。出于可读性可以改成vector_client和llm_client。实际执行时也要注意大模型服务的 context length 是否足够。7. 召回优化实战切分策略、混合检索与重排序RAG 效果不好一半以上问题出在召回阶段。这一章是全文重点按顺序讲切分、查询改写、混合检索和重排序。7.1 切分策略chunk 粒度直接决定检索质量。切太大一个块里包含多个主题语义被稀释切太小单个块承载的信息量不足检索命中但上下文不完整。比较实用的起步配置是 chunk_size 300 到 500overlap 控制在 10% 到 20%。切分要尽量避开句子中间保持语义完整。对于有明确结构的文档优先按标题层级先切再对长段落二次切分。更进阶的做法是父子切分把父块用于上下文理解子块用于精确匹配。实现时可以给每个 chunk 绑定父块 ID检索到子块后把父块一并拼进 Prompt。7.2 查询改写用户问题经常很短比如“它支持吗”“价格是多少”。原样拿去检索召回质量很低。常见的查询改写方案有两种。第一种是 Multi-Query让大模型把原问题改写成多个子问题分别检索后合并结果适合复杂问题。第二种是 HyDE让大模型先根据问题写一段假设性回答再用这段假设回答去检索间接提升语义匹配度。查询改写会多一次大模型调用成本有所增加但很多场景下收益明显。7.3 混合检索与 RRF 融合向量召回擅长语义匹配BM25 擅长精确匹配。混合检索把两条召回链路的结果合并再通过 RRF 算法排序能同时兼顾两种优势。下面给出一个简化示例from rank_bm25 import BM25Okapi # 假设 corpus 是全部 chunk tokenized_corpus [list(chunk) for chunk in corpus] bm25 BM25Okapi(tokenized_corpus) # 向量召回得到 candidate_ids vector_ids [chunk_1, chunk_2] # 示例 # BM25 召回得到 bm25_ids bm25_ids [chunk_2, chunk_3] # 示例 # RRF 融合 def rrf(rank_lists, k60): scores {} for rank_list in rank_lists: for rank, doc_id in enumerate(rank_list): scores[doc_id] scores.get(doc_id, 0) 1.0 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue) final_ranking rrf([vector_ids, bm25_ids])实际项目中直接把 collection.query 返回的文档 ID 列表作为 vector_ids 传入即可。7.4 重排序向量检索得到的 Top-K 候选并不是最终顺序。更稳妥的做法是先用向量或关键词召回 10 到 20 个候选再用 Cross-Encoder 重排序模型逐对计算 query 和文档的相关性最后取前 3 到 5 个拼进 Prompt。from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-base, use_fp16False) candidates [产品A 支持批量导入知识库文档。, 产品A 的问答接口地址是 /api/ask。] pairs [(query, doc) for doc in candidates] scores reranker.compute_score(pairs) # 按分数降序重排 ordered sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue) for doc, score in ordered: print(score, doc)注意Cross-Encoder 重排序本质是让模型逐对打分候选数越多耗时越长一般只对 Top 20 以内的候选精排。7.5 元数据过滤知识库规模变大后所有文档检索一遍会引入大量无关内容。建议在入库时为每个 chunk 绑定来源、部门、文档类型、更新时间等元数据检索时按条件过滤。比如客服场景只检索“客服知识库”来源的文档避免把产品研发文档混进来。优化时不要一上来把所有技巧全叠上。第一步先看坏案例Query 到底是什么检索到的 top-1 chunk 是否包含答案。如果不包含优先调切分如果包含但排名靠后优先加重排序如果是追问和指代类问题优先做查询改写。8. 项目落地接口 API、前端界面与批量入库跑通最小 RAG 后下一步是把它工程化。日常使用不能每次跑 Python 脚本需要用 HTTP 接口暴露服务加上一个简单的 Web 界面并让文档入库支持批量和增量。8.1 FastAPI 包装 RAG 服务# app.py from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class AskRequest(BaseModel): question: str def run_rag(question: str): # 这里调用之前 rag_query.py 中的检索和生成逻辑 # 返回答案和来源 answer 根据检索结果生成的回答 sources [chunk_1, chunk_2] return answer, sources app.post(/api/ask) def ask(request: AskRequest): answer, sources run_rag(request.question) return {answer: answer, sources: sources}启动服务uvicorn app:app --host 0.0.0.0 --port 8000接口服务一旦对外开放需要加上鉴权和访问控制。建议先绑定127.0.0.1或内网地址不要直接暴露到公网。8.2 Streamlit 搭建测试界面# web.py import streamlit as st import requests API_URL http://127.0.0.1:8000/api/ask st.title(RAG 知识库问答 Demo) question st.text_input(请输入问题) if st.button(提问): resp requests.post(API_URL, json{question: question}, timeout60) if resp.status_code 200: st.info(resp.json()[answer]) else: st.error(请求失败请检查后端服务)启动前端streamlit run web.py --server.port 8501浏览器访问http://127.0.0.1:8501就能看到问答页面。8.3 批量入库与增量更新当知识库文档数量变大后手动一条条入库不可行需要用脚本监听输入目录并新增到向量库。import json from pathlib import Path PROCESSED_FILE processed.json def load_processed(): if Path(PROCESSED_FILE).exists(): return set(json.loads(Path(PROCESSED_FILE).read_text(encodingutf-8))) return set() def save_processed(file_path): processed load_processed() processed.add(str(file_path)) Path(PROCESSED_FILE).write_text( json.dumps(list(processed), ensure_asciiFalse, indent2), encodingutf-8 ) def ingest_directory(input_dir: Path): processed load_processed() for file_path in sorted(input_dir.glob(*.txt)): if str(file_path) in processed: continue try: text file_path.read_text(encodingutf-8) chunks split_text(text) add_to_vector_db(chunks, sourcestr(file_path)) save_processed(file_path) print(f[OK] {file_path}) except Exception as e: print(f[FAIL] {file_path} - {e})建议在批量入库时记录文件 hash 或修改时间而不是只记路径避免同名文件更新后漏处理。8.4 数据安全与合规提醒知识库里的文档往往涉及内部流程、客户资料和版权内容。在构建和部署时要注意私域数据不要直接提交到未授权的第三方 API。对外提供问答服务前确认资料使用范围和查看权限。文档内容涉及个人隐私、人脸、声音等敏感信息时必须获得合法授权并做好脱敏。接口服务要限制来源 IP增加访问令牌或登录校验。回答内容可能包含敏感信息时增加输出过滤和人工复核环节。9. 常见问题与排查方法RAG 项目第一次跑通容易但要稳定好用一定会遇到各种问题。下面整理一份高频排查清单问题现象可能原因排查方式解决方案检索结果不相关切分粒度过大或过小、Embedding 不适配打印检索到的 chunk 内容对照问题和答案调整 chunk_size换中文效果更好的 Embedding 模型增加查询改写回答明显在编造上下文没有召回到关键信息或大模型忽略资料打印最终 Prompt检查 context 是否包含答案增大 top_k加强 Prompt 约束要求“无资料必须说明不知道”加入重排序中文场景效果差使用了英文为主的 Embedding 模型对比不同模型在中文测试集上的命中率替换成中文优化模型启动慢或内存高本地大模型和 Embedding 模型同时占用资源查看进程内存和显存占用使用 API 接入大模型对 Embedding 计算做 batch换更小模型导入文件乱码文件编码不是 UTF-8 或 PDF 是扫描件查看原始文件编码统一转 UTF-8扫描件先走 OCR 再入库批量入库中断单个文件格式异常或网络超时查看日志定位失败文件用 try/except 隔离异常支持断点续跑API 请求超时大模型生成时间过长观察单次请求耗时调大 timeout考虑异步接口降低生成长度向量库新增报错向量维度或模型不一致检查传入 embedding 的维度同一项目固定 Embedding 模型版本检索结果重复严重切分 overlap 过大或向量库存在重复入库检查相似 chunk 是否来自同一文档入库时做去重维护 doc id 和 chunk id排查时最常用的一个手段是打印中间结果把 query、检索到的 chunks、最终 Prompt 都打出来看问题到底出在哪一环。不要直接盲目调大模型RAG 的线上问题很多在检索阶段就已经定了。10. 新手学习路线与避坑建议完整跑一遍 RAG 项目后可以按下面这个路径继续深入。第一阶段先跑通最小 RAG 链路熟悉文档入库和检索回答的基础流程。第二阶段用 FastAPI 和 Streamlit 把 Demo 服务化掌握批量入库和增量更新。第三阶段围绕坏案例做召回优化优先调切分和加混合检索再考虑重排序。第四阶段建立自有测试集固定评估指标每次改动后跑一遍对比效果。第五阶段再评估是否需要引入 LangChain、LlamaIndex 等框架来提升开发效率。新手最容易踩的坑有四个一是一上来就堆重排序、查询改写、复杂 Prompt结果问题定位不清二是不做评估靠感觉调参数三是盲目追求大向量库和分布式数据量不大没必要四是把外部文档直接灌进知识库忽视版权和隐私边界。对大多数知识库问答场景来说RAG 的效果瓶颈不在模型参数大小而在召回质量。文本切分的颗粒度、检索到的片段是否命中答案、Prompt 是否让模型学会说“不知道”这三件事决定了最终体验。建议先动手跑通这里的第一个版本再围绕你的失败案例做针对性优化。如果项目遇到问题回到第 9 节排查表对照处理。