从零搞懂 RAG：让大模型不再"胡言乱语"

当大模型开始"编造事实"的时候，我决定给它配一本"参考书"

一、大模型为啥会胡说八道？我踩过的坑

刚开始用大模型的时候，我单纯地以为"问啥答啥，应该挺准的"。直到有一天我问了一个关于我所在公司内部流程的问题，模型信誓旦旦地给我编了一套完整的操作步骤——每一步看起来都很合理，但全错。

这时候我才意识到大模型的几个致命伤：

知识滞后：大模型的训练数据是有截止日期的。你问它"昨天发生的新闻"，它肯定不知道。就像你让一个 2025 年的人预测 2026 年的房价，他只能靠猜。

领域知识缺失：如果你问的是某个公司内部的操作手册、某个小众领域的专业知识，模型根本没学过这些数据。它只能"硬答"，答出来的东西往往是错的。

幻觉：这是最要命的问题。大模型在不确定答案的时候，会编造看起来合理但实际上错误的内容。而且它编得很自信，非专业人士根本分辨不出来。

不可追溯：你永远不知道模型的回答是从哪来的。它说是"根据资料"，但你找不到原文在哪。

尤其在金融、医疗这些领域，一次错误的判断可能就是致命的。但大模型可不管这些，它还是会努力给你一个"听起来很靠谱"的答案。

那有没有办法让大模型"有据可查"呢？这就是 RAG 要做的事。

二、RAG 到底是啥？一句话说清楚

RAG = 给大模型配一本参考书

RAG 全称是 Retrieval-Augmented Generation（检索增强生成）。简单说就是：在让大模型回答问题之前，先从外部资料库里找到相关的内容，把这些内容连问题一起喂给模型，让它"看资料"回答问题。

就像考试的时候，以前你只能凭记忆作答，现在允许你开卷——带一堆参考书进去。RAG 就是那个帮你翻书找答案的过程。

RAG 的完整流程（分成两步走）

整个 RAG 分两个阶段：

第一阶段（离线，提前准备好） ：把文档加载进来 → 切成小段 → 转成向量 → 存进向量数据库。这个阶段只需要做一次，后面查的时候直接用。

第二阶段（在线，每次用户提问时发生） ：用户提问 → 把问题也转成向量 → 去向量库里找最相似的内容 → 把问题 + 找到的内容一起喂给模型 → 模型根据资料生成回答。

说白了就是：先搜再答，让模型有的放矢。

三、文档加载：把各种文件塞进系统

企业里的文档散落在各种格式里：Word、PDF、Markdown、网页、数据库……RAG 的第一步就是把它们统一读进来，转成 LangChain 能处理的 Document 对象。

from langchain_community.document_loaders import UnstructuredWordDocumentLoader

loader = UnstructuredWordDocumentLoader("sample.docx", mode="elements")
docs = loader.load()

不同格式的"友好程度"完全不一样：

• 纯文本（.txt） ：一堆字堆在一起，机器很难判断哪是标题哪是正文
• Markdown（.md） ：用 #、- 这些符号明确标出结构，机器很好认
• Word（.docx） ：内部是 XML，能读出段落，但"是不是标题"很不统一——有人用内置样式，有人直接改字号大小，解析起来很头疼
• 数据库：结构最清晰，列名、类型都固定，机器最喜欢

PDF 是最难搞的。不同来源的 PDF 格式五花八门：扫描版要 OCR、电子版要解析、双栏布局要按顺序读、表格和公式要单独处理……

我用的是 MinerU 来处理复杂 PDF。它用两阶段策略：先在整个页面上"扫一眼"找出标题、正文、表格的位置，再对每个区域做精细识别。用 VLM（视觉语言模型）辅助识别，准确率比传统方式高不少。

四、文档切分：不切没法用

加载完文档之后，面临的第一个问题是：整篇文档太长了，不能直接塞给模型。

模型的上下文窗口有限（GPT-4o 是 128K token，但很多小模型只有 8K），而且一整篇文档里 80% 的内容跟当前问题无关，检索的时候噪声太大。

