1.1 什么是RAG：定义、工作流程与典型应用场景
我第一次接触RAG这个概念的时候，脑子里冒出的第一个问题是：大语言模型不是已经能回答很多问题了吗，为什么还需要额外搞一套检索增强的东西？后来踩了不少坑才明白，LLM本身的知识是冻结在训练数据里的，它不知道你公司上周刚更新的产品文档，也不知道你私人的笔记内容。RAG要解决的就是这个“知识边界”的问题。全称是Retrieval-Augmented Generation，翻译过来叫检索增强生成。核心思路很朴素：在让模型回答问题之前，先去一个外部知识库里把相关的资料找出来，连同问题一起塞给模型，让它基于这些资料来组织答案。
工作流程拆开来看其实就三步。第一步是检索，用户提问后，系统拿着这个问题去向量数据库或者搜索引擎里找最相关的文档片段。第二步是增强，把检索到的内容和原始问题拼成一个完整的提示词，有时候还会加上一些系统指令来约束模型的回答方式。第三步是生成，LLM根据这个增强后的提示词输出最终答案。我自己在调试的时候发现，这三步里最容易出问题的往往是第一步，检索回来的东西不对，后面生成得再漂亮也是白搭。
典型应用场景我列几个自己实际碰到过的。企业内部知识库问答是最常见的，员工问“报销流程是什么”“年假怎么申请”，系统从HR文档里检索相关段落再回答。电商客服机器人也常用RAG，把商品详情、退换货政策这些资料喂进去，用户问什么就检索什么。还有一个场景是法律或者医疗领域的辅助问答，这类场景对准确性要求极高，模型不能瞎编，必须严格基于检索到的条文或指南来回答。我甚至还见过有人用RAG做个人笔记助手，把自己几年的读书笔记向量化，想问什么直接检索自己的笔记来回答，挺有意思的。
1.2 RAG与微调、传统检索方案的对比及选型建议
说到让模型掌握新知识，微调是另一个常被拿出来跟RAG比较的方案。我自己两个都试过，感受挺深的。微调的本质是拿一批标注数据去更新模型的权重，让模型“记住”这些新知识。好处是推理的时候不需要额外检索，速度快，而且模型能学到一些风格和格式上的东西。坏处也很明显：每次知识更新都得重新训练，成本高得吓人；训练数据里如果有错误，模型会把这些错误也学进去；还有个问题是微调后的模型容易“遗忘”原来的通用能力，学术界叫灾难性遗忘。RAG就不一样了，知识存在外部数据库里，更新知识就是更新数据库，跟模型本身没关系，灵活太多了。
传统检索方案比如关键词搜索或者基于BM25的全文检索，这些技术其实很成熟了。它们的优势是速度快、可解释性强，你搜“Python教程”它就给你返回包含“Python教程”这几个字的文档。问题在于语义理解能力弱，用户问“怎么学编程语言”，传统检索可能就找不到“Python入门指南”这篇文档，因为字面上没有重叠。RAG里的向量检索就是来解决这个问题的，它把问题和文档都映射到同一个语义空间里，靠向量距离来判断相关性，能捕捉到字面背后的意思。
选型建议这块我自己的经验是这样的：如果你的知识更新频繁，比如产品文档每周都在变，那RAG几乎是唯一选择。如果你需要模型学会一种特定的输出风格或者说话方式，比如让模型模仿某个客服的语气，微调可能更合适。实际项目中很多人是两者结合，先用微调让模型适应领域语言，再用RAG注入实时知识。传统检索和向量检索也不是非此即彼，混合检索现在很流行，后面讲到优化的时候我会详细说。有个简单的判断标准：知识是“事实性”的还是“风格性”的，事实性知识优先考虑RAG，风格性能力优先考虑微调。
1.3 RAG应用开发所需技能栈与实战学习路径
RAG开发需要的技能栈比我想象中要杂一些。Python是基础中的基础，整个生态几乎都是Python写的，LangChain、LlamaIndex这些框架你绕不开。然后得懂一点向量数据库的操作，不用深入到源码级别，但至少得知道怎么建集合、怎么插入向量、怎么做相似度查询。嵌入模型的选择和使用也是必备技能，OpenAI的embedding API也好，开源的BGE、M3E也好，得知道怎么调、怎么评估效果。Prompt工程也不能忽视，检索回来的内容怎么组织、怎么给模型下指令，直接影响最终回答的质量。
再往上走一层，如果要做生产级的应用，还得会一些工程化的东西。比如FastAPI或者Flask做API服务，Docker做容器化部署，Redis做缓存，这些不是RAG特有的，但实际项目里都会用到。评估和调试能力也很关键，你得知道怎么构建测试集、怎么计算召回率和准确率、怎么定位是检索出了问题还是生成出了问题。我刚开始做的时候就是凭感觉调，后来发现没有量化指标根本不知道优化有没有效果。
学习路径我建议这样走。先花两三天把RAG的核心概念搞清楚，就是前面说的检索、增强、生成这三步。接着动手跑通一个最小的demo，用LangChain加一个向量数据库，几十行代码就能搞定，先感受一下整个流程。然后在这个demo基础上逐步加东西：换不同的嵌入模型对比效果，试试不同的文本切分策略，加一个重排序模块看看检索质量有没有提升。每加一个组件都要有意识地去观察它对最终结果的影响。最后找一个自己真正感兴趣的领域数据，比如你的读书笔记或者某个开源项目的文档，从头到尾做一个完整的应用出来。这个过程里踩的坑越多，学到的东西越扎实。
我特别想强调一点：不要一上来就追求完美架构。我见过太多人包括我自己，一开始就想把什么混合检索、重排序、查询改写全部堆上去，结果调试起来一团乱麻，根本不知道哪个环节出了问题。先把最基础的链路跑通，确保检索能返回相关文档、模型能基于文档给出合理回答，然后再一个模块一个模块地优化。RAG这个东西，入门容易精通难，但入门的那一步其实比想象中简单得多。
2.1 Python环境、嵌入模型与向量数据库选型
我在动手搭RAG应用之前，花了不少时间在环境准备上，踩的坑比写代码还多。Python版本这块，我强烈建议直接用3.10或者3.11，别贪新用3.12。LangChain和一些向量数据库的客户端库对3.12的支持是滞后的，我当初用3.12装Chroma的时候折腾了半天依赖冲突，后来降回3.11一切顺畅。虚拟环境用conda或者venv都行，重点是养成给每个项目单独建环境的习惯，RAG项目依赖又多又杂，全局装迟早出问题。
嵌入模型的选型是我觉得整个准备阶段最需要认真对待的一步。OpenAI的text-embedding-3-small和3-large效果确实好，调用也省心，但成本得算清楚，尤其是文档量大的时候。我自己的做法是先用OpenAI的模型快速验证流程，确认整个链路能跑通之后，再根据预算和数据敏感度决定要不要换成开源的。开源这边BGE系列是目前中文场景下比较稳的选择，bge-large-zh-v1.5我在好几个项目里用过，本地部署不需要联网，数据也不用出内网。M3E也值得一试，体积小一些，推理速度快。选嵌入模型的时候有个容易被忽略的点：不同模型输出的向量维度不一样，这个维度一旦定了，向量数据库的集合结构就跟着定了，后面想换模型得重新入库，所以前期多想一步能省很多事。
向量数据库的选择上，我按使用场景分了三类。本地开发和小规模验证用Chroma就够了，pip install就能跑，不需要额外起服务，数据存在本地文件里，调试的时候直接看得到。中等规模的生产环境我会选Milvus或者Qdrant，这两个都支持分布式部署，性能也经过验证，Qdrant的过滤功能我用得比较多，元数据筛选很灵活。如果完全不想运维，Pinecone这类托管服务可以省掉很多麻烦，代价是数据在别人服务器上，费用也随规模线性增长。FAISS我一般只在对检索速度有极致要求、而且不需要持久化和增删改查的场景下用，它更像一个向量检索库而不是数据库。选哪个没有标准答案，我自己的原则是：先用最简单的把流程跑通，遇到瓶颈再换。
2.2 LangChain核心模块与RAG链路整体架构
LangChain刚上手的时候我觉得它抽象层太多了，一个简单的检索问答被拆成好几个类，理解成本不低。用久了才体会到这种设计的好处。它把RAG链路的每个环节都做成了可替换的组件，你想换嵌入模型就换Embeddings类，想换向量库就换VectorStore类，其他代码基本不用动。这种插拔式的设计在快速实验阶段特别有用，我可以一上午试三四种不同组合，对比效果。
核心模块我挑几个RAG里最常用的说一下。Document Loaders负责从各种数据源加载文档，PDF、Word、网页、数据库都行，加载出来统一成Document对象，里面有page_content和metadata两个字段。Text Splitters把长文档切成小块，切分策略直接影响检索质量，这个后面会细说。Embeddings不用多讲，就是调模型把文本转成向量。Vector Stores封装了向量数据库的操作，add_documents入库，similarity_search查询，接口很统一。Retriever是检索器的抽象，它比Vector Store更灵活，可以实现各种检索策略，比如多路召回、基于元数据过滤、甚至结合关键词搜索。Prompts和LLMs就是提示词模板和语言模型，Chains把这些组件串起来。
RAG链路的整体架构我画过好几遍才理清楚。最左边是数据侧：文档经过加载、切分、向量化，存进向量数据库。这部分通常是一次性的离线处理，但知识更新的时候要重新跑。右边是查询侧：用户提问先经过嵌入模型转成向量，然后在向量库里做相似度检索，拿到Top-K个相关文档片段。这些片段和原始问题一起填入Prompt模板，组成一个增强后的提示词，最后交给LLM生成回答。Retriever在中间起到桥梁作用，它决定了检索策略，也决定了哪些文档片段会被送进Prompt。我自己的经验是，整个链路里Retriever和Prompt模板是调优空间最大的两个地方，其余环节相对固定。
2.3 数据采集、清洗、切分与向量化入库策略
数据采集这一步看着简单，实际做起来脏活累活特别多。我处理过的数据源包括PDF、Markdown、Confluence页面、飞书文档、甚至Excel表格。PDF是最麻烦的，尤其是扫描件和排版复杂的文档，直接解析出来经常是乱的。我一般会用PyMuPDF先试着提取文本，效果不好就上OCR，但OCR的准确率有限，后面清洗的工作量会翻倍。Markdown和网页相对友好，结构清晰，解析出来直接能用。采集的时候我会把来源、时间、作者这些元数据一起存下来，后面检索的时候可以按这些字段做过滤，也能在回答里展示引用来源。
清洗这步我踩过一个大坑。早期我直接把解析出来的文本丢进切分器，结果检索回来的片段里全是页眉页脚、广告栏、导航菜单这些噪声，模型拿着这些内容回答，质量可想而知。后来我加了几条清洗规则：去掉连续重复出现的短行，去掉纯数字和符号组成的行，合并被错误换行拆散的句子。还有一些文档里有大量表格，表格转成文本后结构全丢了，这种我现在的做法是把表格单独提取出来，用Markdown表格的格式重新组织，保留行列关系。清洗没有一劳永逸的方案，不同数据源要写不同的清洗逻辑，但这一步投入的时间在检索质量上的回报非常明显。
切分策略是我认为最容易被低估的一个环节。一开始我用的是最简单的固定长度切分，比如每500个字符切一刀，重叠50个字符。问题很快就暴露了：有时候一句话被切成两半，检索出来的片段语义不完整，模型理解起来也费劲。后来我换成了RecursiveCharacterTextSplitter，它会优先按段落切，段落太长再按句子切，句子还长才按字符切，这样切出来的块语义完整性好很多。切分大小我也试过好几组，中文场景下300到500字是一个比较舒服的区间，太小了信息不够，太大了检索精度会下降。重叠部分保留个10%到15%就行，主要是为了防止关键信息刚好落在切分边界上被截断。
向量化入库看着就是调个接口的事，但有几个细节不注意后面会很痛苦。批量入库比一条一条插快很多，我一般会攒到几百条再批量提交，但也要注意别一次塞太多导致内存爆掉。入库前给每个文档块加上唯一的ID和丰富的元数据，后面更新和删除的时候会方便很多。还有一个我吃了亏的地方：不同嵌入模型对文本长度是有限制的，超过限制的部分会被截断，如果你不检查直接入库，超长文本的关键信息可能就在截断处丢了。我现在入库前会加一步长度检查，超长的文本重新切分或者截断并记录日志，确保每条入库的数据都是完整的。向量数据库建好之后，我会先用几个已知答案的问题去测一下检索效果，看看返回的Top-K结果是不是真的相关，这一步能提前发现很多问题。 from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings loader = PyPDFLoader("./docs/manual.pdf") pages = loader.load() splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=80,
separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""]
) chunks = splitter.split_documents(pages) embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-large-zh-v1.5"
) vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
import os from datetime import datetime from langchain_community.vectorstores import Chroma
def incremental_update(vectorstore, docs_dir, embeddings):
existing = vectorstore.get()
existing_sources = set(
meta.get("source") for meta in existing["metadatas"]
)
to_process = []
for root, _, files in os.walk(docs_dir):
for f in files:
path = os.path.join(root, f)
mtime = datetime.fromtimestamp(os.path.getmtime(path))
if path not in existing_sources:
to_process.append(path)
else:
pass
for source in existing_sources:
if source in [p for p in to_process]:
vectorstore.delete(where={"source": source})
for path in to_process:
loader = DirectoryLoader(path, glob="**/*")
new_docs = loader.load()
chunks = splitter.split_documents(new_docs)
vectorstore.add_documents(chunks)
return vectorstore
from langchain.retrievers.multi_query import MultiQueryRetriever from langchain_openai import ChatOpenAI
llm = ChatOpenAI(temperature=0) retriever_from_llm = MultiQueryRetriever.from_llm(
retriever=vectorstore.as_retriever(),
llm=llm
)
docs = retriever_from_llm.invoke("这个功能收费吗")
from fastapi import FastAPI, Depends, HTTPException from pydantic import BaseModel from typing import Optional
app = FastAPI()
class QueryRequest(BaseModel):
question: str
session_id: str
user_id: str
class QueryResponse(BaseModel):
answer: str
sources: list[dict]
session_id: str
@app.post("/api/chat", response_model=QueryResponse) async def chat(request: QueryRequest, user=Depends(verify_token)):
if not check_permission(user, request.session_id):
raise HTTPException(status_code=403, detail="无权访问该会话")
result = await rag_pipeline(request.question, request.session_id)
return result