所以需要把文档切成小块，每一块叫一个 Chunk（文本块）。

切分要平衡三个东西：

• 块不能太大，否则检索噪声多
• 块不能太小，否则语义不完整（比如一句话被切成两半）
• 要在自然的地方切（段落边界、句子边界），不能强行截断

LangChain 提供了多种切分策略，我用的最多的是 RecursiveCharacterTextSplitter（递归字符切分器） 。它的思路很巧妙：

先定义一堆分隔符的优先级列表：["\n\n", "\n", "。", "，", " "]，然后从最高优先级开始切。如果切出来的块还是太大，就降级用下一级分隔符继续切。这样能最大程度保持语义完整。

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每块最大长度
    chunk_overlap=50,      # 相邻块重叠 50 个字符
    separators=["\n\n", "\n", "。", "，", " "]
)
chunks = splitter.split_documents(docs)

chunk_overlap（重叠）很重要。因为切分会把连续的一段话切开，可能刚好把关键信息切在边界上。重叠保证了边界附近的信息至少出现在两个块里，不会被遗漏。

五、文本嵌入：让计算机"读懂"文字

切分完之后，每个块还是一段中文文字，计算机没法直接比较"哪两段话意思更接近"。需要一个步骤把文字转成计算机能计算的数字——这就是嵌入（Embedding）。

嵌入模型会把一段文本映射成一个高维向量（比如 768 维或 1024 维的浮点数数组）。语义相近的文本，在向量空间里距离更近；语义无关的，距离更远。

我用的是 BGE-M3 这个模型，它有个特别厉害的地方：一次编码，同时生成稠密向量和稀疏向量。

稠密向量：捕捉整体语义。比如"AI"和"人工智能"意思相近，它们的稠密向量就离得近。适合语义匹配。

稀疏向量：记录关键词及其权重。比如"民法典第236条"这种精确匹配，稠密向量可能给我一堆民法相关内容，但稀疏向量能精准找到包含"236条"的文档。适合关键词匹配。

from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel("bge-m3")
result = model.encode(
    ["标量字段用来存储元数据"],
    return_dense=True,
    return_sparse=True
)
# result["dense_vecs"] 稠密向量
# result["lexical_weights"] 稀疏向量

六、向量存储：数据存哪儿？

向量生成之后要存起来，而且得支持"根据语义相似度快速查找"。这就是向量数据库干的事。

传统数据库（MySQL）擅长精确查询，查"等于多少"、"大于多少"非常快。但查"跟这个向量最像的 3 个"就抓瞎了——它只能全表扫描，一个个算距离，数据量一大就慢得要死。

向量数据库专门解决这个问题，通过 ANN（近似最近邻）索引，能在亿级数据里毫秒级返回结果。

我用的是 Milvus，因为它支持稠密向量和稀疏向量的混合检索，而且生产环境用得最多，生态成熟。

from pymilvus import MilvusClient

client = MilvusClient(uri="http://localhost:19530")

理解 Milvus 的几个核心概念就行：

• Collection：相当于 MySQL 里的表
• Schema：定义表有哪些字段（主键、稠密向量、稀疏向量、原始文本、元数据）
• Index：建在向量字段上的索引，用来加速检索。稠密向量用 HNSW，稀疏向量用倒排索引
• Metric Type：怎么算"相似"。稠密向量用 COSINE 或 IP，稀疏向量用 IP

关于 HNSW，我理解它就像查地图：先看全国地图（顶层，节点少）确定大致方向，再看城市地图（中间层）缩小范围，最后在街道地图（底层，包含所有数据）上精确查找。每一步都跳过大量无关数据，所以快。

七、检索：怎么找到最相关的内容

存好数据之后，核心操作就是"搜索"。用户提问，去库里找最相关的文档块。

稠密向量检索（语义匹配）

results = client.search(
    collection_name="demo",
    data=[dense_vec],          # 用户问题的稠密向量
    anns_field="vector",
    limit=3,                   # 返回 Top 3
    output_fields=["text"]
)

用户问"什么是大语言模型"，系统会找到语义最相近的 3 个文档块返回。

稀疏向量检索（关键词匹配）

results = client.search(
    collection_name="demo",
    data=[sparse_vec],         # 用户问题的稀疏向量
    anns_field="sparse_vector",
    limit=3,
    output_fields=["text"]
)

用户搜"民法典第236条"，系统会精准命中包含这个关键词的文档。

混合检索 + RRF 重排序（推荐）

各跑一路，然后把结果合并排序：

from pymilvus import AnnSearchRequest, RRFRanker

dense_req = AnnSearchRequest(data=[dense_vec], anns_field="vector", ...)
sparse_req = AnnSearchRequest(data=[sparse_vec], anns_field="sparse_vector", ...)

results = client.hybrid_search(
    reqs=[dense_req, sparse_req],
    ranker=RRFRanker(k=60),    # RRF 重排序
    limit=3
)

RRF（倒数排名融合）的原理很简单：同样的文档在两个检索结果里排名越高，最终得分越高。这样可以综合语义和关键词的优势。

八、生成：把找到的资料交给模型

检索到相关文档块之后，最后一步就是把检索结果拼成上下文，和用户问题一起喂给大模型。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini")

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的AI助手。请根据以下上下文回答问题：\n{context}"),
    ("human", "{question}")
])

# 检索到的文档块拼成上下文
context = "\n\n---\n\n".join([hit["entity"]["text"] for hit in results])

response = llm.invoke(prompt.format_messages(
    context=context,
    question="什么是大语言模型？"
))
print(response.content)

完整的 RAG 链用 LCEL 写会更优雅：

rag_chain = {
    "context": retriever | format_docs,
    "question": RunnablePassthrough()
} | prompt | llm | StrOutputParser()

result = rag_chain.invoke("什么是大语言模型？")

九、完整端到端案例（代码）

把前面所有的环节串起来：

from langchain_openai import ChatOpenAI
from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 1. 加载文档
loader = TextLoader("sample.txt")
docs = loader.load()

# 2. 切分文档
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
splits = splitter.split_documents(docs)

# 3. 嵌入并存入向量库（这里用 Chroma，轻量级开发用）
vectorstore = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 4. 定义提示词模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "请根据以下上下文回答问题：\n{context}"),
    ("human", "{question}")
])

llm = ChatOpenAI(model="gpt-4o-mini")

def format_docs(docs):
    return "\n\n---\n\n".join([doc.page_content for doc in docs])

# 5. 构建 RAG 链
rag_chain = {
    "context": retriever | format_docs,
    "question": RunnablePassthrough()
} | prompt | llm | StrOutputParser()

# 6. 执行
result = rag_chain.invoke("什么是LangChain？")
print(result)

十、进阶优化方向

跑通基础版 RAG 之后，如果想提升效果，可以从这几个方向优化：

减小 chunk_size：500-800 字符通常比 1000+ 更精准。块越小，检索命中越精确。

混合检索：稠密 + 稀疏向量混合，语义匹配和关键词匹配都覆盖。

RRF 重排序：把多路召回的结果融合排序，提升精准度。

查询重写：用户的问题可能口语化严重，先让 LLM 把它重写成更适合检索的形式。

元数据过滤：在检索时附加过滤条件（只查某份文档、某年之后的内容），缩小范围。

多路召回：用不同的检索策略（不同模型、不同切分粒度）并行查，然后合并排序。

十一、总结

RAG 的核心就六个环节：加载 → 切分 → 嵌入 → 存储 → 检索 → 生成。

让大模型"开卷考试"——给它配一本参考书，答案有据可查，幻觉大幅减少。朗朗上口的一句话总结：RAG = 检索 + 生成，让模型先查后答。

下一步我准备学 Agents（智能代理） ，让模型能自主调用工具完成任务。到时候再来分享。